Data Streams

Send and receive structured data streams with other participants. More...

Data Structures
struct	livekit_data_stream_header_t
	Header information about a data stream. More...
struct	livekit_data_stream_chunk_t
	A received chunk of data stream content. More...
struct	livekit_data_stream_trailer_t
	Trailer information when a data stream closes. More...
struct	livekit_data_stream_options_t
	Options for opening an outgoing data stream. More...
struct	livekit_data_stream_handler_t
	Handler for incoming data streams on a topic. More...

Typedefs
typedef void *	livekit_data_stream_handle_t
	Opaque handle to an open outgoing data stream.
typedef void(*	livekit_data_stream_open_cb_t) (const livekit_data_stream_header_t *header, void *ctx)
	Called when a new data stream is opened.
typedef void(*	livekit_data_stream_recv_cb_t) (const livekit_data_stream_chunk_t *chunk, void *ctx)
	Called for each received chunk of a data stream.
typedef void(*	livekit_data_stream_close_cb_t) (const livekit_data_stream_trailer_t *trailer, void *ctx)
	Called when a data stream is closed.

Functions
livekit_err_t	livekit_room_data_stream_topic_register (livekit_room_handle_t handle, const char *topic, const livekit_data_stream_handler_t *handler)
	Registers a handler for incoming data streams on a given topic.
livekit_err_t	livekit_room_data_stream_topic_unregister (livekit_room_handle_t handle, const char *topic)
	Unregisters a handler for incoming data streams on a given topic.
livekit_err_t	livekit_room_data_stream_open (livekit_room_handle_t handle, const livekit_data_stream_options_t *options, livekit_data_stream_handle_t *stream)
	Opens an outgoing data stream.
livekit_err_t	livekit_room_data_stream_write (livekit_room_handle_t handle, livekit_data_stream_handle_t stream, const uint8_t *data, size_t size)
	Writes data to an open outgoing data stream.
livekit_err_t	livekit_room_data_stream_close (livekit_room_handle_t handle, livekit_data_stream_handle_t stream)
	Closes an open outgoing data stream.

Detailed Description

Send and receive structured data streams with other participants.

A data stream delivers content as a sequence of events: open, recv/write (one or more), and close. Data is automatically chunked into LIVEKIT_DATA_STREAM_CHUNK_SIZE byte pieces.

The maximum number of concurrent streams is controlled by CONFIG_LK_MAX_DATA_STREAM_READERS and CONFIG_LK_MAX_DATA_STREAM_WRITERS in Kconfig (default 4 each).

Receiving

Register a handler for a topic using livekit_room_data_stream_topic_register. The handler's livekit_data_stream_handler_t::on_recv callback is invoked for each received chunk.

static void on_text_chunk(const livekit_data_stream_chunk_t* chunk, void* ctx)
{
    ESP_LOGI(TAG, "%.*s", (int)chunk->content_size, (const char*)chunk->content);
}
 
livekit_data_stream_handler_t handler = { .on_recv = on_text_chunk };
livekit_room_data_stream_topic_register(room_handle, "lk.chat", &handler);

Sending a Text Stream

livekit_data_stream_options_t opts = { .topic = "lk.chat", .is_text = true };
livekit_data_stream_handle_t stream;
livekit_room_data_stream_open(room_handle, &opts, &stream);
livekit_room_data_stream_write(room_handle, stream, (const uint8_t*)"hello ", 6);
livekit_room_data_stream_write(room_handle, stream, (const uint8_t*)"world", 5);
livekit_room_data_stream_close(room_handle, stream);

Sending a Byte Stream

uint8_t image_data[4096];
size_t image_size = read_image(image_data, sizeof(image_data));
 
livekit_data_stream_options_t opts = {
    .topic = "image",
    .is_text = false,
    .total_length = image_size,
    .has_total_length = true,
};
livekit_data_stream_handle_t stream;
livekit_room_data_stream_open(room_handle, &opts, &stream);
livekit_room_data_stream_write(room_handle, stream, image_data, image_size);
livekit_room_data_stream_close(room_handle, stream);

---

Data Structure Documentation

livekit_data_stream_header_t

struct livekit_data_stream_header_t

Header information about a data stream.

Data Fields
bool	has_total_length
bool	is_text	True if this is a text stream (UTF-8 chunks), false if byte stream (raw binary).
const char *	sender_identity
const char *	stream_id
int64_t	timestamp
const char *	topic
uint64_t	total_length

livekit_data_stream_chunk_t

struct livekit_data_stream_chunk_t

A received chunk of data stream content.

