diff --git a/deepgram.toml b/deepgram.toml index bf0652f..342114a 100644 --- a/deepgram.toml +++ b/deepgram.toml @@ -6,7 +6,7 @@ repository = "https://github.com/deepgram-starters/java-flux" useCase = "flux" language = "Java" framework = "Javalin" -sdk = "N/A" +sdk = "deepgram-java-sdk" tags = ["flux", "turn-detection", "real-time-transcription", "streaming-stt", "end-of-turn", "java", "javalin"] [check] diff --git a/pom.xml b/pom.xml index be72cf7..1d03978 100644 --- a/pom.xml +++ b/pom.xml @@ -16,9 +16,18 @@ 21 21 UTF-8 + + 0.7.1 + + + com.deepgram + deepgram-java-sdk + ${deepgram.sdk.version} + + io.javalin @@ -67,13 +76,6 @@ kotlin-stdlib 1.9.25 - - - - org.eclipse.jetty.websocket - websocket-jetty-client - 11.0.24 - diff --git a/src/main/java/com/deepgram/starter/App.java b/src/main/java/com/deepgram/starter/App.java index 8b61e34..e6df534 100644 --- a/src/main/java/com/deepgram/starter/App.java +++ b/src/main/java/com/deepgram/starter/App.java @@ -1,15 +1,20 @@ /** * Java Flux Starter - Backend Server * - * Simple WebSocket proxy to Deepgram's Flux API using Javalin 6.4.0. - * Forwards all messages (JSON and binary) bidirectionally between client and Deepgram. + * WebSocket bridge to Deepgram's Flux API (Listen v2) using the official + * Deepgram Java SDK (`client.listen().v2().v2WebSocket()`) instead of a raw + * Jetty WebSocket client. + * + * The browser-facing contract is unchanged: the browser streams binary PCM + * audio and a `{"type":"CloseStream"}` control message to /api/flux, and the + * backend forwards Deepgram's native Flux JSON (TurnInfo, Connected, ...) back + * to the browser verbatim. * * Key Features: - * - WebSocket proxy endpoint: /api/flux -> Deepgram wss://api.deepgram.com/v2/listen + * - WebSocket bridge endpoint: /api/flux -> Deepgram Flux (SDK Listen v2) * - JWT session auth via access_token. subprotocol * - Session endpoint: GET /api/session * - Metadata endpoint: GET /api/metadata - * - No Deepgram SDK -- direct WebSocket connections via Jetty WebSocket client */ package com.deepgram.starter; @@ -31,21 +36,24 @@ import io.javalin.websocket.WsConfig; import io.javalin.websocket.WsContext; -import org.eclipse.jetty.websocket.api.Session; -import org.eclipse.jetty.websocket.api.StatusCode; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError; -import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage; -import org.eclipse.jetty.websocket.api.annotations.WebSocket; -import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; -import org.eclipse.jetty.websocket.client.WebSocketClient; +import com.deepgram.DeepgramClient; +import com.deepgram.core.Environment; +import com.deepgram.resources.listen.v2.types.ListenV2CloseStream; +import com.deepgram.resources.listen.v2.websocket.V2ConnectOptions; +import com.deepgram.resources.listen.v2.websocket.V2WebSocketClient; +import com.deepgram.types.ListenV2EagerEotThreshold; +import com.deepgram.types.ListenV2Encoding; +import com.deepgram.types.ListenV2EotThreshold; +import com.deepgram.types.ListenV2EotTimeoutMs; +import com.deepgram.types.ListenV2Keyterm; +import com.deepgram.types.ListenV2Model; +import com.deepgram.types.ListenV2SampleRate; +import okio.ByteString; import java.io.File; -import java.net.URI; -import java.nio.ByteBuffer; import java.security.SecureRandom; import java.time.Instant; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Set; @@ -80,9 +88,6 @@ public class App { /** Server host, configurable via HOST env var (default 0.0.0.0) */ private static final String HOST = getEnv("HOST", "0.0.0.0"); - /** Deepgram Flux WebSocket URL (v2 endpoint) */ - private static final String DEEPGRAM_FLUX_URL = "wss://api.deepgram.com/v2/listen"; - /** * Reserved WebSocket close codes that cannot be set by applications. * If Deepgram sends one of these, we fall back to 1000 (normal closure). @@ -170,6 +175,9 @@ private static String validateWsToken(String protocols) { /** The Deepgram API key loaded at startup */ private static String apiKey; + /** One SDK client, reused across connections; the browser never sees the API key. */ + private static DeepgramClient deepgram; + /** * Loads the Deepgram API key from environment variables. * Exits with a helpful error message if not found. @@ -198,15 +206,12 @@ private static String loadApiKey() { } // ======================================================================== - // SECTION 6: SETUP - Track connections and Jetty WebSocket client + // SECTION 6: SETUP - Track connections // ======================================================================== /** Track all active client WebSocket contexts for graceful shutdown */ private static final Set activeConnections = ConcurrentHashMap.newKeySet(); - /** Jetty WebSocket client for outbound connections to Deepgram */ - private static WebSocketClient wsClient; - // ======================================================================== // SECTION 7: HELPER FUNCTIONS // ======================================================================== @@ -251,57 +256,59 @@ private static int getSafeCloseCode(int code) { } /** - * Builds the Deepgram Flux WebSocket URL with query parameters forwarded from the client. + * Builds the Deepgram Flux connect options from the query parameters + * forwarded by the client (the same parameters the previous raw-proxy + * implementation appended to the Deepgram URL). * * @param ctx The client WebSocket context - * @return The fully constructed Deepgram URL string + * @return The V2ConnectOptions for the Deepgram Flux connection */ - private static String buildDeepgramUrl(WsContext ctx) { + private static V2ConnectOptions buildConnectOptions(WsContext ctx) { String model = ctx.queryParam("model"); if (model == null || model.isEmpty()) model = "flux-general-en"; - String sampleRate = ctx.queryParam("sample_rate"); - if (sampleRate == null || sampleRate.isEmpty()) sampleRate = "16000"; - String encoding = ctx.queryParam("encoding"); if (encoding == null || encoding.isEmpty()) encoding = "linear16"; - String channels = ctx.queryParam("channels"); - if (channels == null || channels.isEmpty()) channels = "1"; + String sampleRate = ctx.queryParam("sample_rate"); + if (sampleRate == null || sampleRate.isEmpty()) sampleRate = "16000"; - StringBuilder url = new StringBuilder(DEEPGRAM_FLUX_URL); - url.append("?model=").append(model); - url.append("&sample_rate=").append(sampleRate); - url.append("&encoding=").append(encoding); - url.append("&channels=").append(channels); + V2ConnectOptions._FinalStage opts = V2ConnectOptions.builder() + .model(ListenV2Model.valueOf(model)) + .encoding(ListenV2Encoding.valueOf(encoding)) + .sampleRate(ListenV2SampleRate.of(Integer.parseInt(sampleRate))); - // Forward optional parameters + // Forward optional turn-detection parameters String eotThreshold = ctx.queryParam("eot_threshold"); if (eotThreshold != null && !eotThreshold.isEmpty()) { - url.append("&eot_threshold=").append(eotThreshold); + opts = opts.eotThreshold(ListenV2EotThreshold.of(Double.parseDouble(eotThreshold))); } String eagerEotThreshold = ctx.queryParam("eager_eot_threshold"); if (eagerEotThreshold != null && !eagerEotThreshold.isEmpty()) { - url.append("&eager_eot_threshold=").append(eagerEotThreshold); + opts = opts.eagerEotThreshold(ListenV2EagerEotThreshold.of(Double.parseDouble(eagerEotThreshold))); } String eotTimeoutMs = ctx.queryParam("eot_timeout_ms"); if (eotTimeoutMs != null && !eotTimeoutMs.isEmpty()) { - url.append("&eot_timeout_ms=").append(eotTimeoutMs); + opts = opts.eotTimeoutMs(ListenV2EotTimeoutMs.of(Integer.parseInt(eotTimeoutMs))); } - // Handle keyterm: can appear multiple times, forward each one + // Handle keyterm: can appear multiple times, forward all of them List keyterms = ctx.queryParams("keyterm"); + List filteredKeyterms = new ArrayList<>(); if (keyterms != null) { for (String term : keyterms) { if (term != null && !term.isEmpty()) { - url.append("&keyterm=").append(term); + filteredKeyterms.add(term); } } } + if (!filteredKeyterms.isEmpty()) { + opts = opts.keyterm(ListenV2Keyterm.of(filteredKeyterms)); + } - return url.toString(); + return opts.build(); } // ======================================================================== @@ -319,7 +326,7 @@ private static void handleSession(Context ctx) { } // ======================================================================== - // SECTION 9: API ROUTES & WEBSOCKET PROXY + // SECTION 9: API ROUTES & WEBSOCKET BRIDGE // ======================================================================== /** @@ -355,10 +362,69 @@ private static void handleMetadata(Context ctx) { } } + /** + * Per-connection bridge to a Deepgram Flux WebSocket. Buffers audio that + * arrives from the browser before the Deepgram socket has finished opening. + */ + static final class FluxBridge { + private final V2WebSocketClient dg; + private boolean ready = false; + private boolean closeRequested = false; + private final List pendingAudio = new ArrayList<>(); + + FluxBridge(V2WebSocketClient dg) { + this.dg = dg; + } + + synchronized void sendAudio(ByteString audio) { + if (!ready) { + pendingAudio.add(audio); + return; + } + dg.sendMedia(audio); + } + + synchronized void closeStream() { + if (!ready) { + closeRequested = true; + return; + } + sendCloseStream(); + } + + synchronized void markReady() { + ready = true; + for (ByteString audio : pendingAudio) { + dg.sendMedia(audio); + } + pendingAudio.clear(); + if (closeRequested) { + sendCloseStream(); + closeRequested = false; + } + } + + private void sendCloseStream() { + try { + dg.sendCloseStream(ListenV2CloseStream.builder().build()); + } catch (Exception e) { + System.err.println("Error sending CloseStream to Deepgram: " + e.getMessage()); + } + } + + void disconnect() { + try { + dg.disconnect(); + } catch (Exception ignored) { + // already closed + } + } + } + /** * Configures the /api/flux WebSocket endpoint. - * Validates JWT from subprotocol on upgrade, then creates a bidirectional - * proxy to Deepgram's Flux API. + * Validates JWT from subprotocol on upgrade, then bridges the browser to + * Deepgram's Flux API via the SDK's Listen v2 WebSocket client. * * @param ws Javalin WebSocket config */ @@ -377,46 +443,84 @@ private static void handleFluxWebSocket(WsConfig ws) { System.out.println("Client connected to /api/flux (authenticated)"); activeConnections.add(ctx); - // Build the Deepgram URL with forwarded query parameters - String deepgramUrl = buildDeepgramUrl(ctx); - System.out.println("Connecting to Deepgram Flux: " + deepgramUrl); - - // Create outbound WebSocket connection to Deepgram try { - ClientUpgradeRequest upgradeRequest = new ClientUpgradeRequest(); - upgradeRequest.setHeader("Authorization", "Token " + apiKey); + V2ConnectOptions options = buildConnectOptions(ctx); + + V2WebSocketClient dg = deepgram.listen().v2().v2WebSocket(); + FluxBridge bridge = new FluxBridge(dg); + ctx.attribute("bridge", bridge); + + // Deepgram -> browser: forward the raw Flux JSON verbatim so the + // frontend receives Deepgram's native wire format unchanged. + dg.onMessage(raw -> { + try { + if (ctx.session.isOpen()) { + ctx.send(raw); + } + } catch (Exception e) { + System.err.println("Error forwarding Deepgram message to client: " + e.getMessage()); + } + }); - DeepgramSocket dgSocket = new DeepgramSocket(ctx); - // Store the Deepgram socket on the ctx attribute map for forwarding - ctx.attribute("deepgramSocket", dgSocket); + dg.onError(error -> { + System.err.println("Deepgram socket error: " + error.getMessage()); + if (ctx.session.isOpen()) { + ctx.closeSession(1011, "Deepgram connection error"); + } + }); - wsClient.connect(dgSocket, URI.create(deepgramUrl), upgradeRequest); + dg.onDisconnected(reason -> { + System.out.println("Deepgram connection closed: " + reason.getCode() + " " + reason.getReason()); + if (ctx.session.isOpen()) { + ctx.closeSession(getSafeCloseCode(reason.getCode()), + reason.getReason() != null ? reason.getReason() : ""); + } + }); + + System.out.println("Connecting to Deepgram Flux (SDK Listen v2)"); + + dg.connect(options).whenComplete((v, err) -> { + if (err != null) { + System.err.println("Deepgram connection failed to open: " + err.getMessage()); + if (ctx.session.isOpen()) { + ctx.closeSession(1011, "Failed to connect to Deepgram"); + } + return; + } + System.out.println("Connected to Deepgram Flux API"); + bridge.markReady(); + }); } catch (Exception e) { System.err.println("Failed to connect to Deepgram: " + e.getMessage()); ctx.closeSession(1011, "Failed to connect to Deepgram"); } }); - // Forward client messages to Deepgram + // Forward client control messages (JSON) to Deepgram ws.onMessage(ctx -> { - DeepgramSocket dgSocket = ctx.attribute("deepgramSocket"); - if (dgSocket != null && dgSocket.isOpen()) { - String text = ctx.message(); - dgSocket.sendText(text); + FluxBridge bridge = ctx.attribute("bridge"); + if (bridge == null) return; + try { + JsonNode msg = objectMapper.readTree(ctx.message()); + String type = msg.path("type").asText(""); + if ("CloseStream".equals(type)) { + bridge.closeStream(); + } else { + System.out.println("Ignoring client control message type: " + type); + } + } catch (Exception e) { + System.err.println("Ignoring non-JSON message from client"); } }); - // Forward client binary messages to Deepgram + // Forward client binary audio to Deepgram ws.onBinaryMessage(ctx -> { - DeepgramSocket dgSocket = ctx.attribute("deepgramSocket"); - if (dgSocket != null && dgSocket.isOpen()) { - byte[] data = ctx.data(); - int offset = ctx.offset(); - int length = ctx.length(); - byte[] bytes = new byte[length]; - System.arraycopy(data, offset, bytes, 0, length); - dgSocket.sendBinary(ByteBuffer.wrap(bytes)); - } + FluxBridge bridge = ctx.attribute("bridge"); + if (bridge == null) return; + byte[] data = ctx.data(); + int offset = ctx.offset(); + int length = ctx.length(); + bridge.sendAudio(ByteString.of(data, offset, length)); }); // Handle client disconnect @@ -425,9 +529,9 @@ private static void handleFluxWebSocket(WsConfig ws) { String reason = ctx.reason() != null ? ctx.reason() : ""; System.out.println("Client disconnected: " + code + " " + reason); - DeepgramSocket dgSocket = ctx.attribute("deepgramSocket"); - if (dgSocket != null && dgSocket.isOpen()) { - dgSocket.close(1000, "Client disconnected"); + FluxBridge bridge = ctx.attribute("bridge"); + if (bridge != null) { + bridge.disconnect(); } activeConnections.remove(ctx); }); @@ -438,158 +542,21 @@ private static void handleFluxWebSocket(WsConfig ws) { if (error != null) { System.err.println("Client WebSocket error: " + error.getMessage()); } - DeepgramSocket dgSocket = ctx.attribute("deepgramSocket"); - if (dgSocket != null && dgSocket.isOpen()) { - dgSocket.close(1011, "Client error"); + FluxBridge bridge = ctx.attribute("bridge"); + if (bridge != null) { + bridge.disconnect(); } activeConnections.remove(ctx); }); } - /** - * Jetty WebSocket endpoint for the outbound connection to Deepgram. - * Forwards all messages from Deepgram back to the connected client. - */ - @WebSocket - public static class DeepgramSocket { - - /** Reference to the client-side WsContext for forwarding messages */ - private final WsContext clientCtx; - - /** The Deepgram-side Jetty WebSocket session */ - private volatile Session deepgramSession; - - /** Message counters for debug logging */ - private int clientMessageCount = 0; - private int deepgramMessageCount = 0; - - public DeepgramSocket(WsContext clientCtx) { - this.clientCtx = clientCtx; - } - - @OnWebSocketConnect - public void onOpen(Session session) { - this.deepgramSession = session; - System.out.println("Connected to Deepgram Flux API"); - } - - /** - * Forwards text messages from Deepgram to the client. - */ - @OnWebSocketMessage - public void onTextMessage(Session session, String message) { - deepgramMessageCount++; - if (deepgramMessageCount % 10 == 0) { - System.out.println("<- Deepgram text message #" + deepgramMessageCount - + " (size: " + message.length() + ")"); - } - try { - if (clientCtx.session.isOpen()) { - clientCtx.send(message); - } - } catch (Exception e) { - System.err.println("Error forwarding Deepgram text to client: " + e.getMessage()); - } - } - - /** - * Forwards binary messages from Deepgram to the client. - */ - @OnWebSocketMessage - public void onBinaryMessage(byte[] payload, int offset, int len) { - deepgramMessageCount++; - if (deepgramMessageCount % 10 == 0) { - System.out.println("<- Deepgram binary message #" + deepgramMessageCount - + " (size: " + len + ")"); - } - try { - if (clientCtx.session.isOpen()) { - byte[] data = new byte[len]; - System.arraycopy(payload, offset, data, 0, len); - clientCtx.send(ByteBuffer.wrap(data)); - } - } catch (Exception e) { - System.err.println("Error forwarding Deepgram binary to client: " + e.getMessage()); - } - } - - @OnWebSocketClose - public void onClose(int statusCode, String reason) { - System.out.println("Deepgram connection closed: " + statusCode + " " + reason); - try { - if (clientCtx.session.isOpen()) { - int safeCode = getSafeCloseCode(statusCode); - clientCtx.closeSession(safeCode, reason != null ? reason : ""); - } - } catch (Exception e) { - System.err.println("Error closing client after Deepgram close: " + e.getMessage()); - } - } - - @OnWebSocketError - public void onError(Throwable cause) { - System.err.println("Deepgram WebSocket error: " + cause.getMessage()); - try { - if (clientCtx.session.isOpen()) { - clientCtx.closeSession(1011, "Deepgram connection error"); - } - } catch (Exception e) { - System.err.println("Error closing client after Deepgram error: " + e.getMessage()); - } - } - - /** Check if the Deepgram connection is open */ - public boolean isOpen() { - return deepgramSession != null && deepgramSession.isOpen(); - } - - /** Send text data to Deepgram */ - public void sendText(String text) { - clientMessageCount++; - if (clientMessageCount % 100 == 0) { - System.out.println("-> Client text message #" + clientMessageCount - + " (size: " + text.length() + ")"); - } - if (isOpen()) { - try { - deepgramSession.getRemote().sendString(text); - } catch (Exception e) { - System.err.println("Error sending text to Deepgram: " + e.getMessage()); - } - } - } - - /** Send binary data to Deepgram */ - public void sendBinary(ByteBuffer data) { - clientMessageCount++; - if (clientMessageCount % 100 == 0) { - System.out.println("-> Client binary message #" + clientMessageCount - + " (size: " + data.remaining() + ")"); - } - if (isOpen()) { - try { - deepgramSession.getRemote().sendBytes(data); - } catch (Exception e) { - System.err.println("Error sending binary to Deepgram: " + e.getMessage()); - } - } - } - - /** Close the Deepgram connection */ - public void close(int code, String reason) { - if (isOpen()) { - deepgramSession.close(code, reason); - } - } - } - // ======================================================================== // SECTION 10: SERVER START // ======================================================================== /** * Application entry point. Loads configuration, validates the API key, - * initializes the Jetty WebSocket client, and starts the Javalin server. + * initializes the Deepgram SDK client, and starts the Javalin server. * * @param args Command-line arguments (unused) */ @@ -597,14 +564,22 @@ public static void main(String[] args) { // Load API key (exits if missing) apiKey = loadApiKey(); - // Initialize Jetty WebSocket client for outbound Deepgram connections - wsClient = new WebSocketClient(); - try { - wsClient.start(); - } catch (Exception e) { - System.err.println("Failed to start WebSocket client: " + e.getMessage()); - System.exit(1); + // Build the Deepgram SDK client. DEEPGRAM_BASE_URL (e.g. a staging host + // like wss://api.staging.deepgram.com) overrides the default production + // endpoint used for the /v2/listen Flux websocket. + var builder = DeepgramClient.builder().apiKey(apiKey); + String baseUrl = getEnv("DEEPGRAM_BASE_URL", null); + if (baseUrl != null && !baseUrl.isEmpty()) { + String https = baseUrl.replaceFirst("^wss://", "https://").replaceFirst("^ws://", "http://"); + builder.environment(Environment.custom() + .base(https) + .production(baseUrl) + .agent(baseUrl) + .agentRest(https) + .build()); + System.out.println("Using custom Deepgram base URL: " + baseUrl); } + deepgram = builder.build(); // Create Javalin app with CORS enabled Javalin app = Javalin.create(config -> { @@ -626,7 +601,7 @@ public static void main(String[] args) { ctx.json(Map.of("status", "ok")); }); - // WebSocket proxy route (authenticated via subprotocol) + // WebSocket bridge route (authenticated via subprotocol) app.ws("/api/flux", App::handleFluxWebSocket); // Graceful shutdown hook @@ -643,13 +618,6 @@ public static void main(String[] args) { } } - // Stop the Jetty WebSocket client - try { - wsClient.stop(); - } catch (Exception e) { - System.err.println("Error stopping WebSocket client: " + e.getMessage()); - } - System.out.println("Shutdown complete"); }));