diff --git a/docs/guide/data/data-channels.md b/docs/guide/data/data-channels.md index e11e8ebe..7ace61cc 100644 --- a/docs/guide/data/data-channels.md +++ b/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`: diff --git a/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h b/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h index 8211f00a..a9c54436 100644 --- a/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h +++ b/webrtc-jni/src/main/cpp/include/JNI_RTCDataChannel.h @@ -151,6 +151,22 @@ extern "C" { JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendByteArrayBufferAsync (JNIEnv *, jobject, jbyteArray, jboolean); + /* + * Class: dev_onvoid_webrtc_RTCDataChannel + * Method: sendDirectBufferAsyncWithObserver + * Signature: (Ljava/nio/ByteBuffer;IIZLdev/onvoid/webrtc/RTCDataChannelSendObserver;)V + */ + JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendDirectBufferAsyncWithObserver + (JNIEnv *, jobject, jobject, jint, jint, jboolean, jobject); + + /* + * Class: dev_onvoid_webrtc_RTCDataChannel + * Method: sendByteArrayBufferAsyncWithObserver + * Signature: ([BZLdev/onvoid/webrtc/RTCDataChannelSendObserver;)V + */ + JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendByteArrayBufferAsyncWithObserver + (JNIEnv *, jobject, jbyteArray, jboolean, jobject); + #ifdef __cplusplus } #endif diff --git a/webrtc-jni/src/main/cpp/include/api/RTCDataChannelSendObserver.h b/webrtc-jni/src/main/cpp/include/api/RTCDataChannelSendObserver.h new file mode 100644 index 00000000..a861dc7c --- /dev/null +++ b/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 + +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 observer; + const std::shared_ptr javaClass; + }; +} + +#endif diff --git a/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp b/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp index 4afc434c..226686b1 100644 --- a/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp +++ b/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(env, caller); - CHECK_HANDLE(channel); + if (jObserver == nullptr) { + channel->SendAsync(std::move(buffer), &logSendAsyncError); + return; + } - uint8_t * address = static_cast(env->GetDirectBufferAddress(jBuffer)); + auto observer = std::make_shared(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(env, caller); + CHECK_HANDLE(channel); - if (position < 0 || length < 0 || static_cast(position) + length > capacity) { - env->Throw(jni::JavaError(env, "Buffer position/length out of bounds")); - return; + 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 before returning, so the caller may reuse the buffer. + webrtc::CopyOnWriteBuffer data(address + position, static_cast(length)); + + sendAsync(env, channel, webrtc::DataBuffer(data, static_cast(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(length)); +static void sendByteArrayBufferAsync(JNIEnv * env, jobject caller, jbyteArray jBufferArray, + jboolean isBinary, jobject jObserver) +{ + try { + webrtc::DataChannelInterface * channel = GetHandle(env, caller); + CHECK_HANDLE(channel); + + auto releaseArray = [env, jBufferArray](jbyte * bytes) { + env->ReleaseByteArrayElements(jBufferArray, bytes, JNI_ABORT); + }; + std::unique_ptr 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(isBinary)), &logSendAsyncError); + sendAsync(env, channel, webrtc::DataBuffer(data, static_cast(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(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(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); } \ No newline at end of file diff --git a/webrtc-jni/src/main/cpp/src/api/RTCDataChannelSendObserver.cpp b/webrtc-jni/src/main/cpp/src/api/RTCDataChannelSendObserver.cpp new file mode 100644 index 00000000..b8e1f34d --- /dev/null +++ b/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(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(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 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 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"); + } +} diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java b/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java index 01ed68fd..f00fd201 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java +++ b/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. + *

+ * 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. + *

+ * 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); + } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannelSendObserver.java b/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannelSendObserver.java new file mode 100644 index 00000000..e9b381ba --- /dev/null +++ b/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); +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelSendCompletionTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelSendCompletionTests.java new file mode 100644 index 00000000..cf12c1ed --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelSendCompletionTests.java @@ -0,0 +1,199 @@ +package dev.onvoid.webrtc; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.ByteBuffer; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +class RTCDataChannelSendCompletionTests extends TestBase { + + @Test + void acceptsHeapBufferWindows() throws Exception { + assertAccepted(false); + } + + @Test + void acceptsDirectBufferWindows() throws Exception { + assertAccepted(true); + } + + @Test + void reportsDiscardedSendsBeforeTransportExists() throws Exception { + try (TestDataChannelPair pair = new TestDataChannelPair(factory)) { + assertRejected(pair.sender); + } + } + + @Test + void reportsNativeRejectionAfterClose() throws Exception { + try (TestDataChannelPair pair = new TestDataChannelPair(factory)) { + pair.connect(); + pair.sender.close(); + assertRejected(pair.sender); + } + } + + @Test + void rejectsInvalidArgumentsWithoutCallingObserver() { + SendResult result = new SendResult(); + try (TestDataChannelPair pair = new TestDataChannelPair(factory)) { + assertThrows(NullPointerException.class, () -> pair.sender.sendAsync(null, result)); + assertThrows(NullPointerException.class, + () -> pair.sender.sendAsync(new RTCDataChannelBuffer(null, true), result)); + assertThrows(NullPointerException.class, + () -> pair.sender.sendAsync(new RTCDataChannelBuffer(ByteBuffer.allocate(1), true), null)); + } + assertEquals(0, result.calls.get()); + } + + @Test + void callbackExceptionDoesNotStopLaterSends() throws Exception { + try (TestDataChannelPair pair = new TestDataChannelPair(factory)) { + pair.connect(); + CountDownLatch entered = new CountDownLatch(1); + RTCDataChannelBuffer buffer = new RTCDataChannelBuffer(ByteBuffer.allocate(1), true); + pair.sender.sendAsync(buffer, new RTCDataChannelSendObserver() { + @Override + public void onSuccess() { + entered.countDown(); + throw new IllegalStateException("Send observer test exception"); + } + + @Override + public void onFailure(String error) { + entered.countDown(); + throw new IllegalStateException(error); + } + }); + assertTrue(entered.await(5, TimeUnit.SECONDS), "Observer was not called"); + SendResult next = new SendResult(); + pair.sender.sendAsync(buffer, next); + assertNull(next.result.get(5, TimeUnit.SECONDS)); + assertEquals(1, next.calls.get()); + } + } + + @Test + void failureCallbackExceptionDoesNotStopLaterSends() throws Exception { + try (TestDataChannelPair pair = new TestDataChannelPair(factory)) { + CountDownLatch entered = new CountDownLatch(1); + RTCDataChannelBuffer buffer = new RTCDataChannelBuffer(ByteBuffer.allocate(1), true); + pair.sender.sendAsync(buffer, new RTCDataChannelSendObserver() { + @Override + public void onSuccess() { + entered.countDown(); + throw new IllegalStateException("Unexpected send success"); + } + + @Override + public void onFailure(String error) { + entered.countDown(); + throw new IllegalStateException("Send failure observer test exception"); + } + }); + assertTrue(entered.await(5, TimeUnit.SECONDS), "Observer was not called"); + assertRejected(pair.sender); + } + } + + @Test + void pendingSendsCompleteOnceWhenPeerCloses() throws Exception { + RTCPeerConnection peer = factory.createPeerConnection(new RTCConfiguration(), candidate -> { }); + RTCDataChannel channel = peer.createDataChannel("send", new RTCDataChannelInit()); + SendResult[] results = new SendResult[64]; + try { + try { + for (int i = 0; i < results.length; i++) { + results[i] = new SendResult(); + channel.sendAsync(new RTCDataChannelBuffer(ByteBuffer.allocateDirect(1), true), results[i]); + } + } + finally { + peer.close(); + } + for (SendResult result : results) { + assertNotNull(result.result.get(5, TimeUnit.SECONDS)); + assertEquals(1, result.calls.get()); + } + } + finally { + channel.dispose(); + } + } + + private void assertAccepted(boolean direct) throws Exception { + try (TestDataChannelPair pair = new TestDataChannelPair(factory)) { + pair.connect(); + for (boolean readOnly : new boolean[] { false, true }) { + ByteBuffer storage = direct ? ByteBuffer.allocateDirect(8) : ByteBuffer.allocate(8); + storage.put(new byte[] { 99, 10, 20, 30, 40, 50, 88, 77 }); + ByteBuffer window = storage.duplicate(); + window.position(1); + window.limit(6); + if (readOnly) { + window = window.asReadOnlyBuffer(); + } + SendResult result = new SendResult(); + pair.sender.sendAsync(new RTCDataChannelBuffer(window, true), result); + assertEquals(1, window.position()); + assertEquals(6, window.limit()); + for (int i = 0; i < storage.capacity(); i++) { + storage.put(i, (byte) 0); + } + assertNull(result.result.get(5, TimeUnit.SECONDS)); + assertEquals(1, result.calls.get()); + RTCDataChannelBuffer received = pair.messages.poll(5, TimeUnit.SECONDS); + assertNotNull(received); + assertTrue(received.binary); + assertArrayEquals(new byte[] { 10, 20, 30, 40, 50 }, received.data.array()); + } + SendResult empty = new SendResult(); + ByteBuffer buffer = direct ? ByteBuffer.allocateDirect(0) : ByteBuffer.allocate(0); + pair.sender.sendAsync(new RTCDataChannelBuffer(buffer, false), empty); + assertNull(empty.result.get(5, TimeUnit.SECONDS)); + assertEquals(1, empty.calls.get()); + RTCDataChannelBuffer received = pair.messages.poll(5, TimeUnit.SECONDS); + assertNotNull(received); + assertEquals(false, received.binary); + assertEquals(0, received.data.remaining()); + } + } + + private static void assertRejected(RTCDataChannel channel) throws Exception { + for (ByteBuffer buffer : new ByteBuffer[] { ByteBuffer.allocate(1), ByteBuffer.allocateDirect(1) }) { + SendResult result = new SendResult(); + channel.sendAsync(new RTCDataChannelBuffer(buffer, true), result); + String error = result.result.get(5, TimeUnit.SECONDS); + assertNotNull(error); + assertTrue(error.startsWith("[INVALID_STATE]"), error); + assertEquals(1, result.calls.get()); + } + } + + private static class SendResult implements RTCDataChannelSendObserver { + final CompletableFuture result = new CompletableFuture<>(); + final AtomicInteger calls = new AtomicInteger(); + + @Override + public void onSuccess() { + calls.incrementAndGet(); + result.complete(null); + } + + @Override + public void onFailure(String error) { + calls.incrementAndGet(); + result.complete(error); + } + } +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/TestDataChannelPair.java b/webrtc/src/test/java/dev/onvoid/webrtc/TestDataChannelPair.java new file mode 100644 index 00000000..c5afa1e0 --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/TestDataChannelPair.java @@ -0,0 +1,96 @@ +package dev.onvoid.webrtc; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.ByteBuffer; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +class TestDataChannelPair implements AutoCloseable { + + final RTCDataChannel sender; + final RTCDataChannel receiver; + final BlockingQueue messages = new LinkedBlockingQueue<>(); + + private final RTCPeerConnection caller; + private final RTCPeerConnection callee; + private final CountDownLatch open = new CountDownLatch(2); + + TestDataChannelPair(PeerConnectionFactory factory) { + RTCPeerConnection[] peers = new RTCPeerConnection[2]; + caller = factory.createPeerConnection(new RTCConfiguration(), + candidate -> peers[1].addIceCandidate(candidate)); + callee = factory.createPeerConnection(new RTCConfiguration(), + candidate -> peers[0].addIceCandidate(candidate)); + peers[0] = caller; + peers[1] = callee; + + RTCDataChannelInit config = new RTCDataChannelInit(); + config.negotiated = true; + config.id = 0; + sender = caller.createDataChannel("send", config); + receiver = callee.createDataChannel("send", config); + sender.registerObserver(observer(sender)); + receiver.registerObserver(observer(receiver)); + } + + void connect() throws Exception { + TestCreateDescObserver offer = new TestCreateDescObserver(); + caller.createOffer(new RTCOfferOptions(), offer); + RTCSessionDescription offerDescription = offer.get(5, TimeUnit.SECONDS); + setDescription(caller, offerDescription, true); + setDescription(callee, offerDescription, false); + + TestCreateDescObserver answer = new TestCreateDescObserver(); + callee.createAnswer(new RTCAnswerOptions(), answer); + RTCSessionDescription answerDescription = answer.get(5, TimeUnit.SECONDS); + setDescription(callee, answerDescription, true); + setDescription(caller, answerDescription, false); + assertTrue(open.await(5, TimeUnit.SECONDS), "Data channels did not open"); + } + + private static void setDescription(RTCPeerConnection peer, + RTCSessionDescription description, boolean local) throws Exception { + TestSetDescObserver observer = new TestSetDescObserver(); + if (local) { + peer.setLocalDescription(description, observer); + } + else { + peer.setRemoteDescription(description, observer); + } + observer.get(5, TimeUnit.SECONDS); + } + + private RTCDataChannelObserver observer(RTCDataChannel channel) { + return new RTCDataChannelObserver() { + @Override + public void onBufferedAmountChange(long sentDataSize) { } + + @Override + public void onStateChange() { + if (channel.getState() == RTCDataChannelState.OPEN) { + open.countDown(); + } + } + + @Override + public void onMessage(RTCDataChannelBuffer buffer) { + byte[] bytes = new byte[buffer.data.remaining()]; + buffer.data.duplicate().get(bytes); + messages.add(new RTCDataChannelBuffer(ByteBuffer.wrap(bytes), buffer.binary)); + } + }; + } + + @Override + public void close() { + sender.unregisterObserver(); + receiver.unregisterObserver(); + caller.close(); + callee.close(); + sender.dispose(); + receiver.dispose(); + } +}