watsonb8/fayec
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix overlapped payload values

Browse Source
This commit is contained in:
Marcelo Pires 2018-09-06 11:16:36 +02:00
parent f11de6960b
commit 23ae14adea
1 changed files with 7 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
transport/websocket/websocket.go
View File

@ -53,8 +53,8 @@ func (w *Websocket) Init(options *transport.Options) error {
} }
func (w *Websocket) readWorker() error { func (w *Websocket) readWorker() error {
var payload []message.Message
for { for {
var payload []message.Message
err := w.conn.ReadJSON(&payload) err := w.conn.ReadJSON(&payload)
if err != nil { if err != nil {
return err return err
@ -63,26 +63,23 @@ func (w *Websocket) readWorker() error {
msg := payload[0] msg := payload[0]
if transport.IsControlMsg(msg.Channel) { if transport.IsControlMsg(msg.Channel) {
log.Println("recv control message", debugJson(msg))
//handle it //handle it
switch msg.Channel { switch msg.Channel {
case transport.Subscribe: case transport.Subscribe:
//handle Subscribe resp //handle Subscribe resp
if !msg.Successful { if !msg.Successful {
w.subsMu.Lock() w.subsMu.Lock()
subscription, ok := w.subs[msg.Channel] subscription, ok := w.subs[msg.Subscription]
w.subsMu.Unlock() w.subsMu.Unlock()
log.Println(debugJson(msg))
if !ok { if !ok {
panic("BUG: subscription not registered") panic("BUG: subscription not registered `" + msg.Subscription + "`")
} }
subscription <- &msg subscription <- &msg
close(subscription) close(subscription)
w.subsMu.Lock() w.subsMu.Lock()
delete(w.subs, msg.Channel) delete(w.subs, msg.Channel)
w.subsMu.Unlock() w.subsMu.Unlock()
} }
case transport.Unsubscribe: case transport.Unsubscribe:
//handle Unsubscribe resp //handle Unsubscribe resp
@ -128,8 +125,8 @@ func (w *Websocket) nextMsgID() string {
func (w *Websocket) Options() *transport.Options { func (w *Websocket) Options() *transport.Options {
return w.TransportOpts return w.TransportOpts
} }
func (w *Websocket) Handshake() (err error) {
func (w *Websocket) Handshake() (err error) {
m := message.Message{ m := message.Message{
Channel: transport.Handshake, Channel: transport.Handshake,
Version: "1.0", //todo const Version: "1.0", //todo const
@ -152,7 +149,6 @@ func (w *Websocket) Handshake() (err error) {
if resp.GetError() != nil { if resp.GetError() != nil {
return err return err
} }
log.Println(debugJson(resp))
w.clientID = resp.ClientId w.clientID = resp.ClientId
return nil return nil
} }
@ -165,7 +161,6 @@ func (w *Websocket) Connect() error {
Id: w.nextMsgID(), Id: w.nextMsgID(),
} }
//todo verify if extensions are applied on connect,verify if hs is complete //todo verify if extensions are applied on connect,verify if hs is complete
go w.readWorker() go w.readWorker()
return w.sendMessage(&m) return w.sendMessage(&m)
} }
@ -186,10 +181,11 @@ func (w *Websocket) Subscribe(subscription string, onMessage func(message *messa
} }
//todo validate //todo validate
inMsgCh := make(chan *message.Message, 0) inMsgCh := make(chan *message.Message, 0)
w.subsMu.Lock()
w.subs[subscription] = inMsgCh w.subs[subscription] = inMsgCh
w.subsMu.Unlock()
var inMsg *message.Message var inMsg *message.Message
for inMsg = range inMsgCh { for inMsg = range inMsgCh {