Struct PullSocket 

pub struct PullSocket<S = TcpStream>
where
    S: AsyncRead + AsyncWrite + Unpin,
{ /* private fields */ }

PULL socket for receiving messages in a pipeline.

PULL sockets receive messages from connected PUSH sockets, providing the worker side of the pipeline pattern.

Implementations

impl<S> PullSocket<S>

where S: AsyncRead + AsyncWrite + Unpin,

pub async fn new(stream: S) -> Result<Self>

Create a new PULL socket from a stream with default buffer configuration.

pub async fn with_options(stream: S, options: SocketOptions) -> Result<Self>

Create a new PULL socket with custom buffer configuration and socket options.

pub fn try_recv(&mut self) -> Result<Option<Vec<Bytes>>>

Try to receive a message from the already-buffered input without doing a kernel read.

Decodes from bytes already present in the receive buffer. Returns Ok(None) immediately when the buffer is empty rather than suspending. Use this after recv() to drain all messages from a single read batch before returning to the io_uring submission loop:

// One kernel read may deliver many messages - drain them all before
// going back to the event loop.
if let Some(first) = pull.recv().await? {
    process(first);
    while let Some(msg) = pull.try_recv()? {
        process(msg);
    }
}

When a PING heartbeat command is decoded the corresponding PONG is queued in the send buffer; the next recv() call flushes it. For pure pipeline throughput benchmarks (where heartbeats are inactive) this is never triggered.

pub fn try_recv_into(&mut self, out: &mut Vec<Bytes>) -> Result<bool>

Try to receive a message into a caller-provided buffer, without a kernel read.

The allocation-free counterpart to try_recv: on a complete message the frames are moved into out (reusing its capacity) and Ok(true) is returned; when no complete message is buffered it returns Ok(false) and leaves out untouched. Partial frames stay in the socket’s accumulator, so it interleaves correctly with recv_into for multipart messages split across reads.

pub async fn recv(&mut self) -> Result<Option<Vec<Bytes>>>

Receive a message from a connected PUSH socket.

When multiple PUSH sockets are connected, messages are received in a fair-queued manner (in a multi-connection scenario).

Returns Ok(Some(msg)) if a message was received, Ok(None) if the connection was closed, or an error.

pub async fn recv_into(&mut self, out: &mut Vec<Bytes>) -> Result<bool>

Receive a message into a caller-provided buffer, reusing its allocation.

Identical to recv except the message frames are pushed straight into out instead of a freshly allocated Vec. The caller keeps one Vec and passes it on every call, so a steady recv loop performs no per-message heap allocation (the dominant per-message cost at small message sizes). out is cleared on entry.

Returns Ok(true) when a complete message was read into out, Ok(false) when the connection was closed.

pub async fn recv_batch(&mut self) -> Result<Option<Vec<Vec<Bytes>>>>

Receive a batch of messages with a single .await.

Blocks until at least one message is available (like recv), then drains every further message already decoded from the same kernel read(s) without suspending again. One read frequently delivers many small messages; returning them all from one .await amortizes the per-await overhead that becomes a real fraction of the budget at multi-million-msg/s rates. It is the receive-side counterpart to PushSocket::send_batch.

Returns Ok(Some(batch)) with one or more messages (in arrival order), or Ok(None) if the connection was closed before any message arrived.

pub async fn close(self) -> Result<()>

Close the socket gracefully by shutting down the underlying stream.

pub const fn options(&self) -> &SocketOptions

Get a reference to the socket options.

pub fn options_mut(&mut self) -> &mut SocketOptions

Get a mutable reference to the socket options.

pub fn set_options(&mut self, options: SocketOptions)

Set socket options (builder-style).

impl PullSocket<TcpStream>

pub async fn from_tcp(stream: TcpStream) -> Result<Self>

Create a new PULL socket from a TCP stream with TCP_NODELAY enabled.

pub async fn from_tcp_with_options( stream: TcpStream, options: SocketOptions, ) -> Result<Self>

Create a new PULL socket from a TCP stream with TCP_NODELAY and custom options.

pub async fn connect(addr: impl ToSocketAddrs) -> Result<Self>

Connect to a remote PULL socket, storing the endpoint for automatic reconnection.

pub async fn connect_with_options( addr: impl ToSocketAddrs, options: SocketOptions, ) -> Result<Self>

Connect with custom options, storing the endpoint for reconnection.

pub fn is_connected(&self) -> bool

Check if the socket is currently connected.

pub async fn try_reconnect(&mut self) -> Result<()>

Try to reconnect to the stored endpoint.

pub async fn recv_with_reconnect(&mut self) -> Result<Option<Vec<Bytes>>>

Receive a message with automatic reconnection on EOF or network error.

If the socket was created with connect() and stores an endpoint, this method loops: on EOF or broken-pipe it clears the stream and calls try_reconnect() (which applies exponential backoff), then retries recv().

Respects max_reconnect_attempts - returns NotConnected when exhausted.

Trait Implementations

impl ProxySocket for PullSocket

async fn recv_multipart(&mut self) -> Result<Option<Vec<Bytes>>>

Receive a multipart message from the socket.

async fn send_multipart(&mut self, _msg: Vec<Bytes>) -> Result<()>

Send a multipart message to the socket.

fn socket_desc(&self) -> &'static str

Get a description of the socket for logging.

impl<S> Socket for PullSocket<S>

where S: AsyncRead + AsyncWrite + Unpin + 'static,

async fn send(&mut self, _msg: Vec<Bytes>) -> Result<()>

Send a multipart message on the socket.

async fn recv(&mut self) -> Result<Option<Vec<Bytes>>>

Receive a multipart message from the socket.

fn socket_type(&self) -> SocketType

Get the socket type.

fn has_more(&self) -> bool

Check if socket has more message frames pending.