fix: add OnBufferedAmountChange callback to RTCDataChannelObserver (#206)

Author: devopvoid

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

@@ -37,6 +37,7 @@ namespace jni
 			// DataChannelObserver implementation.
 			void OnStateChange() override;
 			void OnMessage(const webrtc::DataBuffer & buffer) override;
+			void OnBufferedAmountChange(uint64_t sent_data_size) override;
 
 		private:
 			class JavaRTCDataChannelObserverClass : public JavaClass
@@ -46,6 +47,7 @@ namespace jni
 
 					jmethodID onStateChange;
 					jmethodID onMessage;
+					jmethodID onBufferedAmountChange;
 			};
 
 		private:

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

@@ -48,11 +48,21 @@ namespace jni
 		ExceptionCheck(env);
 	}
 
+	void RTCDataChannelObserver::OnBufferedAmountChange(uint64_t sent_data_size)
+	{
+		JNIEnv * env = AttachCurrentThread();
+
+		env->CallVoidMethod(observer, javaClass->onBufferedAmountChange, static_cast<jlong>(sent_data_size));
+
+		ExceptionCheck(env);
+	}
+
 	RTCDataChannelObserver::JavaRTCDataChannelObserverClass::JavaRTCDataChannelObserverClass(JNIEnv * env)
 	{
 		jclass cls = FindClass(env, PKG"RTCDataChannelObserver");
 
 		onStateChange = GetMethod(env, cls, "onStateChange", "()V");
 		onMessage = GetMethod(env, cls, "onMessage", "(L" PKG "RTCDataChannelBuffer;)V");
+		onBufferedAmountChange = GetMethod(env, cls, "onBufferedAmountChange", "(J)V");
 	}
 }

webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelTests.java

@@ -18,18 +18,65 @@
 
 import static java.util.Objects.nonNull;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 class RTCDataChannelTests extends TestBase {
 
+	@Test
+	void bufferedAmountChangeCallback() throws Exception {
+		DataPeerConnection caller = new DataPeerConnection(factory);
+		DataPeerConnection callee = new DataPeerConnection(factory);
+
+		caller.setRemotePeerConnection(callee);
+		callee.setRemotePeerConnection(caller);
+
+		callee.setRemoteDescription(caller.createOffer());
+		caller.setRemoteDescription(callee.createAnswer());
+
+		caller.waitUntilConnected();
+		callee.waitUntilConnected();
+
+		// Prepare a latch-based observer to detect buffered amount change.
+		CountDownLatch latch = new CountDownLatch(1);
+
+		caller.getLocalDataChannel().registerObserver(new RTCDataChannelObserver() {
+			@Override
+			public void onBufferedAmountChange(long previousAmount) {
+				latch.countDown();
+			}
+
+			@Override
+			public void onStateChange() { }
+
+			@Override
+			public void onMessage(RTCDataChannelBuffer buffer) { }
+		});
+
+		// Send a large enough message to cause buffering (increase from 0).
+		byte[] big = new byte[64 * 1024]; // 64 KB
+		ByteBuffer data = ByteBuffer.wrap(big);
+		RTCDataChannelBuffer buffer = new RTCDataChannelBuffer(data, true);
+		caller.getLocalDataChannel().send(buffer);
+
+		// Wait for the callback to fire to avoid flakiness.
+		boolean signaled = latch.await(5, java.util.concurrent.TimeUnit.SECONDS);
+
+		assertTrue(signaled, "onBufferedAmountChange should be called when sending data");
+
+		caller.close();
+		callee.close();
+	}
+
 	@Test
 	void textMessage() throws Exception {
 		DataPeerConnection caller = new DataPeerConnection(factory);
@@ -98,6 +145,10 @@ public void onMessage(RTCDataChannelBuffer buffer) {
 			});
 		}
 
+		RTCDataChannel getLocalDataChannel() {
+			return localDataChannel;
+		}
+
 		List<String> getReceivedTexts() {
 			return receivedTexts;
 		}