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, aborting every still-open stream as [`InputEvent::AbortAllStreams`]
51    /// does.
52    Shutdown,
53}
54
55/// A new stream was opened; its reader should be delivered to the application (or routed
56/// internally for reserved topics). Carries the sender's identity.
57pub struct StreamOpened {
58    pub stream_reader: AnyStreamReader,
59    pub participant_identity: ParticipantIdentity,
60}
61
62/// A stream previously announced via [`StreamOpened`] has terminated and will produce no further
63/// data: its trailer arrived, its inline payload completed, it failed with an error, or it was
64/// aborted.
65///
66/// Emitted exactly once per opened stream. Hosts delivering streams on ordered topics use this to
67/// know when a stream's handler can be considered finished on the wire (a trailer alone is not
68/// enough: inline single-packet streams never receive one).
69pub struct StreamClosed {
70    pub stream_id: StreamId,
71    pub participant_identity: ParticipantIdentity,
72    /// Topic the stream was opened on.
73    pub topic: String,
74}
75
76/// A "raw chunk received" notification, which is used to trigger
77/// the deprecated [RoomEvent:::StreamChunkReceived] event.
78pub struct ChunkReceived {
79    pub chunk: Chunk,
80    pub participant_identity: ParticipantIdentity,
81
82    /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could
83    /// not be mapped to a topic.
84    pub topic: Option<String>,
85}
86
87/// A "raw trailer received" notification, which is used to trigger
88/// the deprecated [RoomEvent:::StreamTrailerReceived] event.
89pub struct TrailerReceived {
90    pub trailer: Trailer,
91    pub participant_identity: ParticipantIdentity,
92
93    /// Topic of the stream this chunk belongs to, or `None` if the associated stream id could
94    /// not be mapped to a topic.
95    ///
96    /// See [`ChunkReceived::topic`].
97    pub topic: Option<String>,
98}
99
100/// An event emitted by [`IncomingStreamManager::run`] for the host crate to surface. The manager
101/// stays decoupled from `RoomEvent`; the host maps these onto its own event types.
102#[derive(FromVariants)]
103pub enum OutputEvent {
104    StreamOpened(StreamOpened),
105    StreamClosed(StreamClosed),
106    ChunkReceived(ChunkReceived),
107    TrailerReceived(TrailerReceived),
108}
livekit_data_stream/incoming/events.rs

livekit_data_stream/incoming/
events.rs
