lightningclient: add custom message subscription

Author: carlaKC

lightning_client.go

@@ -201,6 +201,11 @@ type LightningClient interface {
 
 	// SendCustomMessage sends a custom message to a peer.
 	SendCustomMessage(ctx context.Context, msg CustomMessage) error
+
+	// SubscribeCustomMessages creates a subscription to custom messages
+	// received from our peers.
+	SubscribeCustomMessages(ctx context.Context) (<-chan CustomMessage,
+		<-chan error, error)
 }
 
 // Info contains info about the connected lnd node.
@@ -3757,3 +3762,65 @@ func (s *lightningClient) SendCustomMessage(ctx context.Context,
 	_, err := s.client.SendCustomMessage(rpcCtx, rpcReq)
 	return err
 }
+
+// SubscribeCustomMessages subscribes to a stream of custom messages, optionally
+// filtering by peer and message type. The channels returned will be closed
+// when the subscription exits.
+func (s *lightningClient) SubscribeCustomMessages(ctx context.Context) (
+	<-chan CustomMessage, <-chan error, error) {
+
+	rpcCtx := s.adminMac.WithMacaroonAuth(ctx)
+	rpcReq := &lnrpc.SubscribeCustomMessagesRequest{}
+
+	client, err := s.client.SubscribeCustomMessages(rpcCtx, rpcReq)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var (
+		// Buffer error channel by 1 so that consumer reading from this
+		// channel does not block our exit.
+		errChan = make(chan error, 1)
+		msgChan = make(chan CustomMessage)
+	)
+
+	s.wg.Add(1)
+	go func() {
+		defer func() {
+			// Close channels on exit so that callers know the
+			// subscription has finished.
+			close(errChan)
+			close(msgChan)
+
+			s.wg.Done()
+		}()
+
+		for {
+			msg, err := client.Recv()
+			if err != nil {
+				errChan <- fmt.Errorf("receive failed: %w", err)
+				return
+			}
+
+			peer, err := route.NewVertexFromBytes(msg.Peer)
+			if err != nil {
+				errChan <- fmt.Errorf("invalid peer: %w", err)
+				return
+			}
+
+			customMsg := CustomMessage{
+				Peer:    peer,
+				Data:    msg.Data,
+				MsgType: msg.Type,
+			}
+
+			select {
+			case msgChan <- customMsg:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return msgChan, errChan, nil
+}

testdata/permissions.json

@@ -540,6 +540,14 @@
                 }
             ]
         },
+	"/lnrpc.Lightning/SubscribeCustomMessages": {
+            "permissions": [
+                {
+                    "entity": "peers",
+                    "action": "read"
+                }
+            ]
+        },
         "/lnrpc.Lightning/SubscribeChannelEvents": {
             "permissions": [
                 {