-
Notifications
You must be signed in to change notification settings - Fork 22
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Streaming Flux state to the frontend
- Loading branch information
Showing
7 changed files
with
303 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,213 @@ | ||
| // Copyright 2013 The Gorilla WebSocket Authors. All rights reserved. | ||
| // Use of this source code is governed by a BSD-style | ||
| // license that can be found in the LICENSE file. | ||
|
|
||
| package streaming | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "net/http" | ||
| "time" | ||
|
|
||
| "github.com/gorilla/websocket" | ||
| log "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| const ( | ||
| // Time allowed to write a message to the peer. | ||
| writeWait = 10 * time.Second | ||
|
|
||
| // Time allowed to read the next pong message from the peer. | ||
| pongWait = 60 * time.Second | ||
|
|
||
| // Send pings to peer with this period. Must be less than pongWait. | ||
| pingPeriod = (pongWait * 9) / 15 | ||
|
|
||
| // Maximum message size allowed from peer. | ||
| maxMessageSize = 4096 | ||
| ) | ||
|
|
||
| var ( | ||
| newline = []byte{'\n'} | ||
| space = []byte{' '} | ||
| ) | ||
|
|
||
| var upgrader = websocket.Upgrader{ | ||
| ReadBufferSize: 1024, | ||
| WriteBufferSize: 1024, | ||
| } | ||
|
|
||
| // Client is a middleman between the websocket connection and the hub. | ||
| type Client struct { | ||
| hub *ClientHub | ||
|
|
||
| // The websocket connection. | ||
| conn *websocket.Conn | ||
|
|
||
| // Buffered channel of outbound messages. | ||
| send chan []byte | ||
|
|
||
| clientId string | ||
| } | ||
|
|
||
| // readPump pumps messages from the websocket connection to the hub. | ||
|
|
||
| // The application runs readPump in a per-connection goroutine. The application | ||
| // ensures that there is at most one reader on a connection by executing all | ||
| // reads from this goroutine. | ||
| func (c *Client) readPump() { | ||
| defer func() { | ||
| c.hub.Unregister <- c | ||
| c.conn.Close() | ||
| }() | ||
| c.conn.SetReadLimit(maxMessageSize) | ||
| c.conn.SetReadDeadline(time.Now().Add(pongWait)) | ||
| c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil }) | ||
| for { | ||
| _, message, err := c.conn.ReadMessage() | ||
| if err != nil { | ||
| if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { | ||
| log.Printf("error: %v", err) | ||
| } | ||
| break | ||
| } | ||
| message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1)) | ||
| c.hub.Broadcast <- message | ||
| } | ||
| } | ||
|
|
||
| // writePump pumps messages from the hub to the websocket connection. | ||
| // | ||
| // A goroutine running writePump is started for each connection. The | ||
| // application ensures that there is at most one writer to a connection by | ||
| // executing all writes from this goroutine. | ||
| func (c *Client) writePump() { | ||
| ticker := time.NewTicker(pingPeriod) | ||
| defer func() { | ||
| ticker.Stop() | ||
| c.conn.Close() | ||
| }() | ||
| for { | ||
| select { | ||
| case message, ok := <-c.send: | ||
| c.conn.SetWriteDeadline(time.Now().Add(writeWait)) | ||
| if !ok { | ||
| // The hub closed the channel. | ||
| c.conn.WriteMessage(websocket.CloseMessage, []byte{}) | ||
| return | ||
| } | ||
|
|
||
| w, err := c.conn.NextWriter(websocket.TextMessage) | ||
| if err != nil { | ||
| return | ||
| } | ||
| w.Write(message) | ||
|
|
||
| // Add queued chat messages to the current websocket message. | ||
| n := len(c.send) | ||
| for i := 0; i < n; i++ { | ||
| w.Write(newline) | ||
| w.Write(<-c.send) | ||
| } | ||
|
|
||
| if err := w.Close(); err != nil { | ||
| return | ||
| } | ||
| case <-ticker.C: | ||
| c.conn.SetWriteDeadline(time.Now().Add(writeWait)) | ||
| if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { | ||
| return | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // ServeWs handles websocket requests from the peer. | ||
| func ServeWs(hub *ClientHub, w http.ResponseWriter, r *http.Request) { | ||
| upgrader.CheckOrigin = func(r *http.Request) bool { return true } | ||
| conn, err := upgrader.Upgrade(w, r, nil) | ||
| if err != nil { | ||
| log.Println(err) | ||
| return | ||
| } | ||
|
|
||
| client := &Client{ | ||
| hub: hub, | ||
| conn: conn, | ||
| send: make(chan []byte, 256), | ||
| } | ||
| hub.Register <- client | ||
|
|
||
| // Allow collection of memory referenced by the caller by doing all work in | ||
| // new goroutines. | ||
| go client.writePump() | ||
| go client.readPump() | ||
| } | ||
|
|
||
| // ClientHub maintains the set of active clients and broadcasts messages to the | ||
| // clients. | ||
| type ClientHub struct { | ||
| // Registered clients. | ||
| Clients map[*Client]bool | ||
|
|
||
| // Updates to be broadcasted to the clients. | ||
| Broadcast chan []byte | ||
|
|
||
| // Updates to be sent to a single user. | ||
| Send chan *ClientMessage | ||
|
|
||
| // Register requests from the clients. | ||
| Register chan *Client | ||
|
|
||
| // Unregister requests from clients. | ||
| Unregister chan *Client | ||
| } | ||
|
|
||
| type ClientMessage struct { | ||
| ClientId string | ||
| Message []byte | ||
| } | ||
|
|
||
| func NewClientHub() *ClientHub { | ||
| return &ClientHub{ | ||
| Broadcast: make(chan []byte), | ||
| Send: make(chan *ClientMessage), | ||
| Register: make(chan *Client), | ||
| Unregister: make(chan *Client), | ||
| Clients: make(map[*Client]bool), | ||
| } | ||
| } | ||
|
|
||
| func (h *ClientHub) Run() { | ||
| for { | ||
| select { | ||
| case client := <-h.Register: | ||
| h.Clients[client] = true | ||
| case client := <-h.Unregister: | ||
| if _, ok := h.Clients[client]; ok { | ||
| delete(h.Clients, client) | ||
| close(client.send) | ||
| } | ||
| case message := <-h.Broadcast: | ||
| for client := range h.Clients { | ||
| select { | ||
| case client.send <- message: | ||
| default: | ||
| close(client.send) | ||
| delete(h.Clients, client) | ||
| } | ||
| } | ||
| case clientMessage := <-h.Send: | ||
| for client := range h.Clients { | ||
| if client.clientId == clientMessage.ClientId { | ||
| select { | ||
| case client.send <- clientMessage.Message: | ||
| default: | ||
| close(client.send) | ||
| delete(h.Clients, client) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| import {Component} from "react"; | ||
|
|
||
| let URL = ''; | ||
| if (typeof window !== 'undefined') { | ||
| let protocol = window.location.protocol === 'https:' ? 'wss' : 'ws'; | ||
| URL = protocol + '://' + window.location.hostname; | ||
|
|
||
| let port = window.location.port | ||
| if (window.location.hostname === 'localhost' || window.location.hostname === '127.0.0.1') { | ||
| port = 9000 | ||
| } | ||
| if (port && port !== '') { | ||
| URL = URL + ':' + port | ||
| } | ||
| } | ||
|
|
||
| export default class StreamingBackend extends Component { | ||
| componentDidMount() { | ||
| console.log("Connecting to " + URL + '/ws/') | ||
|
|
||
| this.ws = new WebSocket(URL + '/ws/'); | ||
| this.ws.onopen = this.onOpen; | ||
| this.ws.onmessage = this.onMessage; | ||
| this.ws.onclose = this.onClose; | ||
|
|
||
| this.onClose = this.onClose.bind(this); | ||
| } | ||
|
|
||
| render() { | ||
| return null; | ||
| } | ||
|
|
||
| onOpen = () => { | ||
| console.log('connected'); | ||
| }; | ||
|
|
||
| onClose = (evt) => { | ||
| console.log('disconnected: ' + evt.code + ': ' + evt.reason); | ||
| const ws = new WebSocket(URL + '/ws/'); | ||
| ws.onopen = this.onOpen; | ||
| ws.onmessage = this.onMessage; | ||
| ws.onclose = this.onClose; | ||
| this.setState({ | ||
| ws | ||
| }); | ||
| } | ||
|
|
||
| onMessage = (evt) => { | ||
| evt.data.split('\n').forEach((line) => { | ||
| const message = JSON.parse(line); | ||
| console.log(message) | ||
| }); | ||
| } | ||
| } |