[NOD-1367] Fix race condition with notification listeners.

app/rpc/rpc.go

@@ -44,8 +44,12 @@ func (m *Manager) routerInitializer(router *router.Router, netConnection *netada
 		err := m.handleIncomingMessages(router, incomingRoute)
 		m.handleError(err, netConnection)
 	})
+
+	notificationListener := m.context.NotificationManager.AddListener(router)
 	spawn("routerInitializer-handleOutgoingNotifications", func() {
-		err := m.handleOutgoingNotifications(router)
+		defer m.context.NotificationManager.RemoveListener(router)
+
+		err := m.handleOutgoingNotifications(notificationListener)
 		m.handleError(err, netConnection)
 	})
 }
@@ -72,9 +76,7 @@ func (m *Manager) handleIncomingMessages(router *router.Router, incomingRoute *r
 	}
 }
 
-func (m *Manager) handleOutgoingNotifications(router *router.Router) error {
-	notificationListener := m.context.NotificationManager.AddListener(router)
-	defer m.context.NotificationManager.RemoveListener(router)
+func (m *Manager) handleOutgoingNotifications(notificationListener *rpccontext.NotificationListener) error {
 	for {
 		err := notificationListener.ProcessNextNotification()
 		if err != nil {

infrastructure/network/rpcclient/grpcclient/grpcclient.go

@@ -12,9 +12,12 @@ import (
 	"time"
 )
 
+type OnErrorHandler func(err error)
+
 // GRPCClient is a gRPC-based RPC client
 type GRPCClient struct {
-	stream protowire.RPC_MessageStreamClient
+	stream         protowire.RPC_MessageStreamClient
+	onErrorHandler OnErrorHandler
 }
 
 // Connect connects to the RPC server with the given address
@@ -41,6 +44,10 @@ func (c *GRPCClient) Disconnect() error {
 	return c.stream.CloseSend()
 }
 
+func (c *GRPCClient) SetOnErrorHandler(onErrorHandler OnErrorHandler) {
+	c.onErrorHandler = onErrorHandler
+}
+
 // AttachRouter attaches the given router to the client and starts
 // sending/receiving messages via it
 func (c *GRPCClient) AttachRouter(router *router.Router) {
@@ -101,5 +108,9 @@ func (c *GRPCClient) handleError(err error) {
 		}
 		return
 	}
+	if c.onErrorHandler != nil {
+		c.onErrorHandler(err)
+		return
+	}
 	panic(err)
 }