From b7827962dcbcdfa0cf8a07e85104067014371cd1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Torkel=20=C3=96degaard?= <torkel@grafana.org>
Date: Tue, 20 Dec 2016 16:09:04 +0100
Subject: [PATCH] feat(live): just wanted to checkout how far I got on the
 websocket data source

---
 pkg/api/live/conn.go                               |  1 +
 pkg/api/live/hub.go                                | 12 +++++++-----
 pkg/api/live/live.go                               |  8 ++++++--
 public/app/features/panel/metrics_panel_ctrl.ts    |  8 ++++++++
 .../plugins/datasource/grafana-live/_plugin.json   |  7 -------
 .../grafana-live/partials/query.editor.html        | 14 +++++++-------
 6 files changed, 29 insertions(+), 21 deletions(-)
 delete mode 100644 public/app/plugins/datasource/grafana-live/_plugin.json

diff --git a/pkg/api/live/conn.go b/pkg/api/live/conn.go
index d474fc48a1f..09b66761b42 100644
--- a/pkg/api/live/conn.go
+++ b/pkg/api/live/conn.go
@@ -48,6 +48,7 @@ func (c *connection) readPump() {
 		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 })
diff --git a/pkg/api/live/hub.go b/pkg/api/live/hub.go
index 736f848db2c..bca65d57432 100644
--- a/pkg/api/live/hub.go
+++ b/pkg/api/live/hub.go
@@ -7,6 +7,7 @@ import (
 )
 
 type hub struct {
+	log         log.Logger
 	connections map[*connection]bool
 	streams     map[string]map[*connection]bool
 
@@ -29,10 +30,10 @@ var h = hub{
 	unregister:    make(chan *connection),
 	streamChannel: make(chan *dtos.StreamMessage),
 	subChannel:    make(chan *streamSubscription),
+	log:           log.New("live.hub"),
 }
 
 func (h *hub) removeConnection() {
-
 }
 
 func (h *hub) run() {
@@ -40,17 +41,17 @@ func (h *hub) run() {
 		select {
 		case c := <-h.register:
 			h.connections[c] = true
-			log.Info("Live: New connection (Total count: %v)", len(h.connections))
+			h.log.Info("New connection", "total", len(h.connections))
 
 		case c := <-h.unregister:
 			if _, ok := h.connections[c]; ok {
-				log.Info("Live: Closing Connection (Total count: %v)", len(h.connections))
+				h.log.Info("Closing connection", "total", len(h.connections))
 				delete(h.connections, c)
 				close(c.send)
 			}
 		// hand stream subscriptions
 		case sub := <-h.subChannel:
-			log.Info("Live: Subscribing to: %v, remove: %v", sub.name, sub.remove)
+			h.log.Info("Subscribing", "channel", sub.name, "remove", sub.remove)
 			subscribers, exists := h.streams[sub.name]
 
 			// handle unsubscribe
@@ -63,13 +64,14 @@ func (h *hub) run() {
 				subscribers = make(map[*connection]bool)
 				h.streams[sub.name] = subscribers
 			}
+
 			subscribers[sub.conn] = true
 
 			// handle stream messages
 		case message := <-h.streamChannel:
 			subscribers, exists := h.streams[message.Stream]
 			if !exists || len(subscribers) == 0 {
-				log.Info("Live: Message to stream without subscribers: %v", message.Stream)
+				h.log.Info("Message to stream without subscribers", "stream", message.Stream)
 				continue
 			}
 
diff --git a/pkg/api/live/live.go b/pkg/api/live/live.go
index 4a309740590..a7e73b4b28a 100644
--- a/pkg/api/live/live.go
+++ b/pkg/api/live/live.go
@@ -9,23 +9,27 @@ import (
 )
 
 type LiveConn struct {
+	log log.Logger
 }
 
 func New() *LiveConn {
 	go h.run()
-	return &LiveConn{}
+
+	return &LiveConn{log: log.New("live.server")}
 }
 
 func (lc *LiveConn) Serve(w http.ResponseWriter, r *http.Request) {
-	log.Info("Live: Upgrading to WebSocket")
+	lc.log.Info("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 := newConnection(ws)
 	h.register <- c
+
 	go c.writePump()
 	c.readPump()
 }
diff --git a/public/app/features/panel/metrics_panel_ctrl.ts b/public/app/features/panel/metrics_panel_ctrl.ts
index aba4492401b..5dbd4d4a4e4 100644
--- a/public/app/features/panel/metrics_panel_ctrl.ts
+++ b/public/app/features/panel/metrics_panel_ctrl.ts
@@ -47,6 +47,14 @@ class MetricsPanelCtrl extends PanelCtrl {
 
     this.events.on('refresh', this.onMetricsPanelRefresh.bind(this));
     this.events.on('init-edit-mode', this.onInitMetricsPanelEditMode.bind(this));
+    this.events.on('panel-teardown', this.onPanelTearDown.bind(this));
+  }
+
+  private onPanelTearDown() {
+    if (this.dataSubscription) {
+      this.dataSubscription.unsubscribe();
+      this.dataSubscription = null;
+    }
   }
 
   private onInitMetricsPanelEditMode() {
diff --git a/public/app/plugins/datasource/grafana-live/_plugin.json b/public/app/plugins/datasource/grafana-live/_plugin.json
deleted file mode 100644
index 1f2ec204949..00000000000
--- a/public/app/plugins/datasource/grafana-live/_plugin.json
+++ /dev/null
@@ -1,7 +0,0 @@
-{
-  "type": "datasource",
-  "name": "Grafana Live",
-  "id": "grafana-live",
-
-  "metrics": true
-}
diff --git a/public/app/plugins/datasource/grafana-live/partials/query.editor.html b/public/app/plugins/datasource/grafana-live/partials/query.editor.html
index 912b28a6247..512263ee9ba 100644
--- a/public/app/plugins/datasource/grafana-live/partials/query.editor.html
+++ b/public/app/plugins/datasource/grafana-live/partials/query.editor.html
@@ -1,8 +1,8 @@
-<query-editor-row ctrl="ctrl">
-	<li class="tight-form-item">
-		Stream
-	</li>
-	<li>
-		<input type="text" class="tight-form-input input-large" ng-model="ctrl.target.stream">
-	</li>
+<query-editor-row query-ctrl="ctrl" can-collapse="false">
+	<div class="gf-form-inline">
+		<div class="gf-form gf-form--grow">
+			<label class="gf-form-label width-8">Stream</label>
+			<input type="text" class="gf-form-input" ng-model="ctrl.target.stream" spellcheck='false' placeholder="metric">
+		</div>
+	</div>
 </query-editor-row>