ws: add thread map to clients

Signed-off-by: Sander Pick <[email protected]>

threads.go

@@ -2,7 +2,6 @@ package threads
 
 import (
  "context"
- "encoding/base64"
  "fmt"
  "io"
  "net/http"
@@ -720,10 +719,7 @@ func (t *threads) broadcast(r tserv.Record) error {
  return err
  }
 
- data := r.Value().RawData()
- encoded := make([]byte, base64.StdEncoding.EncodedLen(len(data)))
- base64.StdEncoding.Encode(encoded, data)
- t.ws.Send(encoded)
+ t.ws.Send(r)
  return nil
 }
 

ws/client.go

@@ -2,10 +2,12 @@ package ws
 
 import (
  "bytes"
+ "encoding/json"
  "net/http"
  "time"
 
  "github.com/gorilla/websocket"
+ "github.com/textileio/go-textile-core/thread"
 )
 
 const (
@@ -36,6 +38,11 @@ var (
  space = []byte{' '}
 )
 
+type msg struct {
+ Type string `json:"type"`
+ Threads []string `json:"threads"`
+}
+
 // Client is a middleman between the websocket connection and the hub.
 type Client struct {
  hub *Hub
@@ -45,6 +52,9 @@ type Client struct {
 
  // Buffered channel of outbound messages.
  send chan []byte
+
+ // Active threads.
+ threads map[thread.ID]struct{}
 }
 
 // readPump pumps messages from the websocket connection to the hub.
@@ -70,13 +80,30 @@ func (c *Client) readPump() {
  err,
  websocket.CloseGoingAway,
  websocket.CloseAbnormalClosure) {
- log.Errorf("error: %v", err)
+ log.Errorf("error reading message: %s", err)
  }
  break
  }
  message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1))
- // @todo: don't echo message—it should hit AddRecord instead
- c.hub.broadcast <- message
+
+ var m msg
+ err = json.Unmarshal(message, &m)
+ if err != nil {
+ log.Errorf("error unmarshaling message: %s", err)
+ break
+ }
+
+ switch m.Type {
+ case "subscribe":
+ for _, t := range m.Threads {
+ id, err := thread.Decode(t)
+ if err == nil {
+ log.Debugf("client requested thread %s", id.String())
+
+ c.threads[id] = struct{}{}
+ }
+ }
+ }
  }
 }
 
@@ -133,7 +160,12 @@ func serveWs(hub *Hub, w http.ResponseWriter, r *http.Request) {
  log.Error(err)
  return
  }
- client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)}
+ client := &Client{
+ hub: hub,
+ conn: conn,
+ send: make(chan []byte, 256),
+ threads: make(map[thread.ID]struct{}),
+ }
  client.hub.register <- client
 
  // Allow collection of memory referenced by the caller by doing all work in

ws/hub.go

@@ -1,42 +1,21 @@
-/* BSD 2-Clause "Simplified" License
-
- Copyright (c) 2013 The Gorilla WebSocket Authors. All rights reserved.
-
- Redistribution and use in source and binary forms, with or without
- modification, are permitted provided that the following conditions are met:
-
- Redistributions of source code must retain the above copyright notice, this
- list of conditions and the following disclaimer.
-
- Redistributions in binary form must reproduce the above copyright notice,
- this list of conditions and the following disclaimer in the documentation
- and/or other materials provided with the distribution.
-
- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
- ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
- WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
- DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
- FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
- DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
- SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
- CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
- OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-*/
 // Inspired by https://github.com/gorilla/websocket/tree/master/examples/chat with
-// adaptations for multiple rooms ("threads" in Textile parlance).
+// adaptations for multiple rooms ("threads" in Textile parlance) and authentication.
 package ws
 
+import (
+ "encoding/base64"
+
+ tserv "github.com/textileio/go-textile-core/threadservice"
+)
+
 // Hub maintains the set of active clients and broadcasts messages to the
 // clients.
 type Hub struct {
  // clients currently registered.
- // @todo: migrate to emtpy struct
- // @todo: map clients to threads via thread list
- clients map[*Client]bool
+ clients map[*Client]struct{}
 
- // broadcast message to clients.
- broadcast chan []byte
+ // broadcast records to clients.
+ broadcast chan tserv.Record
 
  // register requests from the clients.
  register chan *Client
@@ -48,10 +27,10 @@ type Hub struct {
 // NewHub creates a new client hub.
 func newHub() *Hub {
  return &Hub{
- broadcast: make(chan []byte),
+ broadcast: make(chan tserv.Record),
  register: make(chan *Client),
  unregister: make(chan *Client),
- clients: make(map[*Client]bool),
+ clients: make(map[*Client]struct{}),
  }
 }
 
@@ -60,16 +39,22 @@ func (h *Hub) run() {
  for {
  select {
  case client := <-h.register:
- h.clients[client] = true
+ h.clients[client] = struct{}{}
  case client := <-h.unregister:
  if _, ok := h.clients[client]; ok {
  delete(h.clients, client)
  close(client.send)
  }
- case message := <-h.broadcast:
+ case rec := <-h.broadcast:
+ data := rec.Value().RawData()
+ msg := make([]byte, base64.StdEncoding.EncodedLen(len(data)))
+ base64.StdEncoding.Encode(msg, data)
  for client := range h.clients {
+ if _, ok := client.threads[rec.ThreadID()]; !ok {
+ continue
+ }
  select {
- case client.send <- message:
+ case client.send <- msg:
  default:
  close(client.send)
  delete(h.clients, client)

ws/index.html

@@ -7,16 +7,28 @@
  <meta charset="utf-8">
  <script>
  window.addEventListener('load', function() {
- const socket = new WebSocket('ws:https://localhost:8080');
+ const socket = new WebSocket('ws:https://localhost:8080')
  socket.addEventListener('open', function (e) {
  console.log('Connection started.')
- });
+
+ // Subscribe to thread in URL query
+ let threads
+ let params = new URLSearchParams(window.location.search);
+ if (params.has('thread')) {
+ threads = params.getAll('thread')
+ }
+ let msg = {
+ type: 'subscribe',
+ threads: threads
+ }
+ socket.send(JSON.stringify(msg))
+ })
  socket.addEventListener('message', function (e) {
- console.log('Record: ', e.data);
- });
+ console.log('Record: ', e.data)
+ })
  socket.addEventListener('error', function (e) {
- console.error("Error: ", e);
- });
+ console.error("Error: ", e)
+ })
  });
  </script>
  <title>Threads</title>

ws/server.go

@@ -6,6 +6,7 @@ import (
  "time"
 
  logging "github.com/ipfs/go-log"
+ tserv "github.com/textileio/go-textile-core/threadservice"
 )
 
 var log = logging.Logger("ws")
@@ -55,8 +56,8 @@ func NewServer(addr string) *Server {
 }
 
 // Send a message to the hub.
-func (s *Server) Send(message []byte) {
- s.hub.broadcast <- message
+func (s *Server) Send(r tserv.Record) {
+ s.hub.broadcast <- r
 }
 
 // Close the server.