public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
(cherry picked from commit 063e1b5ff5)

Co-authored-by: Alexander Emelin <frvzmb@gmail.com>
This commit is contained in:
Grot (@grafanabot)
2021-05-25 20:16:54 +02:00
committed by GitHub
co-authored by Alexander Emelin
parent ef05f687ca
commit e2b6820d42
15 changed files with 84 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/plugins/backendplugin/coreplugin/core_plugin.go
+1 -1
View File
@@ -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
+5 -3
View File
@@ -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
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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
}