Turn adsb-ws into a json repeater, rather than a protocol converter

sinks/adsb-ws/Makefile

@@ -1,9 +1,6 @@
 all: adsb-ws
 
-proto/adsb.pb.go: ../../proto/adsb.proto
+adsb-ws: *.go
-	protoc --gogoslick_out=. --proto_path=../.. $<
-
-adsb-ws: *.go proto/adsb.pb.go
 	go build
 
 clean:

sinks/adsb-ws/hub.go

@@ -0,0 +1,41 @@
+package main
+
+type hub struct {
+	connections map[*connection]bool
+	broadcast chan []byte
+	register chan *connection
+	unregister chan *connection
+}
+
+var h = hub{
+	broadcast:   make(chan []byte),
+	register:    make(chan *connection),
+	unregister:  make(chan *connection),
+	connections: make(map[*connection]bool),
+}
+
+func (h *hub) run() {
+	for {
+		select {
+		case c := <-h.register:
+			h.connections[c] = true
+
+		case c := <-h.unregister:
+			_, ok := h.connections[c]
+			if ok {
+				delete(h.connections, c)
+				close(c.send)
+			}
+
+		case m := <-h.broadcast:
+			for c := range h.connections {
+				select {
+				case c.send <- m:
+				default:
+					close(c.send)
+					delete(h.connections, c)
+				}
+			}
+		}
+	}
+}

sinks/adsb-ws/input.go

@@ -2,66 +2,17 @@ package main
 
 import (
 	"bufio"
-	"errors"
 	"log"
 	"os"
-
-	"github.com/adsb-tools/sinks/adsb-ws/proto"
 )
 
-func decodeVarint(r *bufio.Reader) (n uint64, err error) {
-	var value uint64
-	var shift uint16
-	for {
-		c, err := r.ReadByte()
-		if err != nil {
-			return 0, err
-		}
-		value |= (uint64(c) & 0x7f) << shift
-		if c & 0x80 == 0 {
-			return value, nil
-		}
-		shift += 7
-		if shift > 21 {
-			return 0, errors.New("invalid varint")
-		}
-	}
-}
-
 func readInput() {
 	r := bufio.NewReader(os.Stdin)
 	for {
-		c, err := r.ReadByte()
+		line, err := r.ReadBytes('\n')
 		if err != nil {
-			log.Printf("error: %v", err)
+			log.Fatal("Input read error: ", err)
-			break
 		}
-		if c != 0x0a {
+		h.broadcast <- line
-			log.Printf("invalid message type: %v", c)
-			break
 	}
-		msglen, err := decodeVarint(r)
-		if err != nil {
-			log.Printf("error: %v", err)
-			break
-		}
-		buf := make([]byte, msglen)
-		n, err := r.Read(buf)
-		if err != nil {
-			log.Printf("error: %v", err)
-			break
-		}
-		if uint64(n) != msglen {
-			log.Printf("short read")
-			break
-		}
-		packet := new(adsb.Adsb)
-		err = packet.Unmarshal(buf)
-		if err != nil {
-			log.Printf("error: %v", err)
-			break
-		}
-		log.Println(packet)
-	}
-	os.Exit(1)
 }

sinks/adsb-ws/main.go

@@ -6,15 +6,16 @@ import (
 	"net/http"
 )
 
-var addr = flag.String("addr", ":8080", "http service address")
+var bindaddr = flag.String("bind-address", ":8080", "Address to respond to HTTP requests on")
 
 func main() {
 	log.SetFlags(0)
 	flag.Parse()
+	go h.run()
 	go readInput()
 	http.HandleFunc("/stream", serveStream)
-	err := http.ListenAndServe(*addr, nil)
+	err := http.ListenAndServe(*bindaddr, nil)
 	if err != nil {
-		log.Fatal("ListenAndServe: ", err)
+		log.Fatal("Error starting web server: ", err)
 	}
 }

sinks/adsb-ws/stream.go

@@ -1,11 +1,17 @@
 package main
 
 import (
-	"github.com/gorilla/websocket"
 	"log"
 	"net/http"
+
+	"github.com/gorilla/websocket"
 )
 
+type connection struct {
+	ws *websocket.Conn
+	send chan []byte
+}
+
 var upgrader = websocket.Upgrader{
 	ReadBufferSize:  1024,
 	WriteBufferSize: 1024,
@@ -14,24 +20,31 @@ var upgrader = websocket.Upgrader{
 	},
 }
 
-func readPump(ws *websocket.Conn) {
+func (c *connection) writePump() {
 	for {
-		msgtype, msg, err := ws.ReadMessage()
+		message, ok := <-c.send
+		if !ok {
+			c.ws.WriteMessage(websocket.CloseMessage, []byte{})
+			return
+		}
+		err := c.ws.WriteMessage(websocket.TextMessage, message)
 		if err != nil {
-			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
+			return
-				log.Printf("error: %v", err)
 		}
-			break
-		}
-		log.Println(msgtype, msg)
 	}
 }
 
 func serveStream(w http.ResponseWriter, r *http.Request) {
 	ws, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
-		log.Println(err)
+		log.Printf("%s: Error in websocket handshake: %s", r.RemoteAddr, err)
 		return
 	}
-	readPump(ws)
+	log.Printf("%s: New connection", r.RemoteAddr)
+	c := &connection{send: make(chan []byte, 256), ws: ws}
+	h.register <- c
+	c.writePump()
+	h.unregister <- c
+	c.ws.Close()
+	log.Printf("%s: Connection closed", r.RemoteAddr)
 }