integrate broker into server

cmd/server.go

@@ -37,11 +37,11 @@ type session struct {
 }
 
 func serve() {
+
+	broker := NewBroker(3)
 	// start the can listener
 	go vcanTest()
-	fromCanBus := make(chan can.Frame, 1)
+	go canHandler(broker)
-	toCanBus := make(chan can.Frame)
-	go canHandler(toCanBus, fromCanBus)
 	ln, err := net.Listen("tcp", ":8082")
 	if err != nil {
 		fmt.Printf("Error listening: %v\n", err)
@@ -53,21 +53,53 @@ func serve() {
 		if err != nil {
 			fmt.Printf("error accepting: %v\n", err)
 		}
-		go handleCon(conn)
+		go handleCon(conn, broker)
 	}
 }
 
-func handleCon(conn net.Conn) {
+func handleCon(conn net.Conn, broker *Broker) {
 	//	reader := msgp.NewReader(conn)
+	rxPkts := make(chan gotelem.Data)
+	go func() {
+		// setpu our msgp reader.
+		scann := msgp.NewReader(conn)
+		data := gotelem.Data{}
+		for {
+			err := data.DecodeMsg(scann)
+			if err != nil {
+				break
+			}
+			rxPkts <- data
+		}
+
+	}()
+
+	// subscribe to can packets
+	// TODO: make this unique since remote addr could be non-unique
+	canCh := broker.Subscribe(conn.RemoteAddr().String())
 	writer := msgp.NewWriter(conn)
 	for {
+
+		select {
+		case canFrame := <-canCh:
+			cf := gotelem.CanBody{
+				Id:      canFrame.Id,
+				Payload: canFrame.Data,
+				Source:  "me",
+			}
+			cf.EncodeMsg(writer)
+		case rxBody := <-rxPkts:
+			// do nothing for now.
+			fmt.Printf("got a body %v\n", rxBody)
+		case <-time.NewTimer(1 * time.Second).C: // time out.
+			fmt.Printf("timeout\n")
 			data := gotelem.StatusBody{
 				BatteryPct: 1.2,
 				ErrCode:    0,
 			}
 			data.EncodeMsg(writer)
 			writer.Flush()
-		time.Sleep(1 * time.Second)
+		}
 	}
 }
 
@@ -102,7 +134,7 @@ func xbeeSvc(packets <-chan can.Frame, device string, quit <-chan struct{}) {
 	}
 }
 
-// this spins up a new can socket on vcan0 and broadcasts a packet every second.
+// this spins up a new can socket on vcan0 and broadcasts a packet every second. for testing.
 func vcanTest() {
 	sock, _ := socketcan.NewCanSocket("vcan0")
 	testFrame := &can.Frame{
@@ -118,24 +150,25 @@ func vcanTest() {
 	}
 }
 
-func canHandler(pktToSend <-chan can.Frame, pktRecv chan<- can.Frame) {
+func canHandler(broker *Broker) {
+	rxCh := broker.Subscribe("socketcan")
 	sock, _ := socketcan.NewCanSocket("vcan0")
 
 	// start a simple dispatcher that just relays can frames.
-	r := make(chan can.Frame)
+	rxCan := make(chan can.Frame)
 	go func() {
 		for {
 			pkt, _ := sock.Recv()
-			r <- *pkt
+			rxCan <- *pkt
 		}
 	}()
 	for {
 		select {
-		case msg := <-pktToSend:
+		case msg := <-rxCh:
 			sock.Send(&msg)
-		case msg := <-r:
+		case msg := <-rxCan:
 			fmt.Printf("got a packet from the can %v\n", msg)
-			pktRecv <- msg
+			broker.Publish("socketcan", msg)
 		}
 	}
 }
@@ -157,7 +190,7 @@ type Broker struct {
 	unsubCh chan BrokerClient
 }
 
-func NewBroker(bufsize int) *Broker{
+func NewBroker(bufsize int) *Broker {
 	b := &Broker{
 		subs:      make(map[string]chan can.Frame),
 		publishCh: make(chan BrokerRequest, 3),