Merge pull request #28 from EduardMikhrin/master

glightning: added the emit and subscribe functionality for custom eve…

Author: olegfomenko

glightning/plugin.go

@@ -1091,12 +1091,22 @@ func (fb *FeatureBits) AreSet() bool {
 }
 
 type Manifest struct {
-	Options       []Option     `json:"options"`
-	RpcMethods    []*RpcMethod `json:"rpcmethods"`
-	Dynamic       bool         `json:"dynamic"`
-	Subscriptions []string     `json:"subscriptions,omitempty"`
-	Hooks         []Hook       `json:"hooks,omitempty"`
-	FeatureBits   *FeatureBits `json:"featurebits,omitempty"`
+	Options       []Option            `json:"options"`
+	RpcMethods    []*RpcMethod        `json:"rpcmethods"`
+	Dynamic       bool                `json:"dynamic"`
+	Subscriptions []string            `json:"subscriptions,omitempty"`
+	Hooks         []Hook              `json:"hooks,omitempty"`
+	FeatureBits   *FeatureBits        `json:"featurebits,omitempty"`
+	Notifications []NotificationTopic `json:"notifications,omitempty"`
+}
+
+type NotificationTopic struct {
+	Method      string `json:"method"`
+	Description string `json:"description,omitempty"`
+}
+
+type Notification interface {
+	Name() string
 }
 
 func (gm GetManifestMethod) Name() string {
@@ -1120,21 +1130,13 @@ func (gm GetManifestMethod) Call() (jrpc2.Result, error) {
 		}
 	}
 
-	m.Options = make([]Option, len(gm.plugin.options))
-	i := 0
+	m.Options = make([]Option, 0, len(gm.plugin.options))
 	for _, option := range gm.plugin.options {
-		m.Options[i] = option
-		i++
-	}
-	m.Subscriptions = make([]string, len(gm.plugin.subscriptions))
-	for i, sub := range gm.plugin.subscriptions {
-		m.Subscriptions[i] = sub
-	}
-	m.Hooks = make([]Hook, len(gm.plugin.hooks))
-	for i, hook := range gm.plugin.hooks {
-		m.Hooks[i] = hook
+		m.Options = append(m.Options, option)
 	}
 
+	m.Subscriptions = gm.plugin.subscriptions
+	m.Hooks = gm.plugin.hooks
 	m.Dynamic = gm.plugin.dynamic
 
 	if gm.plugin.features.AreSet() {
@@ -1144,6 +1146,7 @@ func (gm GetManifestMethod) Call() (jrpc2.Result, error) {
 		}
 	}
 	m.FeatureBits = gm.plugin.features
+	m.Notifications = gm.plugin.notifications
 
 	return m, nil
 }
