livekit_data_stream/incoming/
events.rs

1// Copyright 2026 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use from_variants::FromVariants;
16use livekit_common::ParticipantIdentity;
17
18use crate::{
19    incoming::AnyStreamReader,
20    types::{Chunk, Packet, StreamId, Trailer},
21};
22
23pub struct PacketReceived {
24    pub packet: Packet,
25    pub participant_identity: ParticipantIdentity,
26}
27
28impl PacketReceived {
29    pub fn new(packet: Packet, participant_identity: ParticipantIdentity) -> Self {
30        Self { packet, participant_identity }
31    }
32}
33
34/// An event fed into [`IncomingStreamManager::run`] by the host crate. Each corresponds to an
35/// inbound data-stream packet (or a lifecycle signal) and carries everything the manager needs to
36/// process it without reaching back into room state.
37#[derive(FromVariants)]
38#[allow(clippy::large_enum_variant)]
39#[non_exhaustive]
40pub enum InputEvent {
41    PacketReceived(PacketReceived),
42    /// Abort every open stream sent by this participant (they disconnected mid-send).
43    AbortStreamsFrom(ParticipantIdentity),
44    /// Abort every open stream (e.g. the local connection is going away). Unlike
45    /// [`InputEvent::Shutdown`], the run loop keeps going so streams opened later are still handled.
46    AbortAllStreams,
47    /// Reply with the number of currently open streams (registered by a header and awaiting more
48    /// packets). Processed in order with the other events, so the answer reflects everything
49    /// enqueued before it.
50    #[from_variants(skip)]
51    QueryOpenStreamCount(tokio::sync::oneshot::Sender<usize>),
52    /// Stop the run loop, aborting every still-open stream as [`InputEvent::AbortAllStreams`]
53    /// does.
54    Shutdown,
55}
56
57/// A new stream was opened; its reader should be delivered to the application (or routed
58/// internally for reserved topics). Carries the sender's identity.
59pub struct StreamOpened {
60    pub stream_reader: AnyStreamReader,
61    pub participant_identity: ParticipantIdentity,
62}
63
64/// A stream previously announced via [`StreamOpened`] has terminated and will produce no further
65/// data: its trailer arrived, its inline payload completed, it failed with an error, or it was
66/// aborted.
67///
68/// Emitted exactly once per opened stream. Hosts delivering streams on ordered topics use this to
69/// know when a stream's handler can be considered finished on the wire (a trailer alone is not
70/// enough: inline single-packet streams never receive one).
71pub struct StreamClosed {
72    pub stream_id: StreamId,
73    pub participant_identity: ParticipantIdentity,
74    /// Topic the stream was opened on.
75    pub topic: String,
76}
77
78/// A "raw chunk received" notification, which is used to trigger
79/// the deprecated [RoomEvent:::StreamChunkReceived] event.
80pub struct ChunkReceived {
81    pub chunk: Chunk,
82    pub participant_identity: ParticipantIdentity,
83
84    /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could
85    /// not be mapped to a topic.
86    pub topic: Option<String>,
87}
88
89/// A "raw trailer received" notification, which is used to trigger
90/// the deprecated [RoomEvent:::StreamTrailerReceived] event.
91pub struct TrailerReceived {
92    pub trailer: Trailer,
93    pub participant_identity: ParticipantIdentity,
94
95    /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could
96    /// not be mapped to a topic.
97    ///
98    /// See [`ChunkReceived::topic`].
99    pub topic: Option<String>,
100}
101
102/// An event emitted by [`IncomingStreamManager::run`] for the host crate to surface. The manager
103/// stays decoupled from `RoomEvent`; the host maps these onto its own event types.
104#[derive(FromVariants)]
105#[non_exhaustive]
106pub enum OutputEvent {
107    StreamOpened(StreamOpened),
108    StreamClosed(StreamClosed),
109    ChunkReceived(ChunkReceived),
110    TrailerReceived(TrailerReceived),
111}
livekit_data_stream/incoming/events.rs

livekit_data_stream/incoming/
events.rs
