forked from grafana/grafana
-
Notifications
You must be signed in to change notification settings - Fork 0
/
plugin.go
85 lines (72 loc) · 2.58 KB
/
plugin.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
package liveplugin
import (
"context"
"fmt"
"github.com/centrifugal/centrifuge"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/services/datasources"
"github.com/grafana/grafana/pkg/services/live/orgchannel"
"github.com/grafana/grafana/pkg/services/live/pipeline"
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
"github.com/grafana/grafana/pkg/services/user"
)
type ChannelLocalPublisher struct {
node *centrifuge.Node
pipeline *pipeline.Pipeline
}
func NewChannelLocalPublisher(node *centrifuge.Node, pipeline *pipeline.Pipeline) *ChannelLocalPublisher {
return &ChannelLocalPublisher{node: node, pipeline: pipeline}
}
func (p *ChannelLocalPublisher) PublishLocal(channel string, data []byte) error {
if p.pipeline != nil {
orgID, channelID, err := orgchannel.StripOrgID(channel)
if err != nil {
return err
}
ok, err := p.pipeline.ProcessInput(context.Background(), orgID, channelID, data)
if err != nil {
return err
}
if ok {
// if rule found – we are done here. If not - fall through and process as usual.
return nil
}
}
pub := ¢rifuge.Publication{
Data: data,
}
err := p.node.Hub().BroadcastPublication(channel, pub, centrifuge.StreamPosition{})
if err != nil {
return fmt.Errorf("error publishing %s: %w", string(data), err)
}
return nil
}
type NumLocalSubscribersGetter struct {
node *centrifuge.Node
}
func NewNumLocalSubscribersGetter(node *centrifuge.Node) *NumLocalSubscribersGetter {
return &NumLocalSubscribersGetter{node: node}
}
func (p *NumLocalSubscribersGetter) GetNumLocalSubscribers(channelID string) (int, error) {
return p.node.Hub().NumSubscribers(channelID), nil
}
type ContextGetter struct {
pluginContextProvider *plugincontext.Provider
dataSourceCache datasources.CacheService
}
func NewContextGetter(pluginContextProvider *plugincontext.Provider, dataSourceCache datasources.CacheService) *ContextGetter {
return &ContextGetter{
pluginContextProvider: pluginContextProvider,
dataSourceCache: dataSourceCache,
}
}
func (g *ContextGetter) GetPluginContext(ctx context.Context, user *user.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, error) {
if datasourceUID == "" {
return g.pluginContextProvider.Get(ctx, pluginID, user, user.OrgID)
}
ds, err := g.dataSourceCache.GetDatasourceByUID(ctx, datasourceUID, user, skipCache)
if err != nil {
return backend.PluginContext{}, fmt.Errorf("%v: %w", "Failed to get datasource", err)
}
return g.pluginContextProvider.GetWithDataSource(ctx, pluginID, user, ds)
}