From 92e1a22c14749dc867e3a654e679bb3f0d239478 Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Thu, 2 Jul 2026 19:03:32 +0200 Subject: [PATCH 1/3] Add asynchronous sendAsync to RTCDataChannel Send is a blocking proxy call that stalls the caller on the native network thread until the send is processed. sendAsync posts instead, copying the payload out before returning so callers can reuse their buffers immediately. For direct buffers only the bytes between position and limit are sent. Queueing failures are logged natively and fatal errors surface through the data channel observer as a state change. --- .../src/main/cpp/include/JNI_RTCDataChannel.h | 16 ++++++ .../src/main/cpp/src/JNI_RTCDataChannel.cpp | 56 ++++++++++++++++++- .../dev/kastle/webrtc/RTCDataChannel.java | 40 +++++++++++++ 3 files changed, 111 insertions(+), 1 deletion(-) diff --git a/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h b/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h index 065f3c3..8709c5f 100644 --- a/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h +++ b/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h @@ -135,6 +135,22 @@ extern "C" { JNIEXPORT void JNICALL Java_dev_kastle_webrtc_RTCDataChannel_sendByteArrayBuffer (JNIEnv *, jobject, jbyteArray, jboolean); + /* + * Class: dev_kastle_webrtc_RTCDataChannel + * Method: sendDirectBufferAsync + * Signature: (Ljava/nio/ByteBuffer;IIZ)V + */ + JNIEXPORT void JNICALL Java_dev_kastle_webrtc_RTCDataChannel_sendDirectBufferAsync + (JNIEnv *, jobject, jobject, jint, jint, jboolean); + + /* + * Class: dev_kastle_webrtc_RTCDataChannel + * Method: sendByteArrayBufferAsync + * Signature: ([BZ)V + */ + JNIEXPORT void JNICALL Java_dev_kastle_webrtc_RTCDataChannel_sendByteArrayBufferAsync + (JNIEnv *, jobject, jbyteArray, jboolean); + #ifdef __cplusplus } #endif diff --git a/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp b/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp index b1300fb..c32c2eb 100644 --- a/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp +++ b/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp @@ -23,6 +23,7 @@ #include "JavaUtils.h" #include "api/data_channel_interface.h" +#include "rtc_base/logging.h" JNIEXPORT void JNICALL Java_dev_kastle_webrtc_RTCDataChannel_registerObserver (JNIEnv * env, jobject caller, jobject jObserver) @@ -190,11 +191,64 @@ JNIEXPORT void JNICALL Java_dev_kastle_webrtc_RTCDataChannel_sendByteArrayBuffer webrtc::CopyOnWriteBuffer data(arrayPtr, arrayLength); env->ReleaseByteArrayElements(jBufferArray, arrayPtr, JNI_ABORT); - + try { channel->Send(webrtc::DataBuffer(data, static_cast(isBinary))); } catch (...) { ThrowCxxJavaException(env); } +} + +// Completion handler shared by the async send paths. Queueing failures are +// logged; on fatal errors WebRTC closes the channel, which the registered +// data channel observer sees as a state change. +static void logSendAsyncError(webrtc::RTCError error) +{ + if (!error.ok()) { + RTC_LOG(LS_WARNING) << "SendAsync failed: " << error.message(); + } +} + +JNIEXPORT void JNICALL Java_dev_kastle_webrtc_RTCDataChannel_sendDirectBufferAsync +(JNIEnv * env, jobject caller, jobject jBuffer, jint position, jint length, jboolean isBinary) +{ + webrtc::DataChannelInterface * channel = GetHandle(env, caller); + CHECK_HANDLE(channel); + + uint8_t * address = static_cast(env->GetDirectBufferAddress(jBuffer)); + + if (address != NULL) { + jlong capacity = env->GetDirectBufferCapacity(jBuffer); + + if (position < 0 || length < 0 || static_cast(position) + length > capacity) { + env->Throw(jni::JavaError(env, "Buffer position/length out of bounds")); + return; + } + + // The data is copied into the CopyOnWriteBuffer before this call + // returns, so the caller may reuse the direct buffer immediately. + webrtc::CopyOnWriteBuffer data(address + position, static_cast(length)); + + channel->SendAsync(webrtc::DataBuffer(data, static_cast(isBinary)), &logSendAsyncError); + } + else { + env->Throw(jni::JavaError(env, "Non-direct buffer provided")); + } +} + +JNIEXPORT void JNICALL Java_dev_kastle_webrtc_RTCDataChannel_sendByteArrayBufferAsync +(JNIEnv * env, jobject caller, jbyteArray jBufferArray, jboolean isBinary) +{ + webrtc::DataChannelInterface * channel = GetHandle(env, caller); + CHECK_HANDLE(channel); + + int8_t * arrayPtr = env->GetByteArrayElements(jBufferArray, nullptr); + size_t arrayLength = env->GetArrayLength(jBufferArray); + + webrtc::CopyOnWriteBuffer data(arrayPtr, arrayLength); + + env->ReleaseByteArrayElements(jBufferArray, arrayPtr, JNI_ABORT); + + channel->SendAsync(webrtc::DataBuffer(data, static_cast(isBinary)), &logSendAsyncError); } \ No newline at end of file diff --git a/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java b/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java index 3da6414..bcac956 100644 --- a/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java +++ b/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java @@ -175,4 +175,44 @@ public void send(RTCDataChannelBuffer buffer) throws Exception { private native void sendByteArrayBuffer(byte[] buffer, boolean binary); + /** + * Sends data in the provided buffer to the remote peer without blocking + * the calling thread on the native network thread, unlike + * {@link #send(RTCDataChannelBuffer)} whose call is marshalled + * synchronously. The data is copied out of the buffer before this method + * returns, so the buffer may be reused immediately; for direct buffers + * only the bytes between position and limit are sent (the blocking send + * transmits the full capacity of a direct buffer instead). + * + * Errors are reported asynchronously: queueing failures are logged + * natively, and fatal errors close the data channel, which the registered + * {@link RTCDataChannelObserver} sees as a state change. + * + * @param buffer The buffer to be queued for transmission. + */ + public void sendAsync(RTCDataChannelBuffer buffer) { + ByteBuffer data = buffer.data; + + if (data.isDirect()) { + sendDirectBufferAsync(data, data.position(), data.remaining(), buffer.binary); + } + else { + byte[] arrayBuffer; + + if (data.hasArray()) { + arrayBuffer = data.array(); + } + else { + arrayBuffer = new byte[data.remaining()]; + data.get(arrayBuffer); + } + + sendByteArrayBufferAsync(arrayBuffer, buffer.binary); + } + } + + private native void sendDirectBufferAsync(ByteBuffer buffer, int position, int length, boolean binary); + + private native void sendByteArrayBufferAsync(byte[] buffer, boolean binary); + } From 726d4da89a537c5eb91e92c3f84eedd5a6751d48 Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:00:35 +0200 Subject: [PATCH 2/3] Bridge OnIceSelectedCandidatePairChanged to Java observers Exposes the remote address of the ICE nominated candidate pair. The callback fires before DTLS, SCTP, and the data channels open, so consumers can attach the real remote transport address to a connection before it becomes active. Backward compatible default method; existing observers are unaffected. --- .../cpp/include/api/PeerConnectionObserver.h | 2 ++ .../cpp/src/api/PeerConnectionObserver.cpp | 23 +++++++++++++++++++ .../kastle/webrtc/PeerConnectionObserver.java | 16 +++++++++++++ 3 files changed, 41 insertions(+) diff --git a/webrtc-jni/src/main/cpp/include/api/PeerConnectionObserver.h b/webrtc-jni/src/main/cpp/include/api/PeerConnectionObserver.h index 3f2f0d2..21b3e20 100644 --- a/webrtc-jni/src/main/cpp/include/api/PeerConnectionObserver.h +++ b/webrtc-jni/src/main/cpp/include/api/PeerConnectionObserver.h @@ -44,6 +44,7 @@ namespace jni void OnIceCandidateError(const std::string & address, int port, const std::string & url, int error_code, const std::string & error_text) override; void OnIceCandidatesRemoved(const std::vector & candidates) override; void OnIceConnectionReceivingChange(bool receiving) override; + void OnIceSelectedCandidatePairChanged(const webrtc::CandidatePairChangeEvent & event) override; private: class JavaPeerConnectionObserverClass : public JavaClass @@ -61,6 +62,7 @@ namespace jni jmethodID onIceCandidateError; jmethodID onIceCandidatesRemoved; jmethodID onIceConnectionReceivingChange; + jmethodID onSelectedCandidatePairChanged; }; private: diff --git a/webrtc-jni/src/main/cpp/src/api/PeerConnectionObserver.cpp b/webrtc-jni/src/main/cpp/src/api/PeerConnectionObserver.cpp index 31ceb99..ebf0a88 100644 --- a/webrtc-jni/src/main/cpp/src/api/PeerConnectionObserver.cpp +++ b/webrtc-jni/src/main/cpp/src/api/PeerConnectionObserver.cpp @@ -23,6 +23,7 @@ #include "JavaEnums.h" #include "JavaFactories.h" #include "JavaRuntimeException.h" +#include "JavaString.h" #include "JavaUtils.h" #include "JNI_WebRTC.h" @@ -150,6 +151,27 @@ namespace jni ExceptionCheck(env); } + void PeerConnectionObserver::OnIceSelectedCandidatePairChanged(const webrtc::CandidatePairChangeEvent & event) + { + JNIEnv * env = AttachCurrentThread(); + + const webrtc::Candidate & remote = event.selected_candidate_pair.remote_candidate(); + + std::string ip = remote.address().ipaddr().ToString(); + int port = remote.address().port(); + + const auto typeName = remote.type_name(); + std::string type(typeName.data(), typeName.size()); + + JavaLocalRef jAddress = JavaString::toJava(env, ip); + JavaLocalRef jType = JavaString::toJava(env, type); + + env->CallVoidMethod(observer, javaClass->onSelectedCandidatePairChanged, + jAddress.get(), static_cast(port), jType.get()); + + ExceptionCheck(env); + } + PeerConnectionObserver::JavaPeerConnectionObserverClass::JavaPeerConnectionObserverClass(JNIEnv * env) { jclass cls = FindClass(env, PKG"PeerConnectionObserver"); @@ -164,5 +186,6 @@ namespace jni onIceCandidateError = GetMethod(env, cls, "onIceCandidateError", "(L" PKG "RTCPeerConnectionIceErrorEvent;)V"); onIceCandidatesRemoved = GetMethod(env, cls, "onIceCandidatesRemoved", "([L" PKG "RTCIceCandidate;)V"); onIceConnectionReceivingChange = GetMethod(env, cls, "onIceConnectionReceivingChange", "(Z)V"); + onSelectedCandidatePairChanged = GetMethod(env, cls, "onSelectedCandidatePairChanged", "(Ljava/lang/String;ILjava/lang/String;)V"); } } diff --git a/webrtc/src/main/java/dev/kastle/webrtc/PeerConnectionObserver.java b/webrtc/src/main/java/dev/kastle/webrtc/PeerConnectionObserver.java index 91c19c7..498e392 100644 --- a/webrtc/src/main/java/dev/kastle/webrtc/PeerConnectionObserver.java +++ b/webrtc/src/main/java/dev/kastle/webrtc/PeerConnectionObserver.java @@ -110,4 +110,20 @@ default void onDataChannel(RTCDataChannel dataChannel) { default void onRenegotiationNeeded() { } + /** + * ICE has selected (or later re-selected) a candidate pair for this + * connection. Fires once connectivity checks nominate a pair, which is + * before DTLS and SCTP are established and before data channels open, so + * the remote transport address is known before the connection becomes + * usable. May fire again if ICE re-nominates mid session. + * + * @param remoteAddress The remote candidate's IP address. For a relayed + * connection this is the TURN relay's address. + * @param remotePort The remote candidate's port. + * @param candidateType The remote candidate type: "host", "srflx", + * "prflx", or "relay". + */ + default void onSelectedCandidatePairChanged(String remoteAddress, int remotePort, String candidateType) { + } + } From 92b5acaedf2afe1582ee5821e2ee205ad3676052 Mon Sep 17 00:00:00 2001 From: SendableMetatype <263203301+SendableMetatype@users.noreply.github.com> Date: Sun, 2 Aug 2026 20:08:22 +0200 Subject: [PATCH 3/3] Honor the buffer window in every send path send and sendAsync transmitted different bytes depending on the buffer kind: an array backed heap buffer sent its whole backing array, a non array heap buffer sent its readable window, and the blocking send of a direct buffer sent its full capacity. Only the window semantics are coherent, so all paths now send exactly the bytes between position and limit, reading through a duplicate that leaves the caller's position untouched. Buffers whose window already covers their backing storage take the previous zero copy paths unchanged. --- .../dev/kastle/webrtc/RTCDataChannel.java | 58 +++++++++++-------- 1 file changed, 33 insertions(+), 25 deletions(-) diff --git a/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java b/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java index bcac956..ed44863 100644 --- a/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java +++ b/webrtc/src/main/java/dev/kastle/webrtc/RTCDataChannel.java @@ -143,7 +143,10 @@ private RTCDataChannel() { public native void dispose(); /** - * Sends data in the provided buffer to the remote peer. + * Sends data in the provided buffer to the remote peer. Only the bytes + * between the buffer's position and limit are sent, for heap and direct + * buffers alike. The buffer is read through a duplicate, so the caller's + * position is left untouched. * * @param buffer The buffer to be queued for transmission. * @@ -154,20 +157,18 @@ public void send(RTCDataChannelBuffer buffer) throws Exception { ByteBuffer data = buffer.data; if (data.isDirect()) { - sendDirectBuffer(data, buffer.binary); - } - else { - byte[] arrayBuffer; - - if (data.hasArray()) { - arrayBuffer = data.array(); + if (data.position() == 0 && data.limit() == data.capacity()) { + sendDirectBuffer(data, buffer.binary); } else { - arrayBuffer = new byte[data.remaining()]; - data.get(arrayBuffer); + ByteBuffer window = ByteBuffer.allocateDirect(data.remaining()); + window.put(data.duplicate()); + window.flip(); + sendDirectBuffer(window, buffer.binary); } - - sendByteArrayBuffer(arrayBuffer, buffer.binary); + } + else { + sendByteArrayBuffer(copyWindow(data), buffer.binary); } } @@ -180,9 +181,8 @@ public void send(RTCDataChannelBuffer buffer) throws Exception { * the calling thread on the native network thread, unlike * {@link #send(RTCDataChannelBuffer)} whose call is marshalled * synchronously. The data is copied out of the buffer before this method - * returns, so the buffer may be reused immediately; for direct buffers - * only the bytes between position and limit are sent (the blocking send - * transmits the full capacity of a direct buffer instead). + * returns, so the buffer may be reused immediately; only the bytes + * between position and limit are sent. * * Errors are reported asynchronously: queueing failures are logged * natively, and fatal errors close the data channel, which the registered @@ -197,18 +197,26 @@ public void sendAsync(RTCDataChannelBuffer buffer) { sendDirectBufferAsync(data, data.position(), data.remaining(), buffer.binary); } else { - byte[] arrayBuffer; - - if (data.hasArray()) { - arrayBuffer = data.array(); - } - else { - arrayBuffer = new byte[data.remaining()]; - data.get(arrayBuffer); - } + sendByteArrayBufferAsync(copyWindow(data), buffer.binary); + } + } - sendByteArrayBufferAsync(arrayBuffer, buffer.binary); + /** + * Copies the readable window of a heap buffer, position to limit, into a + * fresh array for the byte array send paths, which transmit whole arrays. + * The backing array is handed over directly only when the window covers + * it exactly; bytes outside the window (a nonzero position, a short + * limit, an array offset) must never reach the wire. Reads through a + * duplicate, so the caller's position is left untouched. + */ + private static byte[] copyWindow(ByteBuffer data) { + if (data.hasArray() && data.arrayOffset() == 0 && data.position() == 0 + && data.remaining() == data.array().length) { + return data.array(); } + byte[] window = new byte[data.remaining()]; + data.duplicate().get(window); + return window; } private native void sendDirectBufferAsync(ByteBuffer buffer, int position, int length, boolean binary);