wsRead loops for the duration of a websocket connection, reading messages, decoding them to events, and passing them to the event bus.
(ws *websocket.Conn)
| 406 | // reading messages, decoding them to events, and passing |
| 407 | // them to the event bus. |
| 408 | func (c *Client) wsRead(ws *websocket.Conn) chan error { |
| 409 | errChan := make(chan error, 1) |
| 410 | |
| 411 | go func() { |
| 412 | for { |
| 413 | var data []byte |
| 414 | |
| 415 | err := websocket.Message.Receive(ws, &data) |
| 416 | if err != nil { |
| 417 | errChan <- eris.Wrap(err, "failed to receive websocket message") |
| 418 | return |
| 419 | } |
| 420 | |
| 421 | e, err := ari.DecodeEvent(data) |
| 422 | if err != nil { |
| 423 | c.Options.Logger.Error("failed to decode websocket message to event", "error", err) |
| 424 | // if decode fails, continue to next message, we can't process nil ari.Event anyway |
| 425 | continue |
| 426 | } |
| 427 | |
| 428 | c.bus.Send(e) |
| 429 | } |
| 430 | }() |
| 431 | |
| 432 | return errChan |
| 433 | } |
| 434 | |
| 435 | // stamp imprints the node metadata onto the given Key |
| 436 | func (c *Client) stamp(key *ari.Key) *ari.Key { |