Struct P2PStream 

pub struct P2PStream<S> { /* private fields */ }

A P2PStream wraps over any Stream that yields bytes and makes it compatible with p2p protocol messages.

This stream supports multiple shared capabilities, that were negotiated during the handshake.

Message-ID based multiplexing

Each capability is given as much of the message-ID space as it needs. All such capabilities must statically specify how many message IDs they require. On connection and reception of the Hello message, both peers have equivalent information about what capabilities they share (including versions) and are able to form consensus over the composition of message ID space.

Message IDs are assumed to be compact from ID 0x10 onwards (0x00-0x0f is reserved for the “p2p” capability) and given to each shared (equal-version, equal-name) capability in alphabetic order. Capability names are case-sensitive. Capabilities which are not shared are ignored. If multiple versions are shared of the same (equal name) capability, the numerically highest wins, others are ignored.

See also https://github.com/ethereum/devp2p/blob/master/rlpx.md#message-id-based-multiplexing

This stream emits non-empty Bytes that start with the normalized message id, so that the first byte of each message starts from 0. If this stream only supports a single capability, for example eth then the first byte of each message will match EthMessageID.

Sink behavior

The [Sink] impl batches writes: queued messages are drained into the underlying sink unflushed, and the caller is responsible for driving [Sink::poll_flush] to deliver them to the wire. Queued p2p control messages (ping/pong/disconnect) are the exception: they force a flush from [Sink::poll_ready]. Keepalive pings are generated in poll_ready, so the sink half must be polled regularly even if the caller has nothing to send.

Implementations

impl<S> P2PStream<S>

pub fn new(inner: S, shared_capabilities: SharedCapabilities) -> Self

Create a new P2PStream from the provided stream. New P2PStreams are assumed to have completed the p2p handshake successfully and are ready to send and receive subprotocol messages.

pub const fn inner(&self) -> &S

Returns a reference to the inner stream.

pub const fn set_outgoing_message_buffer_capacity(&mut self, capacity: usize)

Sets a custom outgoing message buffer capacity.

Panics

If the provided capacity is 0.

pub const fn set_max_payload_size(&mut self, max_payload_size: usize)

Sets the maximum size of an uncompressed message payload, which defaults to MAX_PAYLOAD_SIZE. It applies to both incoming and outgoing messages.

pub const fn shared_capabilities(&self) -> &SharedCapabilities

Returns the shared capabilities for this stream.

This includes all the shared capabilities that were negotiated during the handshake and their offsets based on the number of messages of each capability.

pub fn send_ping(&mut self)

Queues in a snappy encoded P2PMessage::Ping message.

impl<S> P2PStream<S>

where S: Sink<Bytes, Error = Error> + Unpin + Send,

pub async fn disconnect( &mut self, reason: DisconnectReason, ) -> Result<(), P2PStreamError>

Disconnects the connection by sending a disconnect message.

This future resolves once the disconnect message has been sent and the stream has been closed.

Trait Implementations

impl<S> CanDisconnect<Bytes> for P2PStream<S>

where S: Sink<Bytes, Error = Error> + Unpin + Send + Sync,

fn disconnect( &mut self, reason: DisconnectReason, ) -> Pin<Box<dyn Future<Output = Result<(), P2PStreamError>> + Send + '_>>

Disconnects from the underlying stream, using a DisconnectReason as disconnect information if the stream implements a protocol that can carry the additional disconnect metadata.

impl<S: Debug> Debug for P2PStream<S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<S> DisconnectP2P for P2PStream<S>

fn start_disconnect( &mut self, reason: DisconnectReason, ) -> Result<(), P2PStreamError>

Starts to gracefully disconnect the connection by sending a Disconnect message and stop reading new messages.

Once disconnect process has started, the [Stream] will terminate immediately.

Errors

Returns an error only if the message fails to compress.

fn is_disconnecting(&self) -> bool

Returns true if the connection is about to disconnect.

impl<S> Sink<Bytes> for P2PStream<S>

where S: Sink<Bytes, Error = Error> + Unpin,

fn poll_flush( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when no buffered items remain.

type Error = P2PStreamError

The type of value produced by the sink when an error occurs.

fn poll_ready( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Attempts to prepare the Sink to receive a value.

fn start_send(self: Pin<&mut Self>, item: Bytes) -> Result<(), Self::Error>

Begin the process of sending a value to the sink. Each call to this function must be preceded by a successful call to poll_ready which returned Poll::Ready(Ok(())).

fn poll_close( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Flush any remaining output and close this sink, if necessary.

impl<S> Stream for P2PStream<S>

where S: Stream<Item = Result<BytesMut>> + Sink<Bytes, Error = Error> + Unpin,

type Item = Result<BytesMut, P2PStreamError>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<'pin, S> Unpin for P2PStream<S>

where PinnedFieldsOf<__P2PStream<'pin, S>>: Unpin,

Layout

Note: Unable to compute type layout, possibly due to this type having generic parameters. Layout can only be computed for concrete, fully-instantiated types.