Data Fields
uint64_t	chunk_index
const uint8_t *	content
size_t	content_size
const char *	stream_id

livekit_data_stream_trailer_t

struct livekit_data_stream_trailer_t

Trailer information when a data stream closes.

Data Fields
const char *	reason	Empty string for normal closure, non-empty for abnormal end.
const char *	stream_id

livekit_data_stream_options_t

struct livekit_data_stream_options_t

Options for opening an outgoing data stream.

Data Fields
bool	has_total_length	Whether total_length is set.
bool	is_text	True for a text stream (UTF-8 chunks), false for a byte stream (raw binary).
const char *	topic	Topic to publish the stream under. Required.
uint64_t	total_length	Total size of the stream in bytes, if known upfront.

livekit_data_stream_handler_t

struct livekit_data_stream_handler_t

Handler for incoming data streams on a topic.

Data Fields
void *	ctx	User context passed to all callbacks.
livekit_data_stream_close_cb_t	on_close	Callback invoked when a stream is closed. Optional, can be NULL.
livekit_data_stream_open_cb_t	on_open	Callback invoked when a new stream is opened. Optional, can be NULL.
livekit_data_stream_recv_cb_t	on_recv	Callback invoked for each received chunk. Required.

Typedef Documentation

livekit_data_stream_close_cb_t

typedef void(* livekit_data_stream_close_cb_t) (const livekit_data_stream_trailer_t *trailer, void *ctx)

Called when a data stream is closed.

livekit_data_stream_handle_t

typedef void* livekit_data_stream_handle_t

Opaque handle to an open outgoing data stream.

livekit_data_stream_open_cb_t

typedef void(* livekit_data_stream_open_cb_t) (const livekit_data_stream_header_t *header, void *ctx)

Called when a new data stream is opened.

livekit_data_stream_recv_cb_t

typedef void(* livekit_data_stream_recv_cb_t) (const livekit_data_stream_chunk_t *chunk, void *ctx)

Called for each received chunk of a data stream.

Function Documentation

livekit_room_data_stream_close()

livekit_err_t livekit_room_data_stream_close	(	livekit_room_handle_t	handle,
		livekit_data_stream_handle_t	stream )

Closes an open outgoing data stream.

Sends the stream trailer and releases the stream slot.

Parameters

handle[in]	Room handle.
stream[in]	Stream handle from livekit_room_data_stream_open.

Returns

LIVEKIT_ERR_NONE if successful, otherwise an error code.

livekit_room_data_stream_open()

livekit_err_t livekit_room_data_stream_open	(	livekit_room_handle_t	handle,
		const livekit_data_stream_options_t *	options,
		livekit_data_stream_handle_t *	stream )

Opens an outgoing data stream.

Sends the stream header and returns a stream handle for writing.

Parameters

handle[in]	Room handle.
options[in]	Stream options (topic, type, optional total length).
stream[out]	Stream handle for subsequent write/close calls.

Returns

LIVEKIT_ERR_NONE if successful, otherwise an error code.

livekit_room_data_stream_topic_register()

livekit_err_t livekit_room_data_stream_topic_register	(	livekit_room_handle_t	handle,
		const char *	topic,
		const livekit_data_stream_handler_t *	handler )

Registers a handler for incoming data streams on a given topic.

Parameters

handle[in]	Room handle.
topic[in]	Topic to handle.
handler[in]	Handler callbacks. The on_recv field is required; on_open and on_close are optional and may be NULL. The struct is copied internally.

Returns

LIVEKIT_ERR_NONE if successful, otherwise an error code.

livekit_room_data_stream_topic_unregister()

livekit_err_t livekit_room_data_stream_topic_unregister	(	livekit_room_handle_t	handle,
		const char *	topic )

Unregisters a handler for incoming data streams on a given topic.

Parameters

handle[in]	Room handle.
topic[in]	Topic to unregister.

Returns

LIVEKIT_ERR_NONE if successful, otherwise an error code.

livekit_room_data_stream_write()

livekit_err_t livekit_room_data_stream_write	(	livekit_room_handle_t	handle,
		livekit_data_stream_handle_t	stream,
		const uint8_t *	data,
		size_t	size )

Writes data to an open outgoing data stream.

Data is automatically chunked into LIVEKIT_DATA_STREAM_CHUNK_SIZE byte pieces. Can be called multiple times.

Parameters

handle[in]	Room handle.
stream[in]	Stream handle from livekit_room_data_stream_open.
data[in]	Data to write.
size[in]	Size of data in bytes.

Returns

LIVEKIT_ERR_NONE if successful, otherwise an error code.