diff --git a/.github/workflows/gradle.yml b/.github/workflows/gradle.yml index 7df7e31..7b2c751 100644 --- a/.github/workflows/gradle.yml +++ b/.github/workflows/gradle.yml @@ -2,7 +2,7 @@ name: Build on: push: - branches: [ '*' ] + branches: [ main ] tags: [ '*' ] pull_request: branches: [ main ] diff --git a/.github/workflows/native-regressions.yml b/.github/workflows/native-regressions.yml new file mode 100644 index 0000000..8dba2a7 --- /dev/null +++ b/.github/workflows/native-regressions.yml @@ -0,0 +1,41 @@ +name: Native transport regressions + +on: + push: + branches: [ nxs-dev ] + pull_request: + branches: [ upstream, nxs-dev ] + workflow_dispatch: + +permissions: + contents: read + +jobs: + native-regressions: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + with: + fetch-depth: 0 + submodules: recursive + - uses: actions/setup-java@v5 + with: + java-version: '17' + distribution: temurin + cache: gradle + - uses: gradle/actions/setup-gradle@v5 + - name: Install native build dependencies + run: sudo apt-get update && sudo apt-get install -y cmake build-essential libssl-dev + - name: Native and JVM regressions + run: | + ./gradlew :nativeTransportProbe :test --no-daemon --max-workers=2 --no-configuration-cache \ + -Plibdatachannel.java-compiler-version=17 \ + -Plibdatachannel.test-native-path="$PWD/build/native-probe/libdatachannel-java.so" + - uses: actions/upload-artifact@v6 + if: always() + with: + name: native-regression-reports + path: | + build/test-results/ + build/reports/tests/ + build/native-probe/libdatachannel/Testing/ diff --git a/.github/workflows/publish-web.yml b/.github/workflows/publish-web.yml index eb2ab1d..b102b68 100644 --- a/.github/workflows/publish-web.yml +++ b/.github/workflows/publish-web.yml @@ -8,6 +8,7 @@ on: jobs: publish-web: + if: github.repository == 'pschichtel/libdatachannel-java' runs-on: ubuntu-latest steps: - uses: actions/checkout@v6 diff --git a/.github/workflows/readme-version.yml b/.github/workflows/readme-version.yml index 617af03..b500dcf 100644 --- a/.github/workflows/readme-version.yml +++ b/.github/workflows/readme-version.yml @@ -7,6 +7,7 @@ on: jobs: update: + if: github.repository == 'pschichtel/libdatachannel-java' runs-on: ubuntu-latest steps: - uses: actions/checkout@v6 diff --git a/.gitmodules b/.gitmodules index b7bd4b7..536b2d8 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,7 +1,6 @@ [submodule "jni/libdatachannel"] path = jni/libdatachannel - url = https://github.com/paullouisageneau/libdatachannel.git - branch = v0.24.1 + url = https://github.com/teamziax/libdatachannel.git [submodule "jni/cmake-conan"] path = jni/cmake-conan url = https://github.com/conan-io/cmake-conan.git diff --git a/build.gradle.kts b/build.gradle.kts index 0ee5e15..1639f18 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -61,7 +61,7 @@ fun produceVersion(): String { } } -version = produceVersion() +version = providers.gradleProperty("libdatachannel.development-version").getOrElse(produceVersion()) val isSnapshot = version.toString().endsWith("-SNAPSHOT") description = "${project.name} is a binding to the libdatachannel that feels native to Java developers." @@ -343,7 +343,22 @@ dependencies { annotationProcessor(libs.jniAccessGenerator) compileOnly(libs.jniAccessGenerator) - testImplementation(files(packageNativeForHost)) + if (providers.gradleProperty("libdatachannel.test-native-path").isPresent) { + // Package the selected binary on the classpath so child JVM lifecycle probes + // load the same checkout's library without relying on inherited properties. + val focusedNative = tasks.register("packageNativeForFocusedTests") { + dependsOn("compileNativeProbe") + archiveFileName = "focused-test-native.jar" + destinationDirectory = layout.buildDirectory.dir("focused-test-native") + from(providers.gradleProperty("libdatachannel.test-native-path")) { + into("native") + rename { "libdatachannel-java.so" } + } + } + testImplementation(files(focusedNative)) + } else { + testImplementation(files(packageNativeForHost)) + } } publishing.publications.withType().configureEach { @@ -424,7 +439,7 @@ val githubActions by tasks.registering(DefaultTask::class) { val deployRefPattern = """^refs/(?:tags/v\d+\.\d+\.\d+\.\d+|heads/main)$""".toRegex() val ref = System.getenv("GITHUB_REF")?.ifBlank { null }?.trim() - if (ref != null && deployRefPattern.matches(ref)) { + if (System.getenv("GITHUB_REPOSITORY") == "pschichtel/libdatachannel-java" && ref != null && deployRefPattern.matches(ref)) { logger.lifecycle("Job in $ref will deploy!") dependsOn(mavenCentralDeploy) } else { @@ -432,3 +447,72 @@ val githubActions by tasks.registering(DefaultTask::class) { dependsOn(tasks.assemble) } } + +// Focused local JNI transport tests using the existing library/package format. +// Uses system OpenSSL; the established dockcross release path remains available. +val configureNativeProbe by tasks.registering(Exec::class) { + dependsOn(tasks.compileJava) + commandLine("cmake", "-S", "jni", "-B", "build/native-probe", "-DCMAKE_POLICY_VERSION_MINIMUM=3.5", + "-DLIBDATACHANNEL_SOURCE_DIR=${project.file("jni/libdatachannel").absolutePath}", "-DUSE_SYSTEM_JUICE=OFF", + "-DCMAKE_BUILD_TYPE=Debug", "-DPROJECT_VERSION=${project.version}", "-DENABLE_LOCALHOST_ADDRESS=ON", + "-DTRANSPORT_TEARDOWN_TESTS=ON", "-DPENDING_MUX_TESTS=ON", "-DICE_UDP_MUX_TESTS=ON", + "-DRTC_ENABLE_TEST_DIAGNOSTICS=ON") +} +val compileNativeProbe by tasks.registering(Exec::class) { + dependsOn(configureNativeProbe) + commandLine("cmake", "--build", "build/native-probe", "--target", "datachannel-java", "transport-teardown-test", "ice-udp-mux-pending-test", "mux-pending-test", "mux-pending-lifetime-test", "mux-authentication-test", "ice-attribute-limits-test", "-j2") +} +val probeSourceSet = sourceSets.create("nativeProbe") { + java.srcDir("native-test") + compileClasspath += sourceSets.main.get().output + configurations.compileClasspath.get() + runtimeClasspath += sourceSets.main.get().output + configurations.runtimeClasspath.get() +} +dependencies { + add(probeSourceSet.implementationConfigurationName, libs.logbackClassic) +} +tasks.named(probeSourceSet.compileJavaTaskName) { + javaCompiler = javaToolchains.compilerFor { languageVersion = JavaLanguageVersion.of(17) } + options.release = 17 +} +val probeIdentity by tasks.registering(Exec::class) { + val dir = layout.buildDirectory.dir("probe-identity") + outputs.dir(dir) + doFirst { dir.get().asFile.mkdirs() } + commandLine("openssl", "req", "-x509", "-newkey", "ec", "-pkeyopt", "ec_paramgen_curve:prime256v1", + "-nodes", "-days", "1", "-subj", "/CN=native-probe", "-keyout", "build/probe-identity/key.pem", "-out", "build/probe-identity/cert.pem") +} +val probeEncryptedIdentity by tasks.registering(Exec::class) { + dependsOn(probeIdentity) + inputs.file("build/probe-identity/key.pem") + outputs.file("build/probe-identity/key-encrypted.pem") + commandLine("openssl", "pkcs8", "-topk8", "-in", "build/probe-identity/key.pem", + "-out", "build/probe-identity/key-encrypted.pem", "-v2", "aes-256-cbc", "-passout", "pass:test-only-password") +} +val runTransportNativeTests by tasks.registering(Exec::class) { + dependsOn(compileNativeProbe) + commandLine("ctest", "--test-dir", "build/native-probe/libdatachannel", "--output-on-failure", "-R", "transport.teardown|mux.pending|mux.authentication|ice.attribute.limits") +} +tasks.register("nativeTransportProbe") { + dependsOn(runTransportNativeTests, probeIdentity, probeEncryptedIdentity, tasks.named(probeSourceSet.classesTaskName), "nativeCallbackCleanupProbe", "nativeLoggingProbe") + javaLauncher = javaToolchains.launcherFor { languageVersion = JavaLanguageVersion.of(17) } + classpath = probeSourceSet.runtimeClasspath + mainClass = "tel.schich.libdatachannel.NativeTransportProbe" + systemProperty("libdatachannel.native.datachannel-java.path", layout.buildDirectory.file("native-probe/libdatachannel-java.so").get().asFile.absolutePath) + args("build/probe-identity/cert.pem", "build/probe-identity/key.pem", "build/probe-identity/key-encrypted.pem") +} + +tasks.register("nativeCallbackCleanupProbe") { + dependsOn(compileNativeProbe, tasks.named(probeSourceSet.classesTaskName)) + javaLauncher = javaToolchains.launcherFor { languageVersion = JavaLanguageVersion.of(17) } + classpath = probeSourceSet.runtimeClasspath + mainClass = "tel.schich.libdatachannel.CallbackCleanupProbe" + systemProperty("libdatachannel.native.datachannel-java.path", layout.buildDirectory.file("native-probe/libdatachannel-java.so").get().asFile.absolutePath) +} + +tasks.register("nativeLoggingProbe") { + dependsOn(compileNativeProbe, tasks.named(probeSourceSet.classesTaskName)) + javaLauncher = javaToolchains.launcherFor { languageVersion = JavaLanguageVersion.of(17) } + classpath = probeSourceSet.runtimeClasspath + mainClass = "tel.schich.libdatachannel.NativeLoggingProbe" + systemProperty("libdatachannel.native.datachannel-java.path", layout.buildDirectory.file("native-probe/libdatachannel-java.so").get().asFile.absolutePath) +} diff --git a/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts b/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts index bb74dfa..139a0e0 100644 --- a/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts +++ b/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts @@ -20,8 +20,9 @@ tasks.test { } tasks.compileJava { + options.release = 11 javaCompiler = javaToolchains.compilerFor { - languageVersion = JavaLanguageVersion.of(11) + languageVersion = JavaLanguageVersion.of(providers.gradleProperty("libdatachannel.java-compiler-version").getOrElse("11").toInt()) } } diff --git a/docs/contribution-provenance.md b/docs/contribution-provenance.md new file mode 100644 index 0000000..8ecb99c --- /dev/null +++ b/docs/contribution-provenance.md @@ -0,0 +1,18 @@ +# Contribution provenance + +This maintained fork compares `nxs-dev` with the upstream mirror at `b31809f14e0d8d62d4c800902e03b4108d95529d`. +The aggregate owned draft records the combined work; future external submissions +should separate independent fixes, acceptance APIs and lifecycle changes. + +| Source | Contribution | +| --- | --- | +| Upstream [`b31809f1`](https://github.com/pschichtel/libdatachannel-java/commit/b31809f14e0d8d62d4c800902e03b4108d95529d) | Baseline with thread attachment, unload synchronization, callback lifetime fixes and regression tests. | +| Zulu `39ec8c6`, `70efb59` | Original identity, ICE, packet-callback and local packaging work. Application-specific probes moved downstream with attribution. | +| Zulu `0812c7e`, `5544964`, `7885652` | Cleanup completion, callback-context checks and callback detachment. | +| Zulu `4a12f67`, `40f2c32` | Earlier deferred packet handling and limits, now replaced by native-owned pending requests. | +| Merge commits `97268df`, `66014e1` | History only; no duplicate patches. | +| Consolidation and current changes | Public C API use, asynchronous metadata callbacks, retained failure ownership, native log filtering, encrypted PEM support and generic regressions. | + +## Review follow-up, 8 September 2026 + +Settle waiting-request cancellation independently of the application executor, bind authenticated peer reuse and asynchronous destruction, add named statistics and an acceptance builder with shared identity configuration, retain compatibility adapters, and move construction diagnostics out of the public Java API. diff --git a/jni/CMakeLists.txt b/jni/CMakeLists.txt index 92b23f8..850e4c3 100644 --- a/jni/CMakeLists.txt +++ b/jni/CMakeLists.txt @@ -10,9 +10,10 @@ set(NO_WEBSOCKET ON CACHE BOOL "configure libdatachannel build") set(NO_MEDIA ON CACHE BOOL "configure libdatachannel build") set(NO_TESTS ON CACHE BOOL "configure libdatachannel build") set(NO_EXAMPLES ON CACHE BOOL "configure libdatachannel build") -add_subdirectory(libdatachannel) +set(LIBDATACHANNEL_SOURCE_DIR "${CMAKE_CURRENT_SOURCE_DIR}/libdatachannel" CACHE PATH "libdatachannel source checkout") +add_subdirectory(${LIBDATACHANNEL_SOURCE_DIR} libdatachannel) -include_directories(libdatachannel/include generated) +include_directories(${LIBDATACHANNEL_SOURCE_DIR}/include generated) include_directories(jdk) if(WIN32) include_directories(jdk/windows) @@ -54,6 +55,10 @@ add_library(datachannel-java SHARED src/util.c src/native_channel.c src/native_peer.c + src/native_mux.c src/native_track.c src/callback.c) target_link_libraries(datachannel-java PRIVATE datachannel-static) +if(CMAKE_SYSTEM_NAME STREQUAL "Linux") + target_link_options(datachannel-java PRIVATE -Wl,--no-undefined) +endif() diff --git a/jni/libdatachannel b/jni/libdatachannel index a02b751..c36c343 160000 --- a/jni/libdatachannel +++ b/jni/libdatachannel @@ -1 +1 @@ -Subproject commit a02b751917ac8afc8c58dc6f4461d25ff9465d48 +Subproject commit c36c34346d5ed6168b2e9f102852239cc1a1eb78 diff --git a/jni/src/init.c b/jni/src/init.c index 5289bd2..bb607c3 100644 --- a/jni/src/init.c +++ b/jni/src/init.c @@ -54,11 +54,18 @@ JNIEXPORT jint JNICALL JNI_OnLoad(JavaVM* jvm, void* reserved) { global_JVM = jvm; JNIEnv* env = get_jni_env_from_jvm(jvm); module_OnLoad(env); - rtcInitLogger(RTC_LOG_VERBOSE, &logger_callback); + rtcLogLevel level = (rtcLogLevel)call_tel_schich_libdatachannel_LibDataChannel_initialNativeLogLevel(env); + if ((*env)->ExceptionCheck(env)) return JNI_ERR; + rtcInitLogger(level, &logger_callback); rtcPreload(); return JNI_VERSION; } +JNIEXPORT void JNICALL Java_tel_schich_libdatachannel_LibDataChannel_setLogLevelNative( + JNIEnv* env, jclass clazz, jint level) { + rtcInitLogger((rtcLogLevel)level, &logger_callback); +} + JNIEXPORT void JNICALL JNI_OnUnload(JavaVM* jvm, void* reserved) { rtcCleanup(); JNIEnv* env = get_jni_env(); diff --git a/jni/src/native_mux.c b/jni/src/native_mux.c new file mode 100644 index 0000000..f86bf8a --- /dev/null +++ b/jni/src/native_mux.c @@ -0,0 +1,97 @@ +#include "util.h" +#include +#include +#include +#include + +struct ice_mux { + int listener; + jobject owner; + jmethodID dispatch; +}; + +JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_listenerIdNative( + JNIEnv *env, jclass clazz, jlong handle) { + return ((struct ice_mux *)(intptr_t)handle)->listener; +} + +static void RTC_API incoming_request(int listener, const rtcIceUdpMuxRequest *request, void *ptr) { + struct ice_mux *mux = ptr; + JNIEnv *env = get_jni_env(); + bool queued = false; + if (env && (*env)->PushLocalFrame(env, 4) == 0) { + jstring local = (*env)->NewStringUTF(env, request->localUfrag); + jstring remote = !(*env)->ExceptionCheck(env) ? (*env)->NewStringUTF(env, request->remoteUfrag) : NULL; + jstring address = !(*env)->ExceptionCheck(env) ? (*env)->NewStringUTF(env, request->remoteAddress) : NULL; + if (!(*env)->ExceptionCheck(env)) queued = (*env)->CallBooleanMethod(env, mux->owner, + mux->dispatch, (jlong)request->id, local, remote, address, (jint)request->remotePort); + if ((*env)->ExceptionCheck(env)) { (*env)->ExceptionClear(env); queued = false; } + (*env)->PopLocalFrame(env, NULL); + } else if (env && (*env)->ExceptionCheck(env)) (*env)->ExceptionClear(env); + if (!queued) rtcRejectIceUdpMuxRequest(listener, request->id); +} + +JNIEXPORT jlong JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_openNative( + JNIEnv *env, jobject self, jstring address, jint port, jint maxPending, jint timeoutMs) { + struct ice_mux *mux = calloc(1, sizeof(*mux)); + if (!mux) return 0; + mux->listener = -1; + mux->owner = (*env)->NewGlobalRef(env, self); + jclass clazz = !(*env)->ExceptionCheck(env) ? (*env)->GetObjectClass(env, self) : NULL; + mux->dispatch = clazz ? (*env)->GetMethodID(env, clazz, "dispatch", "(JLjava/lang/String;Ljava/lang/String;Ljava/lang/String;I)Z") : NULL; + if (clazz) (*env)->DeleteLocalRef(env, clazz); + const char *host = !(*env)->ExceptionCheck(env) ? (*env)->GetStringUTFChars(env, address, NULL) : NULL; + if (host && mux->owner && mux->dispatch) { + rtcIceUdpMuxListenerConfiguration config = {.bindAddress = host, .port = (uint16_t)port, + .maxPendingRequests = (unsigned int)maxPending, .requestTimeoutMs = (unsigned int)timeoutMs}; + mux->listener = rtcCreateIceUdpMuxListener(&config, incoming_request, mux); + } + if (host) (*env)->ReleaseStringUTFChars(env, address, host); + if (mux->listener < 0) { + if (mux->owner) (*env)->DeleteGlobalRef(env, mux->owner); + free(mux); + return 0; + } + return (jlong)(intptr_t)mux; +} + +JNIEXPORT void JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_closeNative( + JNIEnv *env, jclass clazz, jlong handle) { + struct ice_mux *mux = (struct ice_mux *)(intptr_t)handle; + if (rtcDeleteIceUdpMuxListener(mux->listener) != RTC_ERR_SUCCESS) { + throw_native_exception(env, "Failed to close ICE UDP mux listener"); + return; + } + // Native deletion waits for in-flight metadata callbacks before releasing this reference. + (*env)->DeleteGlobalRef(env, mux->owner); + free(mux); +} + +JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_acceptNative( + JNIEnv *env, jclass clazz, jint listener, jlong requestId, jint peer) { + return rtcAcceptIceUdpMuxPeer(listener, (uint64_t)requestId, peer); +} + +JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_rejectNative( + JNIEnv *env, jclass clazz, jint listener, jlong requestId) { + return rtcRejectIceUdpMuxRequest(listener, (uint64_t)requestId); +} + +JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_attachNative( + JNIEnv *env, jclass clazz, jint listener, jlong requestId, jint peer) { + return rtcAttachIceUdpMuxPeer(listener, (uint64_t)requestId, peer); +} + +JNIEXPORT jlongArray JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_statsNative( + JNIEnv *env, jclass clazz, jint listener) { + rtcIceUdpMuxListenerStats stats; + if (rtcGetIceUdpMuxListenerStats(listener, &stats) != RTC_ERR_SUCCESS) { + throw_native_exception(env, "ICE UDP mux statistics unavailable"); + return NULL; + } + jlong values[] = {(jlong)stats.received, (jlong)stats.rejected, (jlong)stats.agents, + (jlong)stats.mappedTuples, (jlong)stats.pendingRequests, (jlong)stats.notifications, (jlong)stats.duplicates}; + jlongArray result = (*env)->NewLongArray(env, 7); + if (result) (*env)->SetLongArrayRegion(env, result, 0, 7, values); + return result; +} diff --git a/jni/src/native_peer.c b/jni/src/native_peer.c index 6094a3f..0c0ec6e 100644 --- a/jni/src/native_peer.c +++ b/jni/src/native_peer.c @@ -48,8 +48,15 @@ void RTC_API handle_track(int pc, int trackHandle, void* ptr) { } SET_CALLBACK_INTERFACE_IMPL(rtcSetTrackCallback, handle_track) -JNIEXPORT jint JNICALL -Java_tel_schich_libdatachannel_LibDataChannelNative_rtcCreatePeerConnection(JNIEnv* env, jclass clazz, +struct incoming_peer { + int listener; + uint64_t request_id; + const char *remote_sdp; + rtcLocalDescriptionInit local_description; + int pc; +}; + +static jint create_peer(JNIEnv* env, jclass clazz, jobjectArray iceServers, jstring proxyServer, jstring bindAddress, jint certificateType, jint iceTransportPolicy, @@ -58,7 +65,7 @@ Java_tel_schich_libdatachannel_LibDataChannelNative_rtcCreatePeerConnection(JNIE jboolean disableAutoNegotiation, jboolean forceMediaTransport, jshort portRangeBegin, jshort portRangeEnd, - jint mtu, jint maxMessageSize) { + jint mtu, jint maxMessageSize, jstring certificateFile, jstring keyFile, jstring keyPassword, struct incoming_peer *incoming) { rtcConfiguration config = { .certificateType = certificateType, .iceTransportPolicy = iceTransportPolicy, @@ -110,7 +117,21 @@ Java_tel_schich_libdatachannel_LibDataChannelNative_rtcCreatePeerConnection(JNIE config.bindAddress = (*env)->GetStringUTFChars(env, bindAddress, NULL); } - jint result = (jint) rtcCreatePeerConnection(&config); + const char *certificate = certificateFile ? (*env)->GetStringUTFChars(env, certificateFile, NULL) : NULL; + const char *key = keyFile && !(*env)->ExceptionCheck(env) ? (*env)->GetStringUTFChars(env, keyFile, NULL) : NULL; + const char *pass = keyPassword && !(*env)->ExceptionCheck(env) ? (*env)->GetStringUTFChars(env, keyPassword, NULL) : NULL; + config.certificatePemFile = certificate; + config.keyPemFile = key; + config.keyPemPass = pass; + jint result = EXCEPTION_THROWN; + if (!(*env)->ExceptionCheck(env)) { + if (incoming) result = rtcPrepareIceUdpMuxPeer(incoming->listener, incoming->request_id, + &config, incoming->remote_sdp, &incoming->local_description, &incoming->pc); + else result = (jint) rtcCreatePeerConnection(&config); + } + if (pass) (*env)->ReleaseStringUTFChars(env, keyPassword, pass); + if (certificate) (*env)->ReleaseStringUTFChars(env, certificateFile, certificate); + if (key) (*env)->ReleaseStringUTFChars(env, keyFile, key); if (proxyServer != NULL) { (*env)->ReleaseStringUTFChars(env, proxyServer, config.proxyServer); @@ -130,6 +151,35 @@ Java_tel_schich_libdatachannel_LibDataChannelNative_rtcCreatePeerConnection(JNIE return result; } + +JNIEXPORT jint JNICALL +Java_tel_schich_libdatachannel_LibDataChannelNative_rtcCreatePeerConnection(JNIEnv* env, jclass clazz, + jobjectArray iceServers, jstring proxyServer, + jstring bindAddress, jint certificateType, + jint iceTransportPolicy, + jboolean enableIceTcp, + jboolean enableIceUdpMux, + jboolean disableAutoNegotiation, + jboolean forceMediaTransport, + jshort portRangeBegin, jshort portRangeEnd, + jint mtu, jint maxMessageSize) { + return create_peer(env, clazz, iceServers, proxyServer, bindAddress, certificateType, iceTransportPolicy, enableIceTcp, enableIceUdpMux, disableAutoNegotiation, forceMediaTransport, portRangeBegin, portRangeEnd, mtu, maxMessageSize, NULL, NULL, NULL, NULL); +} + +JNIEXPORT jint JNICALL +Java_tel_schich_libdatachannel_LibDataChannelNative_rtcCreatePeerConnectionWithIdentity(JNIEnv* env, jclass clazz, + jobjectArray iceServers, jstring proxyServer, + jstring bindAddress, jint certificateType, + jint iceTransportPolicy, + jboolean enableIceTcp, + jboolean enableIceUdpMux, + jboolean disableAutoNegotiation, + jboolean forceMediaTransport, + jshort portRangeBegin, jshort portRangeEnd, + jint mtu, jint maxMessageSize, jstring certificateFile, jstring keyFile, jstring keyPassword) { + return create_peer(env, clazz, iceServers, proxyServer, bindAddress, certificateType, iceTransportPolicy, enableIceTcp, enableIceUdpMux, disableAutoNegotiation, forceMediaTransport, portRangeBegin, portRangeEnd, mtu, maxMessageSize, certificateFile, keyFile, keyPassword, NULL); +} + JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_rtcClosePeerConnection(JNIEnv* env, jclass clazz, jint peerHandle) { return rtcClosePeerConnection(peerHandle); @@ -272,3 +322,95 @@ JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_setup return RTC_ERR_SUCCESS; } +JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_rtcSetLocalDescriptionWithIce( + JNIEnv *env, jclass clazz, jint peer, jstring type, jstring ufrag, jstring password) { + const char *t = type ? (*env)->GetStringUTFChars(env, type, NULL) : NULL; + if (type && !t) return EXCEPTION_THROWN; + const char *u = (*env)->GetStringUTFChars(env, ufrag, NULL); + const char *p = u ? (*env)->GetStringUTFChars(env, password, NULL) : NULL; + rtcLocalDescriptionInit init = {.iceUfrag = u, .icePwd = p}; + int result = p ? rtcSetLocalDescriptionEx(peer, t, &init) : EXCEPTION_THROWN; + if (p) (*env)->ReleaseStringUTFChars(env, password, p); + if (u) (*env)->ReleaseStringUTFChars(env, ufrag, u); + if (t) (*env)->ReleaseStringUTFChars(env, type, t); + return result; +} + +JNIEXPORT jlong JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_rtcGetPeerConnectionCreationAttempts( + JNIEnv *env, jclass clazz) { +#ifdef RTC_ENABLE_TEST_DIAGNOSTICS + return (jlong)rtcGetPeerConnectionCreationAttempts(); +#else + throw_native_exception(env, "Native construction diagnostics require a test build"); + return -1; +#endif +} + +struct peer_close_observer { + jobject peer; + jmethodID completed; +}; + +static void RTC_API peer_close_completed(int pc, void *ptr) { + struct peer_close_observer *observer = ptr; + JNIEnv *env = get_jni_env(); + if (env) { + (*env)->CallVoidMethod(env, observer->peer, observer->completed); + if ((*env)->ExceptionCheck(env)) (*env)->ExceptionClear(env); + (*env)->DeleteGlobalRef(env, observer->peer); + } + free(observer); +} + +JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_PeerConnection_closeAsyncNative( + JNIEnv *env, jclass clazz, jint pc, jobject peer) { + struct peer_close_observer *observer = calloc(1, sizeof(*observer)); + if (!observer) return RTC_ERR_FAILURE; + observer->peer = (*env)->NewGlobalRef(env, peer); + jclass type = observer->peer ? (*env)->GetObjectClass(env, peer) : NULL; + observer->completed = type ? (*env)->GetMethodID(env, type, "nativeCloseCompleted", "()V") : NULL; + if (type) (*env)->DeleteLocalRef(env, type); + int result = RTC_ERR_FAILURE; + if (observer->peer && observer->completed && !(*env)->ExceptionCheck(env)) + result = rtcClosePeerConnectionAsync(pc, peer_close_completed, observer); + if (result != RTC_ERR_SUCCESS) { + if (observer->peer) (*env)->DeleteGlobalRef(env, observer->peer); + free(observer); + } + return result; +} + +JNIEXPORT jint JNICALL +Java_tel_schich_libdatachannel_LibDataChannelNative_rtcClosePeerConnectionAndWait(JNIEnv* env, jclass clazz, jint peerHandle, jint timeoutMs) { + return rtcClosePeerConnectionAndWait(peerHandle, timeoutMs); +} + +JNIEXPORT jintArray JNICALL Java_tel_schich_libdatachannel_IceUdpMuxListener_prepareConfiguredNative( + JNIEnv *env, jclass clazz, jint listener, jlong requestId, + jobjectArray iceServers, jstring proxyServer, jstring bindAddress, jint certificateType, + jint iceTransportPolicy, jboolean enableIceTcp, jboolean enableIceUdpMux, + jboolean disableAutoNegotiation, jboolean forceMediaTransport, + jshort portRangeBegin, jshort portRangeEnd, jint mtu, jint maxMessageSize, + jstring certificateFile, jstring keyFile, jstring keyPassword, + jstring remoteDescription, jstring localUfrag, jstring localPassword) { + // Allocate the result before creating anything: ownership must never be lost on allocation failure. + jintArray result = (*env)->NewIntArray(env, 2); + if (!result) return NULL; + struct incoming_peer incoming = {.listener = listener, .request_id = (uint64_t)requestId, .pc = -1}; + incoming.remote_sdp = (*env)->GetStringUTFChars(env, remoteDescription, NULL); + incoming.local_description.iceUfrag = !(*env)->ExceptionCheck(env) ? (*env)->GetStringUTFChars(env, localUfrag, NULL) : NULL; + incoming.local_description.icePwd = !(*env)->ExceptionCheck(env) ? (*env)->GetStringUTFChars(env, localPassword, NULL) : NULL; + jint status = EXCEPTION_THROWN; + if (!(*env)->ExceptionCheck(env)) status = create_peer(env, clazz, iceServers, proxyServer, bindAddress, + certificateType, iceTransportPolicy, enableIceTcp, enableIceUdpMux, disableAutoNegotiation, + forceMediaTransport, portRangeBegin, portRangeEnd, mtu, maxMessageSize, + certificateFile, keyFile, keyPassword, &incoming); + if (incoming.remote_sdp) (*env)->ReleaseStringUTFChars(env, remoteDescription, incoming.remote_sdp); + if (incoming.local_description.iceUfrag) (*env)->ReleaseStringUTFChars(env, localUfrag, incoming.local_description.iceUfrag); + if (incoming.local_description.icePwd) (*env)->ReleaseStringUTFChars(env, localPassword, incoming.local_description.icePwd); + if (!(*env)->ExceptionCheck(env)) { + jint values[] = {status, incoming.pc}; + (*env)->SetIntArrayRegion(env, result, 0, 2, values); + } + return result; +} diff --git a/native-test/CallbackCleanupProbe.java b/native-test/CallbackCleanupProbe.java new file mode 100644 index 0000000..4b22eba --- /dev/null +++ b/native-test/CallbackCleanupProbe.java @@ -0,0 +1,71 @@ +package tel.schich.libdatachannel; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.AppenderBase; +import org.slf4j.LoggerFactory; + +import java.lang.ref.WeakReference; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** Exercises the production callback layout with both explicit peer close APIs. */ +public final class CallbackCleanupProbe { + private static final List> references = new ArrayList<>(); + + private static void cycle(boolean await) { + PeerConnection peer = PeerConnection.createPeer( + PeerConnectionConfiguration.DEFAULT.withDisableAutoNegotiation(true), Runnable::run); + references.add(new WeakReference<>(peer)); + peer.onStateChange.register((p, state) -> {}); + peer.onDataChannel.register((p, dc) -> {}); + for (String label : new String[] {"ReliableDataChannel", "UnreliableDataChannel"}) { + DataChannel channel = peer.createDataChannel(label); + references.add(new WeakReference<>(channel)); + channel.onMessage.register(DataChannelCallback.Message.handleBinary((dc, buffer) -> {})); + channel.onClosed.register(dc -> {}); + channel.onError.register((dc, error) -> {}); + } + if (await && !peer.closeAndAwait(Duration.ofSeconds(5))) { + throw new AssertionError("Native teardown timed out"); + } + peer.close(); // Also check repeated peer close after closeAndAwait. + } + + public static void main(String[] args) throws Exception { + Logger logger = (Logger) LoggerFactory.getLogger(LibDataChannel.class); + Level previousLevel = logger.getLevel(); + boolean previousAdditive = logger.isAdditive(); + List errors = new CopyOnWriteArrayList<>(); + AppenderBase appender = new AppenderBase<>() { + @Override protected void append(ILoggingEvent event) { + if (event.getLevel().isGreaterOrEqual(Level.ERROR)) errors.add(event.getFormattedMessage()); + } + }; + appender.start(); + logger.addAppender(appender); + logger.setLevel(Level.ERROR); + logger.setAdditive(false); + try { + for (int i = 0; i < 100; i++) cycle(i % 2 == 0); + // JNI keeps the peer listener in a global reference. Explicit deletion + // must release it so peers and their data-channel wrappers can be collected. + for (int i = 0; i < 100 && references.stream().anyMatch(r -> r.get() != null); i++) { + System.gc(); + Thread.sleep(20); + } + long retained = references.stream().filter(r -> r.get() != null).count(); + if (retained != 0) throw new AssertionError("Closed wrappers retained: " + retained); + if (!errors.isEmpty()) throw new AssertionError("Native cleanup errors: " + errors.size() + "; " + errors.get(0)); + System.out.println("callback-cleanup PASS cycles=100 collectedWrappers=300 nativeErrors=0"); + } finally { + logger.detachAppender(appender); + logger.setLevel(previousLevel); + logger.setAdditive(previousAdditive); + appender.stop(); + } + } +} diff --git a/native-test/NativeLoggingProbe.java b/native-test/NativeLoggingProbe.java new file mode 100644 index 0000000..05a759b --- /dev/null +++ b/native-test/NativeLoggingProbe.java @@ -0,0 +1,48 @@ +package tel.schich.libdatachannel; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.AppenderBase; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** Native log filtering must happen before JNI even when Java accepts every level. */ +public final class NativeLoggingProbe { + static void cycle() { + try (PeerConnection peer = PeerConnection.createPeer(PeerConnectionConfiguration.DEFAULT.withDisableAutoNegotiation(true))) { + peer.createDataChannel("log-threshold"); + peer.setLocalDescription("offer", "loggerFixture", "fixturePassword0000000000"); + if (!peer.closeAndAwait(Duration.ofSeconds(5))) throw new AssertionError("logging fixture teardown"); + } + } + public static void main(String[] args) throws Exception { + Logger logger = (Logger) LoggerFactory.getLogger(LibDataChannel.class); + Level previous = logger.getLevel(); + boolean additive = logger.isAdditive(); + List events = new CopyOnWriteArrayList<>(); + AppenderBase appender = new AppenderBase<>() { + @Override protected void append(ILoggingEvent event) { events.add(event); } + }; + appender.start(); logger.addAppender(appender); logger.setLevel(Level.TRACE); logger.setAdditive(false); + try { + if (LibDataChannel.logLevel() != LibDataChannel.LogLevel.WARNING) throw new AssertionError("default native threshold"); + LibDataChannel.setLogLevel(LibDataChannel.LogLevel.NONE); // Before loading/preloading native code. + cycle(); + if (!events.isEmpty()) throw new AssertionError("native logs crossed JNI with logging disabled"); + LibDataChannel.setLogLevel(LibDataChannel.LogLevel.DEBUG); + cycle(); + if (events.stream().noneMatch(e -> e.getLevel() == Level.DEBUG)) throw new AssertionError("native threshold did not update after initialization"); + LibDataChannel.setLogLevel(LibDataChannel.LogLevel.WARNING); + events.clear(); cycle(); + if (events.stream().anyMatch(e -> !e.getLevel().isGreaterOrEqual(Level.WARN))) throw new AssertionError("filtered native transport log crossed JNI"); + System.out.println("native-logging PASS default=WARNING beforeLoad=NONE afterLoad=DEBUG,WARNING nativeFilter=true"); + } finally { + LibDataChannel.setLogLevel(LibDataChannel.LogLevel.WARNING); + logger.detachAppender(appender); logger.setLevel(previous); logger.setAdditive(additive); appender.stop(); + } + } +} diff --git a/native-test/NativeTransportProbe.java b/native-test/NativeTransportProbe.java new file mode 100644 index 0000000..8d481e3 --- /dev/null +++ b/native-test/NativeTransportProbe.java @@ -0,0 +1,392 @@ +package tel.schich.libdatachannel; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.net.*; +import java.nio.*; +import java.nio.charset.StandardCharsets; +import java.nio.file.*; +import java.security.*; +import java.security.cert.CertificateFactory; +import java.time.Duration; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +/** Real UDP regression for asynchronous ICE acceptance and authenticated DTLS. */ +public final class NativeTransportProbe { + static final InetAddress LOOPBACK = InetAddress.getLoopbackAddress(); + static final int PORT = 49184; + static final String CLIENT_UFRAG = "clientFixtureUf", CLIENT_PASSWORD = "p".repeat(24); + static final String SERVER_PASSWORD = "fixedTestPassword0000000000000000"; + static String field(String sdp, String name) { + return sdp.lines().filter(x -> x.startsWith("a=" + name + ":")).findFirst().orElseThrow() + .substring(name.length() + 3).trim(); + } + static void check(boolean ok, String message) { if (!ok) throw new AssertionError(message); } + static void await(java.util.function.BooleanSupplier condition, String message) throws Exception { + for (int i = 0; i < 400 && !condition.getAsBoolean(); i++) Thread.sleep(5); + check(condition.getAsBoolean(), message); + } + static PeerConnection client() { + return PeerConnection.createPeer(PeerConnectionConfiguration.DEFAULT.withDisableAutoNegotiation(true).withBindAddress(LOOPBACK)); + } + static IceUdpMuxListener.Acceptance settings(Path certificate, Path key, String offer, + java.util.function.Consumer initializer) { + return IceUdpMuxListener.Acceptance.builder(offer, SERVER_PASSWORD) + .identity(new DtlsIdentity(certificate, key)).initialize(initializer).build(); + } + static String answer(String ufrag, String fingerprint) { + return "v=0\r\no=- 1 2 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\na=group:BUNDLE 0\r\n" + + "m=application 9 UDP/DTLS/SCTP webrtc-datachannel\r\nc=IN IP4 0.0.0.0\r\na=mid:0\r\na=setup:active\r\n" + + "a=ice-ufrag:" + ufrag + "\r\na=ice-pwd:" + SERVER_PASSWORD + "\r\na=fingerprint:sha-256 " + fingerprint + + "\r\na=sctp-port:5000\r\na=max-message-size:262144\r\na=candidate:1 1 UDP 2130706431 127.0.0.1 " + PORT + + " typ host\r\na=end-of-candidates\r\n"; + } + public static void main(String[] args) throws Exception { + Path certificate = Path.of(args[0]), key = Path.of(args[1]); + byte[] der; + try (var input = Files.newInputStream(certificate)) { + der = CertificateFactory.getInstance("X.509").generateCertificate(input).getEncoded(); + } + String fingerprint = HexFormat.ofDelimiter(":").withUpperCase().formatHex(MessageDigest.getInstance("SHA-256").digest(der)); + try (PeerConnection encrypted = PeerConnection.createPeer(PeerConnectionConfiguration.DEFAULT.withDisableAutoNegotiation(true), + Runnable::run, certificate, Path.of(args[2]), "test-only-password")) { + encrypted.createDataChannel("encrypted-key"); + encrypted.setLocalDescription(null, "encryptedIdentity", "publicTestPassword0000000"); + check(field(encrypted.localDescription(), "fingerprint").equals("sha-256 " + fingerprint), "encrypted key preserves identity"); + check(encrypted.closeAndAwait(Duration.ofSeconds(5)), "encrypted-key peer cleanup"); + } + firstRequest(certificate, key, false); + firstRequest(certificate, key, true); + cancelledRequests(certificate, key); + stalledExecutorCancellation(); + reusedPeer(certificate, key); + failedDecisions(certificate, key); + closeDuringInitialization(certificate, key); + for (int length : new int[] {167, 178, 256}) run(certificate, key, fingerprint, length, false); + run(certificate, key, fingerprint, 167, true); + } + + static byte[] binding(String localUfrag, String password) throws Exception { + byte[] username = (localUfrag + ":" + CLIENT_UFRAG).getBytes(StandardCharsets.US_ASCII); + ByteBuffer packet = ByteBuffer.allocate(2048); + packet.putShort((short) 1).putShort((short) 0).putInt(0x2112a442); + packet.putInt(0x12345678).putLong(0x0102030405060708L); + packet.putShort((short) 6).putShort((short) username.length).put(username); + while (packet.position() % 4 != 0) packet.put((byte) 0); + packet.putShort((short) 0x24).putShort((short) 4).putInt(2130706431); + packet.putShort((short) 0x802a).putShort((short) 8).putLong(0x7071727374757677L); + packet.putShort((short) 0x25).putShort((short) 0); + int integrity = packet.position(); + packet.putShort(2, (short) (integrity + 24 - 20)); + Mac mac = Mac.getInstance("HmacSHA1"); + mac.init(new SecretKeySpec(password.getBytes(StandardCharsets.US_ASCII), "HmacSHA1")); + byte[] digest = mac.doFinal(Arrays.copyOf(packet.array(), integrity)); + packet.putShort((short) 8).putShort((short) 20).put(digest); + return Arrays.copyOf(packet.array(), packet.position()); + } + + static void firstRequest(Path certificate, Path key, boolean forged) throws Exception { + String ufrag = "singleRequestServer"; + CompletableFuture decision = new CompletableFuture<>(); + ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(2); + AtomicInteger notifications = new AtomicInteger(); + try (PeerConnection client = client(); + IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, Runnable::run, request -> { + notifications.incrementAndGet(); arrivals.add(request); return decision; + }); DatagramSocket sender = new DatagramSocket(new InetSocketAddress(LOOPBACK, 0))) { + client.createDataChannel("fixture"); + client.setLocalDescription("offer", CLIENT_UFRAG, CLIENT_PASSWORD); + long before = PeerConnection.nativeCreationAttempts(); + byte[] packet = binding(ufrag, forged ? "wrongPassword0000000000000" : SERVER_PASSWORD); + sender.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); // Exactly one transmission. + IceUdpMuxListener.Request request = arrivals.poll(5, TimeUnit.SECONDS); + check(request != null, "initial request reaches asynchronous listener"); + check(request.localUfrag().equals(ufrag) && request.remoteUfrag().equals(CLIENT_UFRAG), "parsed request metadata"); + Thread.sleep(150); + check(PeerConnection.nativeCreationAttempts() == before && mux.statistics().agents() == 0, "no peer before decision"); + decision.complete(settings(certificate, key, client.localDescription(), peer -> {})); + if (forged) { + try { request.completion().toCompletableFuture().get(5, TimeUnit.SECONDS); throw new AssertionError("forged STUN accepted"); } + catch (ExecutionException expected) { /* Native integrity verification rejected it. */ } + check(PeerConnection.nativeCreationAttempts() == before && mux.statistics().agents() == 0 && mux.statistics().mappedTuples() == 0, + "forged integrity creates no peer or mapping"); + check(mux.failure() == null, "ordinary rejection keeps listener healthy"); + System.out.println("native-transport PASS forgedIntegrity=no-peer"); + } else { + PeerConnection host = request.completion().toCompletableFuture().get(5, TimeUnit.SECONDS); + try { + sender.setSoTimeout(3000); + boolean response = false; + for (int i = 0; i < 20 && !response; i++) { + DatagramPacket received = new DatagramPacket(new byte[2048], 2048); + sender.receive(received); + ByteBuffer data = ByteBuffer.wrap(received.getData(), 0, received.getLength()); + response = received.getLength() >= 20 && data.getShort(0) == 0x0101 && + data.getInt(8) == 0x12345678 && data.getLong(12) == 0x0102030405060708L; + } + check(response && notifications.get() == 1, "retained first request receives response without retransmission"); + System.out.println("native-transport PASS firstRequestSent=1 delayedAcceptance=true response=true"); + } finally { check(host.closeAndAwait(Duration.ofSeconds(5)), "single-request cleanup"); } + } + } + } + + static void cancelledRequests(Path certificate, Path key) throws Exception { + for (boolean close : new boolean[] {false, true}) { + CompletableFuture decision = new CompletableFuture<>(); + ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(1); + try (PeerConnection client = client(); + IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, 1, Duration.ofMillis(200), Runnable::run, + request -> { arrivals.add(request); return decision; }); + DatagramSocket sender = new DatagramSocket(new InetSocketAddress(LOOPBACK, 0))) { + client.createDataChannel("fixture"); client.setLocalDescription("offer", CLIENT_UFRAG, CLIENT_PASSWORD); + long before = PeerConnection.nativeCreationAttempts(); + byte[] packet = binding("cancelledServer", SERVER_PASSWORD); + sender.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); + IceUdpMuxListener.Request request = arrivals.poll(5, TimeUnit.SECONDS); + check(request != null, "pending cancellation fixture"); + if (close) mux.close(); + try { request.completion().toCompletableFuture().get(5, TimeUnit.SECONDS); throw new AssertionError("cancelled request accepted"); } + catch (ExecutionException | CancellationException expected) { } + decision.complete(settings(certificate, key, client.localDescription(), peer -> { throw new AssertionError("late initializer"); })); + Thread.sleep(50); + check(PeerConnection.nativeCreationAttempts() == before, "late decision cannot create a peer"); + } + } + System.out.println("native-transport PASS timeoutAndClose=cancelled lateDecisions=no-peer"); + } + + static void stalledExecutorCancellation() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + CountDownLatch blocked = new CountDownLatch(1), release = new CountDownLatch(1); + ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(1); + AtomicInteger handlers = new AtomicInteger(); + try (IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, 1, Duration.ofMillis(300), executor, + request -> { handlers.incrementAndGet(); arrivals.add(request); return new CompletableFuture<>(); }); + DatagramSocket sender = new DatagramSocket()) { + byte[] packet = binding("executorDeadline", SERVER_PASSWORD); + sender.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); + var request = arrivals.poll(3, TimeUnit.SECONDS); + check(request != null, "executor fixture receives first request"); + executor.execute(() -> { + blocked.countDown(); + try { release.await(); } catch (InterruptedException error) { Thread.currentThread().interrupt(); } + }); + check(blocked.await(3, TimeUnit.SECONDS), "application executor stalled"); + try { request.completion().toCompletableFuture().get(3, TimeUnit.SECONDS); throw new AssertionError("timeout accepted"); } + catch (ExecutionException expected) { } + check(mux.statistics().pendingRequests() == 0, "native timeout finished independently"); + byte[] second = binding("secondExecutorDeadline", SERVER_PASSWORD); + sender.send(new DatagramPacket(second, second.length, LOOPBACK, PORT)); + await(() -> mux.statistics().notifications() == 2, "second notification reaches Java"); + check(mux.statistics().pendingRequests() == 1, "Java timeout freed its admission slot before executor resumed"); + mux.close(); + release.countDown(); + executor.shutdown(); + check(executor.awaitTermination(3, TimeUnit.SECONDS), "executor drains cancelled work"); + check(handlers.get() == 1, "late queued handler cannot revive a closed request"); + } finally { release.countDown(); executor.shutdownNow(); } + System.out.println("native-transport PASS stalledExecutorTimeout=settled pendingSlot=reusable"); + } + + static void reusedPeer(Path certificate, Path key) throws Exception { + AtomicReference accepted = new AtomicReference<>(); + ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(4); + try (PeerConnection client = client()) { + client.createDataChannel("fixture"); client.setLocalDescription("offer", CLIENT_UFRAG, CLIENT_PASSWORD); + try (IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, Runnable::run, request -> { + arrivals.add(request); + return CompletableFuture.completedFuture(accepted.get() == null ? + settings(certificate, key, client.localDescription(), peer -> {}) : IceUdpMuxListener.Acceptance.reuse(accepted.get())); + }); DatagramSocket first = new DatagramSocket(); DatagramSocket second = new DatagramSocket(); DatagramSocket forged = new DatagramSocket()) { + byte[] packet = binding("reuseExistingPeer", SERVER_PASSWORD); + first.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); + var request = arrivals.poll(3, TimeUnit.SECONDS); check(request != null, "new peer notification"); + var peer = request.completion().toCompletableFuture().get(3, TimeUnit.SECONDS); + accepted.set(peer); + try { + await(() -> mux.statistics().mappedTuples() == 1, "first tuple attached"); + long constructions = PeerConnection.nativeCreationAttempts(); + second.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); + var additional = arrivals.poll(3, TimeUnit.SECONDS); check(additional != null, "additional tuple notification"); + check(additional.completion().toCompletableFuture().get(3, TimeUnit.SECONDS) == peer, "reuse returns the same Java wrapper"); + await(() -> mux.statistics().mappedTuples() == 2, "second authenticated tuple attached"); + check(mux.statistics().agents() == 1 && PeerConnection.nativeCreationAttempts() == constructions, + "additional tuple does not allocate a peer"); + byte[] invalid = binding("reuseExistingPeer", "incorrectPassword000000000"); + forged.send(new DatagramPacket(invalid, invalid.length, LOOPBACK, PORT)); + var bad = arrivals.poll(3, TimeUnit.SECONDS); check(bad != null, "forged tuple notification"); + try { bad.completion().toCompletableFuture().get(3, TimeUnit.SECONDS); throw new AssertionError("forged reuse accepted"); } + catch (ExecutionException expected) { } + check(mux.statistics().agents() == 1 && mux.statistics().mappedTuples() == 2, + "rejected tuple leaves caller-owned peer alive"); + peer.closeAsync().toCompletableFuture().get(5, TimeUnit.SECONDS); + check(mux.statistics().agents() == 0, "asynchronous close confirms resource destruction"); + } finally { peer.closeAndAwait(Duration.ofSeconds(5)); } + } + } + System.out.println("native-transport PASS existingPeerReuse=authenticated forgedReuse=preservesPeer asyncClose=complete"); + } + + static void failedDecisions(Path certificate, Path key) throws Exception { + for (String failureKind : new String[] {"handler", "initializer", "closedPeer", "configuration", "expired"}) { + ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(1); + AtomicReference initialized = new AtomicReference<>(); + try (PeerConnection client = client()) { + client.createDataChannel("fixture"); client.setLocalDescription("offer", CLIENT_UFRAG, CLIENT_PASSWORD); + String offer = client.localDescription(); + if (failureKind.equals("configuration")) offer = "v=0\r\no=- 1 1 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\n" + + "a=ice-ufrag:" + CLIENT_UFRAG + "\r\na=ice-pwd:" + CLIENT_PASSWORD + "\r\na=fingerprint:" + + field(offer, "fingerprint") + "\r\na=setup:actpass\r\n"; // Credentials, but no media section. + final String remoteOffer = offer; + try (IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, Runnable::run, request -> { + arrivals.add(request); + if (failureKind.equals("handler")) throw new IllegalStateException("fixture handler failure"); + return CompletableFuture.completedFuture(new IceUdpMuxListener.Acceptance( + PeerConnectionConfiguration.DEFAULT, remoteOffer, SERVER_PASSWORD, certificate, key, null, + Runnable::run, peer -> { + initialized.set(peer); + if (failureKind.equals("closedPeer")) peer.close(); + throw new IllegalStateException("fixture initializer failure"); + }, + failureKind.equals("expired") ? java.time.Instant.EPOCH : java.time.Instant.MAX)); + }); DatagramSocket sender = new DatagramSocket(new InetSocketAddress(LOOPBACK, 0))) { + long before = PeerConnection.nativeCreationAttempts(); + byte[] packet = binding("failureFixtureServer", SERVER_PASSWORD); + sender.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); + IceUdpMuxListener.Request request = arrivals.poll(5, TimeUnit.SECONDS); + check(request != null, "failure fixture metadata"); + try { request.completion().toCompletableFuture().get(10, TimeUnit.SECONDS); throw new AssertionError("failed decision accepted"); } + catch (ExecutionException expected) { } + boolean allocated = failureKind.equals("initializer") || failureKind.equals("closedPeer") || failureKind.equals("configuration"); + check(PeerConnection.nativeCreationAttempts() == before + (allocated ? 1 : 0), "creation ordering for " + failureKind); + check(mux.statistics().agents() == 0 && mux.statistics().mappedTuples() == 0 && mux.statistics().pendingRequests() == 0 && mux.failure() == null, + "failed decision frees native resources and leaves listener healthy: " + failureKind); + if (initialized.get() != null) check(initialized.get().closeAndAwait(Duration.ofMillis(1)), "binding cleanup was completed and is idempotent"); + } + } + } + AtomicInteger handlers = new AtomicInteger(); + try (IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, + task -> { throw new RejectedExecutionException("fixture full executor"); }, + request -> { handlers.incrementAndGet(); return CompletableFuture.completedFuture(null); }); + DatagramSocket sender = new DatagramSocket(new InetSocketAddress(LOOPBACK, 0))) { + long before = PeerConnection.nativeCreationAttempts(); + byte[] packet = binding("rejectedExecutor", SERVER_PASSWORD); + sender.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); + await(() -> mux.statistics().notifications() == 1 && mux.statistics().pendingRequests() == 0, "executor rejection removes pending request"); + check(handlers.get() == 0 && PeerConnection.nativeCreationAttempts() == before && mux.failure() == null, + "executor overload rejects without invoking handler or poisoning listener"); + } + System.out.println("native-transport PASS handlerFailure=true initializerFailure=cleaned closedInitializerPeer=cleaned configurationFailure=cleaned expiry=no-peer executorRejection=no-peer"); + } + + static void closeDuringInitialization(Path certificate, Path key) throws Exception { + AtomicReference listener = new AtomicReference<>(); + AtomicReference prepared = new AtomicReference<>(); + AtomicBoolean closeFinishedInsideInitializer = new AtomicBoolean(); + ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(1); + try (PeerConnection client = client()) { + client.createDataChannel("fixture"); client.setLocalDescription("offer", CLIENT_UFRAG, CLIENT_PASSWORD); + try (IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, Runnable::run, request -> { + arrivals.add(request); + return CompletableFuture.completedFuture(settings(certificate, key, client.localDescription(), peer -> { + prepared.set(peer); + peer.onStateChange.register((p, state) -> { + if (state == PeerState.RTC_CLOSED) listener.get().close(); + }); + try { + CompletableFuture.runAsync(() -> listener.get().close()).get(3, TimeUnit.SECONDS); + closeFinishedInsideInitializer.set(true); + } catch (Exception error) { throw new CompletionException(error); } + })); + }); DatagramSocket sender = new DatagramSocket(new InetSocketAddress(LOOPBACK, 0))) { + listener.set(mux); + byte[] packet = binding("closeDuringSetup", SERVER_PASSWORD); + sender.send(new DatagramPacket(packet, packet.length, LOOPBACK, PORT)); + IceUdpMuxListener.Request request = arrivals.poll(5, TimeUnit.SECONDS); + check(request != null, "concurrent-close fixture metadata"); + try { request.completion().toCompletableFuture().get(8, TimeUnit.SECONDS); throw new AssertionError("closed listener accepted"); } + catch (ExecutionException expected) { } + check(closeFinishedInsideInitializer.get(), "initializer can wait for concurrent listener close without deadlocking"); + check(prepared.get() != null && prepared.get().closeAndAwait(Duration.ofMillis(1)), "concurrent close cleans the prepared peer"); + } + } + System.out.println("native-transport PASS initializerConcurrentClose=no-deadlock preparedPeer=cleaned"); + } + + static void run(Path certificate, Path key, String fingerprint, int ufragLength, boolean wrongFingerprint) throws Exception { + String serverUfrag = "s".repeat(ufragLength); + CompletableFuture decision = new CompletableFuture<>(); + ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(4); + AtomicInteger notifications = new AtomicInteger(), channelMask = new AtomicInteger(), callbackGuards = new AtomicInteger(); + AtomicReference failure = new AtomicReference<>(); + CountDownLatch opened = new CountDownLatch(2), messages = new CountDownLatch(402), failed = new CountDownLatch(1); + PeerConnection host = null; + try (IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, Runnable::run, request -> { + notifications.incrementAndGet(); arrivals.add(request); return decision; + }); PeerConnection client = client()) { + long before = PeerConnection.nativeCreationAttempts(); + try (DatagramSocket noise = new DatagramSocket()) { + byte[] garbage = new byte[40]; noise.send(new DatagramPacket(garbage, garbage.length, LOOPBACK, PORT)); + await(() -> mux.statistics().received() > 0, "native receives garbage"); + check(notifications.get() == 0 && mux.statistics().agents() == 0 && mux.statistics().mappedTuples() == 0, "garbage stays native and creates no state"); + } + for (int channel = 0; channel < 2; channel++) { + String label = channel == 0 ? "ordered" : "unordered"; + var init = DataChannelInitSettings.DEFAULT.withReliability(new DataChannelReliability(channel == 1, false, 0, 0)); + DataChannel dc = client.createDataChannel(label, init); + dc.onOpen.register(d -> { + try { client.closeAndAwait(Duration.ofMillis(1)); failure.set(new AssertionError("callback teardown wait allowed")); } + catch (IllegalStateException expected) { callbackGuards.incrementAndGet(); } + opened.countDown(); + for (int i = 0; i < 201; i++) { + ByteBuffer message = ByteBuffer.allocateDirect(2); + message.put((byte) 0).put((byte) (label.equals("ordered") ? 1 : 2)).flip(); d.sendMessage(message); + } + }); + } + client.setLocalDescription("offer", CLIENT_UFRAG, CLIENT_PASSWORD); + client.setRemoteDescription(answer(serverUfrag, fingerprint), SessionDescriptionType.ANSWER); + IceUdpMuxListener.Request request = arrivals.poll(10, TimeUnit.SECONDS); + check(request != null, "STUN arrives before host peer exists"); + Thread.sleep(1100); // Force normal ICE retransmissions while application approval remains pending. + check(notifications.get() == 1 && mux.statistics().duplicates() > 0 && mux.statistics().agents() == 0 && + PeerConnection.nativeCreationAttempts() == before, "duplicates coalesce before native peer creation"); + String offer = client.localDescription(); + if (wrongFingerprint) { + String old = field(offer, "fingerprint"); char replacement = old.charAt(8) == '0' ? '1' : '0'; + offer = offer.replace(old, old.substring(0, 8) + replacement + old.substring(9)); + } + decision.complete(settings(certificate, key, offer, peer -> { + check(field(peer.localDescription(), "fingerprint").equals("sha-256 " + fingerprint), "published certificate identity"); + check(field(peer.localDescription(), "ice-ufrag").equals(serverUfrag), "explicit ICE username preserved"); + peer.onStateChange.register((p, state) -> { if (state == PeerState.RTC_FAILED) failed.countDown(); }); + peer.onDataChannel.register((p, dc) -> { + int bit = dc.label().equals("ordered") ? 1 : dc.label().equals("unordered") ? 2 : 0; + channelMask.getAndUpdate(mask -> mask | bit); + dc.onMessage.register(DataChannelCallback.Message.handleBinary((d, data) -> { + try { check(bit != 0 && data.remaining() == 2 && data.get() == 0 && data.get() == bit, "distinct channel payload"); messages.countDown(); } + catch (Throwable error) { failure.set(error); } + })); + }); + })); + host = request.completion().toCompletableFuture().get(5, TimeUnit.SECONDS); + if (wrongFingerprint) { + check(failed.await(15, TimeUnit.SECONDS), "DTLS rejects incorrect client fingerprint"); + check(opened.getCount() == 2 && channelMask.get() == 0, "wrong certificate opens no channels"); + System.out.println("native-transport PASS wrongRemoteFingerprint=dtls-rejected channels=0"); + } else { + check(opened.await(10, TimeUnit.SECONDS) && messages.await(10, TimeUnit.SECONDS), "both channels deliver 402 messages"); + check(failure.get() == null && callbackGuards.get() == 2 && channelMask.get() == 3, "callback and data-channel checks"); + long[] stats = mux.stats(); + check(notifications.get() == 1 && stats[5] == 1 && stats[2] == 1 && stats[3] == 1 && stats[0] > 10, + "transport traffic stays native after one admission callback"); + System.out.println("native-transport PASS ufragChars=" + ufragLength + " admissionCallbacks=1 duplicates=" + stats[6] + + " datagrams=" + stats[0] + " channels=2 messages=402"); + } + } finally { if (host != null) check(host.closeAndAwait(Duration.ofSeconds(5)), "accepted peer native cleanup"); } + } +} diff --git a/native-test/README.md b/native-test/README.md new file mode 100644 index 0000000..603e119 --- /dev/null +++ b/native-test/README.md @@ -0,0 +1,128 @@ +# Native transport regressions + +```sh +git submodule update --init --recursive +./gradlew :nativeTransportProbe --no-daemon --max-workers=2 -Plibdatachannel.java-compiler-version=17 +``` + +The focused Linux x86_64 build uses JDK 17, CMake, a C/C++ compiler, system OpenSSL +development files and the `openssl` CLI. Library bytecode remains compatible with +Java 11. The normal dockcross build remains available for portable artifacts. +The Java probes reserve loopback UDP ports 49184 and 49195. Native mux tests also +reserve their documented ports; do not run competing listeners there. + +`NativeTransportProbe` sends an initial STUN request exactly once, delays the +application decision, and checks that the retained request receives a response. +A forged STUN integrity value creates no peer or address mapping. Timeout and +listener closure cancel pending attempts; later decisions cannot create peers. +Handler and initializer failures, executor rejection and expired settings also +reject safely. An initializer can close its peer or wait for another thread to +close the listener without losing native ownership or deadlocking. + +The full connection tests use explicit ICE usernames of 167, 178 and 256 +characters, a supplied PEM certificate, and two data channels carrying 402 +messages. Repeated STUN requests during approval produce one admission callback. +Subsequent transport traffic stays native. An incorrect client certificate +fingerprint fails DTLS before either channel opens. A separate case imports a +password-encrypted PEM key and checks the certificate fingerprint. + +`NativeLoggingProbe` enables every Java log level, then verifies that native +filtering suppresses messages before JNI. It checks configuration before library +loading and changes after initialization. The default native threshold is +`WARNING`; `LibDataChannel.setLogLevel(...)` changes it for the process. + +The suite also checks bounded transport teardown from an external owner thread +and forbids waiting inside native event callbacks. The native teardown test +stalls a worker and retains a transport reference, so completion cannot be +mistaken for task submission or handle removal. `CallbackCleanupProbe` closes +100 peers and verifies that all 300 peer/channel wrappers become collectible +without native invalid-handle errors. To run the normal JNI lifecycle tests with +a focused binary, pass `-Plibdatachannel.test-native-path=/absolute/library.so` +to the Gradle `test` task. + +## Incoming connections + +`IceUdpMuxListener` delivers immutable username fragments and the source address. +The handler returns a `CompletionStage`; a null acceptance rejects the +request. Native code retains the STUN bytes, handles duplicates and verifies +STUN integrity before constructing the peer. JNI uses only libdatachannel's +public C API. + +```java +var listener = new IceUdpMuxListener(bindAddress, port, executor, request -> { + var settings = validate(request.localUfrag(), request.remoteUfrag()); + request.completion().whenComplete((peer, error) -> recordOutcome(peer, error)); + return CompletableFuture.completedFuture(IceUdpMuxListener.Acceptance + .builder(settings.remoteOffer(), settings.localPassword()) + .configuration(configuration) + .identity(new DtlsIdentity(certificatePath, keyPath)) + .peerExecutor(executor) + .initialize(peer -> installCallbacks(peer)) + .expiresAt(settings.expiresAt()).build()); +}); +``` + +The handler and initializer run through the supplied executor. A bounded internal +queue ensures even a direct executor does not run application code inside JNI +request dispatch. The initializer installs peer callbacks before native code +continues the retained request. Accepted peers belong to the caller. Close them +before closing the listener. + +The default limit is 256 pending attempts and a five-second deadline. Both are +configurable, up to 4096 attempts and 30 seconds. An acceptance can also carry an +application expiry time, checked before peer preparation and before final +acceptance. Handler failures, overload and timeouts reject the individual attempt. +A failed prepared peer remains owned until native teardown completes; only then +does the request's completion stage fail. `failure()` reports an unexpected +cleanup failure, without treating ordinary admission rejection as listener failure. + +`statistics()` returns immutable named counters for received datagrams, rejections, +ICE agents, mapped tuples, pending requests, notifications and duplicates. The old +positional `stats()` adapter and `Acceptance` constructors are deprecated and retained +for existing consumers. + +`PeerConnection.closeAsync()` returns a `CompletionStage` after native transport +destruction and Java cleanup. It is safe from event callbacks and uses native completion +notification rather than blocking a Java worker on each closing peer. The blocking +`closeAndAwait(Duration)` convenience returns false on timeout without releasing +ownership. Repeated calls return true after cleanup completes. + +Unprepared deadline/close cancellation releases the Java admission slot independently +of the application executor. Completion continuations run on the common completion +pool, so user code cannot block the deadline or JNI callback thread. An initializer +already running cannot be forcibly stopped: its prepared peer remains owned until +initialization returns and cleanup completes. Cancellation is checked before final +acceptance; native expiry also prevents stale attachment. + +Return `Acceptance.reuse(existingPeer)` to approve another source tuple for a peer. +Native code checks the existing ICE credentials and retains its SDP, DTLS identity, +channels and caller ownership, including on failed attachment. Use the optional +expiry argument when the application approval has its own deadline. Applications +remain responsible for deciding whether an address change is allowed. + +The certificate overload uses upstream `rtcConfiguration` PEM fields; explicit +ICE credentials use upstream `rtcSetLocalDescriptionEx`. Key provisioning and +rotation remain application policy. Peers created without a supplied identity +use native-generated certificates. + +## Local packaging + +From a clean committed tree, `scripts/package-development.sh [maven-directory]` +runs native and JVM regressions and writes an immutable local artifact under +`io.github.teamziax:libdatachannel-java:.0-dev.`. +The classifier is `x86_64`. `provenance.json` records all three source SHAs and +artifact hashes. This binary targets the current host's system OpenSSL/ABI; it is +not a portable dockcross release. Nothing is uploaded. Rebuild headers and JNI +together after changing the pinned native version. + +For local CMake experiments, `LIBDATACHANNEL_SOURCE_DIR` can select a separate +libdatachannel checkout. The Gradle probe and packaging tasks explicitly select +the pinned submodule and bundled libjuice again. + +Construction-attempt diagnostics are package-private test instrumentation and require +`RTC_ENABLE_TEST_DIAGNOSTICS=ON` in the native build. Ordinary builds omit the counter. +The probe suite additionally covers stalled application executors, same-peer tuple +attachment, forged attachment, reentrant listener close, scoped C++ preparation and +asynchronous destruction completion. + +Detailed [contributor and source attribution](../docs/contribution-provenance.md) is retained separately. diff --git a/scripts/package-development.sh b/scripts/package-development.sh new file mode 100755 index 0000000..37cdb4a --- /dev/null +++ b/scripts/package-development.sh @@ -0,0 +1,54 @@ +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "$0")/.." +# An explicit destination is a local Maven repository. This never uploads. +output=${1:-build/development-maven} +output=$(realpath -m "$output") +if test -n "$(git status --porcelain --untracked-files=normal)"; then + echo 'Refusing an immutable development version from a dirty checkout' >&2 + exit 1 +fi +revision=$(git rev-parse HEAD) +native_version=$(sed -n 's/^#define RTC_VERSION "\([^"]*\)"/\1/p' jni/libdatachannel/include/rtc/version.h) +version="${native_version}.0-dev.${revision}" +./gradlew :classes :nativeTransportProbe :test --no-daemon --max-workers=2 --no-configuration-cache \ + -Plibdatachannel.java-compiler-version=17 \ + -Plibdatachannel.development-version="$version" \ + -Plibdatachannel.test-native-path="$PWD/build/native-probe/libdatachannel-java.so" +python3 - "$output" "$version" "$revision" <<'PY' +import hashlib, json, pathlib, subprocess, sys, zipfile +output, version, revision = pathlib.Path(sys.argv[1]), sys.argv[2], sys.argv[3] +root = output / 'io/github/teamziax/libdatachannel-java' / version +root.mkdir(parents=True, exist_ok=True) +def add(jar, path, name): + info = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0)) + info.compress_type = zipfile.ZIP_DEFLATED + info.external_attr = 0o100644 << 16 + jar.writestr(info, path.read_bytes()) +with zipfile.ZipFile(root / f'libdatachannel-java-{version}.jar', 'w') as jar: + for folder in [pathlib.Path('build/classes/java/main'), pathlib.Path('build/resources/main')]: + if folder.exists(): + for path in sorted(folder.rglob('*')): + if path.is_file(): add(jar, path, path.relative_to(folder).as_posix()) +with zipfile.ZipFile(root / f'libdatachannel-java-{version}-x86_64.jar', 'w') as jar: + add(jar, pathlib.Path('build/native-probe/libdatachannel-java.so'), 'native/libdatachannel-java.so') +(root / f'libdatachannel-java-{version}.pom').write_text(f'''4.0.0 +io.github.teamziaxlibdatachannel-java{version} +Mozilla Public License 2.0https://www.mozilla.org/MPL/2.0/ +org.slf4jslf4j-api2.0.17 +\n''') +def head(path): + return subprocess.check_output(['git', '-C', path, 'rev-parse', 'HEAD'], text=True).strip() +provenance = { + 'coordinates': f'io.github.teamziax:libdatachannel-java:{version}', + 'bindingRevision': revision, + 'libdatachannelRevision': head('jni/libdatachannel'), + 'libjuiceRevision': head('jni/libdatachannel/deps/libjuice'), + 'platform': 'linux-x86_64', + 'nativeBuild': 'system OpenSSL, Debug, current host ABI; not a portable release', + 'checks': ['nativeTransportProbe', 'nativeCallbackCleanupProbe', 'nativeLoggingProbe', 'test'], + 'sha256': {path.name: hashlib.sha256(path.read_bytes()).hexdigest() for path in sorted(root.iterdir()) if path.is_file()}, +} +(root / 'provenance.json').write_text(json.dumps(provenance, indent=2) + '\n') +print(root) +PY diff --git a/src/main/java/tel/schich/libdatachannel/DataChannel.java b/src/main/java/tel/schich/libdatachannel/DataChannel.java index 7f1ae0d..536ffd5 100644 --- a/src/main/java/tel/schich/libdatachannel/DataChannel.java +++ b/src/main/java/tel/schich/libdatachannel/DataChannel.java @@ -104,17 +104,18 @@ public void sendMessage(String message) { */ @Override public void close() { - if (rtcClose(channelHandle) != ERR_INVALID) { - rtcDeleteDataChannel(channelHandle); - } - - peer.dropChannelState(channelHandle); + // Listener shutdown unregisters native callbacks, so the handle must still exist. onOpen.close(); onClosed.close(); onError.close(); onMessage.close(); onBufferedAmountLow.close(); onAvailable.close(); + + if (rtcClose(channelHandle) != ERR_INVALID) { + rtcDeleteDataChannel(channelHandle); + } + peer.dropChannelState(channelHandle); } /** diff --git a/src/main/java/tel/schich/libdatachannel/DtlsIdentity.java b/src/main/java/tel/schich/libdatachannel/DtlsIdentity.java new file mode 100644 index 0000000..dec2082 --- /dev/null +++ b/src/main/java/tel/schich/libdatachannel/DtlsIdentity.java @@ -0,0 +1,25 @@ +package tel.schich.libdatachannel; + +import java.nio.file.Path; +import java.util.Objects; +import org.eclipse.jdt.annotation.Nullable; + +/** A paired PEM certificate and private key supplied by the application. */ +public final class DtlsIdentity { + private final Path certificate, privateKey; + private final @Nullable String password; + + public DtlsIdentity(Path certificate, Path privateKey) { + this(certificate, privateKey, null); + } + + public DtlsIdentity(Path certificate, Path privateKey, @Nullable String password) { + this.certificate = Objects.requireNonNull(certificate, "certificate"); + this.privateKey = Objects.requireNonNull(privateKey, "privateKey"); + this.password = password; + } + + public Path certificate() { return certificate; } + public Path privateKey() { return privateKey; } + public @Nullable String password() { return password; } +} diff --git a/src/main/java/tel/schich/libdatachannel/EventListenerContainer.java b/src/main/java/tel/schich/libdatachannel/EventListenerContainer.java index 5023525..65825e6 100644 --- a/src/main/java/tel/schich/libdatachannel/EventListenerContainer.java +++ b/src/main/java/tel/schich/libdatachannel/EventListenerContainer.java @@ -12,6 +12,8 @@ import java.util.function.Consumer; public class EventListenerContainer implements Closeable { + private static final ThreadLocal IN_CALLBACK = ThreadLocal.withInitial(() -> false); + static boolean inCallback() { return IN_CALLBACK.get(); } private static final Logger LOGGER = LoggerFactory.getLogger(EventListenerContainer.class); private final String eventName; @@ -40,6 +42,9 @@ void invoke(Consumer invoker) { return; } executor.execute(() -> { + boolean previous = IN_CALLBACK.get(); + IN_CALLBACK.set(true); + try { for (T listener : this.listeners) { try { invoker.accept(listener); @@ -47,6 +52,7 @@ void invoke(Consumer invoker) { LOGGER.error("Handler for event {} failed!", eventName, t); } } + } finally { if (previous) IN_CALLBACK.set(true); else IN_CALLBACK.remove(); } }); } diff --git a/src/main/java/tel/schich/libdatachannel/IceUdpMuxListener.java b/src/main/java/tel/schich/libdatachannel/IceUdpMuxListener.java new file mode 100644 index 0000000..baca560 --- /dev/null +++ b/src/main/java/tel/schich/libdatachannel/IceUdpMuxListener.java @@ -0,0 +1,417 @@ +package tel.schich.libdatachannel; + +import org.eclipse.jdt.annotation.Nullable; +import tel.schich.jniaccess.JNIAccess; + +import java.net.InetAddress; +import java.nio.file.Path; +import java.time.Duration; +import java.time.Instant; +import java.util.Objects; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; + +/** + * Accepts incoming ICE connections on one UDP endpoint. Native code retains the + * first STUN request while the application decides; packets never pass through Java. + * Accepted peers are caller-owned and should be closed before this listener. + */ +public final class IceUdpMuxListener implements AutoCloseable { + @FunctionalInterface + public interface Handler { + /** Complete with connection settings to accept, or null to reject. */ + CompletionStage onRequest(Request request) throws Exception; + } + + /** Parsed, untrusted metadata from an incoming STUN request. */ + public static final class Request { + private final long id; + private final String localUfrag, remoteUfrag, remoteAddress; + private final int remotePort; + private final CompletableFuture completion = new CompletableFuture<>(); + private final AtomicBoolean settling = new AtomicBoolean(); + private final AtomicReference cancellation = new AtomicReference<>(); + private volatile ScheduledFuture timeout; + + private Request(long id, String localUfrag, String remoteUfrag, String remoteAddress, int remotePort) { + this.id = id; + this.localUfrag = localUfrag; + this.remoteUfrag = remoteUfrag; + this.remoteAddress = remoteAddress; + this.remotePort = remotePort; + } + + public long id() { return id; } + public String localUfrag() { return localUfrag; } + public String remoteUfrag() { return remoteUfrag; } + public String remoteAddress() { return remoteAddress; } + public int remotePort() { return remotePort; } + + /** Completes after acceptance, or after any failed prepared peer has been torn down. */ + public CompletionStage completion() { return completion.minimalCompletionStage(); } + } + + /** Settings returned by application admission. Fingerprint checks remain enabled. */ + public static final class Acceptance { + final PeerConnectionConfiguration configuration; + final String remoteDescription, localPassword; + final @Nullable Path certificate, key; + final @Nullable String keyPassword; + final Executor peerExecutor; + final Consumer initializer; + final Instant expiresAt; + final @Nullable PeerConnection existingPeer; + + /** @deprecated Use {@link #builder(String, String)}. */ + @Deprecated + public Acceptance(PeerConnectionConfiguration configuration, String remoteDescription, String localPassword, + @Nullable Path certificate, @Nullable Path key, @Nullable String keyPassword, + Executor peerExecutor, Consumer initializer) { + this(configuration, remoteDescription, localPassword, certificate, key, keyPassword, + peerExecutor, initializer, Instant.MAX); + } + + /** @deprecated Use {@link #builder(String, String)}. */ + @Deprecated + public Acceptance(PeerConnectionConfiguration configuration, String remoteDescription, String localPassword, + @Nullable Path certificate, @Nullable Path key, @Nullable String keyPassword, + Executor peerExecutor, Consumer initializer, Instant expiresAt) { + this.configuration = Objects.requireNonNull(configuration, "configuration"); + this.remoteDescription = Objects.requireNonNull(remoteDescription, "remoteDescription"); + this.localPassword = Objects.requireNonNull(localPassword, "localPassword"); + if ((certificate == null) != (key == null)) throw new IllegalArgumentException("Certificate/key must be paired"); + if (keyPassword != null && key == null) throw new IllegalArgumentException("A key password requires an identity"); + this.certificate = certificate; + this.key = key; + this.keyPassword = keyPassword; + this.peerExecutor = Objects.requireNonNull(peerExecutor, "peerExecutor"); + this.initializer = Objects.requireNonNull(initializer, "initializer"); + this.expiresAt = Objects.requireNonNull(expiresAt, "expiresAt"); + this.existingPeer = null; + } + + private Acceptance(PeerConnection peer, Instant expiresAt) { + this.configuration = PeerConnectionConfiguration.DEFAULT; + this.remoteDescription = this.localPassword = ""; + this.certificate = this.key = null; + this.keyPassword = null; + this.peerExecutor = Runnable::run; + this.initializer = ignored -> {}; + this.expiresAt = Objects.requireNonNull(expiresAt, "expiresAt"); + this.existingPeer = Objects.requireNonNull(peer, "peer"); + } + + /** Approve another tuple for a caller-owned peer without replacing its identity or channels. */ + public static Acceptance reuse(PeerConnection peer) { return reuse(peer, Instant.MAX); } + public static Acceptance reuse(PeerConnection peer, Instant expiresAt) { return new Acceptance(peer, expiresAt); } + + public static Builder builder(String remoteDescription, String localPassword) { + return new Builder(remoteDescription, localPassword); + } + + public static final class Builder { + private final String remoteDescription, localPassword; + private PeerConnectionConfiguration configuration = PeerConnectionConfiguration.DEFAULT; + private @Nullable DtlsIdentity identity; + private Executor peerExecutor = Runnable::run; + private Consumer initializer = ignored -> {}; + private Instant expiresAt = Instant.MAX; + + private Builder(String remoteDescription, String localPassword) { + this.remoteDescription = Objects.requireNonNull(remoteDescription, "remoteDescription"); + this.localPassword = Objects.requireNonNull(localPassword, "localPassword"); + } + public Builder configuration(PeerConnectionConfiguration value) { configuration = Objects.requireNonNull(value); return this; } + public Builder identity(DtlsIdentity value) { identity = Objects.requireNonNull(value); return this; } + public Builder peerExecutor(Executor value) { peerExecutor = Objects.requireNonNull(value); return this; } + public Builder initialize(Consumer value) { initializer = Objects.requireNonNull(value); return this; } + public Builder expiresAt(Instant value) { expiresAt = Objects.requireNonNull(value); return this; } + public Acceptance build() { + return new Acceptance(configuration, remoteDescription, localPassword, + identity == null ? null : identity.certificate(), identity == null ? null : identity.privateKey(), + identity == null ? null : identity.password(), peerExecutor, initializer, expiresAt); + } + } + } + + /** Snapshot of native listener activity. Counts belong to this endpoint. */ + public static final class Statistics { + private final long received, rejected, agents, mappedTuples, pendingRequests, notifications, duplicates; + private Statistics(long[] values) { + received = values[0]; rejected = values[1]; agents = values[2]; mappedTuples = values[3]; + pendingRequests = values[4]; notifications = values[5]; duplicates = values[6]; + } + public long received() { return received; } + public long rejected() { return rejected; } + public long agents() { return agents; } + public long mappedTuples() { return mappedTuples; } + public long pendingRequests() { return pendingRequests; } + public long notifications() { return notifications; } + public long duplicates() { return duplicates; } + } + + private static final ScheduledThreadPoolExecutor DEADLINES = new ScheduledThreadPoolExecutor(1, task -> { + Thread thread = new Thread(task, "ice-admission-deadlines"); + thread.setDaemon(true); + return thread; + }); + private static final ScheduledThreadPoolExecutor CLEANUP = new ScheduledThreadPoolExecutor(2, task -> { + Thread thread = new Thread(task, "ice-admission-cleanup"); + thread.setDaemon(true); + return thread; + }); + static { + DEADLINES.setRemoveOnCancelPolicy(true); + DEADLINES.setKeepAliveTime(30, TimeUnit.SECONDS); + DEADLINES.allowCoreThreadTimeOut(true); + CLEANUP.setKeepAliveTime(30, TimeUnit.SECONDS); + CLEANUP.allowCoreThreadTimeOut(true); + } + + private final Executor executor; + private final Handler handler; + private final int requestTimeoutMillis, maxPendingRequests; + private final ConcurrentMap requests = new ConcurrentHashMap<>(); + private final AtomicReference failure = new AtomicReference<>(); + private final ThreadPoolExecutor dispatchQueue; + private long handle; + private int listenerId; + + public IceUdpMuxListener(InetAddress bindAddress, int port, Executor executor, Handler handler) { + this(bindAddress, port, 256, Duration.ofSeconds(5), executor, handler); + } + + public IceUdpMuxListener(InetAddress bindAddress, int port, int maxPendingRequests, Duration requestTimeout, + Executor executor, Handler handler) { + if (port < 1 || port > 65535) throw new IllegalArgumentException("Explicit UDP port required"); + if (maxPendingRequests < 1 || maxPendingRequests > 4096) throw new IllegalArgumentException("Pending limit must be 1..4096"); + long timeout = Objects.requireNonNull(requestTimeout, "requestTimeout").toMillis(); + if (timeout < 1 || timeout > 30000) throw new IllegalArgumentException("Request timeout must be 1..30000 ms"); + this.requestTimeoutMillis = (int) timeout; + this.maxPendingRequests = maxPendingRequests; + this.executor = Objects.requireNonNull(executor, "executor"); + this.handler = Objects.requireNonNull(handler, "handler"); + // The trampoline also makes a direct executor safe: user code never runs in JNI dispatch. + this.dispatchQueue = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(maxPendingRequests), task -> { + Thread thread = new Thread(task, "ice-admission-dispatch"); + thread.setDaemon(true); + return thread; + }); + LibDataChannel.initialize(); + synchronized (this) { + handle = openNative(Objects.requireNonNull(bindAddress, "bindAddress").getHostAddress(), port, + maxPendingRequests, requestTimeoutMillis); + if (handle == 0) { + dispatchQueue.shutdownNow(); + throw new IllegalStateException("Cannot acquire ICE UDP mux endpoint"); + } + listenerId = listenerIdNative(handle); + } + } + + // JNI only copies these strings. Native code owns the packet and coalesces duplicates. + @JNIAccess + private boolean dispatch(long id, String localUfrag, String remoteUfrag, String address, int port) { + if (requests.size() >= maxPendingRequests) return false; + Request request = new Request(id, localUfrag, remoteUfrag, address, port); + if (requests.putIfAbsent(id, request) != null) return true; + try { + request.timeout = DEADLINES.schedule(() -> cancel(request, + new TimeoutException("Incoming ICE request expired")), requestTimeoutMillis, TimeUnit.MILLISECONDS); + dispatchQueue.execute(() -> execute(request, () -> decide(request))); + return true; + } catch (Throwable error) { + requests.remove(id, request); + if (request.timeout != null) request.timeout.cancel(false); + complete(request, null, error); + return false; + } + } + + private void execute(Request request, Runnable task) { + try { executor.execute(task); } + catch (Throwable error) { cancel(request, error); } + } + + private void decide(Request request) { + if (!requests.containsKey(request.id)) return; + try { + CompletionStage decision = Objects.requireNonNull(handler.onRequest(request), "decision stage"); + decision.whenComplete((settings, error) -> { + if (error != null || settings == null) + cancel(request, error != null ? error : new CancellationException("Incoming ICE request rejected")); + else execute(request, () -> finish(request, settings, null)); + }); + } catch (Throwable error) { cancel(request, error); } + } + + private synchronized int openListenerId() { + if (handle == 0) throw new CancellationException("ICE listener closed"); + return listenerId; + } + + private static void complete(Request request, @Nullable PeerConnection peer, @Nullable Throwable error) { + // Removing the slot is synchronous. Application continuations use a separate + // completion worker and cannot block the deadline or JNI dispatch thread. + CompletableFuture.runAsync(() -> { + if (error == null) request.completion.complete(peer); + else request.completion.completeExceptionally(error); + }); + } + + private void cancel(Request request, Throwable cause) { + request.cancellation.compareAndSet(null, cause); + if (!request.settling.compareAndSet(false, true)) return; + if (request.timeout != null) request.timeout.cancel(false); + int id; + synchronized (this) { id = handle == 0 ? -1 : listenerId; } + if (id >= 0) rejectNative(id, request.id); + requests.remove(request.id, request); + complete(request, null, cause); + } + + private static void checkCancellation(Request request) { + Throwable cause = request.cancellation.get(); + if (cause != null) throw new CompletionException(cause); + } + + private void finish(Request request, @Nullable Acceptance settings, @Nullable Throwable error) { + if (!request.settling.compareAndSet(false, true)) return; + PeerConnection peer = null; + int preparedHandle = -1; + try { + int id = openListenerId(); + checkCancellation(request); + if (error != null) throw new CompletionException(error); + if (settings == null) throw new CancellationException("Incoming ICE request rejected"); + if (!Instant.now().isBefore(settings.expiresAt)) throw new TimeoutException("Admission settings expired"); + if (settings.existingPeer != null) { + int result = attachNative(id, request.id, settings.existingPeer.peerHandle); + if (result != 0) throw new IllegalStateException("Cannot attach incoming ICE tuple: " + result); + if (request.timeout != null) request.timeout.cancel(false); + requests.remove(request.id, request); + complete(request, settings.existingPeer, null); + return; + } + int[] prepared = prepareNative(id, request.id, settings.configuration, settings.remoteDescription, + request.localUfrag, settings.localPassword, + settings.certificate == null ? null : settings.certificate.toString(), + settings.key == null ? null : settings.key.toString(), settings.keyPassword); + preparedHandle = prepared[1]; + if (preparedHandle >= 0) peer = PeerConnection.fromNative(preparedHandle, settings.peerExecutor); + if (prepared[0] != 0) throw new IllegalStateException("Cannot prepare incoming ICE peer: " + prepared[0]); + if (peer == null) throw new IllegalStateException("Native prepare returned no peer"); + peer.installNativeListener(); + // Application code may close this listener or the peer. Never hold a listener lock here. + settings.initializer.accept(peer); + id = openListenerId(); + checkCancellation(request); + if (!Instant.now().isBefore(settings.expiresAt)) throw new TimeoutException("Admission settings expired"); + if (peer.preparationCloseRequested()) throw new CancellationException("Incoming peer closed during setup"); + int result = acceptNative(id, request.id, peer.peerHandle); + if (result != 0) throw new IllegalStateException("Cannot accept incoming ICE peer: " + result); + if (!peer.releasePreparation()) throw new CancellationException("Incoming peer closed during acceptance"); + if (request.timeout != null) request.timeout.cancel(false); + requests.remove(request.id, request); + complete(request, peer, null); + return; + } catch (Throwable cause) { + error = cause; + int id; + synchronized (this) { id = handle == 0 ? -1 : listenerId; } + if (id >= 0) rejectNative(id, request.id); + } + if (request.timeout != null) request.timeout.cancel(false); + if (preparedHandle < 0) { + requests.remove(request.id, request); + complete(request, null, error); + } else cleanup(preparedHandle, peer, request, error); + } + + private void cleanup(int preparedHandle, @Nullable PeerConnection peer, Request request, Throwable error) { + if (peer != null) { + peer.closeAsync().whenComplete((ignored, closeError) -> { + if (closeError != null) { failure.set(closeError); return; } + try { + peer.releasePreparation(); + peer.close(); + requests.remove(request.id, request); + complete(request, null, error); + } catch (Throwable failureCause) { failure.set(failureCause); } + }); + return; + } + // No wrapper could be constructed. Keep the raw handle until teardown succeeds. + CLEANUP.execute(() -> { + try { + if (LibDataChannelNative.rtcClosePeerConnectionAndWait(preparedHandle, 5000) == 0) { + LibDataChannelNative.rtcDeletePeerConnection(preparedHandle); + requests.remove(request.id, request); + complete(request, null, error); + return; + } + } catch (Throwable closeError) { failure.set(closeError); } + CLEANUP.schedule(() -> cleanup(preparedHandle, null, request, error), 100, TimeUnit.MILLISECONDS); + }); + } + + /** An admission infrastructure failure, for diagnostics. */ + public @Nullable Throwable failure() { return failure.get(); } + + public synchronized Statistics statistics() { + if (handle == 0) throw new IllegalStateException("ICE listener closed"); + return new Statistics(statsNative(listenerId)); + } + + /** @deprecated Use {@link #statistics()} for named counters. */ + @Deprecated + public synchronized long[] stats() { + if (handle == 0) throw new IllegalStateException("ICE listener closed"); + return statsNative(listenerId); + } + + @Override + public void close() { + synchronized (this) { + if (handle == 0) return; + long closing = handle; + handle = 0; + try { closeNative(closing); } + catch (Throwable error) { handle = closing; throw error; } + } + // A direct executor may be running user code on this thread. Closing the + // listener cancels its request; it must not interrupt that application code. + dispatchQueue.getQueue().clear(); + dispatchQueue.shutdown(); + for (Request request : requests.values()) + cancel(request, new CancellationException("ICE listener closed")); + } + + private native long openNative(String address, int port, int maxPendingRequests, int requestTimeoutMillis); + private static native void closeNative(long handle); + private static int[] prepareNative(int handle, long requestId, PeerConnectionConfiguration config, + String remoteDescription, String localUfrag, String localPassword, + @Nullable String certificate, @Nullable String key, @Nullable String keyPassword) { + return prepareConfiguredNative(handle, requestId, PeerConnection.iceUrisToStrings(config.iceServers), + config.proxyServer == null ? null : config.proxyServer.toASCIIString(), + config.bindAddress == null ? null : config.bindAddress.getHostAddress(), + config.certificateType.state, config.iceTransportPolicy.state, config.enableIceTcp, + config.enableIceUdpMux, config.disableAutoNegotiation, config.forceMediaTransport, + config.portRangeBegin, config.portRangeEnd, config.mtu, config.maxMessageSize, + certificate, key, keyPassword, remoteDescription, localUfrag, localPassword); + } + private static native int[] prepareConfiguredNative(int handle, long requestId, + String @Nullable [] iceServers, @Nullable String proxyServer, @Nullable String bindAddress, + int certificateType, int iceTransportPolicy, boolean enableIceTcp, boolean enableIceUdpMux, + boolean disableAutoNegotiation, boolean forceMediaTransport, short portRangeBegin, short portRangeEnd, + int mtu, int maxMessageSize, @Nullable String certificate, @Nullable String key, @Nullable String keyPassword, + String remoteDescription, String localUfrag, String localPassword); + private static native int acceptNative(int handle, long requestId, int peer); + private static native int attachNative(int handle, long requestId, int peer); + private static native int rejectNative(int handle, long requestId); + private static native long[] statsNative(int handle); + private static native int listenerIdNative(long handle); +} diff --git a/src/main/java/tel/schich/libdatachannel/LibDataChannel.java b/src/main/java/tel/schich/libdatachannel/LibDataChannel.java index 4e6d31f..33f60d3 100644 --- a/src/main/java/tel/schich/libdatachannel/LibDataChannel.java +++ b/src/main/java/tel/schich/libdatachannel/LibDataChannel.java @@ -5,16 +5,42 @@ import tel.schich.jniaccess.JNIAccess; import java.lang.ref.Cleaner; +import java.util.Objects; public abstract class LibDataChannel { static final Cleaner CLEANER = Cleaner.create(); private static final Logger LOGGER = LoggerFactory.getLogger(LibDataChannel.class); private static volatile boolean initialized = false; + // Read by JNI_OnLoad before rtcPreload can emit transport logs. + private static int nativeLogLevel = LogLevel.WARNING.value; + + /** Native filtering happens before a message crosses into Java. */ + public enum LogLevel { + NONE(0), FATAL(1), ERROR(2), WARNING(3), INFO(4), DEBUG(5), VERBOSE(6); + + final int value; + LogLevel(int value) { this.value = value; } + } public static final String LIB_NAME = "datachannel-java"; private LibDataChannel() {} + /** Sets the process-wide native log threshold, before or after initialization. */ + public static synchronized void setLogLevel(LogLevel level) { + nativeLogLevel = Objects.requireNonNull(level, "level").value; + if (initialized) setLogLevelNative(nativeLogLevel); + } + + public static synchronized LogLevel logLevel() { + return LogLevel.values()[nativeLogLevel]; + } + + @JNIAccess + private static int initialNativeLogLevel() { return nativeLogLevel; } + + private static native void setLogLevelNative(int level); + /** * Initializes the library by loading the native library. */ diff --git a/src/main/java/tel/schich/libdatachannel/LibDataChannelNative.java b/src/main/java/tel/schich/libdatachannel/LibDataChannelNative.java index d14329a..71971a3 100644 --- a/src/main/java/tel/schich/libdatachannel/LibDataChannelNative.java +++ b/src/main/java/tel/schich/libdatachannel/LibDataChannelNative.java @@ -10,11 +10,15 @@ class LibDataChannelNative { } static native int rtcCreatePeerConnection(String @Nullable [] iceServers, @Nullable String proxyServer, @Nullable String bindAddress, int certificateType, int iceTransportPolicy, boolean enableIceTcp, boolean enableIceUdpMux, boolean disableAutoNegotiation, boolean forceMediaTransport, short portRangeBegin, short portRangeEnd, int mtu, int maxMessageSize); + static native int rtcCreatePeerConnectionWithIdentity(String @Nullable [] iceServers, @Nullable String proxyServer, @Nullable String bindAddress, int certificateType, int iceTransportPolicy, boolean enableIceTcp, boolean enableIceUdpMux, boolean disableAutoNegotiation, boolean forceMediaTransport, short portRangeBegin, short portRangeEnd, int mtu, int maxMessageSize, @Nullable String certificateFile, @Nullable String keyFile, @Nullable String keyPassword); static native int setupPeerConnectionListener(int peerHandle, PeerConnectionListener listener); + static native long rtcGetPeerConnectionCreationAttempts(); static native int rtcClosePeerConnection(int peerHandle); + static native int rtcClosePeerConnectionAndWait(int peerHandle, int timeoutMs); static native int rtcDeletePeerConnection(int peerHandle); static native int rtcSetLocalDescription(int peerHandle, String type); + static native int rtcSetLocalDescriptionWithIce(int peerHandle, @Nullable String type, String ufrag, String password); static native String rtcGetLocalDescription(int peerHandle); static native String rtcGetLocalDescriptionType(int peerHandle); static native int rtcSetRemoteDescription(int peerHandle, String sdp, @Nullable String type); diff --git a/src/main/java/tel/schich/libdatachannel/PeerConnection.java b/src/main/java/tel/schich/libdatachannel/PeerConnection.java index 1b25a2c..edaf4a3 100644 --- a/src/main/java/tel/schich/libdatachannel/PeerConnection.java +++ b/src/main/java/tel/schich/libdatachannel/PeerConnection.java @@ -1,6 +1,10 @@ package tel.schich.libdatachannel; import org.eclipse.jdt.annotation.Nullable; +import tel.schich.jniaccess.JNIAccess; +import java.nio.file.Path; +import static tel.schich.libdatachannel.LibDataChannelNative.rtcCreatePeerConnectionWithIdentity; +import static tel.schich.libdatachannel.LibDataChannelNative.rtcSetLocalDescriptionWithIce; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,6 +49,11 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executor; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; public class PeerConnection implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(PeerConnection.class); @@ -54,6 +63,10 @@ public class PeerConnection implements Closeable { private final ConcurrentMap channels; private final ConcurrentMap tracks; private final Cleaner.Cleanable cleanable; + private volatile boolean nativeTeardownComplete; + private @Nullable CompletableFuture closeCompletion; + private final Object preparationLock = new Object(); + private boolean preparationOwned, preparationCloseRequested; final PeerConnectionListener listener; public final EventListenerContainer onLocalDescription; @@ -89,7 +102,29 @@ private PeerConnection(int peerHandle, final Executor executor) { }); } - private static String @Nullable [] iceUrisToStrings(@Nullable Collection uris) { + static PeerConnection fromNative(int handle, Executor executor) { + PeerConnection peer = new PeerConnection(handle, executor); + peer.preparationOwned = true; + return peer; + } + + boolean preparationCloseRequested() { + synchronized (preparationLock) { return preparationCloseRequested; } + } + + boolean releasePreparation() { + synchronized (preparationLock) { + if (preparationCloseRequested && !nativeTeardownComplete) return false; + preparationOwned = false; + return !preparationCloseRequested; + } + } + + void installNativeListener() { + wrapError("setupPeerConnectionListener", setupPeerConnectionListener(peerHandle, listener)); + } + + static String @Nullable [] iceUrisToStrings(@Nullable Collection uris) { if (uris == null || uris.isEmpty()) { return null; } @@ -111,15 +146,35 @@ private PeerConnection(int peerHandle, final Executor executor) { * @return the peer connection */ public static PeerConnection createPeer(PeerConnectionConfiguration config, Executor executor) { + return createPeer(config, executor, null, null); + } + + public static PeerConnection createPeer(PeerConnectionConfiguration config, Executor executor, DtlsIdentity identity) { + Objects.requireNonNull(identity, "identity"); + return createPeer(config, executor, identity.certificate(), identity.privateKey(), identity.password()); + } + + /** Creates a peer using a paired PEM DTLS certificate/key, or the default identity if both null. */ + public static PeerConnection createPeer(PeerConnectionConfiguration config, Executor executor, + @Nullable Path certificate, @Nullable Path key) { + return createPeer(config, executor, certificate, key, null); + } + + /** Imports an endpoint identity using the upstream native certificate configuration. */ + public static PeerConnection createPeer(PeerConnectionConfiguration config, Executor executor, + @Nullable Path certificate, @Nullable Path key, + @Nullable String keyPassword) { + if ((certificate == null) != (key == null)) throw new IllegalArgumentException("Certificate/key must be paired"); + if (keyPassword != null && key == null) throw new IllegalArgumentException("A key password requires an identity"); String proxyServer = null; if (config.proxyServer != null) { proxyServer = config.proxyServer.toASCIIString(); } String bindAddress = null; if (config.bindAddress != null) { - bindAddress = config.bindAddress.toString(); + bindAddress = config.bindAddress.getHostAddress(); } - int result = rtcCreatePeerConnection( + int result = rtcCreatePeerConnectionWithIdentity( iceUrisToStrings(config.iceServers), proxyServer, bindAddress, @@ -132,7 +187,9 @@ public static PeerConnection createPeer(PeerConnectionConfiguration config, Exec config.portRangeBegin, config.portRangeEnd, config.mtu, - config.maxMessageSize); + config.maxMessageSize, + certificate == null ? null : certificate.toString(), + key == null ? null : key.toString(), keyPassword); final PeerConnection peer = new PeerConnection(wrapError("rtcCreatePeerConnection", result), executor); setupPeerConnectionListener(peer.peerHandle, peer.listener); @@ -140,6 +197,9 @@ public static PeerConnection createPeer(PeerConnectionConfiguration config, Exec return peer; } + /** Diagnostic count at native peer construction, including failed attempts. */ + static long nativeCreationAttempts() { return LibDataChannelNative.rtcGetPeerConnectionCreationAttempts(); } + public static PeerConnection createPeer(PeerConnectionConfiguration config) { return createPeer(config, Runnable::run); } @@ -177,7 +237,7 @@ public void close() { } catch (Exception e) { LOGGER.warn("Failed to close channels of peer connection", e); } - cleanable.clean(); + // Detach callbacks before the cleaner deletes their native peer handle. onLocalDescription.close(); onLocalCandidate.close(); onStateChange.close(); @@ -186,8 +246,68 @@ public void close() { onSignalingStateChange.close(); onDataChannel.close(); onTrack.close(); + boolean deferDeletion; + synchronized (preparationLock) { + deferDeletion = preparationOwned; + if (deferDeletion) preparationCloseRequested = true; + } + if (deferDeletion) rtcClosePeerConnection(peerHandle); + else cleanable.clean(); } + /** + * Force-close and await native transport teardown before deleting the peer. + * Call only from an external owner thread, never a native callback. A timeout + * returns false and retains ownership so the caller can retry or fail closed. + */ + public boolean closeAndAwait(java.time.Duration timeout) { + if (EventListenerContainer.inCallback()) throw new IllegalStateException("Cannot await teardown from a native event callback"); + long millis = timeout.toMillis(); + if (millis < 1 || millis > 30_000) throw new IllegalArgumentException("Teardown timeout must be 1..30000 ms"); + try { + closeAsync().toCompletableFuture().get(millis, TimeUnit.MILLISECONDS); + return true; + } catch (TimeoutException timeoutError) { + return false; + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + return false; + } catch (ExecutionException error) { + throw new IllegalStateException("Native close failed", error.getCause()); + } + } + + /** + * Starts native closure without blocking a worker on teardown. Safe from event + * callbacks. Completion follows transport destruction and Java handle cleanup. + * A peer still being initialized remains owned by its incoming-request handler. + */ + public synchronized CompletionStage closeAsync() { + if (closeCompletion == null) { + closeCompletion = new CompletableFuture<>(); + int result = closeAsyncNative(peerHandle, this); + if (result != 0) closeCompletion.completeExceptionally( + new IllegalStateException("Cannot initiate native close: " + result)); + } + return closeCompletion.minimalCompletionStage(); + } + + @JNIAccess + private void nativeCloseCompleted() { + nativeTeardownComplete = true; + // Neither handle cleanup nor arbitrary user continuations run inside JNI completion. + CompletableFuture.runAsync(() -> { + CompletableFuture completion; + synchronized (this) { completion = closeCompletion; } + try { + close(); + completion.complete(null); + } catch (Throwable error) { completion.completeExceptionally(error); } + }); + } + + private static native int closeAsyncNative(int peerHandle, PeerConnection owner); + /** * Closes all Data Channels. */ @@ -211,6 +331,12 @@ public void closeChannels() { * * @param type (optional): type of the description ("offer", "answer", "pranswer", or "rollback") or NULL for autodetection. */ + /** Installs the exact local ICE credentials before gathering starts. */ + public void setLocalDescription(@Nullable String type, String ufrag, String password) { + if (ufrag.isEmpty() || password.isEmpty()) throw new IllegalArgumentException("ICE credentials required"); + wrapError("rtcSetLocalDescriptionWithIce", rtcSetLocalDescriptionWithIce(peerHandle, type, ufrag, password)); + } + public void setLocalDescription(String type) { wrapError("rtcSetLocalDescription", rtcSetLocalDescription(peerHandle, type)); }