How to receive P2P messages using sc-network-gossip

Question

Overview

I'm interested to understand how to use sc-network-gossip's GossipEngine to receive P2P messages, not just gossip messages. The motivation is that I would like to send specific peers specific messages rather than gossiping all messages to all peers. So far, I haven't found examples for doing so and so am posting here.

Example of gossiping

The BEEFY client receives gossip messages over its provided topic using the following snippet:

let mut votes = Box::pin(self.gossip_engine.lock().messages_for(topic::<B>()).filter_map(
    |notification| async move {
        trace!(target: "beefy", "🥩 Got vote message: {:?}", notification);

        VoteMessage::<NumberFor<B>, AuthorityId, Signature>::decode(
            &mut &notification.message[..],
        )
        .ok()
    },
));

The messages are sent using:

self.gossip_engine.lock().gossip_message(topic::<B>(), encoded_message, false);

API

The GossipEngine API supports this method and another send_message

• https://github.com/paritytech/substrate/blob/master/client/network-gossip/src/bridge.rs#L143-L146
• https://github.com/paritytech/substrate/blob/master/client/network-gossip/src/bridge.rs#L148-L154

Questions

It is clear where messages received over a given topic are able to be processed and consumed using examples from BEEFY. However, it is unclear how to consume messages that are sent P2P using send_message.

1. If we send a message to a peer, how does one receive it and process it.
2. Assuming the message is simply sent without a topic, what structs would I need to access?