messages are being passed along the queue

cmd/echo/server.go

@@ -32,6 +32,8 @@ func main() {
 
 	ctx := context.Background()
 	ts := NewTELNETServer(ctx, l, pubSub)
+	tp := NewTELNETParser(ctx, pubSub)
+	go tp.Handle()
 
 	for {
 		conn, err := ts.l.Accept()

cmd/echo/telnet.go

@@ -11,6 +11,28 @@ import (
 )
 
 type TELNETParser struct {
+	s message.Subscriber
+	c context.Context
+}
+
+func (tp *TELNETParser) Handle() {
+	messages, err := tp.s.Subscribe(tp.c, "telnet.raw")
+	if err != nil {
+		log.Fatalln("couldn't subscribe to telnet.raw")
+	}
+
+	for message := range messages {
+		log.Printf("%#+v\n", message)
+		message.Ack()
+	}
+
+}
+
+func NewTELNETParser(c context.Context, s message.Subscriber) *TELNETParser {
+	return &TELNETParser{
+		c: c,
+		s: s,
+	}
 }
 
 type TELNETServer struct {
@@ -37,7 +59,7 @@ func (ts *TELNETServer) Handle(conn net.Conn) {
 		t := s.Text()
 		err := ts.p.Publish("telnet.raw", message.NewMessage(watermill.NewUUID(), []byte(t)))
 		if err != nil {
-			log.Fatalln("couldn't write to raw_telnet queue")
+			log.Fatalln("couldn't write to telnet.raw")
 		}
 		log.Printf("Received user input: %q\n", t)
 	}