Implementation:ArroyoSystems Arroyo Websocket Connector

Knowledge Sources	ArroyoSystems_Arroyo
Domains	Streaming, Connectors
Last Updated	2026-02-08 08:00 GMT

Overview

WebsocketConnector implements the Arroyo Connector trait for WebSocket connections, providing a source-only connector that connects to a WebSocket server, sends optional subscription messages, and streams incoming messages.

Description

The WebSocket connector uses tokio-tungstenite to establish async WebSocket connections. It supports custom HTTP headers (via colon-separated key-value pairs) and multiple subscription messages sent after connection establishment. The test method validates connectivity by connecting, sending subscription messages, and waiting for at least one incoming message within a 30-second timeout. The connector constructs a WebsocketSourceFunc operator with the endpoint URL, parsed headers, subscription messages, format, and framing configuration. Both single subscription_message and indexed subscription_messages.N options are supported for backwards compatibility.

Usage

Use WebsocketConnector when you need to ingest real-time streaming data from a WebSocket server (such as market data feeds or live event streams) into an Arroyo pipeline.

Code Reference

Source Location

• Repository: ArroyoSystems_Arroyo
• File: crates/arroyo-connectors/src/websocket/mod.rs

Signature

pub struct WebsocketConnector {}

impl Connector for WebsocketConnector {
    type ProfileT = EmptyConfig;
    type TableT = WebsocketTable;

    fn name(&self) -> &'static str;
    fn metadata(&self) -> arroyo_rpc::api_types::connections::Connector;
    fn from_config(&self, id: Option<i64>, name: &str, config: EmptyConfig,
        table: WebsocketTable, schema: Option<&ConnectionSchema>) -> anyhow::Result<Connection>;
    fn make_operator(&self, _: EmptyConfig, table: WebsocketTable,
        config: OperatorConfig) -> anyhow::Result<ConstructedOperator>;
}

Import

use arroyo_connectors::websocket::WebsocketConnector;

I/O Contract

Inputs

Name	Type	Required	Description
endpoint	String	Yes	WebSocket server URL (ws:// or wss://)
headers	Option<VarStr>	No	Comma-separated colon-delimited header key:value pairs
subscription_messages	Vec<SubscriptionMessage>	No	Messages to send after connection for channel subscription
format	Format	Yes	Deserialization format for incoming WebSocket messages

Outputs

Name	Type	Description
records	RecordBatch	Deserialized Arrow record batches from WebSocket messages

Usage Examples

CREATE TABLE ws_source (
    value TEXT
) WITH (
    connector = 'websocket',
    endpoint = 'wss://stream.example.com/feed',
    headers = 'Authorization:Bearer token123',
    subscription_message = '{"subscribe": "trades"}',
    format = 'json'
);

Related Pages

• Environment:ArroyoSystems_Arroyo_Rust_Runtime