@@ -1317,6 +1320,7 @@ type Plugin struct {
 	stopped       bool
 	dynamic       bool
 	features      *FeatureBits
+	notifications []NotificationTopic
 }
 
 func NewPlugin(initHandler func(p *Plugin, o map[string]Option, c *Config)) *Plugin {
@@ -1498,6 +1502,21 @@ func (p *Plugin) UnregisterOption(o Option) error {
 	return nil
 }
 
+func (p *Plugin) RegisterNotificationTopic(topic string, description string) {
+	p.notifications = append(p.notifications, NotificationTopic{
+		Method:      topic,
+		Description: description,
+	})
+}
+
+func (p *Plugin) EmitCustomNotification(payload jrpc2.Method) error {
+	if err := p.server.Notify(payload); err != nil {
+		return fmt.Errorf("failed to emit notification for topic %s: %w", payload.Name(), err)
+	}
+
+	return nil
+}
+
 // this always returns a string option. left as is for legacy reasons
 func (p *Plugin) GetOption(name string) (string, error) {
 	opt := p.options[name]
@@ -1557,60 +1576,60 @@ func (p *Plugin) getOptionSet() map[string]Option {
 }
 
 func (p *Plugin) SubscribeConnect(cb func(c *ConnectEvent)) {
-	p.subscribe(&ConnectEvent{
+	p.Subscribe(&ConnectEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeDisconnect(cb func(c *DisconnectEvent)) {
-	p.subscribe(&DisconnectEvent{
+	p.Subscribe(&DisconnectEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeInvoicePaid(cb func(c *Payment)) {
-	p.subscribe(&InvoicePaidEvent{
+	p.Subscribe(&InvoicePaidEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeChannelOpened(cb func(c *ChannelOpened)) {
-	p.subscribe(&ChannelOpenedEvent{
+	p.Subscribe(&ChannelOpenedEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeWarnings(cb func(c *Warning)) {
-	p.subscribe(&WarnEvent{
+	p.Subscribe(&WarnEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeSendPaySuccess(cb func(c *SendPaySuccess)) {
-	p.subscribe(&SendPaySuccessEvent{
+	p.Subscribe(&SendPaySuccessEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeSendPayFailure(cb func(c *SendPayFailure)) {
-	p.subscribe(&SendPayFailureEvent{
+	p.Subscribe(&SendPayFailureEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeForwardings(cb func(c *Forwarding)) {
-	p.subscribe(&ForwardEvent{
+	p.Subscribe(&ForwardEvent{
 		cb: cb,
 	})
 }
 
 func (p *Plugin) SubscribeShutdown(cb func()) {
-	p.subscribe(&ShutdownEvent{
+	p.Subscribe(&ShutdownEvent{
 		cb: cb,
 	})
 }
 
-func (p *Plugin) subscribe(subscription jrpc2.ServerMethod) {
+func (p *Plugin) Subscribe(subscription jrpc2.ServerMethod) {
 	p.server.Register(subscription)
 	p.subscriptions = append(p.subscriptions, subscription.Name())
 }

glightning/plugin_test.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"log"
 	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -756,6 +757,65 @@ func TestSubscription_Disconnected(t *testing.T) {
 	runTest(t, plugin, msg+"\n\n", "")
 }
 
+type testCustomPayload struct {
+	Message string `json:"message"`
+	Code    int    `json:"code"`
+}
+
+type testCustomEvent struct {
+	Payload testCustomPayload `json:"custom-topic"`
+	cb      func(payload *testCustomPayload)
+}
+
+func (e *testCustomEvent) Name() string {
+	return "custom-topic"
+}
+
+func (e *testCustomEvent) New() interface{} {
+	return &testCustomEvent{cb: e.cb}
+}
+
+func (e *testCustomEvent) Call() (jrpc2.Result, error) {
+	e.cb(&e.Payload)
+	return nil, nil
+}
+
+func TestSubscription_CustomEventNotification(t *testing.T) {
+	var wg sync.WaitGroup
+	defer await(t, &wg)
+
+	wg.Add(1)
+	initFn := getInitFunc(t, func(t *testing.T, options map[string]glightning.Option, config *glightning.Config) {
+		t.Error("Should not have called init when calling get manifest")
+	})
+
+	plugin := glightning.NewPlugin(initFn)
+
+	plugin.Subscribe(&testCustomEvent{
+		cb: func(payload *testCustomPayload) {
+			defer wg.Done()
+
+			assert.Equal(t, "Hello World", payload.Message)
+			assert.Equal(t, 200, payload.Code)
+		},
+	})
+
+	msg := `{
+       "jsonrpc":"2.0",
+       "method":"custom-topic",
+       "params":{
+          "custom-topic":{
+             "message":"Hello World",
+             "code": 200
+          }
+       }
+    }`
+
+	compactMsg := strings.ReplaceAll(strings.ReplaceAll(msg, "\n", ""), "\t", "")
+
+	runTest(t, plugin, compactMsg+"\n\n", "")
+}
+
 func TestSubscription_Shutdown(t *testing.T) {
 	var wg sync.WaitGroup
 	defer await(t, &wg)