pub struct PushClient { /* private fields */ }Expand description
A connected client that consumes server pushes and sends correlated replies.
Construct with PushClient::connect; the background reader starts
immediately and runs until the client is dropped. Pull pushed frames with
PushClient::recv_timeout and answer them with PushClient::reply.
Implementations§
Source§impl PushClient
impl PushClient
Sourcepub fn connect(address: &str) -> Result<Self, SdkError>
pub fn connect(address: &str) -> Result<Self, SdkError>
Connects to address, performs the protocol handshake, and starts the
background reader that drains inbound server pushes.
§Errors
Returns SdkError::Connection when the TCP connection or socket
configuration fails, and SdkError::Protocol when the handshake is
rejected or the socket cannot be cloned for the reader thread.
Examples found in repository?
57fn run() -> Result<(), AppError> {
58 let address = configured("LIMINAL_ADDRESS", DEFAULT_ADDRESS)?;
59 let channel = configured("LIMINAL_DEMO_CHANNEL", DEFAULT_CHANNEL)?;
60 let component_id = ComponentId::new(&configured(
61 "LIMINAL_DEMO_COMPONENT_ID",
62 DEFAULT_COMPONENT_ID,
63 )?)?;
64 let generation_file = PathBuf::from(configured(
65 "LIMINAL_DEMO_GENERATION_FILE",
66 DEFAULT_GENERATION_FILE,
67 )?);
68
69 let authority = FeedAuthority::start(FileGenerationStore::new(generation_file))?;
70
71 // PushWriter is the existing SDK path that preserves these opaque bytes exactly.
72 // Its schema id is zero and its `Result<(), SdkError>` is a write outcome, not a
73 // delivery acknowledgement; periodic snapshots provide demo resynchronization.
74 let client = PushClient::connect(&address).map_err(AppError::Connect)?;
75 let writer = client.writer_handle();
76 let mut cadence = CadenceEngine::new(
77 channel,
78 component_id,
79 authority,
80 FrameEnvelopeCodec,
81 GraphViewState::new()?,
82 writer,
83 SNAPSHOT_PERIOD,
84 )?;
85
86 cadence.emit_initial_snapshot()?;
87 loop {
88 // This wall clock is demo content pacing only. It is not protocol authority,
89 // retry authority, reconnect backoff, or a delivery timer.
90 thread::sleep(TICK_INTERVAL);
91 cadence.emit_tick()?;
92 }
93}Sourcepub fn connect_with_auth(
address: &str,
auth_token: &[u8],
) -> Result<Self, SdkError>
pub fn connect_with_auth( address: &str, auth_token: &[u8], ) -> Result<Self, SdkError>
Connects and handshakes carrying auth_token, then starts the background
reader, for a server gated by an [auth] section. Additive to connect;
an empty token is equivalent to it.
§Errors
Returns SdkError::Connection when the TCP connection or socket
configuration fails or the server rejects the token, and
SdkError::Protocol when the handshake is otherwise rejected or the socket
cannot be cloned for the reader thread.
Sourcepub fn connect_with_registration(
address: &str,
registration: WorkerRegistration,
) -> Result<Self, SdkError>
pub fn connect_with_registration( address: &str, registration: WorkerRegistration, ) -> Result<Self, SdkError>
Connects, performs the handshake, then synchronously registers this client as a worker before starting the background reader.
This mirrors the synchronous Connect/ConnectAck pattern: the
WorkerRegister frame is written and its Frame::WorkerRegisterAck read
on the calling thread, BEFORE the Push-only background reader is spawned, so
the ack is never swallowed by the reader. A connect-variant (rather than a
register() method on a connected client) is the cleanest fit: connect
spawns the reader as its last step, so registration must be threaded into
the connect sequence to land before that spawn; a post-connect method would
race the already-running reader for the ack frame.
§Errors
Returns SdkError::Connection when the TCP connection or socket
configuration fails, and SdkError::Protocol when the handshake is
rejected, the server rejects the registration (the rejection reason is
carried in the error), or the socket cannot be cloned for the reader thread.
Sourcepub fn connect_with_registration_and_auth(
address: &str,
registration: WorkerRegistration,
auth_token: &[u8],
) -> Result<Self, SdkError>
pub fn connect_with_registration_and_auth( address: &str, registration: WorkerRegistration, auth_token: &[u8], ) -> Result<Self, SdkError>
Connects, handshakes carrying auth_token, registers the worker, then starts
the reader — the auth-gated variant of connect_with_registration. Additive;
an empty token is equivalent to it.
§Errors
Returns SdkError::Connection when the TCP connection or socket
configuration fails or the server rejects the token, and
SdkError::Protocol when the handshake is otherwise rejected, the server
rejects the registration (the reason is carried in the error), or the socket
cannot be cloned for the reader thread.
Sourcepub fn recv_timeout(&self, timeout: Duration) -> Result<PushedFrame, SdkError>
pub fn recv_timeout(&self, timeout: Duration) -> Result<PushedFrame, SdkError>
Blocks up to timeout for the next pushed frame from the server.
§Errors
Returns SdkError::Connection when no push arrives within timeout or
the background reader has stopped (e.g. the server closed the connection).
Sourcepub fn reply(
&self,
correlation_id: u64,
payload: Vec<u8>,
) -> Result<(), SdkError>
pub fn reply( &self, correlation_id: u64, payload: Vec<u8>, ) -> Result<(), SdkError>
Sends a correlated reply to a pushed frame, echoing its correlation id so the server matches the reply back to the originating push.
§Errors
Returns SdkError::Protocol when the reply frame cannot be encoded and
SdkError::Connection when it cannot be written to the socket or the
writer lock is poisoned.
Sourcepub fn writer_handle(&self) -> PushWriter
pub fn writer_handle(&self) -> PushWriter
A cheap, cloneable handle to this push connection’s write half, for background tasks that publish out-of-band frames on the same socket without owning the full client (which cannot be cloned — it holds the reader thread join handle).
The returned PushWriter shares the client’s Arc<Mutex<TcpStream>>, so a
frame it writes travels the SAME connection the server pushes on. It is the
worker’s observability-drain leg: a drain task holds one and publishes each
OBSERVABILITY_CHANNEL event live while the client keeps serving pushes.
Examples found in repository?
57fn run() -> Result<(), AppError> {
58 let address = configured("LIMINAL_ADDRESS", DEFAULT_ADDRESS)?;
59 let channel = configured("LIMINAL_DEMO_CHANNEL", DEFAULT_CHANNEL)?;
60 let component_id = ComponentId::new(&configured(
61 "LIMINAL_DEMO_COMPONENT_ID",
62 DEFAULT_COMPONENT_ID,
63 )?)?;
64 let generation_file = PathBuf::from(configured(
65 "LIMINAL_DEMO_GENERATION_FILE",
66 DEFAULT_GENERATION_FILE,
67 )?);
68
69 let authority = FeedAuthority::start(FileGenerationStore::new(generation_file))?;
70
71 // PushWriter is the existing SDK path that preserves these opaque bytes exactly.
72 // Its schema id is zero and its `Result<(), SdkError>` is a write outcome, not a
73 // delivery acknowledgement; periodic snapshots provide demo resynchronization.
74 let client = PushClient::connect(&address).map_err(AppError::Connect)?;
75 let writer = client.writer_handle();
76 let mut cadence = CadenceEngine::new(
77 channel,
78 component_id,
79 authority,
80 FrameEnvelopeCodec,
81 GraphViewState::new()?,
82 writer,
83 SNAPSHOT_PERIOD,
84 )?;
85
86 cadence.emit_initial_snapshot()?;
87 loop {
88 // This wall clock is demo content pacing only. It is not protocol authority,
89 // retry authority, reconnect backoff, or a delivery timer.
90 thread::sleep(TICK_INTERVAL);
91 cadence.emit_tick()?;
92 }
93}Sourcepub fn publish(&self, channel: &str, payload: Vec<u8>) -> Result<(), SdkError>
pub fn publish(&self, channel: &str, payload: Vec<u8>) -> Result<(), SdkError>
Publish payload to channel over this connection (out-of-band from the
push/reply round trip).
Convenience shorthand for self.writer_handle().publish(channel, payload).
§Errors
Returns SdkError::Protocol when the publish frame cannot be encoded and
SdkError::Connection when it cannot be written to the socket or the
writer lock is poisoned.