Publish with pubsub

Enable gossipsub, subscribe and publish signed messages, and consume pubsub events.

The pubsub feature exposes gossipsub. EndpointBuilder::gossipsub() enables it with interoperable defaults and advertises /meshsub/1.1.0 and /meshsub/1.0.0. You need a second peer on the same topic to observe delivery. See Drive events for the caller-driven event model.

Enable gossipsub

cargo add minip2p-rs --features pubsub

Then activate it on the endpoint:

let mut node = minip2p::Endpoint::builder()
    .gossipsub()
    .listen_default()?
    .bind()?;

node.subscribe("news")?;

The Cargo feature alone only compiles the APIs. Calling subscribe on an endpoint that was not built with .gossipsub() or .gossipsub_config(...) returns GossipsubError::NotEnabled.

Publish and keep driving

use std::time::{Duration, Instant};

use minip2p::{EndpointEvent, EndpointWaitOutcome};

node.publish("news", b"hello mesh".to_vec())?;

let deadline = Instant::now() + Duration::from_secs(5);
loop {
    match node.wait(deadline)? {
        EndpointWaitOutcome::Event(EndpointEvent::Gossipsub(event)) => {
            println!("{event:?}");
            break;
        }
        EndpointWaitOutcome::Deadline => break,
        _ => {}
    }
}

A successful publish means the message passed validation and outbound work was accepted. The endpoint must continue being driven to open streams and send frames.

There is no self-delivery. If the local application needs immediate feedback, handle its submitted value locally rather than waiting to receive its own message.

Later delivery problems appear as GossipsubEvent::OutboundFailure or ordinary EndpointEvent::Error runtime events; they are not returned synchronously from an already accepted publish.

Handle messages

use std::time::Duration;

use minip2p::{EndpointEvent, EndpointWaitOutcome, GossipsubEvent};

loop {
    match node.wait(Duration::from_millis(250))? {
        EndpointWaitOutcome::Event(EndpointEvent::Gossipsub(GossipsubEvent::Message {
            from,
            topics,
            data,
            signed,
            ..
        })) => {
            println!(
                "from={from} topics={topics:?} signed={signed} data={:?}",
                String::from_utf8_lossy(&data),
            );
        }
        EndpointWaitOutcome::Event(EndpointEvent::Gossipsub(GossipsubEvent::ProtocolViolation {
            peer,
            reason,
        })) => {
            eprintln!("peer={peer} violated pubsub: {reason}");
        }
        EndpointWaitOutcome::Event(_) => {} // EndpointEvent is non-exhaustive
        _ => {}
    }
}

Gossipsub output arrives as EndpointEvent::Gossipsub on the Endpoint event stream. Discovery beacon-topic traffic never appears there.

Use subscribe and unsubscribe to change local topic membership. Both return false when the requested state already held.

Signing and unsigned compatibility

Published messages are signed with the endpoint identity. By default, gossipsub also rejects unsigned inbound messages. This matches StrictSign behavior without a separate public StrictSign type.

Only relax inbound acceptance when interoperating with a peer that intentionally sends unsigned messages:

use minip2p::{Endpoint, GossipsubConfig};

let config = GossipsubConfig {
    allow_unsigned: true,
    ..GossipsubConfig::default()
};

let node = Endpoint::builder()
    .gossipsub_config(config)
    .listen_default()?
    .bind()?;

Messages that include a signature are still verified. allow_unsigned does not make local publications unsigned.

For automatic peer address exchange on a signed pubsub topic, continue to Discover peers.