feat: Data Tracks (reviewable commit stack of #1004) - #1019
Draft
MaxHeimbrock wants to merge 47 commits into
Draft
MaxHeimbrock wants to merge 47 commits into
MaxHeimbrock wants to merge 47 commits into
Conversation
This commit adds the agentAttributes field to TokenRequestOptions. The field is a map of strings. The token server sets these attributes on the agent participant that it dispatches. The RoomAgentDispatch request body gets the matching attributes field. The toRequest function now compares the dispatch object with a default instance. If the caller sets no dispatch field, the request omits the agents list. Before this change, the function checked each field by hand. The new check stays correct when new fields are added. This change is not related to data tracks. It can move to a separate pull request.
This commit extends the TokenSource test. The test sets an agent attribute in the options. The test then reads the JSON request body. The test checks that the attributes object contains the value. The commit also updates the copyright year of the test file.
This commit moves the protocol submodule from 8381f218 to 2172178d. The new protocol version adds the messages that data tracks need: - DataTrackSchemaId and DataTrackSchemaEncoding - DataBlobKey and DataBlob - StoreDataBlobRequest and StoreDataBlobResponse - GetDataBlobRequest and GetDataBlobResponse - The publish_data_tracks field on SyncState The new version also contains many changes to the agent, SIP, egress, and ingress protos. The Android SDK does not use those changes. Review only that the generated code compiles. This commit contains no other change.
This commit adds the livekit-uniffi-android library. The library contains the Rust data track core and its Kotlin bindings. The commit makes these changes: - The version catalog gets the livekit-uniffi entry. - The SDK module and the test module depend on the library. - The SDK manifest gets a tools:overrideLibrary entry for io.livekit.uniffi. The manifest entry is the important part. The library declares minSdk 24. The SDK declares minSdk 21. Without the override, the manifest merger rejects the build. With the override, the SDK ships a library that is not tested on API 21 to 23. On those API levels the native library can fail to load. A later commit adds code that catches that failure and disables data tracks. Decide if this trade-off is acceptable before the release.
This commit adds two value types. Both types have no dependency on other SDK code. DataTrackSid wraps the identifier that the server gives to a data track. The identifier is not stable across a full reconnect of the publisher. The documentation tells callers to use the track name as the stable key. DataTrackFrame holds the payload of one frame and an optional user timestamp. The SDK does not read the timestamp. The now() function creates a frame with the current time in milliseconds. The durationSinceTimestamp property reads the timestamp as a Unix time in milliseconds. The class has converters to and from the UniFFI frame type. The class also has equals and hashCode based on the payload content.
This commit adds three public types in DataTrackSchema.kt. DataTrackSchemaId names a schema and gives the encoding of the schema definition. The type has converters to the UniFFI type and to the protobuf type. The blobKey property builds the DataBlobKey that stores the schema definition on the server. DataTrackSchemaEncoding lists the well-known encodings of a schema definition. DataTrackFrameEncoding lists the well-known encodings of the frames on a track. Both are sealed classes with a Custom case. The fromIdentifier function maps a string to a case. A well-known identifier always maps to the well-known case. A custom encoding cannot use a well-known identifier. The commit is larger than the target size. All content is a mechanical mapping between the SDK types, the UniFFI types, and the protobuf types. Check the identifier strings against rust-sdks and the Swift SDK. The strings must be the same in all SDKs.
This commit adds DataTrackSchemaTest. One test checks that the blob key of a well-known schema encoding carries the name and the protobuf enum value. One test checks that a custom encoding uses the custom string field of the protobuf message.
This commit adds two public types. DataTrackInfo describes a published data track. It carries the SID, the name, the E2EE flag, the optional schema, and the optional frame encoding. The internal constructor converts the UniFFI type. DataTrackPublishOptions holds the options for a publish call. The frameFormat field is optional. DataTrackFrameFormat groups a frame encoding with an optional schema. The type makes the frame encoding mandatory when a schema is given. A schema always describes frames in one encoding. The secondary constructor of DataTrackPublishOptions accepts the encoding and the schema directly.
…ypes This commit adds four sealed exception hierarchies in DataTrackException.kt: - DataTrackPublishException for publishDataTrack failures - DataTrackPushFrameException for tryPush failures - DataTrackSubscribeException for subscribe failures - DataTrackSchemaException for defineSchema and getSchema failures The QueueFull case of DataTrackPushFrameException carries the rejected frame. The caller can retry the same frame. The commit also adds three internal toSdk functions. They map the UniFFI error types to the SDK types one to one. The mapping is mechanical.
This commit adds two functional interfaces: LocalDataTrackManagerFactory and RemoteDataTrackManagerFactory. Each interface creates one UniFFI manager from a delegate and an optional cryptor. RTCModule provides both factories. The production factories create the real UniFFI LocalDataTrackManager and RemoteDataTrackManager. The interfaces let the tests replace the native managers with fakes. The native library is not loaded in unit tests.
This commit adds test doubles for the UniFFI managers. The doubles do not load native code. MockLocalDataTrackManagerFactory records the encryption provider of the last create call. The factory can throw a LinkageError on create. The factory can make every publish fail with a given PublishException. MockLocalDataTrackManager records the SFU responses it receives. It sends a PublishDataTrackRequest through the delegate on publish. It counts republishTracks calls. MockFfiLocalDataTrack records the pushed frames. MockRemoteDataTrackManagerFactory records the decryption provider of the last create call. The factory can throw a LinkageError on create. MockRemoteDataTrackManager records the join responses, participant updates, subscriber handles, and packets it receives. It counts resendSubscriptionUpdates calls. The simulateTrackPublished and simulateTrackUnpublished functions fire the delegate callbacks. MockFfiRemoteDataTrack does not support subscribe. This commit is larger than the target size. All content is test scaffolding.
This commit wires the mock factories into the test Dagger graph. TestRTCModule provides the mock factories as singletons. It also binds them to the production factory interfaces. TestLiveKitComponent exposes both mock factories. MockE2ETest stores both factories in fields, so tests can inspect the managers that the room creates. The commit also updates the copyright year of the changed files.
…rain This commit adds DataTrackFrameSender. The class sends the packets of data track frames to a channel. It is pure Kotlin and does not use WebRTC classes. DataTrackSendChannel is the interface to the channel. DataChannelManagerSendChannel implements it on top of a DataChannelManager. The interface makes the sender testable without a peer connection. The sender holds at most one queued frame. A new frame replaces the queued frame. The sender logs the replaced frame. The sender sends all packets of the current frame before it starts the next frame. Packets of two frames never mix. The sender sends packets only while the buffered amount of the channel is at or below LOW_WATER_MARK. The mark is 8 KiB. It is the same value as in rust-sdks. The owner calls pump() on each buffered-amount and state change to continue the drain. The design is the same as in rust-sdks and the Swift SDK. It is different from client-sdk-js. The JS SDK blocks the producer when the buffer is full. The Android producer is a callback from the native code. It cannot block. The sender drops the oldest queued frame instead. Review the drop policy with care. The policy decides which frames the subscribers do not receive under load.
This commit adds DataTrackFrameSenderTest. FakeSendChannel is a channel that records sent packets and simulates the buffer. The tests check these rules: - The sender sends immediately when the channel has headroom. - A large frame streams out over many drains without a size limit. - A newer frame replaces the queued frame. - An in-flight frame completes before a newer frame starts. - attach() drops the frames queued for the old channel. - A rejected send drops the rest of the frame and does not block the pump. - An empty packet list does not replace the queued frame. - A closed channel holds the frame until the channel opens.
…frames into it This commit adds DataTrackPublisherChannel and connects it to RTCEngine. DataTrackPublisherChannel owns the publisher-side _data_track transport. It holds one DataTrackFrameSender for the life of the session. attach() gives the channel a new DataChannelManager and starts a pump job. The pump job watches the buffered amount and the state of the channel. detach() stops the pump, drops queued frames, and disposes the manager. sendPackets() queues packets on the RTC thread. awaitOpen() polls until the current manager is open, the session is closed, or the timeout expires. RTCEngine gets these changes: - The DATA_TRACK_DATA_CHANNEL_LABEL constant names the channel "_data_track". - configure() creates the channel on the publisher peer connection after the reliable and lossy channels. The channel is unordered and has zero retransmits. - closeResources() detaches the transport. - sendDataTrackPackets() forwards packets to the transport. A full reconnect replaces the DataChannelManager but keeps the frame sender. A publish that waits in awaitOpen() sees the replacement channel. Review the channel options and the reconnect behavior with care.
This commit adds ensureDataTrackPublisherConnected() to RTCEngine. A publish must not start before the _data_track channel is open. The frame sender queues at most one frame while the channel is not open. The function first checks the publisher peer connection. In subscriber-primary mode the publisher connection is created on demand. If the publisher is not connected and ICE is not checking, the function calls negotiatePublisher(). This also sets hasPublished. The function then waits for the channel with awaitOpen(). It uses MAX_ICE_CONNECT_TIMEOUT_MS as the timeout. If the session closes during the wait, the function throws DataTrackPublishException.Disconnected. If the timeout expires, the function throws DataTrackPublishException.Timeout. Review the negotiation condition with care. It starts a publisher negotiation as a side effect of a publish call.
…responses This commit changes SignalClient so the raw websocket bytes travel with each decoded response. The UniFFI data track managers parse the signal messages themselves. If the SDK decoded and re-encoded a message, the protobuf library would drop fields that the SDK protocol version does not know. The raw bytes avoid that problem. The commit makes these changes: - The private IncomingSignal class replaces the Pair in the response flow. It holds the websocket, the response, and the bytes. - handleSignalResponse() and handleSignalResponseImpl() get an encoded parameter. - onMessage() passes the received bytes through. - lastJoinEncoded keeps the bytes of the last successful Join response. - Listener.onParticipantUpdate() gets an encoded parameter. RTCEngine updates its override. It does not use the bytes yet. The commit does not change the behavior of the existing message handling.
This commit replaces four TODO comments in SignalClient with listener calls. The Listener interface gets four methods with empty default bodies: - onPublishDataTrackResponse - onUnpublishDataTrackResponse - onRequestResponse - onDataTrackSubscriberHandles Each method receives the encoded bytes of the full SignalResponse. The UniFFI managers decode the bytes. The REQUEST_RESPONSE case forwards every request response. The managers ignore the responses that are not about data tracks. RTCEngine does not override the methods yet.
This commit adds the path for signal requests that the native managers build. SignalClient.sendEncodedRequest() parses the bytes as a SignalRequest and sends it through the normal queue. If the bytes do not parse, the function logs an error and returns. It does not throw. The native managers call this function from a callback. An exception here would unwind through the FFI boundary. RTCEngine.sendDataTrackSignalRequest() forwards the bytes to the client. If nothing is published yet, it calls negotiatePublisher() first. Data track signaling needs the publisher peer connection and the _data_track channel.
This commit extends SignalClientTest. The existing join test now checks that lastJoinEncoded equals the bytes of the Join response. Two new tests cover sendEncodedRequest(). One test sends valid bytes and checks that the request reaches the websocket. One test sends invalid bytes and checks that nothing is sent.
This commit adds the public LocalDataTrack class. The class wraps the UniFFI LocalDataTrack. The commit does not include the Flow-based send function. A later commit adds it. The class exposes isPublished, info, tryPush(), unpublish(), and waitForUnpublish(). tryPush() returns a Result. It maps a UniFFI PushFrameErrorReason to DataTrackPushFrameException. The class implements the internal DataTrackFrameSink interface. The interface is the seam that a later commit uses to test the send policy. Review the second catch block in tryPush() with care. The UniFFI bindings cannot decode the PushFrameErrorReason type, because a different UniFFI component defines it. The bindings report an internal error instead. The catch block infers the real cause. If the track is still published, the cause is a full queue. If the track is not published, the cause is the unpublish. This is a workaround for a bindings limitation.
…I local manager This commit adds OutgoingDataTrackManager. The class owns the UniFFI LocalDataTrackManager and connects it to RTCEngine. The class is a Dagger singleton. It gets the engine through a Provider to avoid a dependency cycle. The delegate receives two callbacks from the native manager. onSignalRequest forwards the request bytes to the engine. onPacketsAvailable forwards the packets to the engine. publishTrack() first calls ensureDataTrackPublisherConnected() on the engine. It then creates the native manager on demand and publishes the track. Every failure returns a typed DataTrackPublishException in a Result. The handle functions forward SFU responses to the native manager. The unpublish response is not consumed. The native manager does not use it. republishTracks() and publishResponsesForSyncState() support reconnect. close() disposes the native manager. ensureManager() creates the native manager once. If the native library fails to load, the class remembers the failure. It does not retry. Every later publish fails with DataTrackPublishException.Internal. This commit does not contain E2EE code. The factory receives a null encryption provider. A later commit adds encryption.
This commit injects OutgoingDataTrackManager into RTCEngine. The engine closes the manager in close(). The engine overrides three listener methods. onPublishDataTrackResponse, onUnpublishDataTrackResponse, and onRequestResponse forward the encoded bytes to the manager.
…ataTrack This commit adds the public API for publishing a data track. publishDataTrack() checks that the engine is connected. It then delegates to OutgoingDataTrackManager. The function returns a Result with the LocalDataTrack or a DataTrackPublishException. withDataTrack() publishes a track for the duration of a block. It unpublishes the track when the block returns, throws, or is cancelled. LocalParticipant gets the manager through its constructor. Review the API shape and the documentation with care. The documentation says that a dropped track is unpublished only when the garbage collector runs. Callers must call unpublish() for a predictable end of the publication.
This commit adds OutgoingDataTrackManagerMockE2ETest with eight tests. The tests check these behaviors: - publishDataTrack() uses the injected manager and passes no encryption provider. - A native library failure returns DataTrackPublishException.Internal. - An InvalidSchema error from the core reaches the caller as a typed failure. - publishDataTrack() waits until the _data_track channel is open. - publishDataTrack() fails with Timeout if the channel never opens. - publishDataTrack() fails with Disconnected if the room disconnects during the wait. - Packets wait for the low-water mark and then send. - A newer frame replaces the queued frame. The publisherDataTrackChannel() helper finds the mock channel on the publisher peer connection.
…Flow This commit adds the send() function to LocalDataTrack. The function pushes every frame from a Flow until the Flow ends or the track is unpublished. The FrameDropPolicy enum selects the behavior when the send queue is full. DROP skips the frame. FAIL ends the send with DataTrackPushFrameException.QueueFull. The sendFrames() and sendOne() extension functions on DataTrackFrameSink contain the logic. An unpublish during the send ends the send with success. The documentation describes this outcome.
This commit adds LocalDataTrackSendTest. RecordingSink is a DataTrackFrameSink that rejects frames on demand. It can also unpublish itself after a number of frames. The tests check these behaviors: - DROP skips the rejected frame and sends the rest. - FAIL stops at the rejected frame and returns it in the exception. - An unpublish during the send ends the send with success under both policies. - An unpublished track sends nothing.
…quests This commit adds request and response correlation for data blobs to SignalClient. sendStoreDataBlob() stores a blob under a key. sendGetDataBlob() reads a blob that another participant stored. Both functions return a Result with a DataTrackSchemaException on failure. sendIdCorrelatedRequest() contains the shared logic. It takes a request id from an AtomicInteger. It registers a CompletableDeferred in a map. It sends the request and waits for the response with a deadline of 5 seconds. A timeout returns DataTrackSchemaException.Timeout. The response handling has three parts: - STORE_DATA_BLOB_RESPONSE completes the deferred with an empty array. - GET_DATA_BLOB_RESPONSE completes the deferred with the blob contents. - REQUEST_RESPONSE with a reason other than OK or QUEUED fails the deferred with DataTrackSchemaException.Rejected. close() fails every pending request with DataTrackSchemaException.Disconnected. Review the failure reasons and the timeout with care. The values are not configurable.
This commit adds three tests to SignalClientTest. One test stores a blob and checks that the response completes the request. One test reads a blob and checks that the contents come back. One test sends a RequestResponse with reason NOT_FOUND and checks that the request fails with DataTrackSchemaException.Rejected and the server message.
This commit adds the public schema API to LocalParticipant. defineSchema() stores a schema definition on the server under its DataTrackSchemaId. getSchema() reads the definition that another participant stored. Both functions check that the engine is connected. Both return a Result with a DataTrackSchemaException on failure. getSchema() decodes the stored bytes as strict UTF-8. Invalid bytes return DataTrackSchemaException.InvalidDefinition. The SDK does not parse or validate the definition against its encoding.
This commit adds DataTrackSchemaMockE2ETest. The test calls defineSchema() before the room connects. The test checks that the result is DataTrackSchemaException.Disconnected.
…or flow This commit adds the public DataTrackStream class. The class wraps a UniFFI DataTrackStreamInterface. A subscription creates one stream. next() returns the next frame or null when the stream ends. flow is a Flow of frames. The flow completes when the stream ends or when close() is called. The class drains the native stream in one coroutine. The sharedFrames flow uses shareIn with WhileSubscribed. Every collector receives every frame that arrives while it collects. A late collector does not receive old frames. A slow collector delays the drain for all collectors. The ended flag marks the end of the stream. The public flow merges the shared frames with the ended flag. A collector that starts after the end completes at once. close() sets the flag before it cancels the drain. If it cancelled the drain first, the collectors would not complete. This is the final form of the fix for frames that did not reach all collectors. Review the shareIn and the ended flag logic with care.
This commit adds DataTrackStreamTest. FakeFfiStream is a stream that the test feeds from a Channel. The tests check these behaviors: - Two concurrent collectors each receive every frame. - A collector completes when the stream ends. - A later collector resumes the drain after an earlier collector stopped. - close() completes an active collector. - A collector that starts after the end completes at once.
This commit adds the public RemoteDataTrack class. The class wraps the UniFFI RemoteDataTrack. The class exposes publisherIdentity, name, isPublished, info, and waitForUnpublish(). The name is the stable key across reconnects. The SID is not stable. subscribe() takes a buffer size in frames and returns a Result with a DataTrackStream. Values below 1 are raised to 1. The default is 16 frames. A UniFFI DataTrackSubscribeException maps to the SDK exception type. More than one subscribe call on the same track is allowed. All streams share one pipeline.
…I remote manager This commit adds IncomingDataTrackManager and IncomingDataTrackEvent. The manager owns the UniFFI RemoteDataTrackManager and connects it to RTCEngine. It is a Dagger singleton. IncomingDataTrackEvent has two cases. TrackPublished carries a new RemoteDataTrack. TrackUnpublished carries the SID and the track. The manager posts the events on an internal event bus. The delegate receives callbacks from the native manager. onSignalRequest forwards the bytes to the engine. onTrackPublished wraps the track, stores it in a list, and posts an event. onTrackUnpublished removes every track with that SID and posts an event for each one. The manager keeps every known track in a list. snapshotRemoteTracks() returns a copy. The room uses the snapshot to attach tracks whose publisher joined after the track was published. The handle functions forward join responses, participant updates, subscriber handles, and packets to the native manager. handlePacketReceived() runs on a WebRTC thread. It catches every exception. An exception on that thread would stop the process. ensureManager() creates the native manager once. If the native library fails to load, the class remembers the failure. Every entry point then does nothing. The room stays usable for apps that do not use data tracks. This commit does not contain E2EE code. The factory receives a null decryption provider. A later commit adds decryption.
…to IncomingDataTrackManager This commit injects IncomingDataTrackManager into RTCEngine and feeds it. The engine makes these changes: - close() closes the manager and clears the local identity. - joinImpl() stores the local participant identity from the join response. - joinImpl() forwards the raw join bytes to the manager after the room has processed the participants. If the raw bytes are missing, it re-encodes the join response. It then asks the listener to reattach remote data tracks. - onParticipantUpdate() forwards the raw bytes and the local identity to the manager. It then asks the listener to reattach remote data tracks. - The subscriber peer connection stores the _data_track channel that the SFU opens. - onMessage() routes packets from that channel to the manager. Other channels keep the existing DataPacket path. - onDataTrackSubscriberHandles() forwards the bytes to the manager. - The Listener interface gets reattachRemoteDataTracks() with an empty default body. The order in joinImpl() is important. The listener processes the participants first. The manager processes the join bytes second. The native manager discovers tracks only after the publishers are known. This is the same order as in the Swift SDK.
This commit adds IncomingDataTrackManagerMockE2ETest with three tests. The tests check these behaviors: - A packet on the subscriber _data_track channel reaches the mock remote manager. - connect() forwards the raw join bytes to the mock remote manager. - A native library failure does not break the room. The room connects, a participant joins, and a received packet is dropped without an exception. The helpers open a mock subscriber channel, deliver a packet on it, and find the remote participant.
This commit adds the dataTracks map to RemoteParticipant and the bookkeeping behind it. RemoteDataTrackCollection owns the observable map. The map is keyed by track name. The SID of a track changes when the publisher does a full reconnect. The name does not change. The collection has these operations: - add() attaches a track. It returns false if the same instance is already attached. It fires onPublished on success. - remove() detaches the track with a SID. - unpublish() detaches the track and always fires onUnpublished. This also covers a track that a full reconnect already detached. - unpublishAll() detaches every track and fires onUnpublished for each SID. - detachAll() drops every track without an event. RemoteParticipant creates the collection. The callbacks post ParticipantEvent.DataTrackPublished and ParticipantEvent.DataTrackUnpublished. The public dataTracks property is observable with the flow extension. Five internal extension functions on RemoteParticipant give the room access to the collection. The commit adds the two ParticipantEvent cases. It also updates the copyright year of that file. Review the name-based key and the identity-based duplicate check with care.
This commit connects IncomingDataTrackManager to Room and adds two room events. RoomEvent.DataTrackPublished fires when a remote participant publishes a data track. RoomEvent.DataTrackUnpublished fires when the track is unpublished. Room gets these changes: - The constructor receives IncomingDataTrackManager. - connect() starts a collector for the manager events. - A TrackPublished event attaches the track to its participant. If the participant is not in the room yet, the room logs and waits. - A TrackUnpublished event detaches the track and posts the room event. - reattachRemoteDataTracks() attaches every track from the manager snapshot. The engine calls it after a join and after each participant update. - handleParticipantDisconnect() unpublishes the data tracks of the participant and posts one room event per track. - The participant event collector forwards ParticipantEvent.DataTrackPublished as a room event. The commit also adds a documentation comment to the localParticipant property. RoomTest mocks the new constructor parameter.
This commit adds four tests to IncomingDataTrackManagerMockE2ETest. The tests check these behaviors: - A published track attaches to its participant. The room and the participant each post one published event. - A track published before its participant joins is parked. The track attaches when the participant joins. - An unpublished track leaves the participant map. The room and the participant each post one unpublished event. - A participant disconnect unpublishes its data tracks and posts the events before the disconnect event.
This commit makes data tracks survive a reconnect. RTCEngine gets these changes: - After a full reconnect, reconnect() calls republishTracks() on the outgoing manager. - After every reconnect, reconnect() calls resendSubscriptionUpdates() on the incoming manager. - sendSyncState() adds the published data tracks to the SyncState message. The function becomes suspend, because the outgoing manager reads the tracks with a suspend call. - Listener.onSignalConnected() becomes suspend for the same reason. - joinImpl() negotiates the publisher when hasPublished is true. Room gets these changes: - sendSyncState() and onSignalConnected() become suspend. - onFullReconnecting() detaches the data tracks of every remote participant before it removes the participants. The tracks survive in the incoming manager and reattach after the join. The hasPublished condition is the most important change. In subscriber-primary mode the engine creates the publisher connection only when something is published. After a full reconnect, hasPublished is still true, but the old condition did not negotiate. The publish then waited for ICE and never completed. The new condition negotiates on every full reconnect after a publish. This affects every session that has published, not only sessions that use data tracks. Review the hasPublished change with the most care.
This commit adds four tests to OutgoingDataTrackManagerMockE2ETest. The tests check these behaviors: - A publish that waits for the channel completes on the replacement channel after a full reconnect. - Packets go to the replacement channel after a full reconnect. - A full reconnect calls republishTracks() and reconnects the publisher peer connection. - A soft reconnect includes the published data tracks in the SyncState message. The publisherOfferHandler helper answers publisher offers. The reconnectWebsocket helper reopens the mock websocket and sends the join or reconnect response.
This commit adds four tests to IncomingDataTrackManagerMockE2ETest. The tests check these behaviors: - Packets on a replacement subscriber channel reach the manager after a full reconnect. The manager is not closed. - A full reconnect detaches the data tracks without an unpublished event. - A soft reconnect calls resendSubscriptionUpdates() once. - A full reconnect calls resendSubscriptionUpdates() once and leaves the room connected. The publisherOfferHandler and reconnectWebsocket helpers are the same as in the outgoing test.
This commit adds end-to-end encryption for data track frames. DataTrackCryptor implements the UniFFI EncryptionProvider and DecryptionProvider interfaces. It uses the AES-GCM data path of E2EEManager. This is the same path as for data channel payloads. The cryptor gets the manager from a provider function on every call. A manager that is set after connect still applies. If the room has no manager, the cryptor throws the UniFFI failure exception. E2EEManager gets isDataTrackEncryptionEnabled(). It returns true when E2EE is enabled and the encryption type is not NONE. It does not check dataChannelEncryptionEnabled. That flag applies only to data channels. OutgoingDataTrackManager passes the cryptor to the factory only when isDataTrackEncryptionEnabled() is true. The presence of the provider marks every published track as encrypted. Subscribers read that flag from DataTrackInfo.usesE2ee. The decision is fixed when the native manager is created. IncomingDataTrackManager always passes the cryptor. The native manager decrypts only the tracks that are marked as encrypted. Review the encryption gate and the asymmetry between the two managers with care.
This commit adds DataTrackCryptorTest and one test in each mock end-to-end suite. DataTrackCryptorTest checks that encrypt() and decrypt() throw the UniFFI failure exception when the room has no E2EE manager. The outgoing test enables E2EE with a NoopKeyProvider. It checks that the factory receives an encryption provider. It encrypts three bytes and checks that the reversing test cryptor reverses them. It decrypts the result with the decryption provider and checks the round trip. The incoming test checks that the remote factory always receives a decryption provider.
This commit makes the sample app receive data track frames. The view model subscribes to every data track when the room connects. It also subscribes when RoomEvent.DataTrackPublished fires. A synchronized set of tracks prevents a second subscription to the same track. The set compares tracks by identity. subscribeToDataTrack() starts a coroutine. The coroutine subscribes, collects the frames, and posts each payload as a UTF-8 string to the dataReceived flow. The coroutine removes the track from the set when the stream ends. The commit keeps a commented-out handler for RoomEvent.Reconnected.
This commit updates the detekt baseline for the signatures that changed in this branch: - SignalClient.handleSignalResponse() and handleSignalResponseImpl() have a new parameter. - handleSignalResponseImpl() is now also a LongMethod. - LocalParticipant, RTCEngine, and Room have new constructor parameters. The commit also updates the copyright year in TestData.kt. That file has no other change.
|
Contributor
|
Dependency diff: +--- com.squareup.okhttp3:okhttp:4.12.0
| \--- com.squareup.okio:okio:3.6.0
| \--- com.squareup.okio:okio-jvm:3.6.0
| \--- org.jetbrains.kotlin:kotlin-stdlib-jdk8:1.9.10
| \--- org.jetbrains.kotlin:kotlin-stdlib:1.9.10 -> 1.9.25
-| \--- org.jetbrains:annotations:13.0
+| \--- org.jetbrains:annotations:13.0 -> 23.0.0
+--- com.github.davidliu:audioswitch:039a35aefab7747c557242fa216c9ea11743b604
-| \--- androidx.annotation:annotation:1.3.0 -> 1.7.1
-| \--- androidx.annotation:annotation-jvm:1.7.1
-| \--- org.jetbrains.kotlin:kotlin-stdlib:1.7.10 -> 1.9.25 (*)
+| \--- androidx.annotation:annotation:1.3.0 -> 1.9.0
+| \--- androidx.annotation:annotation-jvm:1.9.0
+| \--- org.jetbrains.kotlin:kotlin-stdlib:1.7.10 -> 1.9.25 (*)
-+--- org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.0 -> 1.6.4
-| +--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.4
-| | \--- org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.6.4
-| | +--- org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.6.4
-| | | +--- org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.4 (c)
-| | | +--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.4 (c)
-| | | \--- org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.6.4 (c)
-| | +--- org.jetbrains.kotlin:kotlin-stdlib-jdk8:1.6.21 -> 1.9.10 (*)
-| | \--- org.jetbrains.kotlin:kotlin-stdlib-common:1.6.21 -> 1.9.25 (*)
-| +--- org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.6.4 (*)
-| \--- org.jetbrains.kotlin:kotlin-stdlib-jdk8:1.6.21 -> 1.9.10 (*)
++--- org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.0 -> 1.8.1
+| +--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1
+| | \--- org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.8.1
+| | +--- org.jetbrains:annotations:23.0.0
+| | +--- org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.8.1
+| | | +--- org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.1 (c)
+| | | +--- org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.8.1 (c)
+| | | \--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1 (c)
+| | \--- org.jetbrains.kotlin:kotlin-stdlib:1.9.21 -> 1.9.25 (*)
+| +--- org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.8.1 (*)
+| \--- org.jetbrains.kotlin:kotlin-stdlib:1.9.21 -> 1.9.25 (*)
-+--- com.auth0.android:jwtdecode:2.0.2
-| \--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
-+--- androidx.annotation:annotation:1.7.1 (*)
-\--- androidx.core:core:1.13.1
- +--- androidx.annotation:annotation:1.6.0 -> 1.7.1 (*)
- +--- androidx.collection:collection:1.0.0
- | \--- androidx.annotation:annotation:1.0.0 -> 1.7.1 (*)
- +--- androidx.concurrent:concurrent-futures:1.0.0 -> 1.1.0
- | \--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
- +--- androidx.interpolator:interpolator:1.0.0
- | \--- androidx.annotation:annotation:1.0.0 -> 1.7.1 (*)
- +--- androidx.lifecycle:lifecycle-runtime:2.6.2
- | +--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
- | +--- androidx.arch.core:core-common:2.2.0
- | | \--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
- | +--- androidx.arch.core:core-runtime:2.2.0
- | | \--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
- | +--- androidx.lifecycle:lifecycle-common:2.6.2
- | | +--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
- | | \--- org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.4 (*)
- | \--- androidx.profileinstaller:profileinstaller:1.3.0
- | +--- androidx.annotation:annotation:1.2.0 -> 1.7.1 (*)
- | \--- androidx.startup:startup-runtime:1.1.1
- | +--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
- | \--- androidx.tracing:tracing:1.0.0
- | \--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
- \--- androidx.versionedparcelable:versionedparcelable:1.1.1
- \--- androidx.annotation:annotation:1.1.0 -> 1.7.1 (*)
++--- io.livekit:livekit-uniffi-android:0.1.9
+| +--- androidx.annotation:annotation:1.9.0 (*)
+| +--- net.java.dev.jna:jna:5.16.0
+| +--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1 (*)
+| \--- org.jetbrains.kotlin:kotlin-stdlib:1.9.22 -> 1.9.25 (*)
++--- com.auth0.android:jwtdecode:2.0.2
+| \--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
++--- androidx.annotation:annotation:1.7.1 -> 1.9.0 (*)
+\--- androidx.core:core:1.13.1
+ +--- androidx.annotation:annotation:1.6.0 -> 1.9.0 (*)
+ +--- androidx.collection:collection:1.0.0
+ | \--- androidx.annotation:annotation:1.0.0 -> 1.9.0 (*)
+ +--- androidx.concurrent:concurrent-futures:1.0.0 -> 1.1.0
+ | \--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
+ +--- androidx.interpolator:interpolator:1.0.0
+ | \--- androidx.annotation:annotation:1.0.0 -> 1.9.0 (*)
+ +--- androidx.lifecycle:lifecycle-runtime:2.6.2
+ | +--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
+ | +--- androidx.arch.core:core-common:2.2.0
+ | | \--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
+ | +--- androidx.arch.core:core-runtime:2.2.0
+ | | \--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
+ | +--- androidx.lifecycle:lifecycle-common:2.6.2
+ | | +--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
+ | | \--- org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.4 -> 1.8.1 (*)
+ | \--- androidx.profileinstaller:profileinstaller:1.3.0
+ | +--- androidx.annotation:annotation:1.2.0 -> 1.9.0 (*)
+ | \--- androidx.startup:startup-runtime:1.1.1
+ | +--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
+ | \--- androidx.tracing:tracing:1.0.0
+ | \--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*)
+ \--- androidx.versionedparcelable:versionedparcelable:1.1.1
+ \--- androidx.annotation:annotation:1.1.0 -> 1.9.0 (*) |
Contributor
|
Diffuse output: AARMANIFESTJAR |
Merged
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds data track support, sharing the Rust UniFFI core (
livekit-uniffi-android) with the other SDKs.Most of
room/datatrack/is a thin wrapper around that core. The Android-specific work is the WebRTC, signaling, and reconnect glue.Actual changes
RTCEngine/DataTrackFrameSender—_data_trackchannel, send backpressure, wait-for-open, reconnectSignalClient— publish / unpublish / subscriber-handle / request-response forwardingRoom,LocalParticipant,RemoteParticipant— public API, attach/park remote tracks, eventsIncomingDataTrackManager/OutgoingDataTrackManager— bridge UniFFI managers to the engineDataTrackCryptor/E2EEManager— E2EE via the existing data-packet cryptorDataChannelManager,RTCModule, test DI — channel metering and injectable factoriesSimple wrappers (FFI type/API mapping, little Android-specific logic)
LocalDataTrack,RemoteDataTrack,DataTrackStreamDataTrackFrame,DataTrackInfo,DataTrackSid,DataTrackSchema,DataTrackPublishOptions,DataTrackExceptionDataTrackManagerFactory,IncomingDataTrackEventE2E tests: https://github.com/livekit/e2e-android/pull/13
API Changes
Proposed SemVer bump: minor (2.27.0 → 2.28.0). Everything user-facing is additive. Two SDK-internal listener interfaces change shape (bottom of the table); they are public in Kotlin but only the SDK implements them.
Bold rows mark the class, interface, or group that the rows below belong to. Members of a class start with a dot.
.publishDataTrack(name, options)Result<LocalDataTrack>.withDataTrack(name, options, block).defineSchema(id, definition).getSchema(id, publishedBy)Result<String>.dataTracksMap<String, RemoteDataTrack>keyed by name;@FlowObservable.DataTrackPublished(room, participant, track).DataTrackUnpublished(room, participant, sid).DataTrackPublished(participant, track).DataTrackUnpublished(participant, sid).isPublished.infoDataTrackInfo.tryPush(frame)Result<Unit>.send(frames, onQueueFull)Flow; defaultFrameDropPolicy.DROP.unpublish().waitForUnpublish().FrameDropPolicyFAIL,DROP.publisherIdentity.name.isPublished.infoDataTrackInfo.subscribe(bufferSize)Result<DataTrackStream>; default 16 frames.waitForUnpublish().DEFAULT_BUFFER_SIZEAutoCloseable.next()nullwhen the stream ends.flow.close()io.livekit.android.room.datatrackDataTrackSidStringDataTrackFrame(payload, userTimestamp)now(),durationSinceTimestampDataTrackInfoDataTrackPublishOptions(frameFormat)(frameEncoding, schema)DataTrackFrameFormat(frameEncoding, schema)DataTrackSchemaId(name, encoding)DataTrackSchemaEncodingCustom;fromIdentifier()DataTrackFrameEncodingCustom;fromIdentifier()io.livekit.android.room.datatrackDataTrackPublishExceptionDataTrackPushFrameExceptionDataTrackSubscribeExceptionDataTrackSchemaExceptionTokenRequestOptions.agentAttributesMap<String, String>?; data class ctor gains a trailing paramRoomAgentDispatch.attributesattributes@suppress; only the SDK implements theseSignalClient.Listener.onParticipantUpdate(updates, encoded)encoded: ByteArrayparamSignalClient.Listener.onPublishDataTrackResponse/onUnpublishDataTrackResponse/onRequestResponse/onDataTrackSubscriberHandlesRTCEngine.Listener.onSignalConnected(isResume)suspendRTCEngine.Listener.reattachRemoteDataTracks()RTCEngine.sendSyncState(...)suspendRTCEngine.DATA_TRACK_DATA_CHANNEL_LABEL"_data_track";@VisibleForTestingRoomconstructorIncomingDataTrackManagerparam (Dagger)OutgoingDataTrackManager,IncomingDataTrackManagerLocalDataTrackManagerFactory,RemoteDataTrackManagerFactoryio.livekit:livekit-uniffi-android:0.1.9implementationdep; manifesttools:overrideLibrarybecause the AAR declares minSdk 24protocolsubmoduleCommits
Group 1. Preliminaries
Group 2. Public model types
5. feat(datatrack): add DataTrackSid and DataTrackFrame
6. feat(datatrack): add schema and frame encoding types
7. test(datatrack): cover schema blob key encoding
8. feat(datatrack): add DataTrackInfo and DataTrackPublishOptions
9. feat(datatrack): add publish, push, subscribe, and schema exception types
Group 3. UniFFI seam and test fakes
10. feat(datatrack): add factory seam for the UniFFI data track managers
11. test(datatrack): add mock UniFFI local and remote data track managers
12. test(datatrack): bind the mock factories in the test component
Group 4. Outbound transport
13. IMPORTANT(datatrack): add DataTrackFrameSender drop-oldest outbound drain
14. test(datatrack): pin DataTrackFrameSender drain semantics
15. IMPORTANT(datatrack): open a publisher _data_track channel and drain frames into it
16. IMPORTANT(datatrack): wait for the publisher channel before publishing
Group 5. Signaling plumbing
17. refactor(signal): keep the encoded websocket bytes alongside decoded responses
18. feat(signal): dispatch data track SFU responses to the listener
19. feat(signal): forward UniFFI-built signal requests
20. test(signal): cover encoded request forwarding and retained join bytes
Group 6. Publishing
21. IMPORTANT(datatrack): add LocalDataTrack wrapper
22. IMPORTANT(datatrack): add OutgoingDataTrackManager bridging the UniFFI local manager
23. feat(datatrack): wire OutgoingDataTrackManager into RTCEngine
24. IMPORTANT(datatrack): add LocalParticipant.publishDataTrack and withDataTrack
25. test(datatrack): cover publishing through the mock local manager
26. feat(datatrack): add LocalDataTrack.send for streaming frames from a Flow
27. test(datatrack): cover send drop and fail policies
Group 7. Schemas
28. IMPORTANT(signal): add id-correlated StoreDataBlob and GetDataBlob requests
29. test(signal): cover blob store, get, and rejection
30. feat(datatrack): add LocalParticipant.defineSchema and getSchema
31. test(datatrack): cover defineSchema when disconnected
Group 8. Subscribing
32. IMPORTANT(datatrack): add DataTrackStream with a shared multi-collector flow
33. test(datatrack): cover DataTrackStream collector semantics
34. feat(datatrack): add RemoteDataTrack wrapper
35. IMPORTANT(datatrack): add IncomingDataTrackManager bridging the UniFFI remote manager
36. IMPORTANT(datatrack): route the subscriber channel and SFU updates into IncomingDataTrackManager
37. test(datatrack): cover incoming transport and join forwarding
38. IMPORTANT(datatrack): expose data tracks on RemoteParticipant
39. feat(datatrack): surface remote data track events on Room
40. test(datatrack): cover attach, park, unpublish, and disconnect
Group 9. Reconnect
41. IMPORTANT(datatrack): restore data track state across reconnects
42. test(datatrack): cover publisher reconnect and sync state
43. test(datatrack): cover subscriber reconnect and reattach
Group 10. E2EE, sample, chores
44. IMPORTANT(datatrack): encrypt data track frames through E2EEManager
45. test(datatrack): cover the E2EE bridge
46. feat(sample): subscribe to remote data tracks in CallViewModel
47. chore: update detekt baseline and copyright headers