Add asynchronous data channel sends and bridge the ICE selected candidate pair

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<webrtc::Candidate> & 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:

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<bool>(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<webrtc::DataChannelInterface>(env, caller);
+	CHECK_HANDLE(channel);
+
+	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 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));
+
+		channel->SendAsync(webrtc::DataBuffer(data, static_cast<bool>(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<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);
+
+	env->ReleaseByteArrayElements(jBufferArray, arrayPtr, JNI_ABORT);
+
+	channel->SendAsync(webrtc::DataBuffer(data, static_cast<bool>(isBinary)), &logSendAsyncError);
 }

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<jstring> jAddress = JavaString::toJava(env, ip);
+		JavaLocalRef<jstring> jType = JavaString::toJava(env, type);
+
+		env->CallVoidMethod(observer, javaClass->onSelectedCandidatePairChanged,
+			jAddress.get(), static_cast<jint>(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");
 	}
 }

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) {
+	}
+
 }

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);
+
 }