commit 2021-12-12]13:48:44

This commit is contained in:
Godopu 2021-12-12 13:48:55 +09:00
parent 31b00fb912
commit a0ff2d6f52
3 changed files with 18 additions and 4 deletions

7
cli.go
View File

@ -3,6 +3,7 @@ package manager
import ( import (
"context" "context"
"log" "log"
"sync"
"git.godopu.net/lab/etri-smartfarm-poc-controller-serial/puserial" "git.godopu.net/lab/etri-smartfarm-poc-controller-serial/puserial"
"github.com/rjeczalik/notify" "github.com/rjeczalik/notify"
@ -21,13 +22,15 @@ func init() {
devWithUUID := map[string]*_device{} devWithUUID := map[string]*_device{}
devWithIface := map[interface{}]*_device{} devWithIface := map[interface{}]*_device{}
chanForSync := map[string]chan map[string]interface{}{} chanForSync := map[string]chan map[string]interface{}{}
var mutex = sync.Mutex{}
_managerObj = &_manager{ _managerObj = &_manager{
devicesWithUUID: devWithUUID, devicesWithUUID: devWithUUID,
devicesWithIface: devWithIface, devicesWithIface: devWithIface,
chanForSync: chanForSync, chanForSync: chanForSync,
SyncListener: &SyncHandler{devices: devWithUUID, chanForSync: chanForSync}, mutex: &mutex,
RecvListener: &RecvHandler{devices: devWithIface, chanForSync: chanForSync}, SyncListener: &SyncHandler{devices: devWithUUID, chanForSync: chanForSync, mutex: &mutex},
RecvListener: &RecvHandler{devices: devWithIface, chanForSync: chanForSync, mutex: &sync.Mutex{}},
} }
registerHandleFunc = nil registerHandleFunc = nil

View File

@ -4,10 +4,12 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"log" "log"
"sync"
) )
type SyncHandler struct { type SyncHandler struct {
devices map[string]*_device devices map[string]*_device
mutex *sync.Mutex
chanForSync map[string]chan map[string]interface{} chanForSync map[string]chan map[string]interface{}
} }
@ -67,8 +69,10 @@ func (sh *SyncHandler) Handle(e Event) {
_, ok := sh.chanForSync[device.IfaceName] _, ok := sh.chanForSync[device.IfaceName]
if ok { if ok {
sh.mutex.Lock()
close(sh.chanForSync[device.IfaceName]) close(sh.chanForSync[device.IfaceName])
delete(sh.chanForSync, device.IfaceName) delete(sh.chanForSync, device.IfaceName)
sh.mutex.Lock()
} }
chanForSync := make(chan map[string]interface{}) chanForSync := make(chan map[string]interface{})
sh.chanForSync[device.IfaceName] = make(chan map[string]interface{}) sh.chanForSync[device.IfaceName] = make(chan map[string]interface{})
@ -76,23 +80,26 @@ func (sh *SyncHandler) Handle(e Event) {
for state := range chanForSync { for state := range chanForSync {
if compareMap(origin, state) { if compareMap(origin, state) {
sh.mutex.Lock()
close(sh.chanForSync[device.IfaceName]) close(sh.chanForSync[device.IfaceName])
delete(sh.chanForSync, device.IfaceName) delete(sh.chanForSync, device.IfaceName)
sh.mutex.Lock()
return return
} }
log.Println("wrong: ", state) log.Println("wrong: ", state)
log.Println("resend: ", origin) log.Println("resend: ", origin)
err := encoder.Encode(origin) err := encoder.Encode(origin)
if err != nil { if err != nil {
log.Println("sync routine is died")
return return
} }
} }
log.Println("sync routine is died")
}() }()
} }
type RecvHandler struct { type RecvHandler struct {
devices map[interface{}]*_device devices map[interface{}]*_device
mutex *sync.Mutex
chanForSync map[string]chan map[string]interface{} chanForSync map[string]chan map[string]interface{}
next EventHandler next EventHandler
} }
@ -110,7 +117,9 @@ func (rh *RecvHandler) Handle(e Event) {
} }
device.states = param device.states = param
rh.mutex.Lock()
channel, ok := rh.chanForSync[device.IfaceName] channel, ok := rh.chanForSync[device.IfaceName]
rh.mutex.Unlock()
if ok { if ok {
channel <- device.states channel <- device.states
} }

View File

@ -6,6 +6,7 @@ import (
"fmt" "fmt"
"io" "io"
"log" "log"
"sync"
"github.com/jacobsa/go-serial/serial" "github.com/jacobsa/go-serial/serial"
) )
@ -14,6 +15,7 @@ type _manager struct {
devicesWithUUID map[string]*_device devicesWithUUID map[string]*_device
devicesWithIface map[interface{}]*_device devicesWithIface map[interface{}]*_device
chanForSync map[string]chan map[string]interface{} chanForSync map[string]chan map[string]interface{}
mutex *sync.Mutex
RegisterListener EventHandler RegisterListener EventHandler
SyncListener EventHandler SyncListener EventHandler
RecvListener EventHandler RecvListener EventHandler
@ -127,5 +129,5 @@ func (m *_manager) onRecv(key interface{}, params map[string]interface{}) {
} }
func (m *_manager) addRecvListener(h EventHandler) { func (m *_manager) addRecvListener(h EventHandler) {
m.RecvListener = &RecvHandler{next: h, devices: m.devicesWithIface, chanForSync: m.chanForSync} m.RecvListener = &RecvHandler{next: h, devices: m.devicesWithIface, chanForSync: m.chanForSync, mutex: m.mutex}
} }