diff --git a/.fernignore b/.fernignore index 4734b1ba..9f670640 100644 --- a/.fernignore +++ b/.fernignore @@ -1,4 +1,4 @@ -# Custom client wrappers (extend generated DeepgramApiClient with Bearer auth, session ID) +# Custom client wrappers (Bearer auth, session ID, and owned HTTP-resource lifecycle) # Flat paths (local generation strips package-prefix) src/main/java/DeepgramClient.java src/main/java/AsyncDeepgramClient.java @@ -20,8 +20,8 @@ src/main/java/com/deepgram/core/ClientOptions.java src/main/java/com/deepgram/core/transport/ # Bug fixes for maxRetries(0) semantics ("connect once, don't retry") and a -# configurable connectionTimeoutMs on ReconnectOptions (was hardcoded 4000ms). -# Pull this back out once the fixes are upstreamed into the Fern generator. +# configurable connectionTimeoutMs on ReconnectOptions (was hardcoded 4000ms), and cancellation +# of in-flight connection attempts. Pull this back out once the fixes are upstreamed into Fern. src/main/java/com/deepgram/core/ReconnectingWebSocketListener.java # Forward-compat patch: Fern's generated dispatcher routes any unrecognized message @@ -50,6 +50,11 @@ src/main/java/com/deepgram/resources/listen/v2/websocket/V2WebSocketClient.java src/main/java/com/deepgram/resources/listen/v1/websocket/V1WebSocketClient.java src/main/java/com/deepgram/resources/speak/v1/websocket/V1WebSocketClient.java +# All streaming WebSocket clients need null guards and terminal disconnect behavior so cleanup is +# safe before or during connect(). The Listen and Speak files are frozen above for other patches; +# Agent V1 needs its own freeze entry until Fern emits the guards. +src/main/java/com/deepgram/resources/agent/v1/websocket/V1WebSocketClient.java + # Restores the FLUX_RENEE_EN constant that generator 4.18.0 dropped. The voice is live: # POST /v2/speak?model=flux-renee-en returns 200 with valid audio, and the name resolves in the # server's model registry (an invented flux-* name is rejected with INVALID_QUERY_PARAMETER), so the diff --git a/AGENTS.md b/AGENTS.md index 39cb6388..2557f3f3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,7 +24,7 @@ How to identify: Current permanently frozen files: -- `src/main/java/com/deepgram/DeepgramClient.java`, `src/main/java/com/deepgram/AsyncDeepgramClient.java`, `src/main/java/com/deepgram/DeepgramClientBuilder.java`, `src/main/java/com/deepgram/AsyncDeepgramClientBuilder.java` - custom wrapper entrypoints that add Bearer auth, session ID support, and custom transport behavior on top of Fern's generated API client +- `src/main/java/com/deepgram/DeepgramClient.java`, `src/main/java/com/deepgram/AsyncDeepgramClient.java`, `src/main/java/com/deepgram/DeepgramClientBuilder.java`, `src/main/java/com/deepgram/AsyncDeepgramClientBuilder.java` - custom wrapper entrypoints that add Bearer auth, session ID support, custom transport behavior, and ownership-aware HTTP-resource lifecycle management on top of Fern's generated API client - `src/main/java/com/deepgram/core/transport/` - hand-written transport abstraction - `build.gradle`, `settings.gradle`, `gradle/`, `gradlew`, `gradlew.bat`, `pom.xml`, `Makefile` - build and project configuration - `README.md`, `CHANGELOG.md`, `CONTRIBUTING.md`, `LICENSE`, `docs/` - docs @@ -48,13 +48,15 @@ How to identify: Current temporarily frozen files: - `src/main/java/com/deepgram/core/ClientOptions.java` - preserves release-please version markers and correct SDK header constants that Fern currently overwrites; use the standard `.bak` swap/restore workflow during regen review. Since generator 4.18.0 Fern emits a `getSdkVersion()` helper reading `Package.getImplementationVersion()` instead of a literal. That *does* resolve in the published artifact (CI publishes via `mvn deploy -P release`, and `pom.xml`'s maven-jar-plugin sets `addDefaultImplementationEntries=true`, so the JAR manifest carries `Implementation-Version`), but it resolves to `null` under Gradle and in tests, where it silently falls back to a hardcoded literal the generator does not keep current. We keep the explicit literals because they are correct in every context and because `.github/release-please-config.json` already lists this file in `extra-files`, so release-please bumps it alongside `pom.xml`, `build.gradle`, and `.fern/metadata.json`. Fern also emits `User-Agent` with a `com.deepgram.` prefix while leaving `X-Fern-SDK-Name` on the `com.deepgram:` Maven-coordinate form; we keep both on the colon form. -- `src/main/java/com/deepgram/core/ReconnectingWebSocketListener.java` - carries bug fixes for `maxRetries(0)` semantics ("connect once, don't retry") and a configurable `connectionTimeoutMs` field (was hardcoded 4000ms), plus an `applyOptionsOverride(...)` hook used by `TransportWebSocketFactory` to apply per-transport reconnect policy; pull this back out once the fixes are upstreamed into the Fern generator. Use the standard `.bak` swap/restore workflow during regen review. +- `src/main/java/com/deepgram/core/ReconnectingWebSocketListener.java` - carries bug fixes for `maxRetries(0)` semantics ("connect once, don't retry"), a configurable `connectionTimeoutMs` field (was hardcoded 4000ms), cancellation of in-flight connection attempts, and an `applyOptionsOverride(...)` hook used by `TransportWebSocketFactory` to apply per-transport reconnect policy; pull this back out once the fixes are upstreamed into the Fern generator. Use the standard `.bak` swap/restore workflow during regen review. - `src/main/java/com/deepgram/resources/speak/v2/websocket/V2WebSocketClient.java` and `src/main/java/com/deepgram/resources/listen/v2/websocket/V2WebSocketClient.java` - forward-compat patch (both clients). Fern's generated `handleIncomingMessage` dispatcher routes any unrecognized message type to `onError` with "Update your SDK version...", which makes a benign new server control frame look fatal to a deployed client. Patched so the unrecognized-type branch is a no-op — the raw frame is already delivered via `onMessage(String)` earlier in the method, so consumers still see it. Mirrors the JS/Python SDKs' forward-compat behavior and is regression-guarded by `src/test/java/com/deepgram/SpeakV2ForwardCompatTest.java` and `src/test/java/com/deepgram/ListenV2ForwardCompatTest.java`. These two clients also carry the streaming query-param patches described in the next entry. Use the standard `.bak` swap/restore workflow during regen review; re-apply the no-op to both after regen, and unfreeze once the generator stops treating unknown frames as errors. - `src/main/java/com/deepgram/resources/listen/v1/websocket/V1WebSocketClient.java` and `src/main/java/com/deepgram/resources/speak/v1/websocket/V1WebSocketClient.java` (and the v2 clients above) - streaming query-param patches on the generated `connect()` builders. Two fixes: (1) multi-value serialization — array-valued params (listen: `keyterm`, `keywords`, `replace`, `search`, `tag`, `extra`, `language_hint`; speak: `tag`) were serialized with `String.valueOf(union.get())`, collapsing a `List` into one param (`keyterm=[a, b]`) instead of repeats (`keyterm=a&keyterm=b`); (2) an `additionalProperties` escape hatch — the builder exposes `additionalProperty(key, value)` for unmodeled params (e.g. `no_delay`) but `connect()` never emitted them to the URL. Both patched to route through `QueryStringMapper(arraysAsRepeats=true)`, matching the REST path. Use the standard `.bak` swap/restore workflow during regen review; re-apply after regen and unfreeze once the generator emits array params as repeats and serializes `additionalProperties` on the WS `connect()` path (tracked as an upstream Fern request). - Fields-less message types carrying a manual `hashCode()` patch (Fern generates `equals()` but no `hashCode()` for these, violating the Object contract): `src/main/java/com/deepgram/resources/listen/v2/types/ListenV2CloseStream.java`, `src/main/java/com/deepgram/resources/listen/v2/types/ListenV2ForceEndTurn.java`, `src/main/java/com/deepgram/resources/speak/v2/types/SpeakV2Close.java`, `src/main/java/com/deepgram/resources/speak/v2/types/SpeakV2Flush.java`, and the `AgentV1*` event types `src/main/java/com/deepgram/resources/agent/v1/types/{AgentV1ListenUpdated,AgentV1SpeakUpdated,AgentV1AgentAudioDone,AgentV1SettingsApplied,AgentV1UserStartedSpeaking,AgentV1KeepAlive,AgentV1ThinkUpdated,AgentV1PromptUpdated,AgentV1ForceEndTurn}.java`. Use the standard `.bak` swap/restore workflow during regen review; drop the patches and unfreeze all of them once the generator emits a matching equals/hashCode pair for fields-less types (tracked as an upstream Fern request). - `src/main/java/com/deepgram/types/DeepgramModel.java` - restores the `FLUX_RENEE_EN` constant that generator 4.18.0 dropped. The voice is live: `POST /v2/speak?model=flux-renee-en` returns 200 with valid audio, and the name resolves in the server's model registry (an invented `flux-*` name is rejected with `INVALID_QUERY_PARAMETER`), so the removal is a spec regression rather than a retirement, and dropping the constant would break 0.8.0 callers for nothing. Five touchpoints: the constant, the `Value` enum entry, the `visit()` case, the `valueOf()` case, and the `Visitor` method. **This file is unlike the other temporarily frozen ones — it receives frequent additive spec changes (4.18.0 alone added 25 constants), so on the next regen do NOT restore the `.bak` wholesale.** Diff the `.bak` against the newly generated file, carry forward every new voice, and re-apply only the `FLUX_RENEE_EN` touchpoints. Drop the patch and unfreeze once the spec lists the voice again (tracked as an upstream spec request). - Union default-variant fix on the agent listen-provider unions: `src/main/java/com/deepgram/resources/agent/v1/types/AgentV1UpdateListenListenProvider.java`, `src/main/java/com/deepgram/resources/agent/v1/types/AgentV1SettingsAgentListenProvider.java`, `src/main/java/com/deepgram/resources/agent/v1/types/AgentV1SettingsAgentContextListenProvider.java`. `version` is an optional discriminator, so a provider payload without it is valid (and is what 0.7.x emits), but Fern points `@JsonTypeInfo` `defaultImpl` at the empty-bodied `_UnknownValue`, so such a payload deserializes to an unknown variant carrying `null` — `getProvider()` returns `null` and re-serialization emits `{"provider":null}`, silently dropping the provider on the wire. Patched to `defaultImpl = V2Value` on each; guarded by `src/test/java/com/deepgram/AgentSettingsProviderDefaultTest.java`. Use the standard `.bak` swap/restore workflow during regen review; drop the patches and unfreeze once the generator stops defaulting unions to the empty `_UnknownValue` (tracked as an upstream Fern request). +- `src/main/java/com/deepgram/resources/listen/v1/websocket/V1WebSocketClient.java`, `src/main/java/com/deepgram/resources/listen/v2/websocket/V2WebSocketClient.java`, `src/main/java/com/deepgram/resources/speak/v1/websocket/V1WebSocketClient.java`, `src/main/java/com/deepgram/resources/speak/v2/websocket/V2WebSocketClient.java`, and `src/main/java/com/deepgram/resources/agent/v1/websocket/V1WebSocketClient.java` - carry null-guarded, terminal `disconnect()` behavior so error-path cleanup is safe before or during `connect()`. The first four are already frozen for their respective patches; Agent V1 is frozen for this guard. Use the standard `.bak` swap/restore workflow during regen review; unfreeze once Fern emits the guard. + ### Prepare repo for regeneration 1. Create a new branch off `main` named `/sdk-gen-` (e.g. `gh/sdk-gen-2026-07-09`). Use your own initials as the prefix. diff --git a/README.md b/README.md index 752d92c1..9f2667f3 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,24 @@ DeepgramClient client = DeepgramClient.builder() .build(); ``` +### Resource lifecycle + +Close an SDK-created client when your application is finished with it. This releases the OkHttp +dispatcher and connection pool, which is especially important for short-lived command-line programs +that use WebSockets. + +```java +try (DeepgramClient client = DeepgramClient.builder().build()) { + // Use the client. +} +``` + +If you provide an `OkHttpClient` through `.httpClient(...)`, you retain ownership and must close its +resources yourself. + +WebSocket clients are terminal after `close()` or `disconnect()`. Create a new WebSocket client to +connect again. + ## Features ### Speech-to-Text (Listen) @@ -240,11 +258,10 @@ import java.nio.file.Path; import java.util.concurrent.TimeUnit; import okio.ByteString; -DeepgramClient client = DeepgramClient.builder().build(); +try (DeepgramClient client = DeepgramClient.builder().build(); + V1WebSocketClient ws = client.listen().v1().v1WebSocket()) { byte[] audioBytes = Files.readAllBytes(Path.of("audio.wav")); -V1WebSocketClient ws = client.listen().v1().v1WebSocket(); - // Register event handlers ws.onResults(results -> { String transcript = results.getChannel() @@ -273,8 +290,7 @@ ws.sendCloseStream(ListenV1CloseStream.builder() .build()) .get(5, TimeUnit.SECONDS); -// Close when done -ws.close(); +} ``` ### Text-to-Speech Streaming (Speak WebSocket) @@ -294,11 +310,10 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.concurrent.TimeUnit; -DeepgramClient client = DeepgramClient.builder().build(); +try (DeepgramClient client = DeepgramClient.builder().build(); + V1WebSocketClient ttsWs = client.speak().v1().v1WebSocket()) { ByteArrayOutputStream audioBuffer = new ByteArrayOutputStream(); -V1WebSocketClient ttsWs = client.speak().v1().v1WebSocket(); - // Register event handlers ttsWs.onSpeakV1Audio(audioData -> { audioBuffer.writeBytes(audioData.toByteArray()); @@ -331,8 +346,7 @@ ttsWs.sendClose(SpeakV1Close.builder() .build()) .get(5, TimeUnit.SECONDS); -// Close when done -ttsWs.close(); +} ``` ### Flux TTS Barge-in (Speak V2 WebSocket) @@ -343,14 +357,17 @@ The Speak V2 WebSocket adds Flux TTS barge-in and mid-stream controls. Open the - **`sendInterrupt(...)`** stops playback (barge-in). Pass a `SpeakV2InterruptPlaybackOffset` with the audio milliseconds played so the `onSpeechInterrupted` event can report `getTextSpoken()` / `getTextRemaining()`. The offset is cumulative from session start, and each interrupt must advance past the previous one. ```java +import com.deepgram.DeepgramClient; import com.deepgram.resources.speak.v2.types.SpeakV2Configure; import com.deepgram.resources.speak.v2.types.SpeakV2Interrupt; import com.deepgram.resources.speak.v2.types.SpeakV2InterruptPlaybackOffset; import com.deepgram.resources.speak.v2.types.SpeakV2Speak; import com.deepgram.resources.speak.v2.websocket.V2ConnectOptions; import com.deepgram.resources.speak.v2.websocket.V2WebSocketClient; +import java.util.concurrent.TimeUnit; -V2WebSocketClient ttsWs = client.speak().v2().v2WebSocket(); +try (DeepgramClient client = DeepgramClient.builder().build(); + V2WebSocketClient ttsWs = client.speak().v2().v2WebSocket()) { // Mid-stream configure acknowledgements ttsWs.onConfigureSuccess(success -> System.out.println("configured: " + success.getApplied())); @@ -373,7 +390,7 @@ ttsWs.sendInterrupt(SpeakV2Interrupt.builder() .playbackOffset(SpeakV2InterruptPlaybackOffset.builder().value(1200).build()) .build()); -ttsWs.close(); +} ``` See [`examples/speak/StreamingTtsV2.java`](examples/speak/StreamingTtsV2.java) for a complete, runnable barge-in example. @@ -397,9 +414,8 @@ import com.deepgram.types.ThinkSettingsV1; import com.deepgram.types.ThinkSettingsV1Provider; import java.util.concurrent.TimeUnit; -DeepgramClient client = DeepgramClient.builder().build(); - -V1WebSocketClient agentWs = client.agent().v1().v1WebSocket(); +try (DeepgramClient client = DeepgramClient.builder().build(); + V1WebSocketClient agentWs = client.agent().v1().v1WebSocket()) { // Register event handlers agentWs.onWelcome(welcome -> { @@ -440,8 +456,7 @@ agentWs.onError(error -> { agentWs.connect().get(10, TimeUnit.SECONDS); Thread.sleep(5000); -// Close when done -agentWs.close(); +} ``` ## Custom Transports @@ -466,6 +481,7 @@ import com.deepgram.DeepgramClient; import com.deepgram.sagemaker.SageMakerConfig; import com.deepgram.sagemaker.SageMakerTransportFactory; import com.deepgram.resources.listen.v1.websocket.V1ConnectOptions; +import com.deepgram.resources.listen.v1.websocket.V1WebSocketClient; import com.deepgram.types.ListenV1Model; import java.nio.file.Files; import java.nio.file.Path; @@ -481,17 +497,21 @@ var factory = new SageMakerTransportFactory( .build() ); -DeepgramClient client = DeepgramClient.builder() - .apiKey("unused") // SageMaker uses AWS credentials, not Deepgram API keys - .transportFactory(factory) - .build(); - -// Use the SDK exactly as normal — the transport is transparent -var ws = client.listen().v1().v1WebSocket(); -ws.onResults(results -> { /* ... */ }); -ws.connect(V1ConnectOptions.builder().model(ListenV1Model.NOVA3).build()) - .get(10, TimeUnit.SECONDS); -ws.sendMedia(ByteString.of(audioBytes)); +try { + try (DeepgramClient client = DeepgramClient.builder() + .apiKey("unused") // SageMaker uses AWS credentials, not Deepgram API keys + .transportFactory(factory) + .build(); + V1WebSocketClient ws = client.listen().v1().v1WebSocket()) { + // Use the SDK exactly as normal — the transport is transparent + ws.onResults(results -> { /* ... */ }); + ws.connect(V1ConnectOptions.builder().model(ListenV1Model.NOVA3).build()) + .get(10, TimeUnit.SECONDS); + ws.sendMedia(ByteString.of(audioBytes)); + } +} finally { + factory.shutdown(); +} ``` See the [SageMaker example](examples/sagemaker/LiveStreamingSageMaker.java) for a complete walkthrough. @@ -509,10 +529,12 @@ DeepgramTransportFactory myFactory = (url, headers) -> { return new MyCustomTransport(url, headers); }; -DeepgramClient client = DeepgramClient.builder() - .apiKey("your-key") - .transportFactory(myFactory) - .build(); +try (DeepgramClient client = DeepgramClient.builder() + .apiKey("your-key") + .transportFactory(myFactory) + .build()) { + // Use the client for streaming. Close each WebSocket before this scope exits. +} ``` The `DeepgramTransport` interface provides bidirectional messaging: `sendText()`, `sendBinary()`, and callback registration for incoming messages, errors, and close events. diff --git a/examples/agent/CustomProviders.java b/examples/agent/CustomProviders.java index a2d4ba03..a161ace4 100644 --- a/examples/agent/CustomProviders.java +++ b/examples/agent/CustomProviders.java @@ -154,7 +154,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } } diff --git a/examples/agent/InjectMessage.java b/examples/agent/InjectMessage.java index f0b6738a..dc728141 100644 --- a/examples/agent/InjectMessage.java +++ b/examples/agent/InjectMessage.java @@ -132,7 +132,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } diff --git a/examples/agent/VoiceAgent.java b/examples/agent/VoiceAgent.java index 5b4d6654..87578c22 100644 --- a/examples/agent/VoiceAgent.java +++ b/examples/agent/VoiceAgent.java @@ -151,7 +151,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } diff --git a/examples/listen/ForceEndTurn.java b/examples/listen/ForceEndTurn.java index 9bc6c8b6..b125f6ce 100644 --- a/examples/listen/ForceEndTurn.java +++ b/examples/listen/ForceEndTurn.java @@ -147,7 +147,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } diff --git a/examples/listen/LiveStreaming.java b/examples/listen/LiveStreaming.java index 194904c3..cc19b697 100644 --- a/examples/listen/LiveStreaming.java +++ b/examples/listen/LiveStreaming.java @@ -103,7 +103,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } } diff --git a/examples/listen/LiveStreamingV2.java b/examples/listen/LiveStreamingV2.java index 1d667d4b..cb025765 100644 --- a/examples/listen/LiveStreamingV2.java +++ b/examples/listen/LiveStreamingV2.java @@ -89,7 +89,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } } diff --git a/examples/sagemaker/LiveStreamingSageMaker.java b/examples/sagemaker/LiveStreamingSageMaker.java index 950cf53c..fbf34295 100644 --- a/examples/sagemaker/LiveStreamingSageMaker.java +++ b/examples/sagemaker/LiveStreamingSageMaker.java @@ -61,92 +61,107 @@ public static void main(String[] args) throws Exception { .build(); SageMakerTransportFactory factory = new SageMakerTransportFactory(config); - // Build the SDK client — apiKey is unused, SageMaker uses AWS credentials - DeepgramClient client = DeepgramClient.builder() - .apiKey("unused") - .transportFactory(factory) - .build(); + try { + // Build the SDK client — apiKey is unused, SageMaker uses AWS credentials + DeepgramClient client = DeepgramClient.builder() + .apiKey("unused") + .transportFactory(factory) + .build(); + + System.out.println("Live transcription via SageMaker transport"); + System.out.println("Endpoint: " + endpointName); + System.out.println("Region: " + region); + System.out.println(); + + // From here, the code is identical to any standard Deepgram SDK usage + V1WebSocketClient wsClient = client.listen().v1().v1WebSocket(); + try { + CountDownLatch done = new CountDownLatch(1); + + wsClient.onResults(result -> { + if (result.getChannel() != null + && result.getChannel().getAlternatives() != null + && !result.getChannel().getAlternatives().isEmpty()) { + ListenV1ResultsChannelAlternativesItem alt = + result.getChannel().getAlternatives().get(0); + String transcript = alt.getTranscript(); + if (transcript != null && !transcript.isEmpty()) { + boolean isFinal = result.getIsFinal().orElse(false); + System.out.printf("%s %s%n", isFinal ? "[final] " : "[interim]", transcript); + } + } + }); + + wsClient.onMetadata(metadata -> { + System.out.println("Metadata: request_id=" + metadata.getRequestId()); + }); + + wsClient.onError(error -> { + System.err.println("Error: " + error.getMessage()); + done.countDown(); + }); + + wsClient.onDisconnected(reason -> { + System.out.println("Closed (code: " + reason.getCode() + ")"); + done.countDown(); + }); + + // Connect + CompletableFuture connectFuture = wsClient.connect( + V1ConnectOptions.builder().model(ListenV1Model.NOVA3).build()); + connectFuture.get(10, TimeUnit.SECONDS); + System.out.println("Connected. Streaming audio...\n"); + + // Parse WAV header for pacing + int sampleRate; + int blockAlign; + try (RandomAccessFile raf = new RandomAccessFile(audioFile.toFile(), "r")) { + raf.skipBytes(24); + byte[] srBytes = new byte[4]; + raf.read(srBytes); + sampleRate = ByteBuffer.wrap(srBytes) + .order(ByteOrder.LITTLE_ENDIAN) + .getInt(); + + raf.skipBytes(4); // skip byte rate + byte[] baBytes = new byte[2]; + raf.read(baBytes); + blockAlign = ByteBuffer.wrap(baBytes) + .order(ByteOrder.LITTLE_ENDIAN) + .getShort() + & 0xFFFF; + } - System.out.println("Live transcription via SageMaker transport"); - System.out.println("Endpoint: " + endpointName); - System.out.println("Region: " + region); - System.out.println(); - - // From here, the code is identical to any standard Deepgram SDK usage - V1WebSocketClient wsClient = client.listen().v1().v1WebSocket(); - CountDownLatch done = new CountDownLatch(1); - - wsClient.onResults(result -> { - if (result.getChannel() != null - && result.getChannel().getAlternatives() != null - && !result.getChannel().getAlternatives().isEmpty()) { - ListenV1ResultsChannelAlternativesItem alt = - result.getChannel().getAlternatives().get(0); - String transcript = alt.getTranscript(); - if (transcript != null && !transcript.isEmpty()) { - boolean isFinal = result.getIsFinal().orElse(false); - System.out.printf("%s %s%n", isFinal ? "[final] " : "[interim]", transcript); + // Read and send audio paced to real-time (including WAV header) + byte[] audio = Files.readAllBytes(audioFile); + int chunkSize = 8192; + double framesPerChunk = (double) chunkSize / blockAlign; + long sleepMicros = (long) (framesPerChunk / sampleRate * 1_000_000); + + for (int i = 0; i < audio.length; i += chunkSize) { + int end = Math.min(i + chunkSize, audio.length); + byte[] chunk = new byte[end - i]; + System.arraycopy(audio, i, chunk, 0, chunk.length); + wsClient.sendMedia(ByteString.of(chunk)); + TimeUnit.MICROSECONDS.sleep(sleepMicros); } - } - }); - - wsClient.onMetadata(metadata -> { - System.out.println("Metadata: request_id=" + metadata.getRequestId()); - }); - - wsClient.onError(error -> { - System.err.println("Error: " + error.getMessage()); - done.countDown(); - }); - - wsClient.onDisconnected(reason -> { - System.out.println("Closed (code: " + reason.getCode() + ")"); - done.countDown(); - }); - - // Connect - CompletableFuture connectFuture = wsClient.connect( - V1ConnectOptions.builder().model(ListenV1Model.NOVA3).build()); - connectFuture.get(10, TimeUnit.SECONDS); - System.out.println("Connected. Streaming audio...\n"); - - // Parse WAV header for pacing - int sampleRate; - int blockAlign; - try (RandomAccessFile raf = new RandomAccessFile(audioFile.toFile(), "r")) { - raf.skipBytes(24); - byte[] srBytes = new byte[4]; - raf.read(srBytes); - sampleRate = ByteBuffer.wrap(srBytes).order(ByteOrder.LITTLE_ENDIAN).getInt(); - - raf.skipBytes(4); // skip byte rate - byte[] baBytes = new byte[2]; - raf.read(baBytes); - blockAlign = ByteBuffer.wrap(baBytes).order(ByteOrder.LITTLE_ENDIAN).getShort() & 0xFFFF; - } - // Read and send audio paced to real-time (including WAV header) - byte[] audio = Files.readAllBytes(audioFile); - int chunkSize = 8192; - double framesPerChunk = (double) chunkSize / blockAlign; - long sleepMicros = (long) (framesPerChunk / sampleRate * 1_000_000); - - for (int i = 0; i < audio.length; i += chunkSize) { - int end = Math.min(i + chunkSize, audio.length); - byte[] chunk = new byte[end - i]; - System.arraycopy(audio, i, chunk, 0, chunk.length); - wsClient.sendMedia(ByteString.of(chunk)); - TimeUnit.MICROSECONDS.sleep(sleepMicros); + // Signal end of audio + wsClient.sendCloseStream(ListenV1CloseStream.builder() + .type(ListenV1CloseStreamType.CLOSE_STREAM) + .build()); + + done.await(60, TimeUnit.SECONDS); + } finally { + try { + wsClient.disconnect(); + } finally { + client.close(); + } + } + } finally { + factory.shutdown(); } - - // Signal end of audio - wsClient.sendCloseStream(ListenV1CloseStream.builder() - .type(ListenV1CloseStreamType.CLOSE_STREAM) - .build()); - - done.await(60, TimeUnit.SECONDS); - wsClient.disconnect(); - factory.shutdown(); System.out.println("Done."); } } diff --git a/examples/speak/StreamingTts.java b/examples/speak/StreamingTts.java index 90158303..f2da2164 100644 --- a/examples/speak/StreamingTts.java +++ b/examples/speak/StreamingTts.java @@ -45,8 +45,7 @@ public static void main(String[] args) { CountDownLatch closeLatch = new CountDownLatch(1); AtomicInteger audioChunks = new AtomicInteger(0); - try { - OutputStream audioOutput = new FileOutputStream(outputFile); + try (OutputStream audioOutput = new FileOutputStream(outputFile)) { final String outputPath = outputFile; // Register event handlers before connecting @@ -83,11 +82,6 @@ public static void main(String[] args) { }); wsClient.onDisconnected(reason -> { - try { - audioOutput.close(); - } catch (Exception e) { - // ignore - } System.out.println( "\nConnection closed (code: " + reason.getCode() + ", reason: " + reason.getReason() + ")"); closeLatch.countDown(); @@ -130,7 +124,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } } diff --git a/examples/speak/StreamingTtsV2.java b/examples/speak/StreamingTtsV2.java index dd9c16d5..d4e2e4e0 100644 --- a/examples/speak/StreamingTtsV2.java +++ b/examples/speak/StreamingTtsV2.java @@ -193,7 +193,11 @@ public static void main(String[] args) { System.err.println("Error: " + e.getMessage()); e.printStackTrace(); } finally { - wsClient.disconnect(); + try { + wsClient.disconnect(); + } finally { + client.close(); + } } } } diff --git a/src/main/java/com/deepgram/AsyncDeepgramClient.java b/src/main/java/com/deepgram/AsyncDeepgramClient.java index 185c556b..57031ff4 100644 --- a/src/main/java/com/deepgram/AsyncDeepgramClient.java +++ b/src/main/java/com/deepgram/AsyncDeepgramClient.java @@ -5,12 +5,32 @@ */ import com.deepgram.core.ClientOptions; -public class AsyncDeepgramClient extends AsyncDeepgramApiClient { +public class AsyncDeepgramClient extends AsyncDeepgramApiClient implements AutoCloseable { + private final boolean ownsHttpClient; + public AsyncDeepgramClient(ClientOptions clientOptions) { + this(clientOptions, false); + } + + AsyncDeepgramClient(ClientOptions clientOptions, boolean ownsHttpClient) { super(clientOptions); + this.ownsHttpClient = ownsHttpClient; } public static AsyncDeepgramClientBuilder builder() { return new AsyncDeepgramClientBuilder(); } + + /** + * Releases resources owned by an SDK-created HTTP client. Clients supplied through + * {@link AsyncDeepgramClientBuilder#httpClient(okhttp3.OkHttpClient)} remain owned by the caller. + */ + @Override + public void close() { + if (!ownsHttpClient) { + return; + } + clientOptions.httpClient().dispatcher().executorService().shutdown(); + clientOptions.httpClient().connectionPool().evictAll(); + } } diff --git a/src/main/java/com/deepgram/AsyncDeepgramClientBuilder.java b/src/main/java/com/deepgram/AsyncDeepgramClientBuilder.java index 074f04cb..ed391da9 100644 --- a/src/main/java/com/deepgram/AsyncDeepgramClientBuilder.java +++ b/src/main/java/com/deepgram/AsyncDeepgramClientBuilder.java @@ -30,6 +30,8 @@ public class AsyncDeepgramClientBuilder extends AsyncDeepgramApiClientBuilder { private DeepgramTransportFactory transportFactory; + private boolean hasCustomHttpClient; + /** * Sets a custom transport factory for all WebSocket connections. When set, WebSocket clients will use this factory * instead of the default OkHttp WebSocket. Use this to route Deepgram API calls through alternative transports such @@ -87,6 +89,7 @@ public AsyncDeepgramClientBuilder maxRetries(int maxRetries) { @Override public AsyncDeepgramClientBuilder httpClient(OkHttpClient httpClient) { + this.hasCustomHttpClient = httpClient != null; super.httpClient(httpClient); return this; } @@ -149,6 +152,6 @@ public AsyncDeepgramClient build() { "Please provide apiKey, accessToken, or set the DEEPGRAM_API_KEY environment variable."); } validateConfiguration(); - return new AsyncDeepgramClient(buildClientOptions()); + return new AsyncDeepgramClient(buildClientOptions(), !hasCustomHttpClient); } } diff --git a/src/main/java/com/deepgram/DeepgramClient.java b/src/main/java/com/deepgram/DeepgramClient.java index 96b08c00..60363008 100644 --- a/src/main/java/com/deepgram/DeepgramClient.java +++ b/src/main/java/com/deepgram/DeepgramClient.java @@ -30,12 +30,32 @@ */ import com.deepgram.core.ClientOptions; -public class DeepgramClient extends DeepgramApiClient { +public class DeepgramClient extends DeepgramApiClient implements AutoCloseable { + private final boolean ownsHttpClient; + public DeepgramClient(ClientOptions clientOptions) { + this(clientOptions, false); + } + + DeepgramClient(ClientOptions clientOptions, boolean ownsHttpClient) { super(clientOptions); + this.ownsHttpClient = ownsHttpClient; } public static DeepgramClientBuilder builder() { return new DeepgramClientBuilder(); } + + /** + * Releases resources owned by an SDK-created HTTP client. Clients supplied through + * {@link DeepgramClientBuilder#httpClient(okhttp3.OkHttpClient)} remain owned by the caller. + */ + @Override + public void close() { + if (!ownsHttpClient) { + return; + } + clientOptions.httpClient().dispatcher().executorService().shutdown(); + clientOptions.httpClient().connectionPool().evictAll(); + } } diff --git a/src/main/java/com/deepgram/DeepgramClientBuilder.java b/src/main/java/com/deepgram/DeepgramClientBuilder.java index 614a23ec..2b74968c 100644 --- a/src/main/java/com/deepgram/DeepgramClientBuilder.java +++ b/src/main/java/com/deepgram/DeepgramClientBuilder.java @@ -29,6 +29,8 @@ public class DeepgramClientBuilder extends DeepgramApiClientBuilder { private DeepgramTransportFactory transportFactory; + private boolean hasCustomHttpClient; + /** * Sets a custom transport factory for all WebSocket connections. When set, WebSocket clients will use this factory * instead of the default OkHttp WebSocket. Use this to route Deepgram API calls through alternative transports such @@ -86,6 +88,7 @@ public DeepgramClientBuilder maxRetries(int maxRetries) { @Override public DeepgramClientBuilder httpClient(OkHttpClient httpClient) { + this.hasCustomHttpClient = httpClient != null; super.httpClient(httpClient); return this; } @@ -148,6 +151,6 @@ public DeepgramClient build() { "Please provide apiKey, accessToken, or set the DEEPGRAM_API_KEY environment variable."); } validateConfiguration(); - return new DeepgramClient(buildClientOptions()); + return new DeepgramClient(buildClientOptions(), !hasCustomHttpClient); } } diff --git a/src/main/java/com/deepgram/core/ReconnectingWebSocketListener.java b/src/main/java/com/deepgram/core/ReconnectingWebSocketListener.java index a5330688..1dd7afcf 100644 --- a/src/main/java/com/deepgram/core/ReconnectingWebSocketListener.java +++ b/src/main/java/com/deepgram/core/ReconnectingWebSocketListener.java @@ -6,6 +6,7 @@ import static java.util.concurrent.TimeUnit.*; import java.util.ArrayList; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutionException; @@ -42,6 +43,12 @@ public abstract class ReconnectingWebSocketListener extends WebSocketListener { private final AtomicBoolean shouldReconnect = new AtomicBoolean(true); + private final Object socketLock = new Object(); + + private volatile CompletableFuture pendingConnection; + + private volatile AtomicBoolean pendingConnectionCancelled; + protected volatile WebSocket webSocket; private volatile long connectionEstablishedTime = 0L; @@ -108,6 +115,11 @@ public void connect() { if (!connectLock.compareAndSet(false, true)) { return; } + if (!shouldReconnect.get()) { + onWebSocketFailure(null, new CancellationException("WebSocket connection was cancelled"), null); + connectLock.set(false); + return; + } // Snapshot the overridable options once so the gate and timeout below use a consistent set. ReconnectOptions opts = this.activeOptions; // retryCount is incremented inside scheduleReconnect() before re-entering connect(), @@ -117,11 +129,46 @@ public void connect() { connectLock.set(false); return; } + CompletableFuture connectionFuture = null; try { - CompletableFuture connectionFuture = CompletableFuture.supplyAsync(connectionSupplier); + AtomicBoolean connectionCancelled = new AtomicBoolean(false); + connectionFuture = CompletableFuture.supplyAsync(() -> { + if (!shouldReconnect.get() || connectionCancelled.get()) { + return null; + } + WebSocket socket = connectionSupplier.get(); + if (!shouldReconnect.get() || connectionCancelled.get()) { + if (socket != null) { + socket.close(1000, "Client disconnecting"); + } + return null; + } + return socket; + }); + synchronized (socketLock) { + pendingConnection = connectionFuture; + pendingConnectionCancelled = connectionCancelled; + } try { - webSocket = connectionFuture.get(opts.connectionTimeoutMs, MILLISECONDS); + WebSocket socket = connectionFuture.get(opts.connectionTimeoutMs, MILLISECONDS); + if (socket == null) { + onWebSocketFailure(null, new CancellationException("WebSocket connection was cancelled"), null); + return; + } + boolean disconnected; + synchronized (socketLock) { + disconnected = !shouldReconnect.get(); + if (!disconnected) { + webSocket = socket; + } + } + if (disconnected) { + socket.close(1000, "Client disconnecting"); + onWebSocketFailure(null, new CancellationException("WebSocket connection was cancelled"), null); + return; + } } catch (TimeoutException e) { + connectionCancelled.set(true); connectionFuture.cancel(true); TimeoutException timeoutError = new TimeoutException("WebSocket connection timeout after " + opts.connectionTimeoutMs @@ -134,6 +181,7 @@ public void connect() { scheduleReconnect(); } } catch (InterruptedException e) { + connectionCancelled.set(true); connectionFuture.cancel(true); Thread.currentThread().interrupt(); InterruptedException interruptError = new InterruptedException("WebSocket connection interrupted" @@ -154,8 +202,16 @@ public void connect() { if (shouldReconnect.get()) { scheduleReconnect(); } + } catch (CancellationException e) { + onWebSocketFailure(null, e, null); } } finally { + synchronized (socketLock) { + if (pendingConnection == connectionFuture) { + pendingConnection = null; + pendingConnectionCancelled = null; + } + } connectLock.set(false); } } @@ -171,11 +227,26 @@ public void connect() { * - Waits up to 5 seconds for executor termination */ public void disconnect() { - shouldReconnect.set(false); + AtomicBoolean connectionCancelled; + CompletableFuture connection; + WebSocket socket; + synchronized (socketLock) { + shouldReconnect.set(false); + connectionCancelled = pendingConnectionCancelled; + connection = pendingConnection; + socket = webSocket; + webSocket = null; + } + if (connectionCancelled != null) { + connectionCancelled.set(true); + } + if (connection != null) { + connection.cancel(true); + } messageQueue.clear(); binaryMessageQueue.clear(); - if (webSocket != null) { - webSocket.close(1000, "Client disconnecting"); + if (socket != null) { + socket.close(1000, "Client disconnecting"); } reconnectExecutor.shutdown(); try { @@ -264,7 +335,17 @@ public WebSocket getWebSocket() { */ @Override public void onOpen(WebSocket webSocket, Response response) { - this.webSocket = webSocket; + boolean disconnected; + synchronized (socketLock) { + disconnected = !shouldReconnect.get(); + if (!disconnected) { + this.webSocket = webSocket; + } + } + if (disconnected) { + webSocket.close(1000, "Client disconnecting"); + return; + } connectionEstablishedTime = System.currentTimeMillis(); retryCount.set(0); flushMessageQueue(); @@ -286,7 +367,11 @@ public void onMessage(WebSocket webSocket, ByteString bytes) { */ @Override public void onFailure(WebSocket webSocket, Throwable t, Response response) { - this.webSocket = null; + synchronized (socketLock) { + if (this.webSocket == webSocket) { + this.webSocket = null; + } + } long uptime = 0L; if (connectionEstablishedTime > 0) { uptime = System.currentTimeMillis() - connectionEstablishedTime; @@ -319,7 +404,11 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) { */ @Override public void onClosed(WebSocket webSocket, int code, String reason) { - this.webSocket = null; + synchronized (socketLock) { + if (this.webSocket == webSocket) { + this.webSocket = null; + } + } if (connectionEstablishedTime > 0) { long uptime = System.currentTimeMillis() - connectionEstablishedTime; if (uptime >= 5000) { diff --git a/src/main/java/com/deepgram/resources/agent/v1/websocket/V1WebSocketClient.java b/src/main/java/com/deepgram/resources/agent/v1/websocket/V1WebSocketClient.java index d4b29108..81835fb3 100644 --- a/src/main/java/com/deepgram/resources/agent/v1/websocket/V1WebSocketClient.java +++ b/src/main/java/com/deepgram/resources/agent/v1/websocket/V1WebSocketClient.java @@ -41,6 +41,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; @@ -62,6 +63,8 @@ public class V1WebSocketClient implements AutoCloseable { private ScheduledExecutorService timeoutExecutor; + private final AtomicBoolean disconnected = new AtomicBoolean(false); + private volatile WebSocketReadyState readyState = WebSocketReadyState.CLOSED; private volatile Runnable onConnectedHandler; @@ -74,7 +77,7 @@ public class V1WebSocketClient implements AutoCloseable { private volatile ReconnectingWebSocketListener.ReconnectOptions reconnectOptions; - private CompletableFuture connectionFuture; + private volatile CompletableFuture connectionFuture; private ReconnectingWebSocketListener reconnectingListener; @@ -130,6 +133,9 @@ public V1WebSocketClient(ClientOptions clientOptions) { * @return a CompletableFuture that completes when the connection is established */ public CompletableFuture connect() { + if (disconnected.get()) { + return disconnectedFuture(); + } connectionFuture = new CompletableFuture<>(); String baseUrl = clientOptions.environment().getAgentURL(); String fullPath = "/v1/agent/converse"; @@ -202,6 +208,11 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { } } }; + if (disconnected.get()) { + reconnectingListener.disconnect(); + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return connectionFuture; + } reconnectingListener.connect(); return connectionFuture; } @@ -210,13 +221,25 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { * Disconnects the WebSocket connection and releases resources. */ public void disconnect() { - reconnectingListener.disconnect(); + disconnected.set(true); + if (connectionFuture != null) { + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + } + if (reconnectingListener != null) { + reconnectingListener.disconnect(); + } if (timeoutExecutor != null) { timeoutExecutor.shutdownNow(); timeoutExecutor = null; } } + private CompletableFuture disconnectedFuture() { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return future; + } + /** * Gets the current state of the WebSocket connection. * diff --git a/src/main/java/com/deepgram/resources/listen/v1/websocket/V1WebSocketClient.java b/src/main/java/com/deepgram/resources/listen/v1/websocket/V1WebSocketClient.java index 9bf053b7..a1cfae10 100644 --- a/src/main/java/com/deepgram/resources/listen/v1/websocket/V1WebSocketClient.java +++ b/src/main/java/com/deepgram/resources/listen/v1/websocket/V1WebSocketClient.java @@ -23,6 +23,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; @@ -44,6 +45,8 @@ public class V1WebSocketClient implements AutoCloseable { private ScheduledExecutorService timeoutExecutor; + private final AtomicBoolean disconnected = new AtomicBoolean(false); + private volatile WebSocketReadyState readyState = WebSocketReadyState.CLOSED; private volatile Runnable onConnectedHandler; @@ -56,7 +59,7 @@ public class V1WebSocketClient implements AutoCloseable { private volatile ReconnectingWebSocketListener.ReconnectOptions reconnectOptions; - private CompletableFuture connectionFuture; + private volatile CompletableFuture connectionFuture; private ReconnectingWebSocketListener reconnectingListener; @@ -83,6 +86,9 @@ public V1WebSocketClient(ClientOptions clientOptions) { * @param options connection options including query parameters */ public CompletableFuture connect(V1ConnectOptions options) { + if (disconnected.get()) { + return disconnectedFuture(); + } connectionFuture = new CompletableFuture<>(); String baseUrl = clientOptions.environment().getProductionURL(); String fullPath = "/v1/listen"; @@ -276,6 +282,11 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { } } }; + if (disconnected.get()) { + reconnectingListener.disconnect(); + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return connectionFuture; + } reconnectingListener.connect(); return connectionFuture; } @@ -284,13 +295,25 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { * Disconnects the WebSocket connection and releases resources. */ public void disconnect() { - reconnectingListener.disconnect(); + disconnected.set(true); + if (connectionFuture != null) { + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + } + if (reconnectingListener != null) { + reconnectingListener.disconnect(); + } if (timeoutExecutor != null) { timeoutExecutor.shutdownNow(); timeoutExecutor = null; } } + private CompletableFuture disconnectedFuture() { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return future; + } + /** * Gets the current state of the WebSocket connection. * diff --git a/src/main/java/com/deepgram/resources/listen/v2/websocket/V2WebSocketClient.java b/src/main/java/com/deepgram/resources/listen/v2/websocket/V2WebSocketClient.java index 445a96aa..82f11ca5 100644 --- a/src/main/java/com/deepgram/resources/listen/v2/websocket/V2WebSocketClient.java +++ b/src/main/java/com/deepgram/resources/listen/v2/websocket/V2WebSocketClient.java @@ -24,6 +24,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; @@ -45,6 +46,8 @@ public class V2WebSocketClient implements AutoCloseable { private ScheduledExecutorService timeoutExecutor; + private final AtomicBoolean disconnected = new AtomicBoolean(false); + private volatile WebSocketReadyState readyState = WebSocketReadyState.CLOSED; private volatile Runnable onConnectedHandler; @@ -57,7 +60,7 @@ public class V2WebSocketClient implements AutoCloseable { private volatile ReconnectingWebSocketListener.ReconnectOptions reconnectOptions; - private CompletableFuture connectionFuture; + private volatile CompletableFuture connectionFuture; private ReconnectingWebSocketListener reconnectingListener; @@ -86,6 +89,9 @@ public V2WebSocketClient(ClientOptions clientOptions) { * @param options connection options including query parameters */ public CompletableFuture connect(V2ConnectOptions options) { + if (disconnected.get()) { + return disconnectedFuture(); + } connectionFuture = new CompletableFuture<>(); String baseUrl = clientOptions.environment().getProductionURL(); String fullPath = "/v2/listen"; @@ -213,6 +219,11 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { } } }; + if (disconnected.get()) { + reconnectingListener.disconnect(); + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return connectionFuture; + } reconnectingListener.connect(); return connectionFuture; } @@ -221,13 +232,25 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { * Disconnects the WebSocket connection and releases resources. */ public void disconnect() { - reconnectingListener.disconnect(); + disconnected.set(true); + if (connectionFuture != null) { + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + } + if (reconnectingListener != null) { + reconnectingListener.disconnect(); + } if (timeoutExecutor != null) { timeoutExecutor.shutdownNow(); timeoutExecutor = null; } } + private CompletableFuture disconnectedFuture() { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return future; + } + /** * Gets the current state of the WebSocket connection. * diff --git a/src/main/java/com/deepgram/resources/speak/v1/websocket/V1WebSocketClient.java b/src/main/java/com/deepgram/resources/speak/v1/websocket/V1WebSocketClient.java index 60a163bc..b485f293 100644 --- a/src/main/java/com/deepgram/resources/speak/v1/websocket/V1WebSocketClient.java +++ b/src/main/java/com/deepgram/resources/speak/v1/websocket/V1WebSocketClient.java @@ -24,6 +24,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; @@ -45,6 +46,8 @@ public class V1WebSocketClient implements AutoCloseable { private ScheduledExecutorService timeoutExecutor; + private final AtomicBoolean disconnected = new AtomicBoolean(false); + private volatile WebSocketReadyState readyState = WebSocketReadyState.CLOSED; private volatile Runnable onConnectedHandler; @@ -57,7 +60,7 @@ public class V1WebSocketClient implements AutoCloseable { private volatile ReconnectingWebSocketListener.ReconnectOptions reconnectOptions; - private CompletableFuture connectionFuture; + private volatile CompletableFuture connectionFuture; private ReconnectingWebSocketListener reconnectingListener; @@ -86,6 +89,9 @@ public V1WebSocketClient(ClientOptions clientOptions) { * @param options connection options including query parameters */ public CompletableFuture connect(V1ConnectOptions options) { + if (disconnected.get()) { + return disconnectedFuture(); + } connectionFuture = new CompletableFuture<>(); String baseUrl = clientOptions.environment().getProductionURL(); String fullPath = "/v1/speak"; @@ -185,6 +191,11 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { } } }; + if (disconnected.get()) { + reconnectingListener.disconnect(); + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return connectionFuture; + } reconnectingListener.connect(); return connectionFuture; } @@ -201,13 +212,25 @@ public CompletableFuture connect() { * Disconnects the WebSocket connection and releases resources. */ public void disconnect() { - reconnectingListener.disconnect(); + disconnected.set(true); + if (connectionFuture != null) { + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + } + if (reconnectingListener != null) { + reconnectingListener.disconnect(); + } if (timeoutExecutor != null) { timeoutExecutor.shutdownNow(); timeoutExecutor = null; } } + private CompletableFuture disconnectedFuture() { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return future; + } + /** * Gets the current state of the WebSocket connection. * diff --git a/src/main/java/com/deepgram/resources/speak/v2/websocket/V2WebSocketClient.java b/src/main/java/com/deepgram/resources/speak/v2/websocket/V2WebSocketClient.java index e0163665..ac156eb6 100644 --- a/src/main/java/com/deepgram/resources/speak/v2/websocket/V2WebSocketClient.java +++ b/src/main/java/com/deepgram/resources/speak/v2/websocket/V2WebSocketClient.java @@ -31,6 +31,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; @@ -52,6 +53,8 @@ public class V2WebSocketClient implements AutoCloseable { private ScheduledExecutorService timeoutExecutor; + private final AtomicBoolean disconnected = new AtomicBoolean(false); + private volatile WebSocketReadyState readyState = WebSocketReadyState.CLOSED; private volatile Runnable onConnectedHandler; @@ -64,7 +67,7 @@ public class V2WebSocketClient implements AutoCloseable { private volatile ReconnectingWebSocketListener.ReconnectOptions reconnectOptions; - private CompletableFuture connectionFuture; + private volatile CompletableFuture connectionFuture; private ReconnectingWebSocketListener reconnectingListener; @@ -105,6 +108,9 @@ public V2WebSocketClient(ClientOptions clientOptions) { * @param options connection options including query parameters */ public CompletableFuture connect(V2ConnectOptions options) { + if (disconnected.get()) { + return disconnectedFuture(); + } connectionFuture = new CompletableFuture<>(); String baseUrl = clientOptions.environment().getProductionURL(); String fullPath = "/v2/speak"; @@ -209,6 +215,11 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { } } }; + if (disconnected.get()) { + reconnectingListener.disconnect(); + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return connectionFuture; + } reconnectingListener.connect(); return connectionFuture; } @@ -217,13 +228,25 @@ protected void onWebSocketClosed(WebSocket webSocket, int code, String reason) { * Disconnects the WebSocket connection and releases resources. */ public void disconnect() { - reconnectingListener.disconnect(); + disconnected.set(true); + if (connectionFuture != null) { + connectionFuture.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + } + if (reconnectingListener != null) { + reconnectingListener.disconnect(); + } if (timeoutExecutor != null) { timeoutExecutor.shutdownNow(); timeoutExecutor = null; } } + private CompletableFuture disconnectedFuture() { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("WebSocket client has been disconnected")); + return future; + } + /** * Gets the current state of the WebSocket connection. * diff --git a/src/test/java/com/deepgram/ClientBuilderTest.java b/src/test/java/com/deepgram/ClientBuilderTest.java index fd7dacd8..aecc42e1 100644 --- a/src/test/java/com/deepgram/ClientBuilderTest.java +++ b/src/test/java/com/deepgram/ClientBuilderTest.java @@ -1,10 +1,16 @@ package com.deepgram; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import com.deepgram.core.ClientOptions; import com.deepgram.core.Environment; +import java.util.concurrent.CompletableFuture; import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.WebSocket; +import okio.ByteString; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; @@ -145,6 +151,201 @@ void testCustomHttpClient() { } } + @Nested + @DisplayName("Client lifecycle") + class ClientLifecycle { + @Test + @DisplayName("closing the default client releases SDK-owned HTTP resources") + void closesDefaultClientResources() { + DeepgramClient client = DeepgramClient.builder().apiKey("test-key").build(); + + client.close(); + + assertThat(client.clientOptions + .httpClient() + .dispatcher() + .executorService() + .isShutdown()) + .isTrue(); + } + + @Test + @DisplayName("closing the default async client releases SDK-owned HTTP resources") + void closesDefaultAsyncClientResources() { + AsyncDeepgramClient client = + AsyncDeepgramClient.builder().apiKey("test-key").build(); + + client.close(); + + assertThat(client.clientOptions + .httpClient() + .dispatcher() + .executorService() + .isShutdown()) + .isTrue(); + } + + @Test + @DisplayName("closing a client does not release caller-owned HTTP resources") + void doesNotCloseCustomClientResources() { + OkHttpClient customHttpClient = new OkHttpClient.Builder().build(); + DeepgramClient client = DeepgramClient.builder() + .apiKey("test-key") + .httpClient(customHttpClient) + .build(); + + try { + client.close(); + + assertThat(customHttpClient.dispatcher().executorService().isShutdown()) + .isFalse(); + } finally { + customHttpClient.dispatcher().executorService().shutdown(); + customHttpClient.connectionPool().evictAll(); + } + } + + @Test + @DisplayName("closing an async client does not release caller-owned HTTP resources") + void doesNotCloseCustomAsyncClientResources() { + OkHttpClient customHttpClient = new OkHttpClient.Builder().build(); + AsyncDeepgramClient client = AsyncDeepgramClient.builder() + .apiKey("test-key") + .httpClient(customHttpClient) + .build(); + + try { + client.close(); + + assertThat(customHttpClient.dispatcher().executorService().isShutdown()) + .isFalse(); + } finally { + customHttpClient.dispatcher().executorService().shutdown(); + customHttpClient.connectionPool().evictAll(); + } + } + + @Test + @DisplayName("disconnecting an unconnected WebSocket is safe") + void disconnectsUnconnectedWebSockets() { + DeepgramClient client = DeepgramClient.builder().apiKey("test-key").build(); + + try { + assertThatCode(() -> client.listen().v1().v1WebSocket().disconnect()) + .doesNotThrowAnyException(); + assertThatCode(() -> client.listen().v2().v2WebSocket().disconnect()) + .doesNotThrowAnyException(); + assertThatCode(() -> client.speak().v1().v1WebSocket().disconnect()) + .doesNotThrowAnyException(); + assertThatCode(() -> client.speak().v2().v2WebSocket().disconnect()) + .doesNotThrowAnyException(); + assertThatCode(() -> client.agent().v1().v1WebSocket().disconnect()) + .doesNotThrowAnyException(); + } finally { + client.close(); + } + } + + @Test + @DisplayName("disconnect prevents a WebSocket from connecting later") + void disconnectPreventsLaterConnections() { + DeepgramClient client = DeepgramClient.builder().apiKey("test-key").build(); + + try { + var listenV1 = client.listen().v1().v1WebSocket(); + listenV1.disconnect(); + assertConnectionRejected( + listenV1.connect(com.deepgram.resources.listen.v1.websocket.V1ConnectOptions.builder() + .model(com.deepgram.types.ListenV1Model.NOVA3) + .build())); + + var listenV2 = client.listen().v2().v2WebSocket(); + listenV2.disconnect(); + assertConnectionRejected( + listenV2.connect(com.deepgram.resources.listen.v2.websocket.V2ConnectOptions.builder() + .model(com.deepgram.types.ListenV2Model.FLUX_GENERAL_EN) + .build())); + + var speakV1 = client.speak().v1().v1WebSocket(); + speakV1.disconnect(); + assertConnectionRejected(speakV1.connect()); + + var speakV2 = client.speak().v2().v2WebSocket(); + speakV2.disconnect(); + assertConnectionRejected( + speakV2.connect(com.deepgram.resources.speak.v2.websocket.V2ConnectOptions.builder() + .model("flux-alexis-en") + .build())); + + var agentV1 = client.agent().v1().v1WebSocket(); + agentV1.disconnect(); + assertConnectionRejected(agentV1.connect()); + } finally { + client.close(); + } + } + + @Test + @DisplayName("disconnect completes an in-flight connection exceptionally") + void disconnectCompletesInFlightConnection() { + ClientOptions options = ClientOptions.builder() + .environment(Environment.PRODUCTION) + .webSocketFactory((request, listener) -> new UnopenedWebSocket()) + .build(); + var socket = new com.deepgram.resources.listen.v2.websocket.V2WebSocketClient(options); + + try { + CompletableFuture connection = + socket.connect(com.deepgram.resources.listen.v2.websocket.V2ConnectOptions.builder() + .model(com.deepgram.types.ListenV2Model.FLUX_GENERAL_EN) + .build()); + + socket.disconnect(); + + assertConnectionRejected(connection); + } finally { + options.httpClient().dispatcher().executorService().shutdown(); + options.httpClient().connectionPool().evictAll(); + } + } + } + + private static void assertConnectionRejected(CompletableFuture connection) { + assertThatThrownBy(connection::join) + .isInstanceOf(java.util.concurrent.CompletionException.class) + .hasCauseInstanceOf(IllegalStateException.class); + } + + private static final class UnopenedWebSocket implements WebSocket { + @Override + public Request request() { + return new Request.Builder().url("ws://localhost/").build(); + } + + @Override + public long queueSize() { + return 0; + } + + @Override + public boolean send(String text) { + return false; + } + + @Override + public boolean send(ByteString bytes) { + return false; + } + + @Override + public boolean close(int code, String reason) { + return true; + } + + @Override + public void cancel() {} + } + @Nested @DisplayName("Custom headers configuration") class CustomHeadersConfiguration { diff --git a/src/test/java/com/deepgram/core/ReconnectingWebSocketListenerTest.java b/src/test/java/com/deepgram/core/ReconnectingWebSocketListenerTest.java index da8001e0..7ea32d7a 100644 --- a/src/test/java/com/deepgram/core/ReconnectingWebSocketListenerTest.java +++ b/src/test/java/com/deepgram/core/ReconnectingWebSocketListenerTest.java @@ -4,6 +4,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import com.deepgram.core.ReconnectingWebSocketListener.ReconnectOptions; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; import okhttp3.Response; @@ -39,6 +41,8 @@ public WebSocket get() { } private static final class FakeWebSocket implements WebSocket { + final CountDownLatch closed = new CountDownLatch(1); + @Override public okhttp3.Request request() { return new okhttp3.Request.Builder().url("ws://localhost/").build(); @@ -61,6 +65,7 @@ public boolean send(ByteString bytes) { @Override public boolean close(int code, String reason) { + closed.countDown(); return true; } @@ -68,6 +73,30 @@ public boolean close(int code, String reason) { public void cancel() {} } + private static final class BlockingSupplier implements Supplier { + final CountDownLatch started = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final FakeWebSocket socket = new FakeWebSocket(); + + @Override + public WebSocket get() { + started.countDown(); + boolean interrupted = false; + while (true) { + try { + release.await(); + break; + } catch (InterruptedException e) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + return socket; + } + } + /** Concrete listener that records callback invocations for assertions. */ private static final class TestListener extends ReconnectingWebSocketListener { final AtomicInteger failures = new AtomicInteger(0); @@ -149,6 +178,41 @@ void initialAttemptProceedsWhenMaxRetriesIsZero() { } } + @Nested + @DisplayName("connection cancellation") + class ConnectionCancellationTests { + @Test + @DisplayName("closes a socket returned after disconnect") + void closesSocketReturnedAfterDisconnect() throws Exception { + BlockingSupplier supplier = new BlockingSupplier(); + TestListener listener = new TestListener(ReconnectOptions.builder().build(), supplier); + Thread connectionThread = new Thread(listener::connect); + connectionThread.start(); + + assertThat(supplier.started.await(1, TimeUnit.SECONDS)).isTrue(); + listener.disconnect(); + supplier.release.countDown(); + + assertThat(supplier.socket.closed.await(1, TimeUnit.SECONDS)).isTrue(); + connectionThread.join(1_000); + assertThat(connectionThread.isAlive()).isFalse(); + assertThat(listener.failures).hasValue(1); + } + + @Test + @DisplayName("does not retain a socket that opens after disconnect") + void doesNotRetainSocketOpenedAfterDisconnect() throws Exception { + TestListener listener = new TestListener(ReconnectOptions.builder().build(), new CountingSupplier(false)); + FakeWebSocket socket = new FakeWebSocket(); + + listener.disconnect(); + listener.onOpen(socket, null); + + assertThat(socket.closed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(listener.getWebSocket()).isNull(); + } + } + @Nested @DisplayName("applyOptionsOverride") class ApplyOverrideTests {