Merge branch 'channels'

Author: rmattam

chat.go

@@ -0,0 +1,120 @@
+package chat
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/gorilla/websocket"
+	"github.com/pkg/errors"
+	"github.com/rmattam/go-websocket-chat-demo/redis"
+)
+
+var (
+	upgrader = websocket.Upgrader{
+		ReadBufferSize:  1024,
+		WriteBufferSize: 1024,
+	}
+	waitTimeout = time.Minute * 10
+	log         = logrus.WithField("cmd", "go-websocket-chat-demo-chat")
+)
+
+//Hub struct
+type Hub struct {
+	channel string
+
+	receive redis.Receiver
+	write   redis.Writer
+}
+
+// message sent to us by the javascript client
+type message struct {
+	Handle string `json:"handle"`
+	Text   string `json:"text"`
+}
+
+// SubsribeRedis : Initialize the redis routines required for pub sub.
+func (c *Hub) SubsribeRedis(redisURL string, channel string) {
+	redisPool, err := redis.NewRedisPoolFromURL(redisURL)
+	if err != nil {
+		log.WithField("url", redisURL).Fatal("Unable to create Redis pool")
+	}
+	c.channel = channel
+	c.receive = redis.NewReceiver(redisPool, c.channel, ValidateRedisMessage)
+	c.write = redis.NewWriter(redisPool, c.channel)
+	go c.receive.Setup(redisURL)
+	go c.write.Setup(redisURL)
+}
+
+//ValidateRedisMessage validates incoming redis messages on the chat channel.
+func ValidateRedisMessage(data []byte) error {
+	_, e := validateMessage(data)
+	return e
+}
+
+// validateMessage so that we know it's valid JSON and contains a Handle and
+// Text
+func validateMessage(data []byte) (message, error) {
+	var msg message
+
+	if err := json.Unmarshal(data, &msg); err != nil {
+		return msg, errors.Wrap(err, "Unmarshaling message")
+	}
+
+	if msg.Handle == "" && msg.Text == "" {
+		return msg, errors.New("Message has no Handle or Text")
+	}
+
+	return msg, nil
+}
+
+// HandleWebsocket connection.
+func HandleWebsocket(c *Hub, w http.ResponseWriter, r *http.Request) {
+	if r.Method != "GET" {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	ws, err := upgrader.Upgrade(w, r, nil)
+	defer func() {
+		ws.Close()
+	}()
+
+	if err != nil {
+		m := "Unable to upgrade to websockets"
+		log.WithField("err", err).Println(m)
+		http.Error(w, m, http.StatusBadRequest)
+		return
+	}
+
+	c.receive.Register(ws)
+
+	for {
+		mt, data, err := ws.ReadMessage()
+		l := log.WithFields(logrus.Fields{"mt": mt, "err": err})
+		if err != nil {
+			if websocket.IsCloseError(err, websocket.CloseGoingAway) || err == io.EOF {
+				l.Info("Websocket closed!")
+				break
+			}
+			l.WithField("data", data).Error("Error reading websocket message")
+		}
+		switch mt {
+		case websocket.TextMessage:
+			msg, err := validateMessage(data)
+			if err != nil {
+				l.WithFields(logrus.Fields{"data": data, "msg": msg, "err": err}).Error("Invalid Message")
+				break
+			}
+			l.WithField("msg", msg).Info("message from client")
+			c.write.Publish(data)
+		default:
+			l.WithField("data", data).Warning("Unknown Message!")
+		}
+	}
+
+	c.receive.DeRegister(ws)
+	ws.WriteMessage(websocket.CloseMessage, []byte{})
+}

chat/client.go

@@ -3,17 +3,12 @@ package main
 import (
 	"net/http"
 	"os"
-	"time"
 
 	"github.com/Sirupsen/logrus"
-	"github.com/heroku/x/redis"
 )
 
 var (
-	waitTimeout = time.Minute * 10
-	log         = logrus.WithField("cmd", "go-websocket-chat-demo")
-	rr          redisReceiver
-	rw          redisWriter
+	log = logrus.WithField("cmd", "go-websocket-chat-demo")
 )
 
 func main() {
@@ -26,44 +21,8 @@ func main() {
 	if redisURL == "" {
 		log.WithField("REDIS_URL", redisURL).Fatal("$REDIS_URL must be set")
 	}
-	redisPool, err := redis.NewRedisPoolFromURL(redisURL)
-	if err != nil {
-		log.WithField("url", redisURL).Fatal("Unable to create Redis pool")
-	}
-
-	rr = newRedisReceiver(redisPool)
-	rw = newRedisWriter(redisPool)
-
-	go func() {
-		for {
-			waited, err := redis.WaitForAvailability(redisURL, waitTimeout, rr.wait)
-			if !waited || err != nil {
-				log.WithFields(logrus.Fields{"waitTimeout": waitTimeout, "err": err}).Fatal("Redis not available by timeout!")
-			}
-			rr.broadcast(availableMessage)
-			err = rr.run()
-			if err == nil {
-				break
-			}
-			log.Error(err)
-		}
-	}()
-
-	go func() {
-		for {
-			waited, err := redis.WaitForAvailability(redisURL, waitTimeout, nil)
-			if !waited || err != nil {
-				log.WithFields(logrus.Fields{"waitTimeout": waitTimeout, "err": err}).Fatal("Redis not available by timeout!")
-			}
-			err = rw.run()
-			if err == nil {
-				break
-			}
-			log.Error(err)
-		}
-	}()
 
-	http.Handle("/", http.FileServer(http.Dir("./public")))
-	http.HandleFunc("/ws", handleWebsocket)
+	router := NewRouter(redisURL)
+	http.Handle("/", router)
 	log.Println(http.ListenAndServe(":"+port, nil))
 }

main.go

@@ -2,3 +2,8 @@
   overflow: auto;
   height: 500px;
 }
+
+#chat-text2 {
+  overflow: auto;
+  height: 500px;
+}

