WebSocket Server (Source)

Description

WebSocket Server (Source) exposes a WebSocket endpoint that publishers connect to, and turns every message they send into a single event in the scenario.

Use it when the publisher wants to push data to Nussknacker over a long-lived connection. When the data lives behind someone else's endpoint that Nussknacker should connect to, use WebSocket Client instead.

The connection details - the URL to connect to and the credentials to use - are shown in the node itself. The URL is derived from the node identifier, which is assigned once when the node is created, so renaming the node does not change the URL publishers have to use.

Parameters and configuration

Name	Description
Content type	Format of the incoming messages. Choose JSON for structured JSON messages or PLAIN for raw text messages.

Advanced parameters

Name	Description
Data sample	Available when Content type is JSON. An example JSON object representing a typical message. Nussknacker analyzes its structure to determine field types and enable type-aware field access in subsequent nodes. Does not affect runtime behavior - only used in the designer for type inference.
Event time	Expression which evaluates to the time when the event was created. For example, if the incoming message contains a timestamp field you can use #input.timestamp. When not specified, processing time is used.
Max out-of-orderness	The maximum amount of time an element is allowed to be late before being ignored when computing the result for time-based stream transformations: aggregates in time windows and joins. To read more about this mechanism see Flink documentation.
Idleness	The time period after which the source is marked as idle if no events are received. To read more about this mechanism see Flink documentation.

Publishing messages

A publisher opens a WebSocket connection to the URL shown in the node and authenticates with Basic Auth during the handshake. Every text or binary frame it sends afterwards is one message - the frame carries the message itself, with nothing to wrap it in.

The node window shows the whole command for websocat, credentials included, so a connection can be opened from a terminal without writing any code:

websocat --basic-auth 'publisher:secret' 'wss://example.nussknacker.io/publish/ws-a1b2c3d4'

Each message is answered once it is safely stored:

{"type": "ack", "seq": 1}

where seq counts the messages sent on this connection, starting at 1. A message that cannot be accepted is answered with an error instead:

{"type": "error", "seq": 1, "code": "RATE_LIMITED", "message": "..."}

RATE_LIMITED means the instance throughput budget is exhausted and TOO_LARGE means the message is over the size limit of the topic. Both leave the connection open, so the message can simply be sent again.

Messages are accepted one at a time and in order, so a publisher that sends faster than Nussknacker can store is slowed down rather than silently losing messages.

Additional considerations

• When Content type is PLAIN: #input is a String containing the raw message text.
• When Content type is JSON: #input is a JSON object. If a Data sample was provided, field types are inferred from the sample and named fields can be accessed directly (e.g. #input.price). Without a data sample, dynamic navigation can still be used (e.g. #input?.["price"].).
• The topic behind this source keeps accepting messages while the scenario is stopped, but they are not delivered: on every deployment the source starts from the newest message, so anything published before it, including while the scenario was stopped, is skipped rather than replayed.
• A Kafka topic backing this source is created automatically when the scenario is deployed.
• The Data sample is only used during scenario editing to provide type hints in the Designer. It has no effect on how messages are deserialized at runtime - all incoming JSON messages are processed in a schema-less manner regardless of the sample.

Event time

The Kafka event timestamp is the default value suggested by Designer for the Event time field. If you do not use this default, read here to learn more about Instant data type. If provided as a numeric value rather than Instant, it must be in epoch milliseconds (not seconds); a Long is the most natural data type for representing this UTC millisecond value.

caution

Getting event time wrong can lead to subtle errors in time-based logic that are difficult to diagnose - if such logic behaves unexpectedly, this should be the first thing to verify.