feat: expose asynchronous data channel send results

docs/guide/data/data-channels.md

@@ -180,6 +180,29 @@ try {
 }
 ```
 
+### Sending Data Asynchronously
+
+`send` queues the message on the calling thread. `sendAsync` hands the message to the native network thread instead and returns immediately, which suits senders that must not block. The readable window of the buffer is copied before the method returns, so the buffer can be reused right away.
+
+To learn whether the native send operation accepted the message, pass an `RTCDataChannelSendObserver`:
+
+```java
+dataChannel.sendAsync(binaryChannelBuffer, new RTCDataChannelSendObserver() {
+    @Override
+    public void onSuccess() {
+        // The local send operation accepted the message.
+    }
+
+    @Override
+    public void onFailure(String error) {
+        // For example "[INVALID_STATE] ..." when the channel is not open.
+        System.err.println("Send failed: " + error);
+    }
+});
+```
+
+The observer is called exactly once, normally on the native network thread, and with a failure if the operation is discarded while the channel shuts down. Success means the message was queued locally, not that the peer received it. Do not block in the callbacks or call other WebRTC methods from them synchronously; dispatch further work to an executor of your own. Without an observer, `sendAsync` logs failures and reports nothing to the caller.
+
 ### Receiving Data
 
 To receive data, implement the `onMessage` method in your `RTCDataChannelObserver`:

webrtc-jni/src/main/cpp/include/api/RTCDataChannelSendObserver.h

@@ -0,0 +1,40 @@
+#ifndef JNI_WEBRTC_API_RTC_DATA_CHANNEL_SEND_OBSERVER_H_
+#define JNI_WEBRTC_API_RTC_DATA_CHANNEL_SEND_OBSERVER_H_
+
+#include "JavaClass.h"
+#include "JavaRef.h"
+#include "api/rtc_error.h"
+
+#include <memory>
+
+namespace jni
+{
+	class RTCDataChannelSendObserver
+	{
+		public:
+			RTCDataChannelSendObserver(JNIEnv * env, jobject observer);
+			~RTCDataChannelSendObserver();
+
+			RTCDataChannelSendObserver(const RTCDataChannelSendObserver &) = delete;
+			RTCDataChannelSendObserver & operator=(const RTCDataChannelSendObserver &) = delete;
+
+			void OnComplete(webrtc::RTCError error) noexcept;
+			void Cancel();
+
+		private:
+			class JavaSendObserverClass : public JavaClass
+			{
+				public:
+					explicit JavaSendObserverClass(JNIEnv * env);
+					jmethodID onSuccess;
+					jmethodID onFailure;
+			};
+
+			void Notify(const char * error) noexcept;
+
+			JavaGlobalRef<jobject> observer;
+			const std::shared_ptr<JavaSendObserverClass> javaClass;
+	};
+}
+
+#endif

webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp

@@ -16,6 +16,7 @@
 
 #include "JNI_RTCDataChannel.h"
 #include "api/RTCDataChannelObserver.h"
+#include "api/RTCDataChannelSendObserver.h"
 #include "JavaEnums.h"
 #include "JavaError.h"
 #include "JavaRef.h"
@@ -212,45 +213,106 @@ static void logSendAsyncError(webrtc::RTCError error)
 	}
 }
 
-JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendDirectBufferAsync
-(JNIEnv * env, jobject caller, jobject jBuffer, jint position, jint length, jboolean isBinary)
+static void sendAsync(JNIEnv * env, webrtc::DataChannelInterface * channel,
+	webrtc::DataBuffer buffer, jobject jObserver)
 {
-	webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
-	CHECK_HANDLE(channel);
+	if (jObserver == nullptr) {
+		channel->SendAsync(std::move(buffer), &logSendAsyncError);
+		return;
+	}
 
-	uint8_t * address = static_cast<uint8_t *>(env->GetDirectBufferAddress(jBuffer));
+	auto observer = std::make_shared<jni::RTCDataChannelSendObserver>(env, jObserver);
+	if (env->ExceptionCheck()) {
+		observer->Cancel();
+		return;
+	}
+	try {
+		channel->SendAsync(std::move(buffer), [observer](webrtc::RTCError error) {
+			observer->OnComplete(std::move(error));
+		});
+	}
+	catch (...) {
+		observer->Cancel();
+		throw;
+	}
+}
 
-	if (address != NULL) {
-		jlong capacity = env->GetDirectBufferCapacity(jBuffer);
+static void sendDirectBufferAsync(JNIEnv * env, jobject caller, jobject jBuffer,
+	jint position, jint length, jboolean isBinary, jobject jObserver)
+{
+	try {
+		webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
+		CHECK_HANDLE(channel);
 
-		if (position < 0 || length < 0 || static_cast<jlong>(position) + length > capacity) {
-			env->Throw(jni::JavaError(env, "Buffer position/length out of bounds"));
-			return;
+		uint8_t * address = static_cast<uint8_t *>(env->GetDirectBufferAddress(jBuffer));
+
+		if (address != NULL) {
+			jlong capacity = env->GetDirectBufferCapacity(jBuffer);
+
+			if (position < 0 || length < 0 || static_cast<jlong>(position) + length > capacity) {
+				env->Throw(jni::JavaError(env, "Buffer position/length out of bounds"));
+				return;
+			}
+
+			// The data is copied before returning, so the caller may reuse the buffer.
+			webrtc::CopyOnWriteBuffer data(address + position, static_cast<size_t>(length));
+
+			sendAsync(env, channel, webrtc::DataBuffer(data, static_cast<bool>(isBinary)), jObserver);
 		}
+		else {
+			env->Throw(jni::JavaError(env, "Non-direct buffer provided"));
+		}
+	}
+	catch (...) {
+		ThrowCxxJavaException(env);
+	}
+}
 
-		// 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<size_t>(length));
+static void sendByteArrayBufferAsync(JNIEnv * env, jobject caller, jbyteArray jBufferArray,
+	jboolean isBinary, jobject jObserver)
+{
+	try {
+		webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
+		CHECK_HANDLE(channel);
+
+		auto releaseArray = [env, jBufferArray](jbyte * bytes) {
+			env->ReleaseByteArrayElements(jBufferArray, bytes, JNI_ABORT);
+		};
+		std::unique_ptr<jbyte, decltype(releaseArray)> bytes(
+			env->GetByteArrayElements(jBufferArray, nullptr), releaseArray);
+		if (!bytes) {
+			return;
+		}
+		webrtc::CopyOnWriteBuffer data(bytes.get(), env->GetArrayLength(jBufferArray));
+		bytes.reset();
 
-		channel->SendAsync(webrtc::DataBuffer(data, static_cast<bool>(isBinary)), &logSendAsyncError);
+		sendAsync(env, channel, webrtc::DataBuffer(data, static_cast<bool>(isBinary)), jObserver);
 	}
-	else {
-		env->Throw(jni::JavaError(env, "Non-direct buffer provided"));
+	catch (...) {
+		ThrowCxxJavaException(env);
 	}
 }
 
+JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendDirectBufferAsync
+(JNIEnv * env, jobject caller, jobject jBuffer, jint position, jint length, jboolean isBinary)
+{
+	sendDirectBufferAsync(env, caller, jBuffer, position, length, isBinary, nullptr);
+}
+
 JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendByteArrayBufferAsync
 (JNIEnv * env, jobject caller, jbyteArray jBufferArray, jboolean isBinary)
 {
-	webrtc::DataChannelInterface * channel = GetHandle<webrtc::DataChannelInterface>(env, caller);
-	CHECK_HANDLE(channel);
-
-	int8_t * arrayPtr = env->GetByteArrayElements(jBufferArray, nullptr);
-	size_t arrayLength = env->GetArrayLength(jBufferArray);
-
-	webrtc::CopyOnWriteBuffer data(arrayPtr, arrayLength);
+	sendByteArrayBufferAsync(env, caller, jBufferArray, isBinary, nullptr);
+}
 
-	env->ReleaseByteArrayElements(jBufferArray, arrayPtr, JNI_ABORT);
+JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendDirectBufferAsyncWithObserver
+(JNIEnv * env, jobject caller, jobject jBuffer, jint position, jint length, jboolean isBinary, jobject jObserver)
+{
+	sendDirectBufferAsync(env, caller, jBuffer, position, length, isBinary, jObserver);
+}
 
-	channel->SendAsync(webrtc::DataBuffer(data, static_cast<bool>(isBinary)), &logSendAsyncError);
+JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendByteArrayBufferAsyncWithObserver
+(JNIEnv * env, jobject caller, jbyteArray jBufferArray, jboolean isBinary, jobject jObserver)
+{
+	sendByteArrayBufferAsync(env, caller, jBufferArray, isBinary, jObserver);
 }

webrtc-jni/src/main/cpp/src/api/RTCDataChannelSendObserver.cpp

@@ -0,0 +1,72 @@
+#include "api/RTCDataChannelSendObserver.h"
+#include "api/WebRTCUtils.h"
+#include "JavaString.h"
+#include "JNI_WebRTC.h"
+
+namespace jni
+{
+	RTCDataChannelSendObserver::RTCDataChannelSendObserver(JNIEnv * env, jobject observer) :
+		observer(env, observer),
+		javaClass(JavaClasses::get<JavaSendObserverClass>(env))
+	{
+	}
+
+	RTCDataChannelSendObserver::~RTCDataChannelSendObserver()
+	{
+		// WebRTC can destroy the completion without calling it after losing its transport.
+		Notify("[INVALID_STATE] Send operation was discarded before completion");
+	}
+
+	void RTCDataChannelSendObserver::Cancel()
+	{
+		observer = JavaGlobalRef<jobject>(nullptr);
+	}
+
+	void RTCDataChannelSendObserver::OnComplete(webrtc::RTCError error) noexcept
+	{
+		try {
+			if (error.ok()) {
+				Notify(nullptr);
+			}
+			else {
+				Notify(RTCErrorToString(error).c_str());
+			}
+		}
+		catch (...) {
+			Notify("[INTERNAL_ERROR] Could not report native send result");
+		}
+	}
+
+	void RTCDataChannelSendObserver::Notify(const char * error) noexcept
+	{
+		JavaGlobalRef<jobject> callback(std::move(observer));
+		if (!callback.get()) {
+			return;
+		}
+		JNIEnv * env = AttachCurrentThread();
+		if (env == nullptr) {
+			return;
+		}
+		if (error == nullptr) {
+			env->CallVoidMethod(callback.get(), javaClass->onSuccess);
+		}
+		else {
+			JavaLocalRef<jstring> message(env, env->NewStringUTF(error));
+			if (!env->ExceptionCheck()) {
+				env->CallVoidMethod(callback.get(), javaClass->onFailure, message.get());
+			}
+		}
+		// A Java exception must not escape into WebRTC's network task or a destructor.
+		if (env->ExceptionCheck()) {
+			env->ExceptionDescribe();
+			env->ExceptionClear();
+		}
+	}
+
+	RTCDataChannelSendObserver::JavaSendObserverClass::JavaSendObserverClass(JNIEnv * env)
+	{
+		jclass cls = FindClass(env, PKG"RTCDataChannelSendObserver");
+		onSuccess = GetMethod(env, cls, "onSuccess", "()V");
+		onFailure = GetMethod(env, cls, "onFailure", "(" STRING_SIG ")V");
+	}
+}

webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java

@@ -19,6 +19,7 @@
 import dev.onvoid.webrtc.internal.DisposableNativeObject;
 
 import java.nio.ByteBuffer;
+import java.util.Objects;
 
 /**
  * Represents a bidirectional data channel between two peers. An RTCDataChannel
@@ -228,4 +229,43 @@ public void sendAsync(RTCDataChannelBuffer buffer) {
 
 	private native void sendByteArrayBufferAsync(byte[] buffer, boolean binary);
 
+	/**
+	 * Sends data asynchronously and reports the native send result. Success
+	 * means the local send operation accepted the message, not that the peer
+	 * received it. The readable buffer window is copied before this method
+	 * returns, without changing its position or limit.
+	 * <p>
+	 * The observer is called once when the operation completes or is discarded
+	 * during channel shutdown. It normally runs on the native network thread;
+	 * a discarded operation can report failure on the thread that destroys it.
+	 * The callback may run before this method returns. It must not block or
+	 * call other WebRTC methods synchronously. Dispatch further work to an
+	 * application executor. No callback is guaranteed during JVM shutdown.
+	 * <p>
+	 * Invalid arguments and failures preparing the native operation are thrown
+	 * on the calling thread. If preparation fails, the observer is not called.
+	 * Exceptions thrown by the observer are printed and cleared on its native
+	 * thread; they do not propagate to the sender.
+	 *
+	 * @param buffer The buffer to be queued for transmission.
+	 * @param observer The observer for this send operation.
+	 * @throws NullPointerException If the buffer, its data, or the observer is null.
+	 */
+	public void sendAsync(RTCDataChannelBuffer buffer, RTCDataChannelSendObserver observer) {
+		Objects.requireNonNull(observer, "observer");
+		ByteBuffer data = Objects.requireNonNull(buffer.data, "buffer.data");
+		if (data.isDirect()) {
+			sendDirectBufferAsyncWithObserver(data, data.position(), data.remaining(), buffer.binary, observer);
+		}
+		else {
+			sendByteArrayBufferAsyncWithObserver(copyWindow(data), buffer.binary, observer);
+		}
+	}
+
+	private native void sendDirectBufferAsyncWithObserver(ByteBuffer buffer, int position,
+			int length, boolean binary, RTCDataChannelSendObserver observer);
+
+	private native void sendByteArrayBufferAsyncWithObserver(byte[] buffer, boolean binary,
+			RTCDataChannelSendObserver observer);
+
 }

webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannelSendObserver.java

@@ -0,0 +1,22 @@
+package dev.onvoid.webrtc;
+
+/**
+ * Receives the result of one {@link RTCDataChannel#sendAsync(RTCDataChannelBuffer,
+ * RTCDataChannelSendObserver)} operation. Callbacks must not block or call WebRTC
+ * synchronously; dispatch further work to an application executor.
+ */
+public interface RTCDataChannelSendObserver {
+
+	/**
+	 * The local send operation accepted the message. This does not confirm
+	 * delivery to the peer.
+	 */
+	void onSuccess();
+
+	/**
+	 * The send failed or was discarded before completion.
+	 *
+	 * @param error The error type in brackets followed by the error message.
+	 */
+	void onFailure(String error);
+}