Struct WsTransport 

pub struct WsTransport {
    pub address: String,
    pub token: Option<String>,
}

Transport implementation using WebSocket.

Fields

address: String

The worker’s base address (ws://host:port); rewritten to http(s):// for the bulk upload/download endpoints.

token: Option<String>

Bearer token appended to every connection when the worker requires authentication.

Implementations

impl WsTransport

pub fn new(address: impl Into<String>, token: Option<String>) -> Self

A transport to the worker at address, authenticating with token if given. Connections are opened per call, not held.

pub fn send_msg(&self, msg: &CoordinatorToWorker) -> Result<WorkerToCoordinator>

Send a CoordinatorToWorker message and wait for the response.

Public because a caller that builds its own plan — the Python bindings decide which worker gets which filters, which is policy, not transport — should not have to open its own socket to ship it. A second connect_async elsewhere is a second place to get the frame-size configuration wrong.

pub fn notify(&self, msg: &CoordinatorToWorker) -> Result<()>

Send a message without waiting for an answer.

Shutdown is the one that needs this: the worker is not going to reply, so WsTransport::send_msg would block until the socket closed.

pub fn upload(&self, value: &Value) -> Result<DataRef>

Upload a value to the worker’s /upload endpoint, for payloads too large to travel inline in a WebSocket message.

pub fn stream_plan( &self, plan: SerializedPlan, chunks: Vec<Value>, ) -> Result<Value>

Ship a plan and a stream of chunks over one WebSocket, collecting results as they come back.

The binary side of the protocol. It lived in the Python bindings, which meant a second hand-rolled connect_async and a second copy of the msgpack StreamMessage framing, a crate away from the enum that defines it.

pub fn resolve_output(&self, delivery: &OutputDelivery) -> Result<Value>

Resolve OutputDelivery — inline or download via HTTP.

Trait Implementations

impl Transport for WsTransport

fn execute( &self, plan: &ExecutionPlan, _filters: &NodeCatalog, input: &Value, mode: &RunMode, seed: Option<i64>, ) -> Result<(Value, HashMap<String, Value>)>

Send a plan for execution and receive the output + trained states.

fn get_state(&self, node_ids: &[String]) -> Result<HashMap<String, Value>>

Request trained states from the remote worker.

fn set_state(&self, states: &HashMap<String, Value>) -> Result<()>

Load states on the remote worker.

fn get_gradients(&self, node_ids: &[String]) -> Result<HashMap<String, Value>>

Request gradients from the remote worker.

fn apply_gradients(&self, gradients: &HashMap<String, Value>) -> Result<()>

Apply aggregated gradients on the remote worker.

fn execute_node( &self, node_id: &str, input: Option<&Value>, ) -> Result<Value, SomaError>

Convenience: execute a single node remotely (used by the plan executor).