commit 2021-12-16]17:46:11

cli.go

@@ -2,8 +2,8 @@ package manager
 
 import (
 	"context"
-	"errors"
 	"log"
+	"sync"
 
 	"git.godopu.net/lab/etri-smartfarm-poc-controller-serial/puserial"
 	"github.com/rjeczalik/notify"
@@ -13,6 +13,8 @@ var ctx context.Context
 var cancel context.CancelFunc
 var ch_discover chan notify.EventInfo
 var _managerObj *_manager
+var registerHandleFunc func(e Event)
+var removeHandleFunc func(e Event)
 
 func init() {
 	ctx, cancel = context.WithCancel(context.Background())
@@ -24,10 +26,21 @@ func init() {
 	_managerObj = &_manager{
 		devicesWithUUID:  devWithUUID,
 		devicesWithIface: devWithIface,
-		SyncListener:     &SyncHandler{devices: devWithUUID, chanForSync: chanForSync},
+		chanForSync:      chanForSync,
+		SyncListener:     &SyncHandler{devices: devWithUUID, chanForSync: chanForSync, mutex: &sync.Mutex{}, states: map[string]map[string]interface{}{}},
 		RecvListener:     &RecvHandler{devices: devWithIface, chanForSync: chanForSync},
 	}
 
+	registerHandleFunc = nil
+	removeHandleFunc = nil
+}
+
+func AddRegisterHandleFunc(h func(e Event)) {
+	registerHandleFunc = h
+}
+
+func AddRemoveHandleFunc(h func(e Event)) {
+	removeHandleFunc = h
 }
 
 func Close() {
@@ -38,24 +51,27 @@ func Sync(key string, param map[string]interface{}) {
 	_managerObj.onSync(key, param)
 }
 
-func SetDevicePropsToSync(uuid string, propsToSync []string) error {
-	device, ok := _managerObj.devicesWithUUID[uuid]
-	if !ok {
-		return errors.New("device not found")
-	}
-	device.propsToSync = propsToSync
-	return nil
+func AddRecvListener(h EventHandler) {
+	_managerObj.addRecvListener(h)
 }
 
+// func SetDevicePropsToSync(uuid string, propsToSync []string) error {
+// 	device, ok := _managerObj.devicesWithUUID[uuid]
+// 	if !ok {
+// 		return errors.New("device not found")
+// 	}
+// 	device.propsToSync = propsToSync
+// 	return nil
+// }
+
 func Run() error {
-	iface, err := puserial.InitDevice()
+	ifaces, err := puserial.InitDevice()
 	if err != nil {
-		if err.Error() != "USB Not found" {
-			return err
-		}
-	} else {
-		// _managerObj.onAdd(iface)
-		_managerObj.onAdded(iface)
+		return err
+	}
+
+	for _, e := range ifaces {
+		go _managerObj.onAdded(e)
 	}
 
 	go puserial.WatchNewDevice(ctx, ch_discover)
@@ -66,6 +82,13 @@ func Run() error {
 			log.Println("manager exit")
 			return nil
 		}
-		_managerObj.onAdded(e.Path())
+		switch e.Event() {
+		case notify.Create:
+			go _managerObj.onAdded(e.Path())
+			// case notify.Remove:
+			// 	log.Println("USB Disconnected!!")
+			// 	_managerObj.onAdded(e.Path())
+		}
+
 	}
 }

communication.go

@@ -15,6 +15,7 @@ func recv(port io.Reader, r Receiver) {
 		if err != nil {
 			if err == io.EOF {
 				log.Println("USB is disconnected")
+				_managerObj.onRemoved(port)
 				return
 			}
 		}

examples/main.go

@@ -11,8 +11,24 @@ import (
 )
 
 func main() {
-	go manager.Run()
+	manager.AddRecvListener(manager.NewEventHandler(func(e manager.Event) {
+		fmt.Println("RECV: ", e.Params())
+	}))
 
+	manager.AddRegisterHandleFunc(func(e manager.Event) {
+		param := e.Params()
+		fmt.Println(param["uuid"].(string), " is registered!! ] ", param["sname"])
+	})
+
+	manager.AddRemoveHandleFunc(func(e manager.Event) {
+		param := e.Params()
+		fmt.Println(param["uuid"].(string), " is removed!!")
+	})
+
+	go manager.Run()
+	defer manager.Close()
+
+	param := map[string]interface{}{}
 	for {
 		fmt.Print("> ")
 		reader := bufio.NewReader(os.Stdin)
@@ -21,9 +37,6 @@ func main() {
 		fmt.Println("cmd: ", cmd)
 		if cmd == "exit" {
 			return
-		} else if cmd == "setup" {
-			manager.SetDevicePropsToSync("DEVICE-A-UUID", []string{"fan", "servo"})
-			continue
 		}
 
 		tkns := strings.Split(cmd, " ")
@@ -32,9 +45,7 @@ func main() {
 		if err != nil {
 			continue
 		}
-		param := map[string]interface{}{
-			tkns[1]: value,
-		}
+		param[tkns[1]] = value
 		manager.Sync(key, param)
 	}
 }

interface.go

@@ -15,3 +15,7 @@ type Receiver interface {
 type Syncronizer interface {
 	onSync(key interface{}, params map[string]interface{})
 }
+
+
+
+

listener.go

@@ -4,34 +4,52 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"reflect"
+	"sync"
 )
 
 type SyncHandler struct {
 	devices     map[string]*_device
+	mutex       *sync.Mutex
 	chanForSync map[string]chan map[string]interface{}
+	states      map[string]map[string]interface{}
 }
 
-func compareMap(src map[string]interface{}, dst map[string]interface{}, props []string) bool {
-	for _, prop := range props {
-		srcV, ok := src[prop].(int)
-		if !ok {
-			srcV = int(src[prop].(float64))
+func compareMap(src map[string]interface{}, dst map[string]interface{}) bool {
+
+	for key, value := range src {
+		if key == "code" {
+			continue
 		}
 
-		dstV, ok := dst[prop].(int)
-		if !ok {
-			dstV = int(dst[prop].(float64))
+		if reflect.TypeOf(value).String() == "string" {
+			dstV, ok := dst[key].(string)
+			if !ok || value != dstV {
+				return false
+			}
+			continue
 		}
+
+		srcV, ok := value.(float64)
+		if !ok {
+			srcV = float64(value.(int))
+		}
+
+		dstV, ok := dst[key].(float64)
+		if !ok {
+			dstV = float64(dst[key].(int))
+		}
+
 		if srcV != dstV {
-			fmt.Println("diff ", prop, "] ", srcV, " vs ", dstV)
+			fmt.Println("diff ", key, "] ", srcV, " vs ", dstV)
 			return false
 		}
 	}
+
 	return true
 }
 
 func (sh *SyncHandler) Handle(e Event) {
-
 	device, ok := sh.devices[e.Key().(string)]
 	// fmt.Println("sync] ", device.IfaceName)
 	// fmt.Println(sh.devices)
@@ -48,46 +66,62 @@ func (sh *SyncHandler) Handle(e Event) {
 		encoder := json.NewEncoder(device.Iface)
 		origin := map[string]interface{}{}
 		origin["code"] = 200
-		props := device.propsToSync
-		// props := []string{"fan", "light", "servo"}
-		for _, pname := range props {
-			prop, ok := params[pname]
-			if !ok {
-				origin[pname] = device.states[pname]
-			} else {
-				origin[pname] = prop
-			}
+		for key, value := range params {
+			origin[key] = value
 		}
 
+		sh.states[device.UUID] = origin
+		// props := []string{"fan", "light", "servo"}
+
 		err := encoder.Encode(origin)
 		if err != nil {
 			return
 		}
 
-		sh.chanForSync[device.IfaceName] = make(chan map[string]interface{})
-		for state := range sh.chanForSync[device.IfaceName] {
-			if compareMap(origin, state, props) {
-				log.Println("Same!!")
+		_, ok := sh.chanForSync[device.IfaceName]
+		if ok {
+			sh.mutex.Lock()
+			close(sh.chanForSync[device.IfaceName])
+			delete(sh.chanForSync, device.IfaceName)
+			sh.mutex.Unlock()
+		}
+
+		chanForSync := make(chan map[string]interface{})
+		sh.mutex.Lock()
+		sh.chanForSync[device.IfaceName] = chanForSync
+		sh.mutex.Unlock()
+
+		for state := range chanForSync {
+			if compareMap(sh.states[device.UUID], state) {
+				sh.mutex.Lock()
 				close(sh.chanForSync[device.IfaceName])
 				delete(sh.chanForSync, device.IfaceName)
+				sh.mutex.Unlock()
 				return
 			}
 			log.Println("wrong: ", state)
-			log.Println("resend: ", origin)
-			err := encoder.Encode(origin)
+			log.Println("resend: ", sh.states[device.UUID])
+			err := encoder.Encode(sh.states[device.UUID])
 			if err != nil {
 				return
 			}
 		}
+
 	}()
 }
 
 type RecvHandler struct {
 	devices     map[interface{}]*_device
 	chanForSync map[string]chan map[string]interface{}
+	next        EventHandler
 }
 
 func (rh *RecvHandler) Handle(e Event) {
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Println("panic recover - ", r)
+		}
+	}()
 	device, ok := rh.devices[e.Key()]
 	if !ok {
 		return
@@ -95,14 +129,19 @@ func (rh *RecvHandler) Handle(e Event) {
 
 	param := e.Params()
 	code, _ := param["code"].(float64)
-	if int(code) != 100 {
-		device.states = param
+	if int(code) == 100 {
+		return
 	}
 
+	device.states = param
 	channel, ok := rh.chanForSync[device.IfaceName]
 	if ok {
 		channel <- device.states
 	}
 
+	if rh.next != nil {
+		rh.next.Handle(e)
+	}
+
 	// fmt.Println("recv] ", device.states)
 }

manager.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log"
 
 	"github.com/jacobsa/go-serial/serial"
@@ -12,13 +13,44 @@ import (
 type _manager struct {
 	devicesWithUUID  map[string]*_device
 	devicesWithIface map[interface{}]*_device
+	chanForSync      map[string]chan map[string]interface{}
+	RegisterListener EventHandler
 	SyncListener     EventHandler
 	RecvListener     EventHandler
 }
 
 func (m *_manager) onRegistered(dev *_device) {
+	// if m.RegisterListener != nil {
+	// 	m.RegisterListener.Handle(&EventStruct{key: })
+	// }
+	if registerHandleFunc != nil {
+		param := map[string]interface{}{}
+		param["uuid"] = dev.UUID
+		param["sname"] = dev.Sname
+
+		registerHandleFunc(&EventStruct{key: dev.UUID, params: param})
+	}
 	go recv(dev.Iface, m)
-	log.Println("> End onRegistered")
+}
+
+func (m *_manager) onRemoved(port io.Reader) {
+	dev := _managerObj.devicesWithIface[port]
+	delete(m.devicesWithUUID, dev.UUID)
+	delete(m.devicesWithIface, port)
+	ch, ok := m.chanForSync[dev.UUID]
+	if ok {
+		close(ch)
+		delete(m.chanForSync, dev.UUID)
+	}
+
+	if removeHandleFunc != nil {
+		param := map[string]interface{}{}
+		param["uuid"] = dev.UUID
+
+		removeHandleFunc(&EventStruct{key: dev.UUID, params: param})
+	}
+	// log.Println(m.devicesWithUUID)
+	// log.Println(m.devicesWithIface)
 }
 
 func (m *_manager) onAdded(iface string) {
@@ -65,6 +97,7 @@ func (m *_manager) onAdded(iface string) {
 					UUID:      rcvMsg["uuid"].(string),
 					IfaceName: iface,
 					Iface:     port,
+					Sname:     rcvMsg["sname"].(string),
 					states:    map[string]interface{}{},
 				}
 				m.devicesWithUUID[rcvMsg["uuid"].(string)] = newDevice
@@ -92,3 +125,7 @@ func (m *_manager) onRecv(key interface{}, params map[string]interface{}) {
 		m.RecvListener.Handle(&EventStruct{key: key, params: params})
 	}
 }
+
+func (m *_manager) addRecvListener(h EventHandler) {
+	m.RecvListener = &RecvHandler{next: h, devices: m.devicesWithIface, chanForSync: m.chanForSync}
+}

model.go

@@ -3,15 +3,13 @@ package manager
 import "io"
 
 type _device struct {
-	UUID        string
-	IfaceName   string
-	Iface       io.ReadWriter
-	states      map[string]interface{}
-	propsToSync []string
+	UUID      string
+	IfaceName string
+	Sname     string
+	Iface     io.ReadWriter
+	states    map[string]interface{}
 }
 
-
-
 type RecvEvent struct {
 	Params map[string]interface{}
 }

puserial/serial.go

@@ -2,7 +2,6 @@ package puserial
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io/ioutil"
 	"path/filepath"
@@ -11,25 +10,27 @@ import (
 	"github.com/rjeczalik/notify"
 )
 
-func InitDevice() (string, error) {
+func InitDevice() ([]string, error) {
 	fs, err := ioutil.ReadDir("/dev")
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
+	var result []string = nil
 	for _, f := range fs {
 		if strings.Contains(f.Name(), "ttyACM") {
-			return filepath.Join("/dev", f.Name()), nil
+			result = append(result, filepath.Join("/dev", f.Name()))
 		}
 	}
 
-	return "", errors.New("USB Not found")
+	return result, nil
 }
+
 func WatchNewDevice(ctx context.Context, ch_discover chan<- notify.EventInfo) error {
 	defer close(ch_discover)
 
 	filter := make(chan notify.EventInfo, 1)
-	if err := notify.Watch("/dev", filter, notify.Create, notify.Remove); err != nil {
+	if err := notify.Watch("/dev", filter, notify.Create); err != nil {
 		return err
 	}
 	defer notify.Stop(filter)

structure.go

@@ -13,3 +13,16 @@ func (es *EventStruct) Params() map[string]interface{} {
 func (es *EventStruct) Key() interface{} {
 	return es.key
 }
+
+type EventHandlerStruct struct {
+	EventHandler
+	HandleFunc func(e Event)
+}
+
+func (ehs *EventHandlerStruct) Handle(e Event) {
+	ehs.HandleFunc(e)
+}
+
+func NewEventHandler(h func(e Event)) *EventHandlerStruct {
+	return &EventHandlerStruct{HandleFunc: h}
+}