Live: use latest changes in Go SDK to deal with streaming frames (#34668)

go.mod

@@ -51,7 +51,7 @@ require (
 	github.com/gosimple/slug v1.9.0
 	github.com/grafana/grafana-aws-sdk v0.4.0
 	github.com/grafana/grafana-live-sdk v0.0.6
-	github.com/grafana/grafana-plugin-sdk-go v0.100.0
+	github.com/grafana/grafana-plugin-sdk-go v0.101.0
 	github.com/grafana/loki v1.6.2-0.20210520072447-15d417efe103
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/hashicorp/go-hclog v0.16.0

go.sum

@@ -922,8 +922,8 @@ github.com/grafana/grafana-live-sdk v0.0.6 h1:P1QFn0ZradOJp3zVpfG0STZMP+pgZrW0e0
 github.com/grafana/grafana-live-sdk v0.0.6/go.mod h1:f15hHmWyLdFjmuWLsjeKeZnq/HnNQ3QkoPcaEww45AY=
 github.com/grafana/grafana-plugin-sdk-go v0.79.0/go.mod h1:NvxLzGkVhnoBKwzkst6CFfpMFKwAdIUZ1q8ssuLeF60=
 github.com/grafana/grafana-plugin-sdk-go v0.91.0/go.mod h1:Ot3k7nY7P6DXmUsDgKvNB7oG1v7PRyTdmnYVoS554bU=
-github.com/grafana/grafana-plugin-sdk-go v0.100.0 h1:BryvIFdx/HrsKMt2hkxN7cJ0WrCgKpgjdJW8y8TSol0=
+github.com/grafana/grafana-plugin-sdk-go v0.101.0 h1:QyXMkgwZXUX9EQjLv5S5uDcvYjwsntqFV/dCC49Fn+w=
-github.com/grafana/grafana-plugin-sdk-go v0.100.0/go.mod h1:D7x3ah+1d4phNXpbnOaxa/osSaZlwh9/ZUnGGzegRbk=
+github.com/grafana/grafana-plugin-sdk-go v0.101.0/go.mod h1:D7x3ah+1d4phNXpbnOaxa/osSaZlwh9/ZUnGGzegRbk=
 github.com/grafana/loki v1.6.2-0.20210520072447-15d417efe103 h1:qCmofFVwQR9QnsinstVqI1NPLMVl33jNCnOCXEAVn6E=
 github.com/grafana/loki v1.6.2-0.20210520072447-15d417efe103/go.mod h1:GHIsn+EohCChsdu5YouNZewqLeV9L2FNw4DEJU3P9qE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=

pkg/plugins/backendplugin/coreplugin/core_plugin.go

@@ -110,7 +110,7 @@ func (cp *corePlugin) PublishStream(ctx context.Context, req *backend.PublishStr
 	return nil, backendplugin.ErrMethodNotImplemented
 }
 
-func (cp *corePlugin) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+func (cp *corePlugin) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 	if cp.StreamHandler != nil {
 		return cp.StreamHandler.RunStream(ctx, req, sender)
 	}

pkg/plugins/backendplugin/grpcplugin/client_v2.go

@@ -194,7 +194,7 @@ func (c *clientV2) PublishStream(ctx context.Context, req *backend.PublishStream
 	return backend.FromProto().PublishStreamResponse(protoResp), nil
 }
 
-func (c *clientV2) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+func (c *clientV2) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 	if c.StreamClient == nil {
 		return backendplugin.ErrMethodNotImplemented
 	}
@@ -209,7 +209,7 @@ func (c *clientV2) RunStream(ctx context.Context, req *backend.RunStreamRequest,
 	}
 
 	for {
-		protoResp, err := protoStream.Recv()
+		p, err := protoStream.Recv()
 		if err != nil {
 			if status.Code(err) == codes.Unimplemented {
 				return backendplugin.ErrMethodNotImplemented
@@ -219,7 +219,9 @@ func (c *clientV2) RunStream(ctx context.Context, req *backend.RunStreamRequest,
 			}
 			return fmt.Errorf("error running stream: %w", err)
 		}
-		if err := sender.Send(backend.FromProto().StreamPacket(protoResp)); err != nil {
+		// From GRPC connection we receive already prepared JSON.
+		err = sender.SendJSON(p.Data)
+		if err != nil {
 			return err
 		}
 	}

pkg/plugins/backendplugin/grpcplugin/grpc_plugin.go

@@ -161,7 +161,7 @@ func (p *grpcPlugin) PublishStream(ctx context.Context, request *backend.Publish
 	return pluginClient.PublishStream(ctx, request)
 }
 
-func (p *grpcPlugin) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+func (p *grpcPlugin) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 	pluginClient, ok := p.getPluginClient()
 	if !ok {
 		return backendplugin.ErrPluginUnavailable

pkg/plugins/backendplugin/manager/manager_test.go

@@ -457,7 +457,7 @@ func (tp *testPlugin) PublishStream(ctx context.Context, request *backend.Publis
 	return nil, backendplugin.ErrMethodNotImplemented
 }
 
-func (tp *testPlugin) RunStream(ctx context.Context, request *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+func (tp *testPlugin) RunStream(ctx context.Context, request *backend.RunStreamRequest, sender *backend.StreamSender) error {
 	return backendplugin.ErrMethodNotImplemented
 }
 

pkg/services/live/features/plugin.go

@@ -95,7 +95,9 @@ func (r *PluginPathRunner) OnSubscribe(ctx context.Context, user *models.SignedI
 
 	reply := models.SubscribeReply{
 		Presence: true,
-		Data:     resp.Data,
+	}
+	if resp.InitialData != nil {
+		reply.Data = resp.InitialData.Data()
 	}
 	return reply, backend.SubscribeStreamStatusOK, nil
 }

pkg/services/live/live.go

@@ -155,9 +155,9 @@ func (g *GrafanaLive) Init() error {
 	g.node = node
 
 	g.contextGetter = newPluginContextGetter(g.PluginContextProvider)
-	packetSender := newPluginPacketSender(node)
+	channelSender := newPluginChannelSender(node)
 	presenceGetter := newPluginPresenceGetter(node)
-	g.runStreamManager = runstream.NewManager(packetSender, presenceGetter, g.contextGetter)
+	g.runStreamManager = runstream.NewManager(channelSender, presenceGetter, g.contextGetter)
 
 	// Initialize the main features
 	dash := &features.DashboardHandler{

pkg/services/live/managedstream/runner.go

@@ -106,11 +106,12 @@ func (s *ManagedStream) ListChannels(orgID int64, prefix string) []util.DynMap {
 // unstableSchema flag can be set to disable schema caching for a path.
 func (s *ManagedStream) Push(orgID int64, path string, frame *data.Frame, unstableSchema bool) error {
 	// Keep schema + data for last packet.
-	frameJSON, err := data.FrameToJSON(frame, true, true)
+	frameJSONWrapper, err := data.FrameToJSON(frame)
 	if err != nil {
 		logger.Error("Error marshaling frame with Schema", "error", err)
 		return err
 	}
+	frameJSON := frameJSONWrapper.Bytes(data.IncludeAll)
 
 	if !unstableSchema {
 		// If schema is stable we can safely cache it, and only send values if
@@ -128,11 +129,12 @@ func (s *ManagedStream) Push(orgID int64, path string, frame *data.Frame, unstab
 		// frame to keep Schema JSON and Values JSON in frame object
 		// to avoid encoding twice.
 		if exists {
-			frameJSON, err = data.FrameToJSON(frame, false, true)
+			frameJSONWrapper, err = data.FrameToJSON(frame)
 			if err != nil {
 				logger.Error("Error marshaling Frame to JSON", "error", err)
 				return err
 			}
+			frameJSON = frameJSONWrapper.Bytes(data.IncludeDataOnly)
 		}
 	} else {
 		// For unstable schema we always need to send everything to a connection.

pkg/services/live/plugin_helpers.go

@@ -10,18 +10,18 @@ import (
 	"github.com/grafana/grafana/pkg/plugins/plugincontext"
 )
 
-type pluginPacketSender struct {
+type pluginChannelSender struct {
 	node *centrifuge.Node
 }
 
-func newPluginPacketSender(node *centrifuge.Node) *pluginPacketSender {
+func newPluginChannelSender(node *centrifuge.Node) *pluginChannelSender {
-	return &pluginPacketSender{node: node}
+	return &pluginChannelSender{node: node}
 }
 
-func (p *pluginPacketSender) Send(channel string, packet *backend.StreamPacket) error {
+func (p *pluginChannelSender) Send(channel string, data []byte) error {
-	_, err := p.node.Publish(channel, packet.Data)
+	_, err := p.node.Publish(channel, data)
 	if err != nil {
-		return fmt.Errorf("error publishing %s: %w", string(packet.Data), err)
+		return fmt.Errorf("error publishing %s: %w", string(data), err)
 	}
 	return nil
 }

pkg/services/live/runstream/manager.go

@@ -18,10 +18,10 @@ var (
 	logger = log.New("live.runstream")
 )
 
-//go:generate mockgen -destination=mock.go -package=runstream github.com/grafana/grafana/pkg/services/live/runstream StreamPacketSender,PresenceGetter,StreamRunner,PluginContextGetter
+//go:generate mockgen -destination=mock.go -package=runstream github.com/grafana/grafana/pkg/services/live/runstream ChannelSender,PresenceGetter,StreamRunner,PluginContextGetter
 
-type StreamPacketSender interface {
+type ChannelSender interface {
-	Send(channel string, packet *backend.StreamPacket) error
+	Send(channel string, data []byte) error
 }
 
 type PluginContextGetter interface {
@@ -33,23 +33,16 @@ type PresenceGetter interface {
 }
 
 type StreamRunner interface {
-	RunStream(ctx context.Context, request *backend.RunStreamRequest, sender backend.StreamPacketSender) error
+	RunStream(ctx context.Context, request *backend.RunStreamRequest, sender *backend.StreamSender) error
 }
 
-type streamSender struct {
+type packetSender struct {
-	channel      string
+	channelSender ChannelSender
-	packetSender StreamPacketSender
+	channel       string
 }
 
-func newStreamSender(channel string, packetSender StreamPacketSender) *streamSender {
+func (p *packetSender) Send(packet *backend.StreamPacket) error {
-	return &streamSender{
+	return p.channelSender.Send(p.channel, packet.Data)
-		channel:      channel,
-		packetSender: packetSender,
-	}
-}
-
-func (p *streamSender) Send(packet *backend.StreamPacket) error {
-	return p.packetSender.Send(p.channel, packet)
 }
 
 // Manager manages streams from Grafana to plugins (i.e. RunStream method).
@@ -60,7 +53,7 @@ type Manager struct {
 	datasourceStreams       map[string]map[string]struct{}
 	presenceGetter          PresenceGetter
 	pluginContextGetter     PluginContextGetter
-	packetSender            StreamPacketSender
+	channelSender           ChannelSender
 	registerCh              chan submitRequest
 	closedCh                chan struct{}
 	checkInterval           time.Duration
@@ -86,11 +79,11 @@ const (
 )
 
 // NewManager creates new Manager.
-func NewManager(packetSender StreamPacketSender, presenceGetter PresenceGetter, pluginContextGetter PluginContextGetter, opts ...ManagerOption) *Manager {
+func NewManager(channelSender ChannelSender, presenceGetter PresenceGetter, pluginContextGetter PluginContextGetter, opts ...ManagerOption) *Manager {
 	sm := &Manager{
 		streams:                 make(map[string]streamContext),
 		datasourceStreams:       map[string]map[string]struct{}{},
-		packetSender:            packetSender,
+		channelSender:           channelSender,
 		presenceGetter:          presenceGetter,
 		pluginContextGetter:     pluginContextGetter,
 		registerCh:              make(chan submitRequest),
@@ -308,7 +301,7 @@ func (s *Manager) runStream(ctx context.Context, cancelFn func(), sr streamReque
 				PluginContext: pluginCtx,
 				Path:          sr.Path,
 			},
-			newStreamSender(sr.Channel, s.packetSender),
+			backend.NewStreamSender(&packetSender{channelSender: s.channelSender, channel: sr.Channel}),
 		)
 		if err != nil {
 			if errors.Is(ctx.Err(), context.Canceled) {

pkg/services/live/runstream/manager_test.go

@@ -27,11 +27,11 @@ func TestStreamManager_Run(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockChannelSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
-	manager := NewManager(mockPacketSender, mockPresenceGetter, mockContextGetter)
+	manager := NewManager(mockChannelSender, mockPresenceGetter, mockContextGetter)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -48,7 +48,7 @@ func TestStreamManager_SubmitStream_Send(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockPacketSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
@@ -84,12 +84,10 @@ func TestStreamManager_SubmitStream_Send(t *testing.T) {
 	mockStreamRunner := NewMockStreamRunner(mockCtrl)
 	mockStreamRunner.EXPECT().RunStream(
 		gomock.Any(), gomock.Any(), gomock.Any(),
-	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		require.Equal(t, "test", req.Path)
 		close(startedCh)
-		err := sender.Send(&backend.StreamPacket{
+		err := sender.SendJSON([]byte("{}"))
-			Data: []byte("test"),
-		})
 		require.NoError(t, err)
 		<-ctx.Done()
 		close(doneCh)
@@ -115,7 +113,7 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockPacketSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
@@ -142,12 +140,10 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
 	mockStreamRunner1 := NewMockStreamRunner(mockCtrl)
 	mockStreamRunner1.EXPECT().RunStream(
 		gomock.Any(), gomock.Any(), gomock.Any(),
-	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		require.Equal(t, "test", req.Path)
 		close(startedCh1)
-		err := sender.Send(&backend.StreamPacket{
+		err := sender.SendJSON([]byte("{}"))
-			Data: []byte("test"),
-		})
 		require.NoError(t, err)
 		<-ctx.Done()
 		close(doneCh1)
@@ -157,12 +153,10 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
 	mockStreamRunner2 := NewMockStreamRunner(mockCtrl)
 	mockStreamRunner2.EXPECT().RunStream(
 		gomock.Any(), gomock.Any(), gomock.Any(),
-	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		require.Equal(t, "test", req.Path)
 		close(startedCh2)
-		err := sender.Send(&backend.StreamPacket{
+		err := sender.SendJSON([]byte("{}"))
-			Data: []byte("test"),
-		})
 		require.NoError(t, err)
 		<-ctx.Done()
 		close(doneCh2)
@@ -190,7 +184,7 @@ func TestStreamManager_SubmitStream_CloseNoSubscribers(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockPacketSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
@@ -218,7 +212,7 @@ func TestStreamManager_SubmitStream_CloseNoSubscribers(t *testing.T) {
 	mockPresenceGetter.EXPECT().GetNumSubscribers("1/test").Return(0, nil).Times(3)
 
 	mockStreamRunner := NewMockStreamRunner(mockCtrl)
-	mockStreamRunner.EXPECT().RunStream(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	mockStreamRunner.EXPECT().RunStream(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		close(startedCh)
 		<-ctx.Done()
 		close(doneCh)
@@ -237,7 +231,7 @@ func TestStreamManager_SubmitStream_ErrorRestartsRunStream(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockPacketSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
@@ -271,7 +265,7 @@ func TestStreamManager_SubmitStream_ErrorRestartsRunStream(t *testing.T) {
 	mockStreamRunner := NewMockStreamRunner(mockCtrl)
 	mockStreamRunner.EXPECT().RunStream(
 		gomock.Any(), gomock.Any(), gomock.Any(),
-	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		if currentErrors >= numErrors {
 			return nil
 		}
@@ -290,7 +284,7 @@ func TestStreamManager_SubmitStream_NilErrorStopsRunStream(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockPacketSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
@@ -309,7 +303,7 @@ func TestStreamManager_SubmitStream_NilErrorStopsRunStream(t *testing.T) {
 	mockStreamRunner := NewMockStreamRunner(mockCtrl)
 	mockStreamRunner.EXPECT().RunStream(
 		gomock.Any(), gomock.Any(), gomock.Any(),
-	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		return nil
 	}).Times(1)
 
@@ -323,7 +317,7 @@ func TestStreamManager_HandleDatasourceUpdate(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockPacketSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
@@ -359,7 +353,7 @@ func TestStreamManager_HandleDatasourceUpdate(t *testing.T) {
 	mockStreamRunner := NewMockStreamRunner(mockCtrl)
 	mockStreamRunner.EXPECT().RunStream(
 		gomock.Any(), gomock.Any(), gomock.Any(),
-	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		if isFirstCall {
 			// first RunStream will wait till context done.
 			isFirstCall = false
@@ -389,7 +383,7 @@ func TestStreamManager_HandleDatasourceDelete(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	mockPacketSender := NewMockStreamPacketSender(mockCtrl)
+	mockPacketSender := NewMockChannelSender(mockCtrl)
 	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
 	mockContextGetter := NewMockPluginContextGetter(mockCtrl)
 
@@ -422,7 +416,7 @@ func TestStreamManager_HandleDatasourceDelete(t *testing.T) {
 	mockStreamRunner := NewMockStreamRunner(mockCtrl)
 	mockStreamRunner.EXPECT().RunStream(
 		gomock.Any(), gomock.Any(), gomock.Any(),
-	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+	).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
 		close(doneCh)
 		<-ctx.Done()
 		return ctx.Err()

pkg/services/live/runstream/mock.go

@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: github.com/grafana/grafana/pkg/services/live/runstream (interfaces: StreamPacketSender,PresenceGetter,StreamRunner,PluginContextGetter)
+// Source: github.com/grafana/grafana/pkg/services/live/runstream (interfaces: ChannelSender,PresenceGetter,StreamRunner,PluginContextGetter)
 
 // Package runstream is a generated GoMock package.
 package runstream
@@ -13,31 +13,31 @@ import (
 	models "github.com/grafana/grafana/pkg/models"
 )
 
-// MockStreamPacketSender is a mock of StreamPacketSender interface.
+// MockChannelSender is a mock of ChannelSender interface.
-type MockStreamPacketSender struct {
+type MockChannelSender struct {
 	ctrl     *gomock.Controller
-	recorder *MockStreamPacketSenderMockRecorder
+	recorder *MockChannelSenderMockRecorder
 }
 
-// MockStreamPacketSenderMockRecorder is the mock recorder for MockStreamPacketSender.
+// MockChannelSenderMockRecorder is the mock recorder for MockChannelSender.
-type MockStreamPacketSenderMockRecorder struct {
+type MockChannelSenderMockRecorder struct {
-	mock *MockStreamPacketSender
+	mock *MockChannelSender
 }
 
-// NewMockStreamPacketSender creates a new mock instance.
+// NewMockChannelSender creates a new mock instance.
-func NewMockStreamPacketSender(ctrl *gomock.Controller) *MockStreamPacketSender {
+func NewMockChannelSender(ctrl *gomock.Controller) *MockChannelSender {
-	mock := &MockStreamPacketSender{ctrl: ctrl}
+	mock := &MockChannelSender{ctrl: ctrl}
-	mock.recorder = &MockStreamPacketSenderMockRecorder{mock}
+	mock.recorder = &MockChannelSenderMockRecorder{mock}
 	return mock
 }
 
 // EXPECT returns an object that allows the caller to indicate expected use.
-func (m *MockStreamPacketSender) EXPECT() *MockStreamPacketSenderMockRecorder {
+func (m *MockChannelSender) EXPECT() *MockChannelSenderMockRecorder {
 	return m.recorder
 }
 
 // Send mocks base method.
-func (m *MockStreamPacketSender) Send(arg0 string, arg1 *backend.StreamPacket) error {
+func (m *MockChannelSender) Send(arg0 string, arg1 []byte) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Send", arg0, arg1)
 	ret0, _ := ret[0].(error)
@@ -45,9 +45,9 @@ func (m *MockStreamPacketSender) Send(arg0 string, arg1 *backend.StreamPacket) e
 }
 
 // Send indicates an expected call of Send.
-func (mr *MockStreamPacketSenderMockRecorder) Send(arg0, arg1 interface{}) *gomock.Call {
+func (mr *MockChannelSenderMockRecorder) Send(arg0, arg1 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockStreamPacketSender)(nil).Send), arg0, arg1)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockChannelSender)(nil).Send), arg0, arg1)
 }
 
 // MockPresenceGetter is a mock of PresenceGetter interface.
@@ -112,7 +112,7 @@ func (m *MockStreamRunner) EXPECT() *MockStreamRunnerMockRecorder {
 }
 
 // RunStream mocks base method.
-func (m *MockStreamRunner) RunStream(arg0 context.Context, arg1 *backend.RunStreamRequest, arg2 backend.StreamPacketSender) error {
+func (m *MockStreamRunner) RunStream(arg0 context.Context, arg1 *backend.RunStreamRequest, arg2 *backend.StreamSender) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "RunStream", arg0, arg1, arg2)
 	ret0, _ := ret[0].(error)

pkg/tsdb/testdatasource/scenarios_test.go

@@ -278,8 +278,9 @@ func TestReadCSV(t *testing.T) {
 	require.NoError(t, err)
 
 	frame := data.NewFrame("", fBool, fBool2, fNum, fStr)
-	out, err := data.FrameToJSON(frame, true, true)
+	frameToJSON, err := data.FrameToJSON(frame)
 	require.NoError(t, err)
+	out := frameToJSON.Bytes(data.IncludeAll)
 
 	// require.Equal(t, "", string(out))
 

pkg/tsdb/testdatasource/stream_handler.go

@@ -9,7 +9,6 @@ import (
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 
-	"github.com/grafana/grafana/pkg/cmd/grafana-cli/logger"
 	"github.com/grafana/grafana/pkg/infra/log"
 )
 
@@ -32,14 +31,14 @@ func newTestStreamHandler(logger log.Logger) *testStreamHandler {
 }
 
 func (p *testStreamHandler) SubscribeStream(_ context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
-	schema, err := data.FrameToJSON(p.frame, true, false)
+	p.logger.Debug("Allowing access to stream", "path", req.Path, "user", req.PluginContext.User)
+	initialData, err := backend.NewInitialFrame(p.frame, data.IncludeSchemaOnly)
 	if err != nil {
 		return nil, err
 	}
-	p.logger.Debug("Allowing access to stream", "path", req.Path, "user", req.PluginContext.User)
 	return &backend.SubscribeStreamResponse{
-		Status: backend.SubscribeStreamStatusOK,
+		Status:      backend.SubscribeStreamStatusOK,
-		Data:   schema,
+		InitialData: initialData,
 	}, nil
 }
 
@@ -50,7 +49,7 @@ func (p *testStreamHandler) PublishStream(_ context.Context, req *backend.Publis
 	}, nil
 }
 
-func (p *testStreamHandler) RunStream(ctx context.Context, request *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+func (p *testStreamHandler) RunStream(ctx context.Context, request *backend.RunStreamRequest, sender *backend.StreamSender) error {
 	p.logger.Debug("New stream call", "path", request.Path)
 	var conf testStreamConfig
 	switch request.Path {
@@ -78,7 +77,7 @@ type testStreamConfig struct {
 	Drop     float64
 }
 
-func (p *testStreamHandler) runTestStream(ctx context.Context, path string, conf testStreamConfig, sender backend.StreamPacketSender) error {
+func (p *testStreamHandler) runTestStream(ctx context.Context, path string, conf testStreamConfig, sender *backend.StreamSender) error {
 	spread := 50.0
 	walker := rand.Float64() * 100
 
@@ -101,17 +100,7 @@ func (p *testStreamHandler) runTestStream(ctx context.Context, path string, conf
 			p.frame.Fields[1].Set(0, walker)                                // Value
 			p.frame.Fields[2].Set(0, walker-((rand.Float64()*spread)+0.01)) // Min
 			p.frame.Fields[3].Set(0, walker+((rand.Float64()*spread)+0.01)) // Max
-
+			if err := sender.SendFrame(p.frame, data.IncludeDataOnly); err != nil {
-			bytes, err := data.FrameToJSON(p.frame, false, true)
-			if err != nil {
-				logger.Warn("unable to marshal line", "error", err)
-				continue
-			}
-
-			packet := &backend.StreamPacket{
-				Data: bytes,
-			}
-			if err := sender.Send(packet); err != nil {
 				return err
 			}
 		}