cln_plugin: custommsg support

cln_plugin/cln_messages.go

@@ -115,3 +115,8 @@ type LogNotification struct {
 	Level   string `json:"level"`
 	Message string `json:"message"`
 }
+
+type CustomMessageRequest struct {
+	PeerId  string `json:"peer_id"`
+	Payload string `json:"payload"`
+}

cln_plugin/cln_plugin.go

@@ -149,6 +149,30 @@ func (c *ClnPlugin) htlcListenServer() {
 	}
 }
 
+// Listens to responses to custommsg requests from the grpc server.
+func (c *ClnPlugin) custommsgListenServer() {
+	for {
+		select {
+		case <-c.done:
+			return
+		default:
+			id, result := c.server.ReceiveCustomMessageResponse()
+
+			// The server may return nil if it is stopped.
+			if result == nil {
+				continue
+			}
+
+			serid, _ := json.Marshal(&id)
+			c.sendToCln(&Response{
+				Id:      serid,
+				JsonRpc: SpecVersion,
+				Result:  result,
+			})
+		}
+	}
+}
+
 // processes a single message from cln. Sends the message to the appropriate
 // handler.
 func (c *ClnPlugin) processMsg(msg []byte) {
@@ -277,6 +301,7 @@ func (c *ClnPlugin) handleGetManifest(request *Request) {
 			},
 			Dynamic: true,
 			Hooks: []Hook{
+				{Name: "custommsg"},
 				{Name: "htlc_accepted"},
 				{Name: "openchannel"},
 				{Name: "openchannel2"},
@@ -406,8 +431,9 @@ func (c *ClnPlugin) handleInit(request *Request) {
 		return
 	}
 
-	// Listen for htlc responses from the grpc server.
+	// Listen for htlc and custommsg responses from the grpc server.
 	go c.htlcListenServer()
+	go c.custommsgListenServer()
 
 	// Let cln know the plugin is initialized.
 	c.sendToCln(&Response{
@@ -466,6 +492,25 @@ func (c *ClnPlugin) handleSetChannelAcceptScript(request *Request) {
 	})
 }
 
+func (c *ClnPlugin) handleCustomMsg(request *Request) {
+	var custommsg CustomMessageRequest
+	err := json.Unmarshal(request.Params, &custommsg)
+	if err != nil {
+		c.sendError(
+			request.Id,
+			ParseError,
+			fmt.Sprintf(
+				"Failed to unmarshal custommsg params:%s [%s]",
+				err.Error(),
+				request.Params,
+			),
+		)
+		return
+	}
+
+	c.server.SendCustomMessage(idToString(request.Id), &custommsg)
+}
+
 func unmarshalOpenChannel(request *Request) (r json.RawMessage, err error) {
 	switch request.Method {
 	case "openchannel":

cln_plugin/proto/cln_plugin.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.28.1
-// 	protoc        v3.21.12
+// 	protoc        v4.23.4
 // source: cln_plugin.proto
 
 package proto
@@ -549,6 +549,99 @@ func (x *HtlcResolve) GetPaymentKey() string {
 	return ""
 }
 
+type CustomMessageRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *CustomMessageRequest) Reset() {
+	*x = CustomMessageRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_cln_plugin_proto_msgTypes[7]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CustomMessageRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CustomMessageRequest) ProtoMessage() {}
+
+func (x *CustomMessageRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_cln_plugin_proto_msgTypes[7]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CustomMessageRequest.ProtoReflect.Descriptor instead.
+func (*CustomMessageRequest) Descriptor() ([]byte, []int) {
+	return file_cln_plugin_proto_rawDescGZIP(), []int{7}
+}
+
+type CustomMessage struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	PeerId  string `protobuf:"bytes,1,opt,name=peer_id,json=peerId,proto3" json:"peer_id,omitempty"`
+	Payload string `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"`
+}
+
+func (x *CustomMessage) Reset() {
+	*x = CustomMessage{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_cln_plugin_proto_msgTypes[8]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *CustomMessage) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*CustomMessage) ProtoMessage() {}
+
+func (x *CustomMessage) ProtoReflect() protoreflect.Message {
+	mi := &file_cln_plugin_proto_msgTypes[8]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use CustomMessage.ProtoReflect.Descriptor instead.
+func (*CustomMessage) Descriptor() ([]byte, []int) {
+	return file_cln_plugin_proto_rawDescGZIP(), []int{8}
+}
+
+func (x *CustomMessage) GetPeerId() string {
+	if x != nil {
+		return x.PeerId
+	}
+	return ""
+}
+
+func (x *CustomMessage) GetPayload() string {
+	if x != nil {
+		return x.Payload
+	}
+	return ""
+}
+
 var File_cln_plugin_proto protoreflect.FileDescriptor
 
 var file_cln_plugin_proto_rawDesc = []byte{
@@ -618,14 +711,23 @@ var file_cln_plugin_proto_rawDesc = []byte{
 	0x75, 0x72, 0x65, 0x22, 0x2e, 0x0a, 0x0b, 0x48, 0x74, 0x6c, 0x63, 0x52, 0x65, 0x73, 0x6f, 0x6c,
 	0x76, 0x65, 0x12, 0x1f, 0x0a, 0x0b, 0x70, 0x61, 0x79, 0x6d, 0x65, 0x6e, 0x74, 0x5f, 0x6b, 0x65,
 	0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x70, 0x61, 0x79, 0x6d, 0x65, 0x6e, 0x74,
-	0x4b, 0x65, 0x79, 0x32, 0x3d, 0x0a, 0x09, 0x43, 0x6c, 0x6e, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e,
+	0x4b, 0x65, 0x79, 0x22, 0x16, 0x0a, 0x14, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x4d, 0x65, 0x73,
-	0x12, 0x30, 0x0a, 0x0a, 0x48, 0x74, 0x6c, 0x63, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x0f,
+	0x73, 0x61, 0x67, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x42, 0x0a, 0x0d, 0x43,
-	0x2e, 0x48, 0x74, 0x6c, 0x63, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x1a,
+	0x75, 0x73, 0x74, 0x6f, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x17, 0x0a, 0x07,
-	0x0d, 0x2e, 0x48, 0x74, 0x6c, 0x63, 0x41, 0x63, 0x63, 0x65, 0x70, 0x74, 0x65, 0x64, 0x28, 0x01,
+	0x70, 0x65, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x70,
-	0x30, 0x01, 0x42, 0x28, 0x5a, 0x26, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d,
+	0x65, 0x65, 0x72, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64,
-	0x2f, 0x62, 0x72, 0x65, 0x65, 0x7a, 0x2f, 0x6c, 0x73, 0x70, 0x64, 0x2f, 0x63, 0x6c, 0x6e, 0x5f,
+	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x32,
-	0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72,
+	0x79, 0x0a, 0x09, 0x43, 0x6c, 0x6e, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x12, 0x30, 0x0a, 0x0a,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x48, 0x74, 0x6c, 0x63, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x0f, 0x2e, 0x48, 0x74, 0x6c,
+	0x63, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x75, 0x74, 0x69, 0x6f, 0x6e, 0x1a, 0x0d, 0x2e, 0x48, 0x74,
+	0x6c, 0x63, 0x41, 0x63, 0x63, 0x65, 0x70, 0x74, 0x65, 0x64, 0x28, 0x01, 0x30, 0x01, 0x12, 0x3a,
+	0x0a, 0x0f, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x4d, 0x73, 0x67, 0x53, 0x74, 0x72, 0x65, 0x61,
+	0x6d, 0x12, 0x15, 0x2e, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
+	0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0e, 0x2e, 0x43, 0x75, 0x73, 0x74, 0x6f,
+	0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x30, 0x01, 0x42, 0x28, 0x5a, 0x26, 0x67, 0x69,
+	0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x62, 0x72, 0x65, 0x65, 0x7a, 0x2f, 0x6c,
+	0x73, 0x70, 0x64, 0x2f, 0x63, 0x6c, 0x6e, 0x5f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2f, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -640,7 +742,7 @@ func file_cln_plugin_proto_rawDescGZIP() []byte {
 	return file_cln_plugin_proto_rawDescData
 }
 
-var file_cln_plugin_proto_msgTypes = make([]protoimpl.MessageInfo, 7)
+var file_cln_plugin_proto_msgTypes = make([]protoimpl.MessageInfo, 9)
 var file_cln_plugin_proto_goTypes = []interface{}{
 	(*HtlcAccepted)(nil),         // 0: HtlcAccepted
 	(*Onion)(nil),                // 1: Onion
@@ -649,6 +751,8 @@ var file_cln_plugin_proto_goTypes = []interface{}{
 	(*HtlcContinue)(nil),         // 4: HtlcContinue
 	(*HtlcFail)(nil),             // 5: HtlcFail
 	(*HtlcResolve)(nil),          // 6: HtlcResolve
+	(*CustomMessageRequest)(nil), // 7: CustomMessageRequest
+	(*CustomMessage)(nil),        // 8: CustomMessage
 }
 var file_cln_plugin_proto_depIdxs = []int32{
 	1, // 0: HtlcAccepted.onion:type_name -> Onion
@@ -657,9 +761,11 @@ var file_cln_plugin_proto_depIdxs = []int32{
 	4, // 3: HtlcResolution.continue:type_name -> HtlcContinue
 	6, // 4: HtlcResolution.resolve:type_name -> HtlcResolve
 	3, // 5: ClnPlugin.HtlcStream:input_type -> HtlcResolution
-	0, // 6: ClnPlugin.HtlcStream:output_type -> HtlcAccepted
+	7, // 6: ClnPlugin.CustomMsgStream:input_type -> CustomMessageRequest
-	6, // [6:7] is the sub-list for method output_type
+	0, // 7: ClnPlugin.HtlcStream:output_type -> HtlcAccepted
-	5, // [5:6] is the sub-list for method input_type
+	8, // 8: ClnPlugin.CustomMsgStream:output_type -> CustomMessage
+	7, // [7:9] is the sub-list for method output_type
+	5, // [5:7] is the sub-list for method input_type
 	5, // [5:5] is the sub-list for extension type_name
 	5, // [5:5] is the sub-list for extension extendee
 	0, // [0:5] is the sub-list for field type_name
@@ -755,6 +861,30 @@ func file_cln_plugin_proto_init() {
 				return nil
 			}
 		}
+		file_cln_plugin_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CustomMessageRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_cln_plugin_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*CustomMessage); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	file_cln_plugin_proto_msgTypes[3].OneofWrappers = []interface{}{
 		(*HtlcResolution_Fail)(nil),
@@ -772,7 +902,7 @@ func file_cln_plugin_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_cln_plugin_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   7,
+			NumMessages:   9,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

cln_plugin/proto/cln_plugin.proto

@@ -3,6 +3,7 @@ option go_package="github.com/breez/lspd/cln_plugin/proto";
 
 service ClnPlugin {
     rpc HtlcStream(stream HtlcResolution) returns (stream HtlcAccepted);
+    rpc CustomMsgStream(CustomMessageRequest) returns (stream CustomMessage);
 }
 
 message HtlcAccepted {
@@ -54,3 +55,9 @@ message HtlcFail {
 message HtlcResolve {
     string payment_key = 1;
 }
+
+message CustomMessageRequest {}
+message CustomMessage {
+    string peer_id = 1;
+    string payload = 2;
+}

cln_plugin/proto/cln_plugin_grpc.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.2.0
-// - protoc             v3.21.12
+// - protoc             v4.23.4
 // source: cln_plugin.proto
 
 package proto
@@ -23,6 +23,7 @@ const _ = grpc.SupportPackageIsVersion7
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type ClnPluginClient interface {
 	HtlcStream(ctx context.Context, opts ...grpc.CallOption) (ClnPlugin_HtlcStreamClient, error)
+	CustomMsgStream(ctx context.Context, in *CustomMessageRequest, opts ...grpc.CallOption) (ClnPlugin_CustomMsgStreamClient, error)
 }
 
 type clnPluginClient struct {
@@ -64,11 +65,44 @@ func (x *clnPluginHtlcStreamClient) Recv() (*HtlcAccepted, error) {
 	return m, nil
 }
 
+func (c *clnPluginClient) CustomMsgStream(ctx context.Context, in *CustomMessageRequest, opts ...grpc.CallOption) (ClnPlugin_CustomMsgStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, &ClnPlugin_ServiceDesc.Streams[1], "/ClnPlugin/CustomMsgStream", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &clnPluginCustomMsgStreamClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type ClnPlugin_CustomMsgStreamClient interface {
+	Recv() (*CustomMessage, error)
+	grpc.ClientStream
+}
+
+type clnPluginCustomMsgStreamClient struct {
+	grpc.ClientStream
+}
+
+func (x *clnPluginCustomMsgStreamClient) Recv() (*CustomMessage, error) {
+	m := new(CustomMessage)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // ClnPluginServer is the server API for ClnPlugin service.
 // All implementations must embed UnimplementedClnPluginServer
 // for forward compatibility
 type ClnPluginServer interface {
 	HtlcStream(ClnPlugin_HtlcStreamServer) error
+	CustomMsgStream(*CustomMessageRequest, ClnPlugin_CustomMsgStreamServer) error
 	mustEmbedUnimplementedClnPluginServer()
 }
 
@@ -79,6 +113,9 @@ type UnimplementedClnPluginServer struct {
 func (UnimplementedClnPluginServer) HtlcStream(ClnPlugin_HtlcStreamServer) error {
 	return status.Errorf(codes.Unimplemented, "method HtlcStream not implemented")
 }
+func (UnimplementedClnPluginServer) CustomMsgStream(*CustomMessageRequest, ClnPlugin_CustomMsgStreamServer) error {
+	return status.Errorf(codes.Unimplemented, "method CustomMsgStream not implemented")
+}
 func (UnimplementedClnPluginServer) mustEmbedUnimplementedClnPluginServer() {}
 
 // UnsafeClnPluginServer may be embedded to opt out of forward compatibility for this service.
@@ -118,6 +155,27 @@ func (x *clnPluginHtlcStreamServer) Recv() (*HtlcResolution, error) {
 	return m, nil
 }
 
+func _ClnPlugin_CustomMsgStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(CustomMessageRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(ClnPluginServer).CustomMsgStream(m, &clnPluginCustomMsgStreamServer{stream})
+}
+
+type ClnPlugin_CustomMsgStreamServer interface {
+	Send(*CustomMessage) error
+	grpc.ServerStream
+}
+
+type clnPluginCustomMsgStreamServer struct {
+	grpc.ServerStream
+}
+
+func (x *clnPluginCustomMsgStreamServer) Send(m *CustomMessage) error {
+	return x.ServerStream.SendMsg(m)
+}
+
 // ClnPlugin_ServiceDesc is the grpc.ServiceDesc for ClnPlugin service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -132,6 +190,11 @@ var ClnPlugin_ServiceDesc = grpc.ServiceDesc{
 			ServerStreams: true,
 			ClientStreams: true,
 		},
+		{
+			StreamName:    "CustomMsgStream",
+			Handler:       _ClnPlugin_CustomMsgStream_Handler,
+			ServerStreams: true,
+		},
 	},
 	Metadata: "cln_plugin.proto",
 }

cln_plugin/server.go

@@ -25,6 +25,19 @@ type htlcResultMsg struct {
 	result interface{}
 }
 
+// Internal custommsg message meant for the sendQueue.
+type custommsgMsg struct {
+	id        string
+	custommsg *CustomMessageRequest
+	timeout   time.Time
+}
+
+// Internal custommsg result message meant for the recvQueue.
+type custommsgResultMsg struct {
+	id     string
+	result interface{}
+}
+
 type server struct {
 	proto.ClnPluginServer
 	listenAddress          string
@@ -39,6 +52,10 @@ type server struct {
 	htlcStream             proto.ClnPlugin_HtlcStreamServer
 	htlcSendQueue          chan *htlcAcceptedMsg
 	htlcRecvQueue          chan *htlcResultMsg
+	custommsgNewSubscriber chan struct{}
+	custommsgStream        proto.ClnPlugin_CustomMsgStreamServer
+	custommsgSendQueue     chan *custommsgMsg
+	custommsgRecvQueue     chan *custommsgResultMsg
 }
 
 // Creates a new grpc server
@@ -49,10 +66,12 @@ func NewServer(listenAddress string, subscriberTimeout time.Duration) *server {
 		subscriberTimeout: subscriberTimeout,
 		// The send queue exists to buffer messages until a subscriber is active.
 		htlcSendQueue:      make(chan *htlcAcceptedMsg, 10000),
+		custommsgSendQueue: make(chan *custommsgMsg, 10000),
 		// The receive queue exists mainly to allow returning timeouts to the
 		// cln plugin. If there is no subscriber active within the subscriber
 		// timeout period these results can be put directly on the receive queue.
 		htlcRecvQueue:      make(chan *htlcResultMsg, 10000),
+		custommsgRecvQueue: make(chan *custommsgResultMsg, 10000),
 		started:            make(chan struct{}),
 		startError:         make(chan error, 1),
 	}
@@ -79,6 +98,7 @@ func (s *server) Start() error {
 	s.done = make(chan struct{})
 	s.completed = make(chan struct{})
 	s.htlcnewSubscriber = make(chan struct{})
+	s.custommsgNewSubscriber = make(chan struct{})
 	s.grpcServer = grpc.NewServer(
 		grpc.KeepaliveParams(keepalive.ServerParameters{
 			Time:    time.Duration(1) * time.Second,
@@ -94,6 +114,7 @@ func (s *server) Start() error {
 	log.Printf("Server starting to listen on %s.", s.listenAddress)
 	go s.listenHtlcRequests()
 	go s.listenHtlcResponses()
+	go s.listenCustomMsgRequests()
 	close(s.started)
 	err = s.grpcServer.Serve(lis)
 	close(s.completed)
@@ -306,13 +327,13 @@ func (s *server) recvHtlcResolution() *proto.HtlcResolution {
 		s.mtx.Unlock()
 
 		if stream == nil {
-			log.Printf("Got no subscribers for receive. Waiting for subscriber.")
+			log.Printf("Got no subscribers for htlc receive. Waiting for subscriber.")
 			select {
 			case <-s.done:
-				log.Printf("Done signalled, stopping receive.")
+				log.Printf("Done signalled, stopping htlc receive.")
 				return nil
 			case <-ns:
-				log.Printf("New subscription available for receive, continue receive.")
+				log.Printf("New subscription available for htlc receive, continue receive.")
 				continue
 			}
 		}
@@ -393,6 +414,147 @@ func (s *server) mapHtlcResult(outcome interface{}) interface{} {
 	return s.defaultResult()
 }
 
+// Grpc method that is called when a new client subscribes. There can only be
+// one subscriber active at a time. If there is an error receiving or sending
+// from or to the subscriber, the subscription is closed.
+func (s *server) CustomMsgStream(
+	_ *proto.CustomMessageRequest,
+	stream proto.ClnPlugin_CustomMsgStreamServer,
+) error {
+
+	s.mtx.Lock()
+	if s.custommsgStream == nil {
+		log.Printf("Got a new custommsg stream subscription request.")
+	} else {
+		s.mtx.Unlock()
+		log.Printf("Got a custommsg stream subscription request, but " +
+			"subscription was already active.")
+		return fmt.Errorf("already subscribed")
+	}
+
+	s.custommsgStream = stream
+
+	// Notify listeners that a new subscriber is active. Replace the chan with
+	// a new one immediately in case this subscriber is dropped later.
+	close(s.custommsgNewSubscriber)
+	s.custommsgNewSubscriber = make(chan struct{})
+	s.mtx.Unlock()
+
+	<-stream.Context().Done()
+	log.Printf(
+		"CustomMsgStream context is done. Return: %v",
+		stream.Context().Err(),
+	)
+
+	// Remove the subscriber.
+	s.mtx.Lock()
+	s.custommsgStream = nil
+	s.mtx.Unlock()
+
+	return stream.Context().Err()
+}
+
+// Enqueues a htlc_accepted message for send to the grpc client.
+func (s *server) SendCustomMessage(id string, c *CustomMessageRequest) {
+	s.custommsgSendQueue <- &custommsgMsg{
+		id:        id,
+		custommsg: c,
+		timeout:   time.Now().Add(s.subscriberTimeout),
+	}
+}
+
+// Receives the next custommsg response message from the grpc client. Returns id
+// and message. Blocks until a message is available. Returns a nil message if
+// the server is done. This function effectively waits until a subscriber is
+// active and has sent a message.
+func (s *server) ReceiveCustomMessageResponse() (string, interface{}) {
+	select {
+	case <-s.done:
+		return "", nil
+	case msg := <-s.custommsgRecvQueue:
+		return msg.id, msg.result
+	}
+}
+
+// Listens to sendQueue for custommsg requests from cln. The message will be
+// held until a subscriber is active, or the subscriber timeout expires. The
+// messages are sent to the grpc client in fifo order.
+func (s *server) listenCustomMsgRequests() {
+	for {
+		select {
+		case <-s.done:
+			log.Printf("listenCustomMsgRequests received done. Stop listening.")
+			return
+		case msg := <-s.custommsgSendQueue:
+			s.handleCustomMsg(msg)
+		}
+	}
+}
+
+// Attempts to send a custommsg message to the grpc client. The message will
+// be held until a subscriber is active, or the subscriber timeout expires.
+func (s *server) handleCustomMsg(msg *custommsgMsg) {
+	for {
+		s.mtx.Lock()
+		stream := s.custommsgStream
+		ns := s.custommsgNewSubscriber
+		s.mtx.Unlock()
+
+		// If there is no active subscription, wait until there is a new
+		// subscriber, or the message times out.
+		if stream == nil {
+			select {
+			case <-s.done:
+				log.Printf("handleCustomMsg received server done. Stop processing.")
+				return
+			case <-ns:
+				log.Printf("got a new subscriber. continue handleCustomMsg.")
+				continue
+			case <-time.After(time.Until(msg.timeout)):
+				log.Printf(
+					"WARNING: custommsg with id '%s' timed out after '%v' waiting "+
+						"for grpc subscriber: %+v",
+					msg.id,
+					s.subscriberTimeout,
+					msg.custommsg,
+				)
+
+				// If the subscriber timeout expires while holding the custommsg
+				// we ignore the message by sending the default result
+				// (continue) to cln.
+				s.custommsgRecvQueue <- &custommsgResultMsg{
+					id:     msg.id,
+					result: s.defaultResult(),
+				}
+
+				return
+			}
+		}
+
+		// There is a subscriber. Attempt to send the custommsg message.
+		err := stream.Send(&proto.CustomMessage{
+			PeerId:  msg.custommsg.PeerId,
+			Payload: msg.custommsg.Payload,
+		})
+
+		// If there is no error, we're done, mark the message as handled.
+		if err == nil {
+			s.custommsgRecvQueue <- &custommsgResultMsg{
+				id:     msg.id,
+				result: s.defaultResult(),
+			}
+			return
+		}
+
+		// If we end up here, there was an error sending the message to the
+		// grpc client.
+		// TODO: If the Send errors, but the context is not done, this will
+		// currently retry immediately. Check whether the context is really
+		// done on an error!
+		log.Printf("Error sending custommsg message to subscriber. Retrying: %v", err)
+	}
+}
+
 // Returns a result: continue message.
 func (s *server) defaultResult() interface{} {
 	return map[string]interface{}{