Adds Application Specific RPC Inspector

pubsub.go

@@ -170,6 +170,11 @@ type PubSub struct {
 	protoMatchFunc ProtocolMatchFn
 
 	ctx context.Context
+
+	// appSpecificRpcInspector is an auxiliary that may be set by the application to inspect incoming RPCs prior to
+	// processing them. The inspector is invoked on an accepted RPC right prior to handling it.
+	// The return value of the inspector function is a boolean indicating whether the RPC should be processed or not.
+	appSpecificRpcInspector func(peer.ID, *RPC) bool
 }
 
 // PubSubRouter is the message router component of PubSub.
@@ -527,6 +532,13 @@ func WithSeenMessagesTTL(ttl time.Duration) Option {
 	}
 }
 
+func WithAppSpecificRpcInspector(inspector func(peer.ID, *RPC) bool) Option {
+	return func(ps *PubSub) error {
+		ps.appSpecificRpcInspector = inspector
+		return nil
+	}
+}
+
 // processLoop handles all inputs arriving on the channels
 func (p *PubSub) processLoop(ctx context.Context) {
 	defer func() {
@@ -1005,6 +1017,14 @@ func (p *PubSub) notifyLeave(topic string, pid peer.ID) {
 }
 
 func (p *PubSub) handleIncomingRPC(rpc *RPC) {
+	// pass the rpc through app specific validation (if any available).
+	if p.appSpecificRpcInspector != nil {
+		// check if the RPC is allowed by the external inspector
+		if accept := p.appSpecificRpcInspector(rpc.from, rpc); !accept {
+			return // reject the RPC
+		}
+	}
+
 	p.tracer.RecvRPC(rpc)
 
 	subs := rpc.GetSubscriptions()