public/css/application.css

@@ -24,10 +24,25 @@ <h1>Go Websocket Chat Demo</h1>
         <button class="btn btn-primary" type="submit">Send</button>
       </form>
       <div class="page-header">
-        <h1>Chat Log</h1>
+        <h1>Chat Log (Channel 1)</h1>
       </div>
       <div id="chat-text">
       </div>
+
+      <form id="input-form2" class="form-inline">
+        <div class="form-group">
+          <input id="input-handle2" type="text" class="form-control" placeholder="Enter handle" autofocus />
+        </div>
+        <div class="form-group">
+          <input id="input-text2" type="text" class="form-control" placeholder="Enter chat text here!" autofocus />
+        </div>
+        <button class="btn btn-primary" type="submit">Send</button>
+      </form>
+      <div class="page-header">
+        <h1>Chat Log (Channel 2)</h1>
+      </div>
+      <div id="chat-text2">
+      </div>
     </div>
 
     <a href="https://github.com/heroku-examples/go-websocket-chat-demo"><img style="position: absolute; top: 0; right: 0; border: 0;" src="https://s3.amazonaws.com/github/ribbons/forkme_right_darkblue_121621.png" alt="Fork me on GitHub"></a>

public/index.html

@@ -1,4 +1,4 @@
-var box = new ReconnectingWebSocket(location.protocol.replace("http","ws") + "//" + location.host + "/ws");
+var box = new ReconnectingWebSocket(location.protocol.replace("http","ws") + "//" + location.host + "/ws/chat1");
 
 box.onmessage = function(message) {
   var data = JSON.parse(message.data);
@@ -21,3 +21,27 @@ $("#input-form").on("submit", function(event) {
   box.send(JSON.stringify({ handle: handle, text: text }));
   $("#input-text")[0].value = "";
 });
+
+
+var box2 = new ReconnectingWebSocket(location.protocol.replace("http","ws") + "//" + location.host + "/ws/chat2");
+box2.onmessage = function(message) {
+  var data = JSON.parse(message.data);
+  $("#chat-text2").append("<div class='panel panel-default'><div class='panel-heading'>" + $('<span/>').text(data.handle).html() + "</div><div class='panel-body'>" + $('<span/>').text(data.text).html() + "</div></div>");
+  $("#chat-text2").stop().animate({
+    scrollTop: $('#chat-text2')[0].scrollHeight
+  }, 800);
+};
+
+box2.onclose = function(){
+    console.log('box2 closed');
+    this.box2 = new WebSocket(box2.url);
+
+};
+
+$("#input-form2").on("submit", function(event) {
+  event.preventDefault();
+  var handle = $("#input-handle2")[0].value;
+  var text   = $("#input-text2")[0].value;
+  box2.send(JSON.stringify({ handle: handle, text: text }));
+  $("#input-text2")[0].value = "";
+});