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)]
38pub enum InputEvent {
39    PacketReceived(PacketReceived),
40    /// Abort every open stream sent by this participant (they disconnected mid-send).
41    AbortStreamsFrom(ParticipantIdentity),
42    /// Abort every open stream (e.g. the local connection is going away). Unlike
43    /// [`InputEvent::Shutdown`], the run loop keeps going so streams opened later are still handled.
44    AbortAllStreams,
45    /// Reply with the number of currently open streams (registered by a header and awaiting more
46    /// packets). Processed in order with the other events, so the answer reflects everything
47    /// enqueued before it.
48    #[from_variants(skip)]
49    QueryOpenStreamCount(tokio::sync::oneshot::Sender<usize>),
50    /// Stop the run loop.
51    Shutdown,
52}
53
54/// A new stream was opened; its reader should be delivered to the application (or routed
55/// internally for reserved topics). Carries the sender's identity.
56pub struct StreamOpened {
57    pub stream_reader: AnyStreamReader,
58    pub participant_identity: ParticipantIdentity,
59}
60
61/// A stream previously announced via [`StreamOpened`] has terminated and will produce no further
62/// data: its trailer arrived, its inline payload completed, it failed with an error, or it was
63/// aborted.
64///
65/// Emitted exactly once per opened stream. Hosts delivering streams on ordered topics use this to
66/// know when a stream's handler can be considered finished on the wire (a trailer alone is not
67/// enough: inline single-packet streams never receive one).
68pub struct StreamClosed {
69    pub stream_id: StreamId,
70    pub participant_identity: ParticipantIdentity,
71    /// Topic the stream was opened on.
72    pub topic: String,
73}
74
75/// A "raw chunk received" notification, which is used to trigger
76/// the deprecated [RoomEvent:::StreamChunkReceived] event.
77pub struct ChunkReceived {
78    pub chunk: Chunk,
79    pub participant_identity: ParticipantIdentity,
80
81    /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could
82    /// not be mapped to a topic.
83    pub topic: Option<String>,
84}
85
86/// A "raw trailer received" notification, which is used to trigger
87/// the deprecated [RoomEvent:::StreamTrailerReceived] event.
88pub struct TrailerReceived {
89    pub trailer: Trailer,
90    pub participant_identity: ParticipantIdentity,
91
92    /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could
93    /// not be mapped to a topic.
94    ///
95    /// See [`ChunkReceived::topic`].
96    pub topic: Option<String>,
97}
98
99/// An event emitted by [`IncomingStreamManager::run`] for the host crate to surface. The manager
100/// stays decoupled from `RoomEvent`; the host maps these onto its own event types.
101#[derive(FromVariants)]
102pub enum OutputEvent {
103    StreamOpened(StreamOpened),
104    StreamClosed(StreamClosed),
105    ChunkReceived(ChunkReceived),
106    TrailerReceived(TrailerReceived),
107}
livekit_data_stream/incoming/events.rs

livekit_data_stream/incoming/
events.rs
