From fc877ae0f4f3550aadb35fd1daadaf05a6e7a8d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Torkel=20=C3=96degaard?= Date: Thu, 3 Mar 2016 23:05:08 +0100 Subject: [PATCH] poc(websockets): websocket poc --- pkg/api/api.go | 7 +- pkg/api/search.go | 3 + pkg/live/conn.go | 105 +++++++++++++++++++++++++++++ pkg/live/hub.go | 52 ++++++++++++++ public/app/features/admin/admin.ts | 11 ++- 5 files changed, 176 insertions(+), 2 deletions(-) create mode 100644 pkg/live/conn.go create mode 100644 pkg/live/hub.go diff --git a/pkg/api/api.go b/pkg/api/api.go index ed029a5171a..7242db9713a 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -4,6 +4,7 @@ import ( "github.com/go-macaron/binding" "github.com/grafana/grafana/pkg/api/avatar" "github.com/grafana/grafana/pkg/api/dtos" + "github.com/grafana/grafana/pkg/live" "github.com/grafana/grafana/pkg/middleware" m "github.com/grafana/grafana/pkg/models" "gopkg.in/macaron.v1" @@ -35,6 +36,7 @@ func Register(r *macaron.Macaron) { r.Get("/org/users/", reqSignedIn, Index) r.Get("/org/apikeys/", reqSignedIn, Index) r.Get("/dashboard/import/", reqSignedIn, Index) + r.Get("/admin", reqGrafanaAdmin, Index) r.Get("/admin/settings", reqGrafanaAdmin, Index) r.Get("/admin/users", reqGrafanaAdmin, Index) r.Get("/admin/users/create", reqGrafanaAdmin, Index) @@ -230,7 +232,10 @@ func Register(r *macaron.Macaron) { avt := avatar.CacheServer() r.Get("/avatar/:hash", avt.ServeHTTP) + // Websocket + liveConn := live.New() + r.Any("/ws", liveConn.Serve) + InitAppPluginRoutes(r) - r.NotFound(NotFoundHandler) } diff --git a/pkg/api/search.go b/pkg/api/search.go index 035e59fa3f4..8f190dd6bf1 100644 --- a/pkg/api/search.go +++ b/pkg/api/search.go @@ -2,6 +2,7 @@ package api import ( "github.com/grafana/grafana/pkg/bus" + "github.com/grafana/grafana/pkg/live" "github.com/grafana/grafana/pkg/middleware" "github.com/grafana/grafana/pkg/services/search" ) @@ -32,4 +33,6 @@ func Search(c *middleware.Context) { } c.JSON(200, searchQuery.Result) + + live.SendMessage(query) } diff --git a/pkg/live/conn.go b/pkg/live/conn.go new file mode 100644 index 00000000000..90ce0af9cae --- /dev/null +++ b/pkg/live/conn.go @@ -0,0 +1,105 @@ +package live + +import ( + "net/http" + "time" + + "github.com/gorilla/websocket" + "github.com/grafana/grafana/pkg/log" +) + +const ( + // Time allowed to write a message to the peer. + writeWait = 10 * time.Second + + // Time allowed to read the next pong message from the peer. + pongWait = 60 * time.Second + + // Send pings to peer with this period. Must be less than pongWait. + pingPeriod = (pongWait * 9) / 10 + + // Maximum message size allowed from peer. + maxMessageSize = 512 +) + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, +} + +type connection struct { + ws *websocket.Conn + send chan []byte +} + +func (c *connection) readPump() { + defer func() { + h.unregister <- c + c.ws.Close() + }() + c.ws.SetReadLimit(maxMessageSize) + c.ws.SetReadDeadline(time.Now().Add(pongWait)) + c.ws.SetPongHandler(func(string) error { c.ws.SetReadDeadline(time.Now().Add(pongWait)); return nil }) + for { + _, message, err := c.ws.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) { + log.Info("error: %v", err) + } + break + } + h.broadcast <- message + } +} + +func (c *connection) write(mt int, payload []byte) error { + c.ws.SetWriteDeadline(time.Now().Add(writeWait)) + return c.ws.WriteMessage(mt, payload) +} + +// writePump pumps messages from the hub to the websocket connection. +func (c *connection) writePump() { + ticker := time.NewTicker(pingPeriod) + defer func() { + ticker.Stop() + c.ws.Close() + }() + for { + select { + case message, ok := <-c.send: + if !ok { + c.write(websocket.CloseMessage, []byte{}) + return + } + if err := c.write(websocket.TextMessage, message); err != nil { + return + } + case <-ticker.C: + if err := c.write(websocket.PingMessage, []byte{}); err != nil { + return + } + } + } +} + +type LiveConn struct { +} + +func New() *LiveConn { + go h.run() + return &LiveConn{} +} + +func (lc *LiveConn) Serve(w http.ResponseWriter, r *http.Request) { + log.Info("Live: Upgrading to WebSocket") + + ws, err := upgrader.Upgrade(w, r, nil) + if err != nil { + log.Error(3, "Live: Failed to upgrade connection to WebSocket", err) + return + } + c := &connection{send: make(chan []byte, 256), ws: ws} + h.register <- c + go c.writePump() + c.readPump() +} diff --git a/pkg/live/hub.go b/pkg/live/hub.go new file mode 100644 index 00000000000..9e276eb46d1 --- /dev/null +++ b/pkg/live/hub.go @@ -0,0 +1,52 @@ +package live + +import "github.com/grafana/grafana/pkg/log" + +type hub struct { + // Registered connections. + connections map[*connection]bool + + // Inbound messages from the connections. + broadcast chan []byte + + // Register requests from the connections. + register chan *connection + + // Unregister requests from connections. + 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: + if _, ok := h.connections[c]; ok { + delete(h.connections, c) + close(c.send) + } + case m := <-h.broadcast: + log.Info("Live: broadcasting") + for c := range h.connections { + select { + case c.send <- m: + default: + close(c.send) + delete(h.connections, c) + } + } + } + } +} + +func SendMessage(message string) { + h.broadcast <- []byte(message) +} diff --git a/public/app/features/admin/admin.ts b/public/app/features/admin/admin.ts index 47757520afe..a76ed6ea6b1 100644 --- a/public/app/features/admin/admin.ts +++ b/public/app/features/admin/admin.ts @@ -19,7 +19,16 @@ class AdminSettingsCtrl { class AdminHomeCtrl { /** @ngInject **/ - constructor() {} + constructor() { + + var conn = new WebSocket("ws://localhost:3000/ws"); + conn.onclose = function(evt) { + console.log("Connection closed"); + }; + conn.onmessage = function(evt) { + console.log("message", evt.data); + }; + } } export class AdminStatsCtrl {