Live: improve broadcast semantics and avoid double posting (#28765)

pkg/models/live.go

@@ -13,8 +13,9 @@ type ChannelHandler interface {
 	// Called when a client wants to subscribe to a channel
 	OnSubscribe(c *centrifuge.Client, e centrifuge.SubscribeEvent) error
 
-	// Called when something writes into the channel.  The returned value will be broadcast if len() > 0
-	OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) ([]byte, error)
+	// AllowBroadcast is called when a client writes a message to the channel websocket.
+	// Returning an error will cancel the broadcast.
+	AllowBroadcast(c *centrifuge.Client, e centrifuge.PublishEvent) error
 }
 
 // ChannelHandlerFactory should be implemented by all core features.

pkg/services/live/features/broadcast.go

@@ -26,8 +26,7 @@ func (b *BroadcastRunner) OnSubscribe(c *centrifuge.Client, e centrifuge.Subscri
 	return nil
 }
 
-// OnPublish called when an event is received from the websocket
-func (b *BroadcastRunner) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) ([]byte, error) {
-	// expect the data to be the right shape?
-	return e.Data, nil
+// AllowBroadcast checks if a message can be broadcast on this channel
+func (b *BroadcastRunner) AllowBroadcast(c *centrifuge.Client, e centrifuge.PublishEvent) error {
+	return nil
 }

pkg/services/live/features/dashboard.go

@@ -39,10 +39,10 @@ func (g *DashboardHandler) OnSubscribe(c *centrifuge.Client, e centrifuge.Subscr
 	return nil
 }
 
-// OnPublish called when an event is received from the websocket
-func (g *DashboardHandler) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) ([]byte, error) {
-	// TODO -- verify and keep track of editors?
-	return e.Data, nil
+// AllowBroadcast checks if a message from the websocket can be broadcast on this channel
+// currently messages are sent when a dashboard starts editing
+func (g *DashboardHandler) AllowBroadcast(c *centrifuge.Client, e centrifuge.PublishEvent) error {
+	return nil
 }
 
 // DashboardSaved should broadcast to the appropriate stream

pkg/services/live/features/measurements.go

@@ -33,9 +33,8 @@ func (m *MeasurementsRunner) OnSubscribe(c *centrifuge.Client, e centrifuge.Subs
 	return nil
 }
 
-// OnPublish is called when an event is received from the websocket.
-func (m *MeasurementsRunner) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) ([]byte, error) {
-	// currently generic... but should be stricter
-	// logger.Debug("Measurements runner got event on channel", "channel", e.Channel)
-	return e.Data, nil
+// AllowBroadcast checks if a message from the websocket can be broadcast on this channel
+// Currently this sends measurements over websocket -- should be replaced with the HTTP interface
+func (m *MeasurementsRunner) AllowBroadcast(c *centrifuge.Client, e centrifuge.PublishEvent) error {
+	return nil
 }

pkg/services/live/features/testdata.go

@@ -73,9 +73,9 @@ func (g *testDataRunner) OnSubscribe(c *centrifuge.Client, e centrifuge.Subscrib
 	return nil
 }
 
-// OnPublish is called when an event is received from the websocket.
-func (g *testDataRunner) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) ([]byte, error) {
-	return nil, fmt.Errorf("can not publish to testdata")
+// AllowBroadcast checks if a message from the websocket can be broadcast on this channel
+func (g *testDataRunner) AllowBroadcast(c *centrifuge.Client, e centrifuge.PublishEvent) error {
+	return fmt.Errorf("can not publish to testdata")
 }
 
 // runRandomCSV is just for an example.

pkg/services/live/live.go

 package live
 
 import (
-	"encoding/json"
 	"fmt"
 	"strings"
 	"sync"
@@ -155,7 +154,9 @@ func (g *GrafanaLive) Init() error {
 		logger.Debug("unsubscribe from channel", "channel", e.Channel, "user", c.UserID())
 	})
 
-	// Called when something is written to the websocket
+	// Called when a client writes to the websocket channel.
+	// In general, we should prefer writing to the HTTP API, but this
+	// allows some simple prototypes to work quickly
 	node.OnPublish(func(c *centrifuge.Client, e centrifuge.PublishEvent) (centrifuge.PublishReply, error) {
 		reply := centrifuge.PublishReply{}
 		handler, err := g.GetChannelHandler(e.Channel)
@@ -163,14 +164,8 @@ func (g *GrafanaLive) Init() error {
 			return reply, err
 		}
 
-		data, err := handler.OnPublish(c, e)
-		if err != nil {
-			return reply, err
-		}
-		if len(data) > 0 {
-			_, err = node.Publish(e.Channel, e.Data)
-		}
-		return centrifuge.PublishReply{}, err // returns an error if it could not publish
+		err = handler.AllowBroadcast(c, e)
+		return centrifuge.PublishReply{}, err
 	})
 
 	// Run node. This method does not block.
@@ -199,26 +194,8 @@ func (g *GrafanaLive) Init() error {
 			return
 		}
 
-		dto := models.UserProfileDTO{
-			Id:             user.UserId,
-			Name:           user.Name,
-			Email:          user.Email,
-			Login:          user.Login,
-			IsGrafanaAdmin: user.IsGrafanaAdmin,
-			OrgId:          user.OrgId,
-		}
-
-		jsonData, err := json.Marshal(dto)
-		if err != nil {
-			logger.Debug("error reading user", "dto", dto)
-			ctx.Resp.WriteHeader(404)
-			return
-		}
-		logger.Info("Logged in user", "user", user)
-
 		cred := &centrifuge.Credentials{
 			UserID: fmt.Sprintf("%d", user.UserId),
-			Info:   jsonData,
 		}
 		newCtx := centrifuge.SetCredentials(ctx.Req.Context(), cred)
 

pkg/services/live/pluginHandler.go

@@ -26,7 +26,7 @@ func (g *PluginHandler) OnSubscribe(c *centrifuge.Client, e centrifuge.Subscribe
 	return nil // anyone can subscribe
 }
 
-// OnPublish called when an event is received from the websocket
-func (g *PluginHandler) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) ([]byte, error) {
-	return e.Data, nil // broadcast any event
+// AllowBroadcast checks if a message from the websocket can be broadcast on this channel
+func (g *PluginHandler) AllowBroadcast(c *centrifuge.Client, e centrifuge.PublishEvent) error {
+	return nil // broadcast any event
 }

pkg/tsdb/cloudwatch/live.go

@@ -80,9 +80,9 @@ func (r *logQueryRunner) OnSubscribe(c *centrifuge.Client, e centrifuge.Subscrib
 	return nil
 }
 
-// OnPublish is called when an event is received from the websocket.
-func (r *logQueryRunner) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) ([]byte, error) {
-	return nil, fmt.Errorf("can not publish")
+// AllowBroadcast checks if a message from the websocket can be broadcast on this channel
+func (r *logQueryRunner) AllowBroadcast(c *centrifuge.Client, e centrifuge.PublishEvent) error {
+	return fmt.Errorf("can not publish")
 }
 
 func (r *logQueryRunner) publishResults(channelName string) error {