Merge pull request #3397 from carlaKC/peernotifier-addsubscribepeerrpc

lnrpc: Add subscribe peer events rpc

lnrpc/rpc.proto

@@ -359,6 +359,13 @@ service Lightning {
         };
     }
 
+    /**
+    SubscribePeerEvents creates a uni-directional stream from the server to
+    the client in which any events relevant to the state of peers are sent
+    over. Events include peers going online and offline.
+    */
+    rpc SubscribePeerEvents (PeerEventSubscription) returns (stream PeerEvent);
+
     /** lncli: `getinfo`
     GetInfo returns general information concerning the lightning node including
     it's identity pubkey, alias, the chains it is connected to, and information
@@ -1461,6 +1468,21 @@ message ListPeersResponse {
     repeated Peer peers = 1 [json_name = "peers"];
 }
 
+message PeerEventSubscription {
+}
+
+message PeerEvent {
+    /// The identity pubkey of the peer.
+    string pub_key = 1 [json_name = "pub_key"];
+
+    enum EventType {
+        PEER_ONLINE = 0;
+        PEER_OFFLINE = 1;
+    }
+
+    EventType type = 2 [ json_name = "type" ];
+}
+
 message GetInfoRequest {
 }
 message GetInfoResponse {

lnrpc/rpc.swagger.json

@@ -1454,6 +1454,14 @@
       ],
       "default": "UNKNOWN"
     },
+    "PeerEventEventType": {
+      "type": "string",
+      "enum": [
+        "PEER_ONLINE",
+        "PEER_OFFLINE"
+      ],
+      "default": "PEER_ONLINE"
+    },
     "PeerSyncType": {
       "type": "string",
       "enum": [
@@ -3443,6 +3451,18 @@
         }
       }
     },
+    "lnrpcPeerEvent": {
+      "type": "object",
+      "properties": {
+        "pub_key": {
+          "type": "string",
+          "description": "/ The identity pubkey of the peer."
+        },
+        "type": {
+          "$ref": "#/definitions/PeerEventEventType"
+        }
+      }
+    },
     "lnrpcPendingChannelsResponse": {
       "type": "object",
       "properties": {
@@ -4160,6 +4180,18 @@
       },
       "title": "Stream result of lnrpcOpenStatusUpdate"
     },
+    "lnrpcPeerEvent": {
+      "type": "object",
+      "properties": {
+        "result": {
+          "$ref": "#/definitions/lnrpcPeerEvent"
+        },
+        "error": {
+          "$ref": "#/definitions/runtimeStreamError"
+        }
+      },
+      "title": "Stream result of lnrpcPeerEvent"
+    },
     "lnrpcSendResponse": {
       "type": "object",
       "properties": {

rpcserver.go

@@ -50,6 +50,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/macaroons"
 	"github.com/lightningnetwork/lnd/monitoring"
+	"github.com/lightningnetwork/lnd/peernotifier"
 	"github.com/lightningnetwork/lnd/record"
 	"github.com/lightningnetwork/lnd/routing"
 	"github.com/lightningnetwork/lnd/routing/route"
@@ -2286,6 +2287,51 @@ func (r *rpcServer) ListPeers(ctx context.Context,
 	return resp, nil
 }
 
+// SubscribePeerEvents returns a uni-directional stream (server -> client)
+// for notifying the client of peer online and offline events.
+func (r *rpcServer) SubscribePeerEvents(req *lnrpc.PeerEventSubscription,
+	eventStream lnrpc.Lightning_SubscribePeerEventsServer) error {
+
+	peerEventSub, err := r.server.peerNotifier.SubscribePeerEvents()
+	if err != nil {
+		return err
+	}
+	defer peerEventSub.Cancel()
+
+	for {
+		select {
+		// A new update has been sent by the peer notifier, we'll
+		// marshal it into the form expected by the gRPC client, then
+		// send it off to the client.
+		case e := <-peerEventSub.Updates():
+			var event *lnrpc.PeerEvent
+
+			switch peerEvent := e.(type) {
+			case peernotifier.PeerOfflineEvent:
+				event = &lnrpc.PeerEvent{
+					PubKey: hex.EncodeToString(peerEvent.PubKey[:]),
+					Type:   lnrpc.PeerEvent_PEER_OFFLINE,
+				}
+
+			case peernotifier.PeerOnlineEvent:
+				event = &lnrpc.PeerEvent{
+					PubKey: hex.EncodeToString(peerEvent.PubKey[:]),
+					Type:   lnrpc.PeerEvent_PEER_ONLINE,
+				}
+
+			default:
+				return fmt.Errorf("unexpected peer event: %v", event)
+			}
+
+			if err := eventStream.Send(event); err != nil {
+				return err
+			}
+		case <-r.quit:
+			return nil
+		}
+	}
+}
+
 // WalletBalance returns total unspent outputs(confirmed and unconfirmed), all
 // confirmed unspent outputs and all unconfirmed unspent outputs under control
 // by the wallet. This method can be modified by having the request specify