From 9868b29dee27171527a6dfc9691b5c6be3c068d5 Mon Sep 17 00:00:00 2001 From: kyle-fitzp Date: Tue, 30 Jun 2026 09:31:01 -0500 Subject: [PATCH 1/4] Initial unfinished RTMP rework --- .../impl/sensor/rtmp/RtmpPortSingleton.java | 65 -- .../sensor/rtmp/config/ConnectionConfig.java | 43 - .../impl/sensor/rtmp/config/HostType.java | 26 - .../sensor/{rtmp => rtmpcam}/Activator.java | 2 +- .../{rtmp => rtmpcam}/RtmpDescriptor.java | 4 +- .../sensor/{rtmp => rtmpcam}/RtmpDriver.java | 41 +- .../rtmpcam/config/ConnectionConfig.java | 70 ++ .../{rtmp => rtmpcam}/config/RtmpConfig.java | 2 +- .../connection/RtmpConnectionContext.java | 9 + .../connection/RtmpConnectionHandler.java | 157 ++++ .../rtmpcam/connection/RtmpListener.java | 36 + .../connection/RtmpListenerManager.java | 97 +++ .../rtmpcam/connection/RtmpNegotiator.java | 815 ++++++++++++++++++ .../rtmpcam/connection/RtmpPortServer.java | 60 ++ .../rtmpcam/event/RtmpConnectEvent.java | 17 + .../rtmpcam/event/RtmpDisconnectEvent.java | 8 + .../impl/sensor/rtmpcam/event/RtmpEvent.java | 5 + .../rtmpcam/event/RtmpEventCallback.java | 5 + .../rtmpcam/event/RtmpReconnectEvent.java | 8 + .../sensor/rtmpcam/event/RtmpStreamEvent.java | 16 + .../sensor/rtmpcam/helpers/RtmpClientSet.java | 92 ++ .../rtmpcam/helpers/RtmpStringParser.java | 49 ++ .../sensor/rtmpcam/stream/StreamInfo.java | 8 + .../org.sensorhub.api.module.IModuleProvider | 2 +- 24 files changed, 1492 insertions(+), 145 deletions(-) delete mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpPortSingleton.java delete mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/ConnectionConfig.java delete mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/HostType.java rename sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/{rtmp => rtmpcam}/Activator.java (95%) rename sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/{rtmp => rtmpcam}/RtmpDescriptor.java (91%) rename sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/{rtmp => rtmpcam}/RtmpDriver.java (94%) create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java rename sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/{rtmp => rtmpcam}/config/RtmpConfig.java (97%) create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionContext.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpConnectEvent.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEvent.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEventCallback.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpReconnectEvent.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpStreamEvent.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java create mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/stream/StreamInfo.java diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpPortSingleton.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpPortSingleton.java deleted file mode 100644 index 1f1bc4098..000000000 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpPortSingleton.java +++ /dev/null @@ -1,65 +0,0 @@ -/***************************** BEGIN LICENSE BLOCK *************************** - The contents of this file are subject to the Mozilla Public License, v. 2.0. - If a copy of the MPL was not distributed with this file, You can obtain one - at http://mozilla.org/MPL/2.0/. - - Software distributed under the License is distributed on an "AS IS" basis, - WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License - for the specific language governing rights and limitations under the License. - - Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. - ******************************* END LICENSE BLOCK ***************************/ - -package org.sensorhub.impl.sensor.rtmp; - -import java.util.HashMap; -import java.util.Map; - -/** - * Tracks RTMP listener ports currently reserved by RTMP driver modules. - *

- * The singleton prevents multiple RTMP driver instances from attempting to listen - * on the same port at the same time. All public methods are synchronized to - * provide simple thread-safe access to the port reservation map. - *

- */ -public final class RtmpPortSingleton { - private static final RtmpPortSingleton instance = new RtmpPortSingleton(); - - private final Map urls = new HashMap<>(); - - public static RtmpPortSingleton getInstance() { - return instance; - } - - /** - * Attempts to reserve a port for the specified module. - *

- * If the port is not already reserved, this method records the module unique - * identifier and returns {@code null}. If the port is already reserved, this - * method returns the unique identifier of the module that currently owns it. - *

- * - * @param url RTMP listener port to reserve - * @param moduleUid unique identifier of the module requesting the port - * @return {@code null} if the reservation succeeded; otherwise the unique - * identifier of the module currently using the port - */ - public synchronized String addConnection(int url, String moduleUid) { - if (urls.containsKey(url)) { - return urls.get(url); - } else { - urls.put(url, moduleUid); - return null; - } - } - - /** - * Releases a previously reserved RTMP listener port. - * - * @param url RTMP listener port to release - */ - public synchronized void removeConnection(int url) { - urls.remove(url); - } -} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/ConnectionConfig.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/ConnectionConfig.java deleted file mode 100644 index 5f8900d5f..000000000 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/ConnectionConfig.java +++ /dev/null @@ -1,43 +0,0 @@ -/***************************** BEGIN LICENSE BLOCK *************************** - The contents of this file are subject to the Mozilla Public License, v. 2.0. - If a copy of the MPL was not distributed with this file, You can obtain one - at http://mozilla.org/MPL/2.0/. - - Software distributed under the License is distributed on an "AS IS" basis, - WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License - for the specific language governing rights and limitations under the License. - - Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. - ******************************* END LICENSE BLOCK ***************************/ - -package org.sensorhub.impl.sensor.rtmp.config; - -import org.sensorhub.api.config.DisplayInfo; - -public class ConnectionConfig { - - /* - @DisplayInfo.Required - @DisplayInfo(label = "Generate Random Stream Key", desc = "Enable to generate and append a random hex string to the path. " + - "Recommended for security. Only enable on first init, otherwise path will include multiple keys. ") - public boolean generateRandomStreamKey = true; - - */ - - @DisplayInfo.Required - @DisplayInfo(label = "Host", desc = "Domain listening for an RTMP connection request. Unspecified should work " + - "for most cases.") - public HostType host = HostType.UNSPECIFIED; - - @DisplayInfo.Required - @DisplayInfo(label = "Port", desc = "Port listening for an RTMP connection request.") - @DisplayInfo.ValueRange(min = 1, max = 65535) - public int port = 1935; - - /* - @DisplayInfo(label = "Path", desc = "(Optional) Path to listen for an RTMP connection request. I.e. everything in the URL " + - "after the port.") - public String path = ""; - - */ -} \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/HostType.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/HostType.java deleted file mode 100644 index cd72f7f5b..000000000 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/HostType.java +++ /dev/null @@ -1,26 +0,0 @@ -/***************************** BEGIN LICENSE BLOCK *************************** - The contents of this file are subject to the Mozilla Public License, v. 2.0. - If a copy of the MPL was not distributed with this file, You can obtain one - at http://mozilla.org/MPL/2.0/. - - Software distributed under the License is distributed on an "AS IS" basis, - WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License - for the specific language governing rights and limitations under the License. - - Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. - ******************************* END LICENSE BLOCK ***************************/ - -package org.sensorhub.impl.sensor.rtmp.config; - -public enum HostType { - UNSPECIFIED("0.0.0.0"), - LOCALHOST("localhost"), - DOCKER_INTERNAL("host.docker.internal")/*, - OVERRIDE("")*/; - - public final String host; - - HostType(String host) { - this.host = host; - } -} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/Activator.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/Activator.java similarity index 95% rename from sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/Activator.java rename to sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/Activator.java index 8f3900c58..935857b5f 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/Activator.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/Activator.java @@ -10,7 +10,7 @@ Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.sensor.rtmp; +package org.sensorhub.impl.sensor.rtmpcam; import org.osgi.framework.BundleActivator; import org.sensorhub.utils.OshBundleActivator; diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDescriptor.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDescriptor.java similarity index 91% rename from sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDescriptor.java rename to sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDescriptor.java index 85396aed1..3f3af6b82 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDescriptor.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDescriptor.java @@ -10,13 +10,13 @@ Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.sensor.rtmp; +package org.sensorhub.impl.sensor.rtmpcam; import org.sensorhub.api.module.IModule; import org.sensorhub.api.module.IModuleProvider; import org.sensorhub.api.module.ModuleConfig; import org.sensorhub.impl.module.JarModuleProvider; -import org.sensorhub.impl.sensor.rtmp.config.RtmpConfig; +import org.sensorhub.impl.sensor.rtmpcam.config.RtmpConfig; public class RtmpDescriptor extends JarModuleProvider implements IModuleProvider diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDriver.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java similarity index 94% rename from sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDriver.java rename to sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java index f28a7f409..9d055997f 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDriver.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java @@ -10,7 +10,7 @@ Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.sensor.rtmp; +package org.sensorhub.impl.sensor.rtmpcam; import org.sensorhub.api.common.SensorHubException; import org.sensorhub.api.module.ModuleEvent; @@ -18,7 +18,11 @@ import org.sensorhub.impl.sensor.AbstractSensorModule; import org.sensorhub.impl.sensor.ffmpeg.outputs.AudioOutput; import org.sensorhub.impl.sensor.ffmpeg.outputs.VideoOutput; -import org.sensorhub.impl.sensor.rtmp.config.RtmpConfig; +import org.sensorhub.impl.sensor.rtmpcam.config.ConnectionConfig; +import org.sensorhub.impl.sensor.rtmpcam.config.RtmpConfig; +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpConnectionContext; +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpListener; +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpListenerManager; import org.sensorhub.mpegts.MpegTsProcessor; import org.sensorhub.utils.Async; @@ -39,10 +43,10 @@ *

*

* Only one RTMP driver instance may use a given port at a time. Port ownership - * is tracked through a shared {@link RtmpPortSingleton}. + * is tracked through a shared {@link RtmpListenerManager}. *

*/ -public class RtmpDriver extends AbstractSensorModule { +public class RtmpDriver extends AbstractSensorModule implements RtmpListener { private static final String COMMAND_LINE_ARGS = "-timeout 0 -listen 1 -username test -password test"; private static final int EXECUTOR_JOIN_TIMEOUT = 10; private static final TimeUnit EXECUTOR_JOIN_TIME_UNIT = TimeUnit.SECONDS; @@ -50,13 +54,12 @@ public class RtmpDriver extends AbstractSensorModule { private static final TimeUnit HEARTBEAT_TIME_UNIT = TimeUnit.SECONDS; private static final int MAX_STARTUP_WAIT_TIME_MS = 5000; - private final RtmpPortSingleton portSingleton = RtmpPortSingleton.getInstance(); + private final RtmpListenerManager portSingleton = RtmpListenerManager.getInstance(); private ExecutorService executorService; private ExecutorService videoExecutorService; private ExecutorService audioExecutorService; private ScheduledExecutorService heartbeatExecutorService; - final AtomicReference mpegTsProcessor = new AtomicReference<>(); final AtomicReference> videoOutput = new AtomicReference<>(); final AtomicReference> audioOutput = new AtomicReference<>(); @@ -94,6 +97,8 @@ protected void doInit() throws SensorHubException { generateXmlID("RTMP_", config.serialNumber); } + + portSingleton.removeConnection(connectionPort); setConnectionUrl(); @@ -501,4 +506,28 @@ public void cleanup() throws SensorHubException { public boolean isConnected() { return isConnected; } + + @Override + public ConnectionConfig config() { + return config.connectionConfig; + } + + @Override + public void publish(byte[] data, int streamIndex, long pts, boolean isVideo) { + if (isVideo) { + videoOutput.get(). + } + } + + @Override + public void onStreamConnected(RtmpConnectionContext ctx) { + + } + + public void onStreamInfo() + + @Override + public void onStreamDisconnected() { + + } } \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java new file mode 100644 index 000000000..86cda0855 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java @@ -0,0 +1,70 @@ +/***************************** BEGIN LICENSE BLOCK *************************** + The contents of this file are subject to the Mozilla Public License, v. 2.0. + If a copy of the MPL was not distributed with this file, You can obtain one + at http://mozilla.org/MPL/2.0/. + + Software distributed under the License is distributed on an "AS IS" basis, + WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License + for the specific language governing rights and limitations under the License. + + Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. + ******************************* END LICENSE BLOCK ***************************/ + +package org.sensorhub.impl.sensor.rtmpcam.config; + +import org.sensorhub.api.config.DisplayInfo; + +public class ConnectionConfig { + + /* + @DisplayInfo.Required + @DisplayInfo(label = "Generate Random Stream Key", desc = "Enable to generate and append a random hex string to the path. " + + "Recommended for security. Only enable on first init, otherwise path will include multiple keys. ") + public boolean generateRandomStreamKey = true; + + */ + + @DisplayInfo.Required + @DisplayInfo(label = "Username") + public String username = ""; + + @DisplayInfo.Required + @DisplayInfo(label = "Password") + public String password = ""; + + @DisplayInfo.Required + @DisplayInfo(label = "Port", desc = "Port listening for an RTMP connection request.") + @DisplayInfo.ValueRange(min = 1, max = 65535) + public int port = 1935; + + @DisplayInfo.Required + @DisplayInfo(label = "Path") + public String path = ""; + + @DisplayInfo(label = "Stream Key", desc = "(Optional) Stream key to use for the RTMP connection.") + public String streamKey = ""; + + @DisplayInfo(label = "Generate Random Stream Key", desc = "Overwrite the stream key field with a random string of characters.") + public boolean generateRandomKey = false; + + /** Key for this config's exact fields — used as the map key on registration. */ + public String compositeKey() { + return compositeKey(username, password, port, path, streamKey); + } + + /** + * Static form used by the router to generate wildcard candidate keys + * (null fields become empty strings, producing a distinct key per specificity level). + * + * Format: "username:password:port:path:streamKey" + * Example: "alice:secret:1935:live:cam1" + * Catch-all: "::1935::" + */ + public static String compositeKey(String username, String password, int port, String path, String streamKey) { + return (username != null ? username : "") + ":" + + (password != null ? password : "") + ":" + + port + ":" + + (path != null ? path : "") + ":" + + (streamKey != null ? streamKey : ""); + } +} \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/RtmpConfig.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/RtmpConfig.java similarity index 97% rename from sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/RtmpConfig.java rename to sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/RtmpConfig.java index d466a4ba0..43ae02853 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/config/RtmpConfig.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/RtmpConfig.java @@ -10,7 +10,7 @@ Copyright (C) 2026 GeoRobotix Innovative Research, Inc. All Rights Reserved. ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.sensor.rtmp.config; +package org.sensorhub.impl.sensor.rtmpcam.config; import org.sensorhub.api.config.DisplayInfo; import org.sensorhub.api.sensor.PositionConfig; diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionContext.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionContext.java new file mode 100644 index 000000000..eb38056fd --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionContext.java @@ -0,0 +1,9 @@ +package org.sensorhub.impl.sensor.rtmpcam.connection; + +public record RtmpConnectionContext( + int port, + String username, + String password, + String path, + String streamKey +) {} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java new file mode 100644 index 000000000..34ae623f3 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java @@ -0,0 +1,157 @@ +package org.sensorhub.impl.sensor.rtmpcam.connection; + +import org.bytedeco.ffmpeg.avcodec.AVPacket; +import org.bytedeco.ffmpeg.avformat.*; +import org.bytedeco.ffmpeg.avutil.AVDictionary; +import org.bytedeco.javacpp.BytePointer; +import org.bytedeco.javacpp.Pointer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.*; +import java.net.Socket; +import java.util.Optional; + +import static org.bytedeco.ffmpeg.global.avcodec.*; +import static org.bytedeco.ffmpeg.global.avformat.*; +import static org.bytedeco.ffmpeg.global.avutil.*; + +/** + * Handles one RTMP connection: + * 1. Handshake + * 2. AMF0 negotiation → {@link RtmpConnectionContext} + * 3. Route to a matching {@link RtmpListener} + * 4. RTMP chunks → FLV pipe → FFmpeg custom AVIO + * 5. Deliver encoded packets via {@link RtmpListener#publish} + */ +class RtmpConnectionHandler { + + private static final int AVIO_BUF = 64 * 1024; + private static final Logger logger = LoggerFactory.getLogger(RtmpConnectionHandler.class); + + private final Socket socket; + private final int port; + private final RtmpListenerManager manager; + + RtmpConnectionHandler(Socket socket, int port, RtmpListenerManager manager) { + this.socket = socket; + this.port = port; + this.manager = manager; + } + + void handle() { + try (socket) { + var in = new DataInputStream(socket.getInputStream()); + var out = new DataOutputStream(socket.getOutputStream()); + + // One negotiator instance owns all state across all three phases + RtmpNegotiator negotiator = new RtmpNegotiator(in, out, port); + + negotiator.doHandshake(); + RtmpConnectionContext ctx = negotiator.negotiate(); + + Optional match = manager.route(ctx); + if (match.isEmpty()) { + // No listener registered for this combination — drop silently + System.out.printf("[RTMP:%d] No listener for path='%s' key='%s'%n", + port, ctx.path(), ctx.streamKey()); + return; + } + + RtmpListener listener = match.get(); + listener.onConnected(ctx); + try { + pipeToFfmpeg(negotiator.buildFlvStream(), listener); + } finally { + listener.onDisconnected(); + } + + } catch (Exception e) { + System.err.printf("[RTMP:%d] connection fault: %s%n", port, e.getMessage()); + } + } + + // ── FFmpeg pipeline ──────────────────────────────────────────────────── + + private void pipeToFfmpeg(InputStream flvStream, RtmpListener listener) { + + // Both must stay reachable for the pipeline's lifetime: + // readCb — stored as a raw native function pointer inside AVIOContext + // avioBuf — FFmpeg takes ownership; free via ctx.buffer(), not this reference + Read_packet_Pointer_BytePointer_int readCb = buildReadCb(flvStream); + BytePointer avioBuf = new BytePointer(av_malloc(AVIO_BUF)).capacity(AVIO_BUF); + + AVIOContext avioCtx = avio_alloc_context( + avioBuf, AVIO_BUF, + 0, // read-only + (Pointer) null, // opaque + (Read_packet_Pointer_BytePointer_int) readCb, (Write_packet_Pointer_BytePointer_int) null, (Seek_Pointer_long_int) null); // no write, no seek (live stream) + + AVFormatContext fmtCtx = avformat_alloc_context(); + fmtCtx.pb(avioCtx); // must be set before avformat_open_input + + int ret = avformat_open_input(fmtCtx, (String) null, + av_find_input_format("flv"), null); + + if (ret < 0) { logError("avformat_open_input", ret); freeAVIO(avioCtx); return; } + + avformat_find_stream_info(fmtCtx, (AVDictionary) null); + packetLoop(fmtCtx, listener); + + avformat_close_input(fmtCtx); + freeAVIO(avioCtx); + // readCb and avioBuf are now safe to collect + } + + private void packetLoop(AVFormatContext fmtCtx, RtmpListener listener) { + AVPacket pkt = av_packet_alloc(); + try { + int ret; + while ((ret = av_read_frame(fmtCtx, pkt)) >= 0) { + AVStream stream = fmtCtx.streams(pkt.stream_index()); + int mediaType = stream.codecpar().codec_type(); + if (mediaType == AVMEDIA_TYPE_VIDEO || mediaType == AVMEDIA_TYPE_AUDIO) { + byte[] data = new byte[pkt.size()]; + pkt.data().get(data); + listener.publish(data, pkt.stream_index(), pkt.pts(), + mediaType == AVMEDIA_TYPE_VIDEO); + } + av_packet_unref(pkt); + } + if (ret != AVERROR_EOF) logError("av_read_frame", ret); + } finally { + av_packet_free(pkt); + } + } + + private Read_packet_Pointer_BytePointer_int buildReadCb(InputStream src) { + byte[] tmp = new byte[AVIO_BUF]; + return new Read_packet_Pointer_BytePointer_int() { + @Override + public int call(Pointer opaque, BytePointer dst, int requested) { + try { + int n = src.read(tmp, 0, Math.min(requested, tmp.length)); + if (n <= 0) return AVERROR_EOF; + dst.put(tmp, 0, n); + return n; + } catch (IOException e) { + return AVERROR_EOF; + } + } + }; + } + + private static void freeAVIO(AVIOContext ctx) { + if (ctx == null || ctx.isNull()) return; + BytePointer buf = ctx.buffer(); + if (buf != null && !buf.isNull()) av_freep(buf); + avio_context_free(ctx); + } + + private static void logError(String fn, int code) { + try (BytePointer buf = new BytePointer(128)) { + av_strerror(code, buf, buf.capacity()); + logger.warn("FFmpeg returned error code {} from {}: {}", code, fn, buf.getString()); + } + } +} \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java new file mode 100644 index 000000000..6bf2de14a --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java @@ -0,0 +1,36 @@ +package org.sensorhub.impl.sensor.rtmpcam.connection; + +import org.sensorhub.impl.sensor.rtmpcam.config.ConnectionConfig; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpConnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpDisconnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpReconnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpStreamEvent; + +/** + * Extend this class and register it with {@link RtmpListenerManager} to + * receive encoded packets from a matched RTMP stream. + */ +public interface RtmpListener { + + public ConnectionConfig config(); + + /** + * Receives one encoded packet demuxed from the RTMP stream. + * + * @param data raw encoded bytes (H.264 Annex B, AAC, etc.) + * @param streamIndex FFmpeg stream index within the FLV container + * @param pts presentation timestamp in stream timebase units + * @param isVideo true for video, false for audio + */ + public void publish(byte[] data, int streamIndex, long pts, boolean isVideo); + + /** Called once after negotiation succeeds and this listener is selected. */ + public void onConnected(RtmpConnectEvent event); + + /** Called once when the client disconnects or the pipeline faults. */ + public void onDisconnected(RtmpDisconnectEvent event); + + public void onReconnected(RtmpReconnectEvent event); + + public void onStreamInfo(RtmpStreamEvent event); +} \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java new file mode 100644 index 000000000..36aefebc1 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java @@ -0,0 +1,97 @@ +package org.sensorhub.impl.sensor.rtmpcam.connection; + +import org.sensorhub.impl.sensor.rtmpcam.config.ConnectionConfig; + +import java.util.Comparator; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.IntStream; + +public class RtmpListenerManager { + + private static final RtmpListenerManager INSTANCE = new RtmpListenerManager(); + + public static RtmpListenerManager getInstance() { return INSTANCE; } + + /** + * All 16 bitmasks for the 4 optional fields (username, password, path, streamKey), + * pre-sorted from most specific (0b1111 = all four) to least (0b0000 = none). + * The router iterates these in order so the first map hit is always the best match. + * + * Bit layout: bit3=username, bit2=password, bit1=path, bit0=streamKey + */ + private static final int[] CANDIDATE_MASKS = IntStream.range(0, 16) + .boxed() + .sorted(Comparator.comparingInt(Integer::bitCount).reversed()) + .mapToInt(Integer::intValue) + .toArray(); + + // One entry per unique composite key — second registration with the same + // config overwrites the first (unlike the old CopyOnWriteArrayList). + private final ConcurrentHashMap listeners = + new ConcurrentHashMap<>(); + + private final ConcurrentHashMap portServers = + new ConcurrentHashMap<>(); + + private final Object portLock = new Object(); + + // ── Registration ─────────────────────────────────────────────────────── + + public void addListener(RtmpListener listener) { + listeners.put(listener.config().compositeKey(), listener); + + synchronized (portLock) { + portServers.computeIfAbsent(listener.config().port, port -> { + RtmpPortServer srv = new RtmpPortServer(port, this); + srv.start(); + return srv; + }); + } + } + + public void removeListener(RtmpListener listener) { + // Two-arg remove: only deletes if the value still matches this exact listener, + // so a replacement registered under the same key isn't accidentally removed. + listeners.remove(listener.config().compositeKey(), listener); + + int port = listener.config().port; + synchronized (portLock) { + boolean anyRemaining = listeners.values().stream() + .anyMatch(l -> l.config().port == port); + if (!anyRemaining) { + RtmpPortServer srv = portServers.remove(port); + if (srv != null) srv.stop(); + } + } + } + + // ── Routing ──────────────────────────────────────────────────────────── + + /** + * Tries all 16 composite key variants for the connection in specificity + * order (most fields → fewest fields) and returns the first map hit. + * + * A listener registered as "::1935::" (catch-all, specificity 0) is only + * reached if no more-specific listener claims the connection. + */ + Optional route(RtmpConnectionContext ctx) { + String u = ctx.username(); + String pw = ctx.password(); + String pa = ctx.path(); + String sk = ctx.streamKey(); + + for (int mask : CANDIDATE_MASKS) { + String key = ConnectionConfig.compositeKey( + (mask & 0b1000) != 0 ? u : null, + (mask & 0b0100) != 0 ? pw : null, + ctx.port(), + (mask & 0b0010) != 0 ? pa : null, + (mask & 0b0001) != 0 ? sk : null); + + RtmpListener found = listeners.get(key); + if (found != null) return Optional.of(found); + } + return Optional.empty(); + } +} \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java new file mode 100644 index 000000000..b2a7b9469 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java @@ -0,0 +1,815 @@ +package org.sensorhub.impl.sensor.rtmpcam.connection; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.*; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.*; +import java.util.concurrent.ThreadLocalRandom; + +/** + * Owns the complete RTMP protocol for one publisher connection. + * + * Phase 1 doHandshake() — C0/C1/C2 ↔ S0/S1/S2 + * Phase 2 negotiate() — AMF0 command loop → RtmpConnectionContext + * Phase 3 buildFlvStream() — chunk → FLV pump on a virtual thread → InputStream + * + * All three phases share chunk-size and per-stream reassembly state. + */ +public class RtmpNegotiator { + + // ── Protocol constants ───────────────────────────────────────────────── + + private static final int RTMP_VERSION = 3; + private static final int HANDSHAKE_SIZE = 1536; + private static final int DEFAULT_CHUNK_SIZE = 128; + private static final int SERVER_CHUNK_SIZE = 4096; + private static final int WINDOW_ACK_SIZE = 2_500_000; + + // Inbound message type IDs + private static final int MSG_SET_CHUNK_SIZE = 1; + private static final int MSG_ABORT = 2; + private static final int MSG_ACK = 3; + private static final int MSG_USER_CONTROL = 4; + private static final int MSG_WINDOW_ACK_SIZE = 5; + private static final int MSG_SET_PEER_BANDWIDTH = 6; + private static final int MSG_AUDIO = 8; + private static final int MSG_VIDEO = 9; + private static final int MSG_DATA_AMF3 = 15; + private static final int MSG_COMMAND_AMF3 = 17; + private static final int MSG_DATA_AMF0 = 18; + private static final int MSG_COMMAND_AMF0 = 20; + + // User control event types + private static final int EVENT_PING_REQUEST = 6; + + // Chunk stream IDs used for server-sent messages (all < 64 → 1-byte basic header) + private static final int CS_PROTOCOL = 2; // protocol control + private static final int CS_COMMAND = 3; // AMF0 command responses + private static final int CS_STREAM = 5; // per-stream status (onStatus) + + private static final byte[] FLV_HEADER = { + 'F', 'L', 'V', + 0x01, // version + 0x05, // flags: audio | video + 0x00, 0x00, 0x00, 0x09, // data offset = 9 + 0x00, 0x00, 0x00, 0x00 // PreviousTagSize0 = 0 + }; + private static final Logger logger = LoggerFactory.getLogger(RtmpNegotiator.class); + + // ── Instance state ───────────────────────────────────────────────────── + + private final DataInputStream in; + private final DataOutputStream out; + private final int port; + + private int readChunkSize = DEFAULT_CHUNK_SIZE; + private int windowAckSize = WINDOW_ACK_SIZE; + private long bytesReceived = 0; + private long bytesAcked = 0; + + private final Map chunkStreams = new HashMap<>(); + + public RtmpNegotiator(DataInputStream in, DataOutputStream out, int port) { + this.in = in; + this.out = out; + this.port = port; + } + + // ── Phase 1: Handshake ───────────────────────────────────────────────── + + /** + * Performs the C0/C1/C2 ↔ S0/S1/S2 handshake (simple mode). + * Modern publishers (OBS, FFmpeg, etc.) accept the simple echo handshake. + */ + public void doHandshake() throws IOException { + // C0: version byte + int c0 = in.readUnsignedByte(); + if (c0 != RTMP_VERSION) throw new IOException("Unsupported RTMP version: " + c0); + + // C1: [4 time] [4 zeros] [1528 random] + byte[] c1 = new byte[HANDSHAKE_SIZE]; + in.readFully(c1); + + // S0 + out.writeByte(RTMP_VERSION); + + // S1: [4 time=0] [4 zeros] [1528 random] + byte[] s1 = new byte[HANDSHAKE_SIZE]; + ThreadLocalRandom.current().nextBytes(s1); + Arrays.fill(s1, 0, 8, (byte) 0); // zero time and reserved fields + + // S2: echo of C1 — [4 C1-time] [4 server-time] [1528 C1-random] + byte[] s2 = Arrays.copyOf(c1, HANDSHAKE_SIZE); + long now = System.currentTimeMillis(); + s2[4] = (byte)(now >> 24); + s2[5] = (byte)(now >> 16); + s2[6] = (byte)(now >> 8); + s2[7] = (byte) now; + + out.write(s1); + out.write(s2); + out.flush(); + + // C2: echo of S1 — read and discard (no validation for simple mode) + in.readFully(new byte[HANDSHAKE_SIZE]); + } + + // ── Phase 2: AMF0 Negotiation ────────────────────────────────────────── + + /** + * Sends server control messages then loops over incoming RTMP chunks, + * processing AMF0 commands until a {@code publish} command is confirmed. + * + * Handles: connect, releaseStream, FCPublish, createStream, publish. + * Ignores: getStreamLength, FCSubscribe, and any unknown commands. + */ + public RtmpConnectionContext negotiate() throws IOException { + // Server control sent immediately — expected before any AMF0 exchange + sendWindowAckSize(WINDOW_ACK_SIZE); + sendSetPeerBandwidth(WINDOW_ACK_SIZE); + sendSetChunkSize(SERVER_CHUNK_SIZE); + + String username = null; + String password = null; + String path = null; + String streamKey = null; + + negotiateLoop: + while (streamKey == null) { + RtmpMessage msg = readMessage(); + + switch (msg.type()) { + case MSG_SET_CHUNK_SIZE -> readChunkSize = parseUInt32(msg.data()); + case MSG_WINDOW_ACK_SIZE -> windowAckSize = parseUInt32(msg.data()); + case MSG_ACK -> { /* client ack — no action needed */ } + case MSG_ABORT -> chunkStreams.remove(parseUInt32(msg.data())); + case MSG_USER_CONTROL -> handleUserControl(msg.data()); + case MSG_SET_PEER_BANDWIDTH -> { /* ignore */ } + case MSG_DATA_AMF0 -> { /* @setDataFrame before publish — ignore */ } + case MSG_DATA_AMF3 -> { /* ignore */ } + + case MSG_COMMAND_AMF0 -> { + DataInputStream amf = wrapBytes(msg.data()); + Object nameObj = readAmf0Value(amf); + if (!(nameObj instanceof String cmdName)) continue; + + switch (cmdName) { + case "connect" -> { + double txId = asDouble(readAmf0Value(amf)); + + @SuppressWarnings("unchecked") + Map info = + (Map) readAmf0Value(amf); + + path = getString(info, "app"); + + // Credentials: prefer tcUrl embed, then explicit fields + String tcUrl = getString(info, "tcUrl"); + if (tcUrl != null) { + String[] creds = extractCredentials(tcUrl); + username = creds[0]; + password = creds[1]; + } + if (username == null) + username = coalesce(getString(info, "user"), + getString(info, "username")); + if (password == null) + password = coalesce(getString(info, "pass"), + getString(info, "password")); + + sendConnectResult(txId); + } + + case "releaseStream", "getStreamLength", "FCSubscribe" -> { /* no-op */ } + + case "FCPublish" -> { + double txId = asDouble(readAmf0Value(amf)); // txId + readAmf0Value(amf); // null + String name = readOptionalString(amf); + sendFCPublishResult(txId, name); + } + + case "createStream" -> { + double txId = asDouble(readAmf0Value(amf)); + sendCreateStreamResult(txId, 1); // always stream ID 1 + } + + case "publish" -> { + readAmf0Value(amf); // txId (0 for publish commands) + readAmf0Value(amf); // null command object + streamKey = (String) readAmf0Value(amf); // stream name + // publish type ("live", "record", "append") — ignored + sendPublishStart(1, streamKey); + break negotiateLoop; + } + + default -> + System.out.printf("[RTMP:%d] Unknown negotiate command: %s%n", + port, cmdName); + } + } + + case MSG_COMMAND_AMF3 -> { + // AMF3 command: skip leading 0x00 compatibility byte, then read name + byte[] d = msg.data(); + if (d.length > 1) { + Object nameObj = readAmf0Value(wrapBytes(d, 1)); + System.out.printf("[RTMP:%d] AMF3 command ignored: %s%n", port, nameObj); + } + } + } + } + + return new RtmpConnectionContext( + port, + username != null ? username : "", + password != null ? password : "", + path != null ? path : "", + streamKey); + } + + // ── Phase 3: FLV pump ────────────────────────────────────────────────── + + /** + * Starts a virtual thread that reads RTMP chunks and writes FLV-framed + * bytes into a pipe. The returned {@link InputStream} is consumed by + * FFmpeg via custom AVIO. + * + * The background thread exits cleanly when the publisher disconnects + * (IOException from the socket) or sends deleteStream / FCUnpublish. + */ + public InputStream buildFlvStream() throws IOException { + PipedOutputStream pipeOut = new PipedOutputStream(); + PipedInputStream pipeIn = new PipedInputStream(pipeOut, 1 << 20); + + Thread pumpThread = new Thread(() -> { + try (pipeOut) { + pipeOut.write(FLV_HEADER); + pumpFlv(pipeOut); + } catch (IOException ignored) { + // Normal exit: publisher dropped or pipe closed + } + }, "rtmp-flv-pump-" + port); + pumpThread.setDaemon(true); + pumpThread.start(); + + return pipeIn; + } + + private void pumpFlv(OutputStream flvOut) throws IOException { + while (true) { + RtmpMessage msg = readMessage(); + switch (msg.type()) { + case MSG_AUDIO, MSG_VIDEO -> writeFlvMediaTag(flvOut, msg); + case MSG_DATA_AMF0 -> writeFlvScriptTag(flvOut, msg); + case MSG_SET_CHUNK_SIZE -> readChunkSize = parseUInt32(msg.data()); + case MSG_WINDOW_ACK_SIZE -> windowAckSize = parseUInt32(msg.data()); + case MSG_ACK -> { /* ignore */ } + case MSG_ABORT -> chunkStreams.remove(parseUInt32(msg.data())); + case MSG_USER_CONTROL -> handleUserControl(msg.data()); + + case MSG_COMMAND_AMF0 -> { + Object nameObj = readAmf0Value(wrapBytes(msg.data())); + if (nameObj instanceof String cmdName) { + switch (cmdName) { + case "deleteStream", + "closeStream", + "FCUnpublish" -> { return; } + default -> { /* ignore mid-stream commands */ } + } + } + } + } + } + } + + // ── FLV tag emission ─────────────────────────────────────────────────── + + private void writeFlvMediaTag(OutputStream out, RtmpMessage msg) throws IOException { + writeFlvTag(out, msg.type(), msg.data(), (int) msg.timestamp()); + } + + /** + * Data messages carry "@setDataFrame" + "onMetaData" + ECMA array. + * FLV script tags expect "onMetaData" + ECMA array — strip the prefix. + */ + private void writeFlvScriptTag(OutputStream out, RtmpMessage msg) throws IOException { + DataInputStream src = wrapBytes(msg.data()); + Object first = readAmf0Value(src); + + byte[] payload = "@setDataFrame".equals(first) + ? src.readAllBytes() // remaining bytes start with "onMetaData" + : msg.data(); // already starts with event name + + writeFlvTag(out, 0x12 /* script */, payload, (int) msg.timestamp()); + } + + /** + * Writes one complete FLV tag: + * [1 type][3 dataSize][3 ts_low][1 ts_high][3 streamId=0][N data][4 prevTagSize] + */ + private static void writeFlvTag(OutputStream out, int tagType, + byte[] payload, int ts) throws IOException { + int dataSize = payload.length; + int tagTotal = 11 + dataSize; + + out.write(tagType); + writeUInt24(out, dataSize); + writeUInt24(out, ts & 0x00FFFFFF); // lower 24 bits + out.write((ts >> 24) & 0xFF); // TimestampExtended (upper 8 bits) + writeUInt24(out, 0); // StreamID always 0 in FLV + out.write(payload); + writeUInt32(out, tagTotal); // PreviousTagSize + } + + // ── Chunk reassembly ─────────────────────────────────────────────────── + + /** + * Reads and reassembles RTMP chunks until a complete RTMP message is ready. + * + * Chunk wire format (per Adobe spec): + * [basic header 1-3B] [message header 0/3/7/11B] [ext timestamp 0/4B] [payload ≤ chunkSize] + * + * fmt=0: full 11-byte header → new message with absolute timestamp + * fmt=1: 7-byte header → new message, inherits stream ID, delta timestamp + * fmt=2: 3-byte header → continues with same type+length, delta timestamp + * fmt=3: no header → continuation chunk, same everything + */ + private RtmpMessage readMessage() throws IOException { + while (true) { + + // ── Basic header ─────────────────────────────────────────────── + int byte0 = readByte(); + int fmt = (byte0 >> 6) & 0x3; + int csId = byte0 & 0x3F; + + // Extended chunk stream IDs + if (csId == 0) csId = readByte() + 64; + else if (csId == 1) csId = readByte() * 256 + readByte() + 64; + + ChunkStream cs = chunkStreams.computeIfAbsent(csId, k -> new ChunkStream()); + boolean prevDone = cs.payload == null || cs.bytesRead >= cs.messageLength; + + // ── Message header ───────────────────────────────────────────── + switch (fmt) { + case 0 -> { + // New message — absolute timestamp, full header + long ts = readUInt24(); + cs.messageLength = (int) readUInt24(); + cs.messageType = readByte(); + cs.messageStreamId = readLittleEndianInt(); + cs.hasExtTimestamp = (ts >= 0xFFFFFF); + cs.timestamp = cs.hasExtTimestamp ? readUInt32() : ts; + cs.timestampDelta = 0; + cs.payload = new byte[cs.messageLength]; + cs.bytesRead = 0; + } + case 1 -> { + // New message — inherits stream ID, delta timestamp + long delta = readUInt24(); + cs.messageLength = (int) readUInt24(); + cs.messageType = readByte(); + cs.hasExtTimestamp = (delta >= 0xFFFFFF); + cs.timestampDelta = cs.hasExtTimestamp ? readUInt32() : delta; + cs.timestamp += cs.timestampDelta; + cs.payload = new byte[cs.messageLength]; + cs.bytesRead = 0; + } + case 2 -> { + // Delta timestamp only — inherits type and length + long delta = readUInt24(); + cs.hasExtTimestamp = (delta >= 0xFFFFFF); + cs.timestampDelta = cs.hasExtTimestamp ? readUInt32() : delta; + cs.timestamp += cs.timestampDelta; + if (prevDone) { + cs.payload = new byte[cs.messageLength]; + cs.bytesRead = 0; + } + } + case 3 -> { + // Spec: re-read 4-byte ext timestamp if the last fmt 0/1/2 had one + if (cs.hasExtTimestamp) readUInt32(); + if (prevDone) { + cs.timestamp += cs.timestampDelta; + cs.payload = new byte[cs.messageLength]; + cs.bytesRead = 0; + } + } + } + + if (cs.payload == null) cs.payload = new byte[cs.messageLength]; // safety + + // ── Payload bytes for this chunk ─────────────────────────────── + int remaining = cs.messageLength - cs.bytesRead; + int toRead = Math.min(readChunkSize, remaining); + in.readFully(cs.payload, cs.bytesRead, toRead); + cs.bytesRead += toRead; + bytesReceived += toRead; + maybeAck(); + + if (cs.bytesRead >= cs.messageLength) { + byte[] data = cs.payload.clone(); + cs.payload = null; // ready for next message on this chunk stream + cs.bytesRead = 0; + return new RtmpMessage(cs.messageType, cs.messageStreamId, cs.timestamp, data); + } + // Message spans more chunks — keep looping + } + } + + // ── Server → Client: protocol control ───────────────────────────────── + + private void sendWindowAckSize(int size) throws IOException { + sendMessage(CS_PROTOCOL, MSG_WINDOW_ACK_SIZE, 0, 0, encodeUInt32(size)); + } + + private void sendSetPeerBandwidth(int bandwidth) throws IOException { + byte[] b = Arrays.copyOf(encodeUInt32(bandwidth), 5); + b[4] = 0x02; // limit type: dynamic + sendMessage(CS_PROTOCOL, MSG_SET_PEER_BANDWIDTH, 0, 0, b); + } + + private void sendSetChunkSize(int size) throws IOException { + sendMessage(CS_PROTOCOL, MSG_SET_CHUNK_SIZE, 0, 0, encodeUInt32(size)); + } + + private void sendStreamBegin(int streamId) throws IOException { + byte[] payload = new byte[6]; + payload[0] = 0x00; payload[1] = 0x00; // event type: StreamBegin + payload[2] = (byte)(streamId >> 24); payload[3] = (byte)(streamId >> 16); + payload[4] = (byte)(streamId >> 8); payload[5] = (byte) streamId; + sendMessage(CS_PROTOCOL, MSG_USER_CONTROL, 0, 0, payload); + } + + private void sendAck(long seq) throws IOException { + sendMessage(CS_PROTOCOL, MSG_ACK, 0, 0, encodeUInt32((int) seq)); + } + + private void sendPingResponse(long token) throws IOException { + byte[] payload = new byte[6]; + payload[0] = 0x00; payload[1] = 0x07; // event type: PingResponse + payload[2] = (byte)(token >> 24); payload[3] = (byte)(token >> 16); + payload[4] = (byte)(token >> 8); payload[5] = (byte) token; + sendMessage(CS_PROTOCOL, MSG_USER_CONTROL, 0, 0, payload); + } + + // ── Server → Client: AMF0 command responses ──────────────────────────── + + private void sendConnectResult(double txId) throws IOException { + ByteArrayOutputStream buf = new ByteArrayOutputStream(256); + DataOutputStream d = new DataOutputStream(buf); + + writeAmf0Str(d, "_result"); + writeAmf0Num(d, txId); + + writeAmf0ObjStart(d); + writeAmf0Field(d, "fmsVer", "FMS/3,0,1,123"); + writeAmf0Field(d, "capabilities", 31.0); + writeAmf0Field(d, "mode", 1.0); + writeAmf0ObjEnd(d); + + writeAmf0ObjStart(d); + writeAmf0Field(d, "level", "status"); + writeAmf0Field(d, "code", "NetConnection.Connect.Success"); + writeAmf0Field(d, "description", "Connection succeeded."); + writeAmf0Field(d, "objectEncoding", 0.0); + writeAmf0ObjEnd(d); + + sendMessage(CS_COMMAND, MSG_COMMAND_AMF0, 0, 0, buf.toByteArray()); + + // onBWDone — expected by OBS and other publishers after _result + buf.reset(); + writeAmf0Str(d, "onBWDone"); + writeAmf0Num(d, 0); + writeAmf0Null(d); + sendMessage(CS_COMMAND, MSG_COMMAND_AMF0, 0, 0, buf.toByteArray()); + } + + private void sendFCPublishResult(double txId, String streamKey) throws IOException { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + DataOutputStream d = new DataOutputStream(buf); + + writeAmf0Str(d, "onFCPublish"); + writeAmf0Num(d, txId); + writeAmf0Null(d); + writeAmf0ObjStart(d); + writeAmf0Field(d, "code", "NetStream.Publish.Start"); + writeAmf0Field(d, "description", streamKey + " is now published."); + writeAmf0ObjEnd(d); + + sendMessage(CS_COMMAND, MSG_COMMAND_AMF0, 0, 0, buf.toByteArray()); + } + + private void sendCreateStreamResult(double txId, int streamId) throws IOException { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + DataOutputStream d = new DataOutputStream(buf); + + writeAmf0Str(d, "_result"); + writeAmf0Num(d, txId); + writeAmf0Null(d); + writeAmf0Num(d, streamId); + + sendMessage(CS_COMMAND, MSG_COMMAND_AMF0, 0, 0, buf.toByteArray()); + } + + private void sendPublishStart(int streamId, String streamKey) throws IOException { + sendStreamBegin(streamId); + + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + DataOutputStream d = new DataOutputStream(buf); + + writeAmf0Str(d, "onStatus"); + writeAmf0Num(d, 0); + writeAmf0Null(d); + writeAmf0ObjStart(d); + writeAmf0Field(d, "level", "status"); + writeAmf0Field(d, "code", "NetStream.Publish.Start"); + writeAmf0Field(d, "description", streamKey + " is now published."); + writeAmf0Field(d, "details", streamKey); + writeAmf0ObjEnd(d); + + sendMessage(CS_STREAM, MSG_COMMAND_AMF0, streamId, 0, buf.toByteArray()); + } + + // ── Low-level RTMP message write ─────────────────────────────────────── + + /** + * Serialises one RTMP message. Uses fmt=0 for the first chunk (full + * header) and fmt=3 (no header) for any continuation chunks. + * Assumes csId < 64 (1-byte basic header) for all server-sent chunk streams. + */ + private void sendMessage(int csId, int msgType, int msgStreamId, + long timestamp, byte[] payload) throws IOException { + boolean extTs = (timestamp >= 0xFFFFFF); + + // Basic header (fmt=00 | csId) + out.writeByte(csId & 0x3F); + + // 11-byte message header + writeUInt24(out, extTs ? 0xFFFFFF : (int) timestamp); + writeUInt24(out, payload.length); + out.writeByte(msgType); + // Stream ID is little-endian in the RTMP spec + out.writeByte( msgStreamId & 0xFF); + out.writeByte((msgStreamId >> 8) & 0xFF); + out.writeByte((msgStreamId >> 16) & 0xFF); + out.writeByte((msgStreamId >> 24) & 0xFF); + + if (extTs) out.writeInt((int) timestamp); + + // Payload — split into SERVER_CHUNK_SIZE-byte chunks + int offset = 0; + while (offset < payload.length) { + if (offset > 0) out.writeByte(0xC0 | (csId & 0x3F)); // fmt=11 continuation + int n = Math.min(SERVER_CHUNK_SIZE, payload.length - offset); + out.write(payload, offset, n); + offset += n; + } + out.flush(); + } + + // ── User control ─────────────────────────────────────────────────────── + + private void handleUserControl(byte[] data) throws IOException { + if (data.length < 2) return; + int eventType = ((data[0] & 0xFF) << 8) | (data[1] & 0xFF); + if (eventType == EVENT_PING_REQUEST && data.length >= 6) { + long token = ((long)(data[2] & 0xFF) << 24) + | ((long)(data[3] & 0xFF) << 16) + | ((long)(data[4] & 0xFF) << 8) + | (long)(data[5] & 0xFF); + sendPingResponse(token); + } + } + + // ── Acknowledgement ──────────────────────────────────────────────────── + + private void maybeAck() throws IOException { + if (windowAckSize > 0 && (bytesReceived - bytesAcked) >= windowAckSize) { + bytesAcked = bytesReceived; + sendAck(bytesReceived); + } + } + + // ── AMF0 read ────────────────────────────────────────────────────────── + + private Object readAmf0Value(DataInputStream src) throws IOException { + int type = src.readUnsignedByte(); + return switch (type) { + case 0 -> src.readDouble(); + case 1 -> src.readUnsignedByte() != 0; + case 2 -> readAmf0Utf8(src); + case 3 -> readAmf0Object(src); + case 5, 6 -> null; // Null, Undefined + case 8 -> readAmf0EcmaArray(src); + case 9 -> null; // ObjectEnd (context-terminator) + case 10 -> readAmf0StrictArray(src); + case 11 -> { src.skipBytes(10); yield null; } // Date — skip + case 12 -> readAmf0LongString(src); + default -> throw new IOException("Unknown AMF0 type: 0x" + Integer.toHexString(type)); + }; + } + + private String readAmf0Utf8(DataInputStream src) throws IOException { + byte[] b = new byte[src.readUnsignedShort()]; + src.readFully(b); + return new String(b, StandardCharsets.UTF_8); + } + + private String readAmf0LongString(DataInputStream src) throws IOException { + byte[] b = new byte[src.readInt()]; + src.readFully(b); + return new String(b, StandardCharsets.UTF_8); + } + + private Map readAmf0Object(DataInputStream src) throws IOException { + Map map = new LinkedHashMap<>(); + while (true) { + int keyLen = src.readUnsignedShort(); + if (keyLen == 0) { src.readUnsignedByte(); break; } // consume 0x09 end marker + byte[] kb = new byte[keyLen]; + src.readFully(kb); + map.put(new String(kb, StandardCharsets.UTF_8), readAmf0Value(src)); + } + return map; + } + + private Map readAmf0EcmaArray(DataInputStream src) throws IOException { + src.readInt(); // array count — informational only + return readAmf0Object(src); // same layout as Object after the count + } + + private List readAmf0StrictArray(DataInputStream src) throws IOException { + int count = src.readInt(); + List list = new ArrayList<>(count); + for (int i = 0; i < count; i++) list.add(readAmf0Value(src)); + return list; + } + + /** Reads one AMF0 value if bytes remain in the stream; otherwise returns "". */ + private String readOptionalString(DataInputStream src) throws IOException { + if (src.available() <= 0) return ""; + Object v = readAmf0Value(src); + return v instanceof String s ? s : ""; + } + + // ── AMF0 write ───────────────────────────────────────────────────────── + + private static void writeAmf0Str(DataOutputStream d, String s) throws IOException { + byte[] b = s.getBytes(StandardCharsets.UTF_8); + d.writeByte(2); d.writeShort(b.length); d.write(b); + } + + private static void writeAmf0Num(DataOutputStream d, double n) throws IOException { + d.writeByte(0); d.writeDouble(n); + } + + private static void writeAmf0Null(DataOutputStream d) throws IOException { + d.writeByte(5); + } + + private static void writeAmf0Bool(DataOutputStream d, boolean b) throws IOException { + d.writeByte(1); d.writeByte(b ? 1 : 0); + } + + private static void writeAmf0ObjStart(DataOutputStream d) throws IOException { + d.writeByte(3); + } + + private static void writeAmf0ObjEnd(DataOutputStream d) throws IOException { + d.writeShort(0); d.writeByte(9); // empty key + 0x09 end marker + } + + /** Writes one key-value property inside an AMF0 object. */ + private static void writeAmf0Field(DataOutputStream d, String key, Object value) + throws IOException { + byte[] kb = key.getBytes(StandardCharsets.UTF_8); + d.writeShort(kb.length); + d.write(kb); + + if (value == null) { + writeAmf0Null(d); + } else if (value instanceof String s) { + writeAmf0Str(d, s); + } else if (value instanceof Double n) { + writeAmf0Num(d, n); + } else if (value instanceof Boolean b) { + writeAmf0Bool(d, b); + } else { + writeAmf0Null(d); + logger.warn("Unsupported AMF0 value type: {}", value.getClass().getName()); + } + } + + // ── Primitive I/O ────────────────────────────────────────────────────── + + /** Reads one byte and increments the ACK counter. */ + private int readByte() throws IOException { + bytesReceived++; + return in.readUnsignedByte(); + } + + private long readUInt24() throws IOException { + long a = readByte(), b = readByte(), c = readByte(); + return (a << 16) | (b << 8) | c; + } + + private long readUInt32() throws IOException { + long a = readByte(), b = readByte(), c = readByte(), d = readByte(); + return (a << 24) | (b << 16) | (c << 8) | d; + } + + /** RTMP message stream IDs are written little-endian. */ + private int readLittleEndianInt() throws IOException { + int a = readByte(), b = readByte(), c = readByte(), d = readByte(); + return a | (b << 8) | (c << 16) | (d << 24); + } + + private static void writeUInt24(OutputStream out, int v) throws IOException { + out.write((v >> 16) & 0xFF); + out.write((v >> 8) & 0xFF); + out.write( v & 0xFF); + } + + private static void writeUInt32(OutputStream out, int v) throws IOException { + out.write((v >> 24) & 0xFF); + out.write((v >> 16) & 0xFF); + out.write((v >> 8) & 0xFF); + out.write( v & 0xFF); + } + + private static byte[] encodeUInt32(int v) { + return new byte[]{ (byte)(v >> 24), (byte)(v >> 16), (byte)(v >> 8), (byte) v }; + } + + // ── Utilities ────────────────────────────────────────────────────────── + + private static DataInputStream wrapBytes(byte[] data) { + return new DataInputStream(new ByteArrayInputStream(data)); + } + + /** Wraps {@code data[offset..]} — used to skip the AMF3 compatibility byte. */ + private static DataInputStream wrapBytes(byte[] data, int offset) { + return new DataInputStream( + new ByteArrayInputStream(data, offset, data.length - offset)); + } + + private static int parseUInt32(byte[] b) { + return ((b[0] & 0xFF) << 24) | ((b[1] & 0xFF) << 16) + | ((b[2] & 0xFF) << 8) | (b[3] & 0xFF); + } + + private static String getString(Map m, String key) { + Object v = m.get(key); return v instanceof String s ? s : null; + } + + private static double asDouble(Object v) { + return v instanceof Number n ? n.doubleValue() : 0; + } + + @SafeVarargs + private static T coalesce(T... values) { + for (T v : values) if (v != null) return v; + return null; + } + + /** + * Extracts [username, password] from an RTMP URL. + * "rtmp://user:pass@host:1935/app" → ["user", "pass"] + * "rtmp://host:1935/app" → [null, null] + */ + private static String[] extractCredentials(String tcUrl) { + try { + // URI doesn't understand rtmp:// — swap scheme for parsing only + URI uri = new URI(tcUrl.replaceFirst("^rtmps?://", "http://")); + String info = uri.getUserInfo(); + if (info == null) return new String[]{null, null}; + int c = info.indexOf(':'); + return c < 0 + ? new String[]{info, null} + : new String[]{info.substring(0, c), info.substring(c + 1)}; + } catch (Exception e) { + return new String[]{null, null}; + } + } + + // ── Inner types ──────────────────────────────────────────────────────── + + /** Per-chunk-stream reassembly state, persisted across chunk reads. */ + private static final class ChunkStream { + int messageType; + int messageStreamId; + int messageLength; + long timestamp; + long timestampDelta; + boolean hasExtTimestamp; // true → re-read ext ts on fmt=3 continuation + byte[] payload; + int bytesRead; + } + + /** A fully reassembled RTMP message. */ + record RtmpMessage(int type, int streamId, long timestamp, byte[] data) {} +} \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java new file mode 100644 index 000000000..e6aadce15 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java @@ -0,0 +1,60 @@ +package org.sensorhub.impl.sensor.rtmpcam.connection; + +import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +class RtmpPortServer { + + private final int port; + private final RtmpListenerManager manager; + private final ExecutorService connectionExecutor; + + private volatile ServerSocket serverSocket; + private volatile boolean running; + + RtmpPortServer(int port, RtmpListenerManager manager) { + this.port = port; + this.manager = manager; + + // Each accepted connection gets its own daemon thread from a cached pool. + // AtomicInteger gives unique names without synchronisation overhead. + AtomicInteger counter = new AtomicInteger(); + this.connectionExecutor = Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "rtmp-conn-" + port + "-" + counter.incrementAndGet()); + t.setDaemon(true); + return t; + }); + } + + void start() { + running = true; + Thread acceptThread = new Thread(this::acceptLoop, "rtmp-accept-" + port); + acceptThread.setDaemon(true); + acceptThread.start(); + } + + void stop() { + running = false; + try { if (serverSocket != null) serverSocket.close(); } + catch (IOException ignored) {} + connectionExecutor.shutdown(); + } + + private void acceptLoop() { + try { + serverSocket = new ServerSocket(port); + System.out.println("[RTMP] Listening on port " + port); + while (running) { + Socket client = serverSocket.accept(); + connectionExecutor.execute( + () -> new RtmpConnectionHandler(client, port, manager).handle()); + } + } catch (IOException e) { + if (running) System.err.println("[RTMP] Port " + port + " fault: " + e.getMessage()); + } + } +} \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpConnectEvent.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpConnectEvent.java new file mode 100644 index 000000000..b1dae1887 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpConnectEvent.java @@ -0,0 +1,17 @@ +package org.sensorhub.impl.sensor.rtmpcam.event; + +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpConnectionContext; + +public class RtmpConnectEvent implements RtmpEvent { + + private final RtmpConnectionContext payload; + + public RtmpConnectEvent(RtmpConnectionContext payload) { + this.payload = payload; + } + + @Override + public RtmpConnectionContext getPayload() { + return payload; + } +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java new file mode 100644 index 000000000..d42772008 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java @@ -0,0 +1,8 @@ +package org.sensorhub.impl.sensor.rtmpcam.event; + +public class RtmpDisconnectEvent implements RtmpEvent { + @Override + public Object getPayload() { + return null; + } +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEvent.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEvent.java new file mode 100644 index 000000000..3fc3dc6a7 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEvent.java @@ -0,0 +1,5 @@ +package org.sensorhub.impl.sensor.rtmpcam.event; + +public interface RtmpEvent { + public T getPayload(); +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEventCallback.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEventCallback.java new file mode 100644 index 000000000..abefe4d01 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpEventCallback.java @@ -0,0 +1,5 @@ +package org.sensorhub.impl.sensor.rtmpcam.event; + +public interface RtmpEventCallback { + void call(RtmpEvent event); +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpReconnectEvent.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpReconnectEvent.java new file mode 100644 index 000000000..e7dad6bc6 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpReconnectEvent.java @@ -0,0 +1,8 @@ +package org.sensorhub.impl.sensor.rtmpcam.event; + +public class RtmpReconnectEvent implements RtmpEvent { + @Override + public Object getPayload() { + return null; + } +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpStreamEvent.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpStreamEvent.java new file mode 100644 index 000000000..a60cb6bc1 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpStreamEvent.java @@ -0,0 +1,16 @@ +package org.sensorhub.impl.sensor.rtmpcam.event; + +import org.sensorhub.impl.sensor.rtmpcam.stream.StreamInfo; + +public class RtmpStreamEvent implements RtmpEvent { + + private final StreamInfo payload; + + public RtmpStreamEvent(StreamInfo payload) { + this.payload = payload; + } + @Override + public StreamInfo getPayload() { + return payload; + } +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java new file mode 100644 index 000000000..76eae4bf6 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java @@ -0,0 +1,92 @@ +package org.sensorhub.impl.sensor.rtmpcam.helpers; + +import org.sensorhub.impl.sensor.rtmpcam.RtmpListener; + +import java.net.URL; +import java.util.*; + +public class RtmpClientSet implements Set { + private final Map stringAndClient = new HashMap<>(); + private final Map clientAndString = new HashMap<>(); + + @Override + public int size() { + return stringAndClient.size(); + } + + @Override + public boolean isEmpty() { + return stringAndClient.isEmpty(); + } + + @Override + public boolean contains(Object item) { + if (item instanceof RtmpListener client) { + return clientAndString.containsKey(client); + } else if (item instanceof String string) { + return stringAndClient.containsKey(string); + } else if (item instanceof URL url) { + return stringAndClient.containsKey(url.toString()); + } else { + return false; + } + } + + @Override + public Iterator iterator() { + return null; + } + + @Override + public Object[] toArray() { + return null; + } + + @Override + public T[] toArray(T[] a) { + return null; + } + + @Override + public boolean add(RtmpListener rtmpListener) { + String url = rtmpListener.getUrl().toString(); + if (rtmpListener == null || url == null || contains(rtmpListener) || contains(url)) { + return false; + } + stringAndClient.put(url, rtmpListener); + clientAndString.put(rtmpListener, url); + return true; + } + + @Override + public boolean remove(Object item) { + return (item instanceof RtmpListener client) + && (stringAndClient.remove(clientAndString.remove(client)) != null); + } + + @Override + public boolean containsAll(Collection c) { + return false; + } + + @Override + public boolean addAll(Collection c) { + return false; + } + + @Override + public boolean retainAll(Collection c) { + return false; + } + + @Override + public boolean removeAll(Collection c) { + return false; + } + + + @Override + public void clear() { + + } +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java new file mode 100644 index 000000000..298831c89 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java @@ -0,0 +1,49 @@ +package org.sensorhub.impl.sensor.rtmpcam.helpers; + +import java.net.MalformedURLException; +import java.net.URL; + +public class RtmpStringParser { + public static boolean isRtmp(String str) { + return str.startsWith("rtmp"); + } + + public static String getUsername(String str) { + URL url; + try { + url = new URL(str); + } catch (MalformedURLException e) { + return null; + } + + if (url.getUserInfo() == null) { + return null; + } else { + return url.getUserInfo().split(":")[0]; + } + } + + public static String getPassword(String str) { + URL url; + try { + url = new URL(str); + } catch (MalformedURLException e) { + return null; + } + if (url.getUserInfo() == null) { + return null; + } else { + return url.getUserInfo().split(":")[1]; + } + } + + public static String getPath(String str) { + URL url; + try { + url = new URL(str); + } catch (MalformedURLException e) { + return null; + } + return url.getPath(); + } +} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/stream/StreamInfo.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/stream/StreamInfo.java new file mode 100644 index 000000000..e36876e3a --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/stream/StreamInfo.java @@ -0,0 +1,8 @@ +package org.sensorhub.impl.sensor.rtmpcam.stream; + +public record StreamInfo( + int[] videoDimensions, + String videoCodec, + int audioSampleRate, + String audioCodec +) {} diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider b/sensors/video/sensorhub-driver-rtmp/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider index e9ee1b390..b06642b44 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider +++ b/sensors/video/sensorhub-driver-rtmp/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider @@ -1 +1 @@ -org.sensorhub.impl.sensor.rtmp.RtmpDescriptor \ No newline at end of file +org.sensorhub.impl.sensor.rtmpcam.RtmpDescriptor \ No newline at end of file From cd7acd8f700e6082efcee7c35ed2d20bfe74590f Mon Sep 17 00:00:00 2001 From: kyle-fitzp Date: Wed, 1 Jul 2026 12:22:31 -0500 Subject: [PATCH 2/4] Additional RTMP changes --- .../impl/sensor/rtmpcam/RtmpDriver.java | 428 ++---------------- .../rtmpcam/config/ConnectionConfig.java | 3 - .../connection/RtmpConnectionHandler.java | 95 +++- .../rtmpcam/connection/RtmpListener.java | 20 +- .../rtmpcam/connection/RtmpNegotiator.java | 2 +- .../rtmpcam/event/RtmpDisconnectEvent.java | 14 +- .../sensor/rtmpcam/helpers/RtmpClientSet.java | 92 ---- 7 files changed, 145 insertions(+), 509 deletions(-) delete mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java index 9d055997f..b87beb309 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java @@ -13,18 +13,18 @@ package org.sensorhub.impl.sensor.rtmpcam; import org.sensorhub.api.common.SensorHubException; -import org.sensorhub.api.module.ModuleEvent; -import org.sensorhub.api.sensor.SensorException; import org.sensorhub.impl.sensor.AbstractSensorModule; import org.sensorhub.impl.sensor.ffmpeg.outputs.AudioOutput; import org.sensorhub.impl.sensor.ffmpeg.outputs.VideoOutput; import org.sensorhub.impl.sensor.rtmpcam.config.ConnectionConfig; import org.sensorhub.impl.sensor.rtmpcam.config.RtmpConfig; -import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpConnectionContext; import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpListener; import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpListenerManager; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpConnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpDisconnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpReconnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpStreamEvent; import org.sensorhub.mpegts.MpegTsProcessor; -import org.sensorhub.utils.Async; import java.security.SecureRandom; import java.util.HexFormat; @@ -54,12 +54,15 @@ public class RtmpDriver extends AbstractSensorModule implements Rtmp private static final TimeUnit HEARTBEAT_TIME_UNIT = TimeUnit.SECONDS; private static final int MAX_STARTUP_WAIT_TIME_MS = 5000; - private final RtmpListenerManager portSingleton = RtmpListenerManager.getInstance(); + private volatile boolean doStreamProcessing = false; + private final AtomicReference mpegTsProcessor = new AtomicReference<>(); + private final RtmpListenerManager rtmpListenerManager = RtmpListenerManager.getInstance(); private ExecutorService executorService; private ExecutorService videoExecutorService; private ExecutorService audioExecutorService; private ScheduledExecutorService heartbeatExecutorService; + // TODO: Create a DataOutput final AtomicReference> videoOutput = new AtomicReference<>(); final AtomicReference> audioOutput = new AtomicReference<>(); @@ -97,13 +100,14 @@ protected void doInit() throws SensorHubException { generateXmlID("RTMP_", config.serialNumber); } + rtmpListenerManager.removeListener(this); + } - - portSingleton.removeConnection(connectionPort); - - setConnectionUrl(); - - //createMpegTsProcessor(); + @Override + protected void doStart() throws SensorHubException { + super.doStart(); + rtmpListenerManager.addListener(this); + doStreamProcessing = true; } private static String generateStreamKey() { @@ -112,303 +116,10 @@ private static String generateStreamKey() { return HexFormat.of().formatHex(bytes); } - /** - * Creates and stores the FFmpeg-backed MPEG-TS processor for the configured - * RTMP listener URL. - */ - private void createMpegTsProcessor() { - var mpegts = new MpegTsProcessor(connectionUrl, COMMAND_LINE_ARGS/* + " -rtmp_app /live -rtmp_playpath " + path*/); - mpegts.setInjectVideoExtradata(true); - mpegTsProcessor.set(mpegts); - } - - /** - * Builds the RTMP listener URL from the configured host and port. - * - * @throws SensorException if the connection configuration is invalid - */ - private void setConnectionUrl() throws SensorException { - - var connectionConfig = config.connectionConfig; - StringBuilder sb = new StringBuilder("rtmp://"); - - - /* - if (connectionConfig.host == HostType.OVERRIDE) { - if (connectionConfig.hostOverride == null || connectionConfig.hostOverride.isBlank()) { - throw new SensorException("Domain override is not set"); - } - sb.append(connectionConfig.hostOverride); - } else { - sb.append(connectionConfig.host.host); - } - - */ - - sb.append(connectionConfig.host.host); - - sb.append(":").append(connectionConfig.port); - - /* - if (connectionConfig.generateRandomStreamKey) { - connectionConfig.generateRandomStreamKey = false; - String streamKey = generateStreamKey(); - if (!connectionConfig.path.isBlank() && !connectionConfig.path.endsWith("/")) { - connectionConfig.path += "/"; - } - connectionConfig.path += streamKey; - } - - if (connectionConfig.path != null && !connectionConfig.path.isBlank()) { - if (!connectionConfig.path.startsWith("/")) { - connectionConfig.path = "/" + connectionConfig.path; - } - sb.append(connectionConfig.path); - } - - path = connectionConfig.path; - - */ - connectionUrl = sb.toString(); - connectionPort = connectionConfig.port; - } - - /** - * Starts the driver by reserving the configured RTMP port. - * - * @throws SensorHubException if the configured port is already in use by - * another RTMP driver module - */ - @Override - protected void doStart() throws SensorHubException { - String moduleUid; - if ((moduleUid = portSingleton.addConnection(connectionPort, this.getUniqueIdentifier())) != null) { - throw new SensorException("Port "+ connectionPort + " already in use by module: " + moduleUid); - } - } - - /** - * Performs post-start setup for RTMP listening and heartbeat monitoring. - *

- * This method creates a fresh MPEG-TS processor, stops any previous executor - * services, reports the listening URL, starts the stream listener thread, and - * schedules periodic heartbeat checks. - *

- * - * @throws SensorHubException if executor shutdown is interrupted - */ - @Override - protected void afterStart() throws SensorHubException { - super.afterStart(); - //stopStream(); - hasConnected = false; - - synchronized (mpegTsProcessor) { - var mpegts = mpegTsProcessor.get(); - if (mpegts != null && mpegts.getState() != Thread.State.NEW) { - stopStream(); - } - createMpegTsProcessor(); - } - - try { - stopExecutors(); - } catch (InterruptedException e) { - throw new SensorHubException("Interrupted while stopping executors", e); - } - - reportStatus("Listening on: " + connectionUrl); - executorService = Executors.newSingleThreadExecutor(); - executorService.submit(this::startStream); - heartbeatExecutorService = Executors.newSingleThreadScheduledExecutor(); - heartbeatExecutorService.scheduleAtFixedRate(this::heartbeat, HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL, HEARTBEAT_TIME_UNIT); - heartbeatExecutorService.submit(this::heartbeat); + public boolean doStreamProcessing() { + return doStreamProcessing; } - /** - * Determines whether the module is currently stopping or stopped. - * - * @return {@code true} if the module is stopping or stopped; otherwise {@code false} - */ - private boolean isStopping () { - return getCurrentState() == ModuleEvent.ModuleState.STOPPING || getCurrentState() == ModuleEvent.ModuleState.STOPPED; - } - - /* - private boolean isMatchingPath(String url) { - boolean isMatching = url != null && url.trim().contains(path); - if (!isMatching) { - logger.warn("Received stream on: {} but expected: {}", url, config.connectionConfig.path); - } - return isMatching; - } - - */ - - /** - * Opens the RTMP listener, waits for an incoming stream, creates audio/video - * outputs for detected streams, and begins processing stream data. - *

- * The method waits until the module reaches the {@code STARTED} state before - * adding outputs. If the stream cannot be opened, an error is reported and - * processing is not started. - *

- */ - private void startStream() { - // Need to wait for STARTED state so that outputs can be added - try { - Async.waitForCondition(() -> getCurrentState() == ModuleEvent.ModuleState.STARTED, MAX_STARTUP_WAIT_TIME_MS); - } catch (TimeoutException e) { - reportError("Failed to start stream; timed out waiting for startup", e); - return; - } - - boolean status; - - var mpegts = mpegTsProcessor.get(); - - if (mpegts == null) { - logger.error("Could not start; stream processor is null"); - return; - } - - /* - // Reject connections that don't match the configured path - // Thread will sit here until a matching connection is made - do { - if (Thread.currentThread().isInterrupted() || isStopping()) { - return; - } - mpegts.closeStream(); - status = mpegts.openStream(); - String path = mpegts.getPrivDataString("rtmp_app") + "/" + mpegts.getPrivDataString("rtmp_playpath"); - } while (!isMatchingPath(path)); - - */ - - mpegts.closeStream(); - status = mpegts.openStream(); - - if (isStopping()) { - return; - } - - if (!status) { - String error = "Failed to connect to " + connectionUrl; - reportError(error, new SensorException(error)); - return; - } - - synchronized (mpegTsProcessor) { - mpegts = mpegTsProcessor.get(); - - if (mpegts == null) { - reportError("Stream could not be opened", new SensorException("MpegTs processor is null")); - return; - } - - if (mpegts.isStreamOpened()) { - if (mpegts.hasVideoStream()) { - createVideoOutput(mpegts.getVideoStreamFrameDimensions(), mpegts.getVideoCodecName()); - mpegts.setVideoDataBufferListener(videoOutput.get()); - } - - if (mpegts.hasAudioStream()) { - createAudioOutput(mpegts.getAudioSampleRate(), mpegts.getAudioCodecName()); - mpegts.setAudioDataBufferListener(audioOutput.get()); - } - - } else { - reportError("Stream could not be opened", new SensorException("RTMP stream connected but not opened")); - return; - } - clearStatus(); - reportStatus("RTMP stream for " + connectionUrl + " opened."); - hasConnected = true; - isConnected = true; - mpegts.processStream(); - /* - executorService.submit(() -> { - MpegTsProcessor processor; - while ((processor = mpegTsProcessor.get()) != null) { - - while (processor.isStreamOpened()) { - processor.processP(); - } - if (!Thread.currentThread().isInterrupted()) { - reportStatus("RTMP stream " + connectionUrl + " lost connection. Reconnecting..."); - processor.openStream(); - } else { - return; - } - } - reportStatus("RTMP stream closed."); - - }); - - */ - //mpegts.processStream(); - } - } - - /** - * Checks the active stream and attempts to reconnect if a previously - * connected stream has been lost. - */ - private void heartbeat() { - var mpegts = mpegTsProcessor.get(); - if (mpegts == null || !hasConnected) { return; } - - if (!mpegts.isStreamOpened()) { - reportStatus("RTMP stream " + connectionUrl + " lost connection. Reconnecting..."); - isConnected = false; - createMpegTsProcessor(); - startStream(); - } - } - - /** - * Creates and registers the video output for the detected RTMP video stream. - * - * @param videoDims video frame dimensions, usually width and height - * @param codecName name of the detected video codec - */ - protected void createVideoOutput(int[] videoDims, String codecName) { - synchronized (videoOutput) { - var videoOut = new VideoOutput<>(this, videoDims, codecName); - videoOutput.set(videoOut); - - if (videoExecutorService != null) { - videoExecutorService.shutdown(); - } - - videoExecutorService = Executors.newSingleThreadExecutor(); - videoOut.setExecutor(videoExecutorService); - videoOut.doInit(); - addOutput(videoOut, false); - } - } - - /** - * Creates and registers the audio output for the detected RTMP audio stream. - * - * @param sampleRate detected audio sample rate in hertz - * @param codecName name of the detected audio codec - */ - protected void createAudioOutput(int sampleRate, String codecName) { - synchronized (audioOutput) { - var audioOut = new AudioOutput<>(this, sampleRate, codecName); - audioOutput.set(audioOut); - - if (audioExecutorService != null) { - audioExecutorService.shutdown(); - } - audioExecutorService = Executors.newSingleThreadExecutor(); - audioOut.setExecutor(audioExecutorService); - audioOut.doInit(); - addOutput(audioOut, false); - } - } /** * Stops the driver and releases all RTMP stream resources. @@ -418,82 +129,8 @@ protected void createAudioOutput(int sampleRate, String codecName) { @Override protected void doStop() throws SensorHubException { super.doStop(); - shutdown(); - } - - /** - * Releases the reserved port, stops the stream, and terminates executor services. - * - * @throws SensorHubException if executor shutdown is interrupted - */ - private void shutdown() throws SensorHubException { - portSingleton.removeConnection(config.connectionConfig.port); - stopStream(); - try { - stopExecutors(); - } catch (InterruptedException e) { - throw new SensorHubException("Interrupted while stopping executors", e); - } - - } - - /** - * Stops stream processing, waits for the MPEG-TS processor thread to finish, - * closes the stream, and clears the processor reference. - */ - private void stopStream() { - isConnected = false; - synchronized (mpegTsProcessor) { - var mpegts = mpegTsProcessor.get(); - if (mpegts != null) { - mpegts.stopProcessingStream(); - if (mpegts.isAlive()) { - try { - logger.info("Waiting for stream to stop."); - mpegts.join(); - } catch (InterruptedException e) { - logger.error("Interrupted while waiting for stream to stop.", e); - } - } - mpegts.closeStream(); - } - mpegTsProcessor.set(null); - } - } - - /** - * Stops all executor services used by the driver and waits for termination. - * - * @throws InterruptedException if interrupted while waiting for executor termination - */ - private void stopExecutors() throws InterruptedException { - if (executorService != null) { - executorService.shutdownNow(); - executorService.awaitTermination(EXECUTOR_JOIN_TIMEOUT, EXECUTOR_JOIN_TIME_UNIT); - } - if (videoExecutorService != null) { - videoExecutorService.shutdownNow(); - videoExecutorService.awaitTermination(EXECUTOR_JOIN_TIMEOUT, EXECUTOR_JOIN_TIME_UNIT); - } - if (audioExecutorService != null) { - audioExecutorService.shutdownNow(); - audioExecutorService.awaitTermination(EXECUTOR_JOIN_TIMEOUT, EXECUTOR_JOIN_TIME_UNIT); - } - if (heartbeatExecutorService != null) { - heartbeatExecutorService.shutdownNow(); - heartbeatExecutorService.awaitTermination(EXECUTOR_JOIN_TIMEOUT, EXECUTOR_JOIN_TIME_UNIT); - } - } - - /** - * Cleans up module resources before disposal. - * - * @throws SensorHubException if cleanup or shutdown fails - */ - @Override - public void cleanup() throws SensorHubException { - super.cleanup(); - shutdown(); + doStreamProcessing = false; + rtmpListenerManager.removeListener(this); } /** @@ -513,21 +150,34 @@ public ConnectionConfig config() { } @Override - public void publish(byte[] data, int streamIndex, long pts, boolean isVideo) { - if (isVideo) { - videoOutput.get(). - } + public void onConnected(RtmpConnectEvent event) { + reportStatus("Connected to: " + connectionUrl); } @Override - public void onStreamConnected(RtmpConnectionContext ctx) { + public void onStreamConnected(RtmpStreamEvent event) { + addOutput(videoOutput.get(), false); + addOutput(audioOutput.get(), false); + } + @Override + public void onDisconnected(RtmpDisconnectEvent event) { + removeAllOutputs(); + reportStatus("Disconnected from: " + connectionUrl); } - public void onStreamInfo() + @Override + public void onReconnected(RtmpReconnectEvent event) { + + } @Override - public void onStreamDisconnected() { + public VideoOutput getVideoOutput() { + return this.videoOutput.get(); + } + @Override + public AudioOutput getAudioOutput() { + return this.audioOutput.get(); } } \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java index 86cda0855..2e3347401 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java @@ -24,11 +24,9 @@ public class ConnectionConfig { */ - @DisplayInfo.Required @DisplayInfo(label = "Username") public String username = ""; - @DisplayInfo.Required @DisplayInfo(label = "Password") public String password = ""; @@ -37,7 +35,6 @@ public class ConnectionConfig { @DisplayInfo.ValueRange(min = 1, max = 65535) public int port = 1935; - @DisplayInfo.Required @DisplayInfo(label = "Path") public String path = ""; diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java index 34ae623f3..756d24ad3 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java @@ -3,13 +3,23 @@ import org.bytedeco.ffmpeg.avcodec.AVPacket; import org.bytedeco.ffmpeg.avformat.*; import org.bytedeco.ffmpeg.avutil.AVDictionary; +import org.bytedeco.ffmpeg.avutil.AVRational; +import org.bytedeco.ffmpeg.global.avutil; import org.bytedeco.javacpp.BytePointer; import org.bytedeco.javacpp.Pointer; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpConnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpDisconnectEvent; +import org.sensorhub.impl.sensor.rtmpcam.event.RtmpStreamEvent; +import org.sensorhub.mpegts.DataBufferListener; +import org.sensorhub.mpegts.StreamContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nonnull; import java.io.*; import java.net.Socket; +import java.util.HashMap; +import java.util.Map; import java.util.Optional; import static org.bytedeco.ffmpeg.global.avcodec.*; @@ -33,6 +43,12 @@ class RtmpConnectionHandler { private final int port; private final RtmpListenerManager manager; + private final StreamContext videoStreamContext = new StreamContext(); + private final StreamContext audioStreamContext = new StreamContext(); + private final StreamContext dataStreamContext = new StreamContext(); + + private final Map streamContextMap = new HashMap<>(); + RtmpConnectionHandler(Socket socket, int port, RtmpListenerManager manager) { this.socket = socket; this.port = port; @@ -59,11 +75,13 @@ void handle() { } RtmpListener listener = match.get(); - listener.onConnected(ctx); + RtmpConnectEvent connectEvent = new RtmpConnectEvent(ctx); + listener.onConnected(connectEvent); + try { pipeToFfmpeg(negotiator.buildFlvStream(), listener); } finally { - listener.onDisconnected(); + listener.onDisconnected(new RtmpDisconnectEvent(ctx)); } } catch (Exception e) { @@ -96,6 +114,9 @@ private void pipeToFfmpeg(InputStream flvStream, RtmpListener listener) { if (ret < 0) { logError("avformat_open_input", ret); freeAVIO(avioCtx); return; } avformat_find_stream_info(fmtCtx, (AVDictionary) null); + + streamContextSetup(fmtCtx, listener); + packetLoop(fmtCtx, listener); avformat_close_input(fmtCtx); @@ -103,18 +124,72 @@ private void pipeToFfmpeg(InputStream flvStream, RtmpListener listener) { // readCb and avioBuf are now safe to collect } + private void queryEmbeddedStreams(AVFormatContext avFormatContext) { + streamContextMap.clear(); + + for (int streamId = 0; streamId < avFormatContext.nb_streams(); ++streamId) { + int codecType = avFormatContext.streams(streamId).codecpar().codec_type(); + + AVRational timeBase = avFormatContext.streams(streamId).time_base(); + double timeBaseUnits = (double) timeBase.num() / timeBase.den(); + + if (!videoStreamContext.hasStream() && codecType == avutil.AVMEDIA_TYPE_VIDEO) { + logger.debug("Video stream present with id: {}", streamId); + + videoStreamContext.setStreamId(streamId); + videoStreamContext.setStreamTimeBase(timeBaseUnits); + streamContextMap.put(streamId, videoStreamContext); + } else if (!audioStreamContext.hasStream() && codecType == avutil.AVMEDIA_TYPE_AUDIO) { + logger.debug("Audio stream present with id: {}", streamId); + + audioStreamContext.setStreamId(streamId); + audioStreamContext.setStreamTimeBase(timeBaseUnits); + streamContextMap.put(streamId, audioStreamContext); + } else if (!dataStreamContext.hasStream() && codecType == avutil.AVMEDIA_TYPE_DATA) { + logger.debug("Data stream present with id: {}", streamId); + + dataStreamContext.setStreamId(streamId); + dataStreamContext.setStreamTimeBase(timeBaseUnits); + streamContextMap.put(streamId, dataStreamContext); + } + } + } + + private void streamContextSetup(AVFormatContext avFormatContext, RtmpListener listener) { + queryEmbeddedStreams(avFormatContext); + + videoStreamContext.openCodecContext(avFormatContext); + audioStreamContext.openCodecContext(avFormatContext); + dataStreamContext.openCodecContext(avFormatContext); + + videoStreamContext.setDataBufferListener(listener.getVideoOutput()); + audioStreamContext.setDataBufferListener(listener.getAudioOutput()); + + // TODO Probably don't need the parameters added to the RtmpStreamEvent. Remove + RtmpStreamEvent streamEvent = new RtmpStreamEvent(null); + listener.onStreamConnected(streamEvent); + } + + public void setVideoBufferListener(@Nonnull DataBufferListener videoDataBufferListener) { + videoStreamContext.setDataBufferListener(videoDataBufferListener); + } + + public void setAudioBufferListener(@Nonnull DataBufferListener audioDataBufferListener) { + audioStreamContext.setDataBufferListener(audioDataBufferListener); + } + + public void setDataBufferListener(@Nonnull DataBufferListener dataBufferListener) { + dataStreamContext.setDataBufferListener(dataBufferListener); + } + private void packetLoop(AVFormatContext fmtCtx, RtmpListener listener) { AVPacket pkt = av_packet_alloc(); try { int ret; - while ((ret = av_read_frame(fmtCtx, pkt)) >= 0) { - AVStream stream = fmtCtx.streams(pkt.stream_index()); - int mediaType = stream.codecpar().codec_type(); - if (mediaType == AVMEDIA_TYPE_VIDEO || mediaType == AVMEDIA_TYPE_AUDIO) { - byte[] data = new byte[pkt.size()]; - pkt.data().get(data); - listener.publish(data, pkt.stream_index(), pkt.pts(), - mediaType == AVMEDIA_TYPE_VIDEO); + while ((ret = av_read_frame(fmtCtx, pkt)) >= 0 && listener.doStreamProcessing()) { + StreamContext streamContext = streamContextMap.get(pkt.stream_index()); + if (streamContext != null) { + streamContext.processPacket(pkt); } av_packet_unref(pkt); } diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java index 6bf2de14a..691838b5b 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java @@ -1,5 +1,7 @@ package org.sensorhub.impl.sensor.rtmpcam.connection; +import org.sensorhub.impl.sensor.ffmpeg.outputs.AudioOutput; +import org.sensorhub.impl.sensor.ffmpeg.outputs.VideoOutput; import org.sensorhub.impl.sensor.rtmpcam.config.ConnectionConfig; import org.sensorhub.impl.sensor.rtmpcam.event.RtmpConnectEvent; import org.sensorhub.impl.sensor.rtmpcam.event.RtmpDisconnectEvent; @@ -14,23 +16,19 @@ public interface RtmpListener { public ConnectionConfig config(); - /** - * Receives one encoded packet demuxed from the RTMP stream. - * - * @param data raw encoded bytes (H.264 Annex B, AAC, etc.) - * @param streamIndex FFmpeg stream index within the FLV container - * @param pts presentation timestamp in stream timebase units - * @param isVideo true for video, false for audio - */ - public void publish(byte[] data, int streamIndex, long pts, boolean isVideo); - /** Called once after negotiation succeeds and this listener is selected. */ public void onConnected(RtmpConnectEvent event); + public void onStreamConnected(RtmpStreamEvent event); + /** Called once when the client disconnects or the pipeline faults. */ public void onDisconnected(RtmpDisconnectEvent event); public void onReconnected(RtmpReconnectEvent event); - public void onStreamInfo(RtmpStreamEvent event); + public VideoOutput getVideoOutput(); + + public AudioOutput getAudioOutput(); + + public boolean doStreamProcessing(); } \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java index b2a7b9469..f4e6af094 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java @@ -183,7 +183,7 @@ public RtmpConnectionContext negotiate() throws IOException { sendConnectResult(txId); } - case "releaseStream", "getStreamLength", "FCSubscribe" -> { /* no-op */ } + case "releaseStream", "getStreamLength", "FCSubscribe", "_checkbw" -> { /* no-op */ } case "FCPublish" -> { double txId = asDouble(readAmf0Value(amf)); // txId diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java index d42772008..83ffcb1aa 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java @@ -1,8 +1,16 @@ package org.sensorhub.impl.sensor.rtmpcam.event; -public class RtmpDisconnectEvent implements RtmpEvent { +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpConnectionContext; + +public class RtmpDisconnectEvent implements RtmpEvent { + RtmpConnectionContext payload; + + public RtmpDisconnectEvent(RtmpConnectionContext payload) { + this.payload = payload; + } + @Override - public Object getPayload() { - return null; + public RtmpConnectionContext getPayload() { + return payload; } } diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java deleted file mode 100644 index 76eae4bf6..000000000 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpClientSet.java +++ /dev/null @@ -1,92 +0,0 @@ -package org.sensorhub.impl.sensor.rtmpcam.helpers; - -import org.sensorhub.impl.sensor.rtmpcam.RtmpListener; - -import java.net.URL; -import java.util.*; - -public class RtmpClientSet implements Set { - private final Map stringAndClient = new HashMap<>(); - private final Map clientAndString = new HashMap<>(); - - @Override - public int size() { - return stringAndClient.size(); - } - - @Override - public boolean isEmpty() { - return stringAndClient.isEmpty(); - } - - @Override - public boolean contains(Object item) { - if (item instanceof RtmpListener client) { - return clientAndString.containsKey(client); - } else if (item instanceof String string) { - return stringAndClient.containsKey(string); - } else if (item instanceof URL url) { - return stringAndClient.containsKey(url.toString()); - } else { - return false; - } - } - - @Override - public Iterator iterator() { - return null; - } - - @Override - public Object[] toArray() { - return null; - } - - @Override - public T[] toArray(T[] a) { - return null; - } - - @Override - public boolean add(RtmpListener rtmpListener) { - String url = rtmpListener.getUrl().toString(); - if (rtmpListener == null || url == null || contains(rtmpListener) || contains(url)) { - return false; - } - stringAndClient.put(url, rtmpListener); - clientAndString.put(rtmpListener, url); - return true; - } - - @Override - public boolean remove(Object item) { - return (item instanceof RtmpListener client) - && (stringAndClient.remove(clientAndString.remove(client)) != null); - } - - @Override - public boolean containsAll(Collection c) { - return false; - } - - @Override - public boolean addAll(Collection c) { - return false; - } - - @Override - public boolean retainAll(Collection c) { - return false; - } - - @Override - public boolean removeAll(Collection c) { - return false; - } - - - @Override - public void clear() { - - } -} From 8ea61a490956ac8dfc22a04573fe12b2b3eb08c6 Mon Sep 17 00:00:00 2001 From: kyle-fitzp Date: Thu, 2 Jul 2026 11:44:50 -0500 Subject: [PATCH 3/4] Changes to routing, memory allocation. Other fixes. --- .../impl/sensor/rtmpcam/RtmpDriver.java | 31 +++++++-- .../rtmpcam/config/ConnectionConfig.java | 3 + .../connection/RtmpConnectionHandler.java | 67 ++++++++++++++----- .../rtmpcam/connection/RtmpListener.java | 4 ++ .../connection/RtmpListenerManager.java | 38 +---------- 5 files changed, 87 insertions(+), 56 deletions(-) diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java index b87beb309..afe45b090 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java @@ -58,8 +58,8 @@ public class RtmpDriver extends AbstractSensorModule implements Rtmp private final AtomicReference mpegTsProcessor = new AtomicReference<>(); private final RtmpListenerManager rtmpListenerManager = RtmpListenerManager.getInstance(); private ExecutorService executorService; - private ExecutorService videoExecutorService; - private ExecutorService audioExecutorService; + private ExecutorService videoExecutorService = Executors.newSingleThreadExecutor(); + private ExecutorService audioExecutorService = Executors.newSingleThreadExecutor(); private ScheduledExecutorService heartbeatExecutorService; // TODO: Create a DataOutput @@ -99,6 +99,10 @@ protected void doInit() throws SensorHubException { generateUniqueID("urn:osh:sensor:rtmp:", config.serialNumber); generateXmlID("RTMP_", config.serialNumber); } + if (config.connectionConfig.generateRandomKey) { + config.connectionConfig.streamKey = generateStreamKey(); + config.connectionConfig.generateRandomKey = false; + } rtmpListenerManager.removeListener(this); } @@ -156,8 +160,27 @@ public void onConnected(RtmpConnectEvent event) { @Override public void onStreamConnected(RtmpStreamEvent event) { - addOutput(videoOutput.get(), false); - addOutput(audioOutput.get(), false); + var streamInfo = event.getPayload(); + + if (streamInfo == null) { + logger.warn("StreamInfo is null"); + return; + } + if (streamInfo.videoCodec() != null) { + videoOutput.set(new VideoOutput<>(this, streamInfo.videoDimensions(), streamInfo.videoCodec())); + videoExecutorService = Executors.newSingleThreadExecutor(); + videoOutput.get().setExecutor(videoExecutorService); + videoOutput.get().doInit(); + addOutput(videoOutput.get(), false); + + } + if (streamInfo.audioCodec() != null) { + audioOutput.set(new AudioOutput<>(this, streamInfo.audioSampleRate(), streamInfo.audioCodec())); + audioExecutorService = Executors.newSingleThreadExecutor(); + audioOutput.get().setExecutor(audioExecutorService); + audioOutput.get().doInit(); + addOutput(audioOutput.get(), false); + } } @Override diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java index 2e3347401..fa7fe3671 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/config/ConnectionConfig.java @@ -13,6 +13,7 @@ package org.sensorhub.impl.sensor.rtmpcam.config; import org.sensorhub.api.config.DisplayInfo; +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpConnectionContext; public class ConnectionConfig { @@ -64,4 +65,6 @@ public static String compositeKey(String username, String password, int port, St (path != null ? path : "") + ":" + (streamKey != null ? streamKey : ""); } + + public static String compositeKey(RtmpConnectionContext ctx) { return compositeKey(ctx.username(), ctx.password(), ctx.port(), ctx.path(), ctx.streamKey()); } } \ No newline at end of file diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java index 756d24ad3..67e542b86 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java @@ -1,5 +1,6 @@ package org.sensorhub.impl.sensor.rtmpcam.connection; +import org.bytedeco.ffmpeg.avcodec.AVCodec; import org.bytedeco.ffmpeg.avcodec.AVPacket; import org.bytedeco.ffmpeg.avformat.*; import org.bytedeco.ffmpeg.avutil.AVDictionary; @@ -10,6 +11,7 @@ import org.sensorhub.impl.sensor.rtmpcam.event.RtmpConnectEvent; import org.sensorhub.impl.sensor.rtmpcam.event.RtmpDisconnectEvent; import org.sensorhub.impl.sensor.rtmpcam.event.RtmpStreamEvent; +import org.sensorhub.impl.sensor.rtmpcam.stream.StreamInfo; import org.sensorhub.mpegts.DataBufferListener; import org.sensorhub.mpegts.StreamContext; import org.slf4j.Logger; @@ -21,6 +23,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Optional; +import java.util.stream.Stream; import static org.bytedeco.ffmpeg.global.avcodec.*; import static org.bytedeco.ffmpeg.global.avformat.*; @@ -85,7 +88,7 @@ void handle() { } } catch (Exception e) { - System.err.printf("[RTMP:%d] connection fault: %s%n", port, e.getMessage()); + logger.error("[RTMP:{}] Error handling RTMP connection", port, e); } } @@ -97,7 +100,7 @@ private void pipeToFfmpeg(InputStream flvStream, RtmpListener listener) { // readCb — stored as a raw native function pointer inside AVIOContext // avioBuf — FFmpeg takes ownership; free via ctx.buffer(), not this reference Read_packet_Pointer_BytePointer_int readCb = buildReadCb(flvStream); - BytePointer avioBuf = new BytePointer(av_malloc(AVIO_BUF)).capacity(AVIO_BUF); + BytePointer avioBuf = new BytePointer(av_malloc(AVIO_BUF)).capacity(AVIO_BUF); AVIOContext avioCtx = avio_alloc_context( avioBuf, AVIO_BUF, @@ -113,9 +116,12 @@ private void pipeToFfmpeg(InputStream flvStream, RtmpListener listener) { if (ret < 0) { logError("avformat_open_input", ret); freeAVIO(avioCtx); return; } + logger.debug("Here 1"); avformat_find_stream_info(fmtCtx, (AVDictionary) null); + logger.debug("Here 2"); streamContextSetup(fmtCtx, listener); + logger.debug("Here 3"); packetLoop(fmtCtx, listener); @@ -124,28 +130,57 @@ private void pipeToFfmpeg(InputStream flvStream, RtmpListener listener) { // readCb and avioBuf are now safe to collect } - private void queryEmbeddedStreams(AVFormatContext avFormatContext) { + private StreamInfo queryEmbeddedStreams(AVFormatContext avFormatContext) { streamContextMap.clear(); + int[] videoDimensions = new int[2]; + String videoCodec = null; + int audioSampleRate = 0; + String audioCodec = null; + for (int streamId = 0; streamId < avFormatContext.nb_streams(); ++streamId) { - int codecType = avFormatContext.streams(streamId).codecpar().codec_type(); + var stream = avFormatContext.streams(streamId); + var codecpar = stream.codecpar(); + int codecType = codecpar.codec_type(); AVRational timeBase = avFormatContext.streams(streamId).time_base(); double timeBaseUnits = (double) timeBase.num() / timeBase.den(); - if (!videoStreamContext.hasStream() && codecType == avutil.AVMEDIA_TYPE_VIDEO) { + if (!videoStreamContext.hasStream() && codecType == AVMEDIA_TYPE_VIDEO) { logger.debug("Video stream present with id: {}", streamId); + try (AVCodec avCodec = avcodec_find_decoder(codecpar.codec_id())) { + if (avCodec == null) { + logger.error("Unsupported codec: {}", codecpar.codec_id()); + continue; + } else { + videoCodec = avCodec.name().getString(); + } + } + + videoDimensions[0] = codecpar.width(); + videoDimensions[1] = codecpar.height(); + videoStreamContext.setStreamId(streamId); videoStreamContext.setStreamTimeBase(timeBaseUnits); streamContextMap.put(streamId, videoStreamContext); - } else if (!audioStreamContext.hasStream() && codecType == avutil.AVMEDIA_TYPE_AUDIO) { + } else if (!audioStreamContext.hasStream() && codecType == AVMEDIA_TYPE_AUDIO) { logger.debug("Audio stream present with id: {}", streamId); + try (AVCodec avCodec = avcodec_find_decoder(codecpar.codec_id())) { + if (avCodec == null) { + logger.error("Unsupported codec: {}", codecpar.codec_id()); + continue; + } else { + audioCodec = avCodec.name().getString(); + } + } + audioSampleRate = codecpar.sample_rate(); + audioStreamContext.setStreamId(streamId); audioStreamContext.setStreamTimeBase(timeBaseUnits); streamContextMap.put(streamId, audioStreamContext); - } else if (!dataStreamContext.hasStream() && codecType == avutil.AVMEDIA_TYPE_DATA) { + } else if (!dataStreamContext.hasStream() && codecType == AVMEDIA_TYPE_DATA) { logger.debug("Data stream present with id: {}", streamId); dataStreamContext.setStreamId(streamId); @@ -153,21 +188,25 @@ private void queryEmbeddedStreams(AVFormatContext avFormatContext) { streamContextMap.put(streamId, dataStreamContext); } } + + return new StreamInfo(videoDimensions, videoCodec, audioSampleRate, audioCodec); } private void streamContextSetup(AVFormatContext avFormatContext, RtmpListener listener) { - queryEmbeddedStreams(avFormatContext); + var streamInfo = queryEmbeddedStreams(avFormatContext); + videoStreamContext.setInjectingExtradata(true); videoStreamContext.openCodecContext(avFormatContext); audioStreamContext.openCodecContext(avFormatContext); dataStreamContext.openCodecContext(avFormatContext); - videoStreamContext.setDataBufferListener(listener.getVideoOutput()); - audioStreamContext.setDataBufferListener(listener.getAudioOutput()); - - // TODO Probably don't need the parameters added to the RtmpStreamEvent. Remove - RtmpStreamEvent streamEvent = new RtmpStreamEvent(null); + RtmpStreamEvent streamEvent = new RtmpStreamEvent(streamInfo); listener.onStreamConnected(streamEvent); + + if (listener.getVideoOutput() != null) + videoStreamContext.setDataBufferListener(listener.getVideoOutput()); + if (listener.getAudioOutput() != null) + audioStreamContext.setDataBufferListener(listener.getAudioOutput()); } public void setVideoBufferListener(@Nonnull DataBufferListener videoDataBufferListener) { @@ -218,8 +257,6 @@ public int call(Pointer opaque, BytePointer dst, int requested) { private static void freeAVIO(AVIOContext ctx) { if (ctx == null || ctx.isNull()) return; - BytePointer buf = ctx.buffer(); - if (buf != null && !buf.isNull()) av_freep(buf); avio_context_free(ctx); } diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java index 691838b5b..1e7de3f44 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java @@ -19,6 +19,10 @@ public interface RtmpListener { /** Called once after negotiation succeeds and this listener is selected. */ public void onConnected(RtmpConnectEvent event); + /** + * Called once when a new stream is connected. Implementations MUST set up the video and audio outputs here. + * @param event Event containing the stream info. + */ public void onStreamConnected(RtmpStreamEvent event); /** Called once when the client disconnects or the pipeline faults. */ diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java index 36aefebc1..87b3c75da 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java @@ -13,19 +13,6 @@ public class RtmpListenerManager { public static RtmpListenerManager getInstance() { return INSTANCE; } - /** - * All 16 bitmasks for the 4 optional fields (username, password, path, streamKey), - * pre-sorted from most specific (0b1111 = all four) to least (0b0000 = none). - * The router iterates these in order so the first map hit is always the best match. - * - * Bit layout: bit3=username, bit2=password, bit1=path, bit0=streamKey - */ - private static final int[] CANDIDATE_MASKS = IntStream.range(0, 16) - .boxed() - .sorted(Comparator.comparingInt(Integer::bitCount).reversed()) - .mapToInt(Integer::intValue) - .toArray(); - // One entry per unique composite key — second registration with the same // config overwrites the first (unlike the old CopyOnWriteArrayList). private final ConcurrentHashMap listeners = @@ -68,30 +55,7 @@ public void removeListener(RtmpListener listener) { // ── Routing ──────────────────────────────────────────────────────────── - /** - * Tries all 16 composite key variants for the connection in specificity - * order (most fields → fewest fields) and returns the first map hit. - * - * A listener registered as "::1935::" (catch-all, specificity 0) is only - * reached if no more-specific listener claims the connection. - */ Optional route(RtmpConnectionContext ctx) { - String u = ctx.username(); - String pw = ctx.password(); - String pa = ctx.path(); - String sk = ctx.streamKey(); - - for (int mask : CANDIDATE_MASKS) { - String key = ConnectionConfig.compositeKey( - (mask & 0b1000) != 0 ? u : null, - (mask & 0b0100) != 0 ? pw : null, - ctx.port(), - (mask & 0b0010) != 0 ? pa : null, - (mask & 0b0001) != 0 ? sk : null); - - RtmpListener found = listeners.get(key); - if (found != null) return Optional.of(found); - } - return Optional.empty(); + return Optional.of(listeners.get(ConnectionConfig.compositeKey(ctx))); } } \ No newline at end of file From e75766fe5b8383431fcd23a1460213ceb15fc928 Mon Sep 17 00:00:00 2001 From: kyle-fitzp Date: Thu, 23 Jul 2026 09:37:14 -0500 Subject: [PATCH 4/4] Code cleanup, documentation --- .../impl/sensor/rtmpcam/RtmpDriver.java | 41 +++++------- .../connection/RtmpConnectionHandler.java | 29 ++++---- .../connection/RtmpListenerManager.java | 15 +++-- .../rtmpcam/connection/RtmpNegotiator.java | 66 ++++++++++--------- .../rtmpcam/connection/RtmpPortServer.java | 17 +++++ .../rtmpcam/helpers/RtmpStringParser.java | 49 -------------- 6 files changed, 91 insertions(+), 126 deletions(-) delete mode 100644 sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java index afe45b090..c0539a4ec 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java @@ -33,42 +33,24 @@ import static org.bytedeco.ffmpeg.global.avutil.av_log_set_callback; + /** - * OpenSensorHub sensor module that listens for and processes RTMP streams. - *

- * The driver creates an FFmpeg-backed MPEG-TS processor configured as an RTMP - * listener. Once a publisher connects, the driver detects available audio and - * video streams, creates matching OpenSensorHub outputs, and forwards stream - * data through those outputs. - *

- *

- * Only one RTMP driver instance may use a given port at a time. Port ownership - * is tracked through a shared {@link RtmpListenerManager}. - *

+ * RtmpDriver is a class that provides the implementation for managing an RTMP stream as part of + * a sensor module. This class initiates, monitors, and handles connections to RTMP streams. It also + * handles video and audio stream processing based on the RTMP protocol. */ public class RtmpDriver extends AbstractSensorModule implements RtmpListener { - private static final String COMMAND_LINE_ARGS = "-timeout 0 -listen 1 -username test -password test"; - private static final int EXECUTOR_JOIN_TIMEOUT = 10; - private static final TimeUnit EXECUTOR_JOIN_TIME_UNIT = TimeUnit.SECONDS; - private static final int HEARTBEAT_INTERVAL = 5; - private static final TimeUnit HEARTBEAT_TIME_UNIT = TimeUnit.SECONDS; - private static final int MAX_STARTUP_WAIT_TIME_MS = 5000; private volatile boolean doStreamProcessing = false; - private final AtomicReference mpegTsProcessor = new AtomicReference<>(); private final RtmpListenerManager rtmpListenerManager = RtmpListenerManager.getInstance(); - private ExecutorService executorService; private ExecutorService videoExecutorService = Executors.newSingleThreadExecutor(); private ExecutorService audioExecutorService = Executors.newSingleThreadExecutor(); - private ScheduledExecutorService heartbeatExecutorService; // TODO: Create a DataOutput final AtomicReference> videoOutput = new AtomicReference<>(); final AtomicReference> audioOutput = new AtomicReference<>(); - int connectionPort = -1; String connectionUrl = ""; - //String path = ""; /** * Indicates whether the driver has successfully connected to an RTMP stream at least once since starting. @@ -110,6 +92,7 @@ protected void doInit() throws SensorHubException { @Override protected void doStart() throws SensorHubException { super.doStart(); + reportStatus("RTMP: Listening for connection"); rtmpListenerManager.addListener(this); doStreamProcessing = true; } @@ -158,6 +141,15 @@ public void onConnected(RtmpConnectEvent event) { reportStatus("Connected to: " + connectionUrl); } + /** + * Handles the event triggered when an RTMP stream is connected. + * This method initializes both video and audio output streams based on the given stream information + * contained in the event payload. If the stream payload is null, a warning will be logged, and no further + * action will be taken. + * + * @param event the event containing the payload with stream information, including video and audio codec + * details and video dimensions or audio sample rate information required to initialize the outputs. + */ @Override public void onStreamConnected(RtmpStreamEvent event) { var streamInfo = event.getPayload(); @@ -181,17 +173,18 @@ public void onStreamConnected(RtmpStreamEvent event) { audioOutput.get().doInit(); addOutput(audioOutput.get(), false); } + reportStatus("RTMP: Connected"); } @Override public void onDisconnected(RtmpDisconnectEvent event) { removeAllOutputs(); - reportStatus("Disconnected from: " + connectionUrl); + reportStatus("RTMP: Disconnected"); } @Override public void onReconnected(RtmpReconnectEvent event) { - + reportStatus("RTMP: Connected"); } @Override diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java index 67e542b86..00d9c6046 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java @@ -35,7 +35,7 @@ * 2. AMF0 negotiation → {@link RtmpConnectionContext} * 3. Route to a matching {@link RtmpListener} * 4. RTMP chunks → FLV pipe → FFmpeg custom AVIO - * 5. Deliver encoded packets via {@link RtmpListener#publish} + * 5. Deliver encoded packets */ class RtmpConnectionHandler { @@ -72,7 +72,7 @@ void handle() { Optional match = manager.route(ctx); if (match.isEmpty()) { // No listener registered for this combination — drop silently - System.out.printf("[RTMP:%d] No listener for path='%s' key='%s'%n", + logger.warn("[RTMP:{}] No listener for path={} key={}", port, ctx.path(), ctx.streamKey()); return; } @@ -92,8 +92,19 @@ void handle() { } } - // ── FFmpeg pipeline ──────────────────────────────────────────────────── + /** + * Creates a pipeline to feed FLV data from the given input stream to FFmpeg for processing. + * This method sets up an in-memory AVIOContext for FFmpeg to read the stream, initializes + * the FFmpeg format context, and manages the processing of the incoming stream data + * through registered {@link RtmpListener} callbacks. The pipeline processes the stream + * until the end of the input or an error occurs. + * + * @param flvStream The input stream containing the FLV data to be processed. This must + * remain open and readable for the duration of the pipeline's lifetime. + * @param listener An implementation of {@link RtmpListener} that will handle stream + * connection events, data processing, and stream disconnection events. + */ private void pipeToFfmpeg(InputStream flvStream, RtmpListener listener) { // Both must stay reachable for the pipeline's lifetime: @@ -209,18 +220,6 @@ private void streamContextSetup(AVFormatContext avFormatContext, RtmpListener li audioStreamContext.setDataBufferListener(listener.getAudioOutput()); } - public void setVideoBufferListener(@Nonnull DataBufferListener videoDataBufferListener) { - videoStreamContext.setDataBufferListener(videoDataBufferListener); - } - - public void setAudioBufferListener(@Nonnull DataBufferListener audioDataBufferListener) { - audioStreamContext.setDataBufferListener(audioDataBufferListener); - } - - public void setDataBufferListener(@Nonnull DataBufferListener dataBufferListener) { - dataStreamContext.setDataBufferListener(dataBufferListener); - } - private void packetLoop(AVFormatContext fmtCtx, RtmpListener listener) { AVPacket pkt = av_packet_alloc(); try { diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java index 87b3c75da..61798a488 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java @@ -7,14 +7,21 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.stream.IntStream; +/** + * Manages the registration and lifecycle of RTMP listeners and port servers. + * + * This class provides a singleton instance for handling RTMP listeners, ensuring that + * multiple registrations with the same composite key will overwrite the prior registration. + * It also manages the mapping between RTMP port servers and their associated listeners, + * ensuring resources allocated for specific ports are initialized or cleaned up as + * necessary. + */ public class RtmpListenerManager { private static final RtmpListenerManager INSTANCE = new RtmpListenerManager(); public static RtmpListenerManager getInstance() { return INSTANCE; } - // One entry per unique composite key — second registration with the same - // config overwrites the first (unlike the old CopyOnWriteArrayList). private final ConcurrentHashMap listeners = new ConcurrentHashMap<>(); @@ -23,8 +30,6 @@ public class RtmpListenerManager { private final Object portLock = new Object(); - // ── Registration ─────────────────────────────────────────────────────── - public void addListener(RtmpListener listener) { listeners.put(listener.config().compositeKey(), listener); @@ -53,8 +58,6 @@ public void removeListener(RtmpListener listener) { } } - // ── Routing ──────────────────────────────────────────────────────────── - Optional route(RtmpConnectionContext ctx) { return Optional.of(listeners.get(ConnectionConfig.compositeKey(ctx))); } diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java index f4e6af094..d626d1d87 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java @@ -20,8 +20,7 @@ */ public class RtmpNegotiator { - // ── Protocol constants ───────────────────────────────────────────────── - + // Protocol constants private static final int RTMP_VERSION = 3; private static final int HANDSHAKE_SIZE = 1536; private static final int DEFAULT_CHUNK_SIZE = 128; @@ -59,8 +58,7 @@ public class RtmpNegotiator { }; private static final Logger logger = LoggerFactory.getLogger(RtmpNegotiator.class); - // ── Instance state ───────────────────────────────────────────────────── - + // Instance state private final DataInputStream in; private final DataOutputStream out; private final int port; @@ -78,8 +76,6 @@ public RtmpNegotiator(DataInputStream in, DataOutputStream out, int port) { this.port = port; } - // ── Phase 1: Handshake ───────────────────────────────────────────────── - /** * Performs the C0/C1/C2 ↔ S0/S1/S2 handshake (simple mode). * Modern publishers (OBS, FFmpeg, etc.) accept the simple echo handshake. @@ -117,8 +113,6 @@ public void doHandshake() throws IOException { in.readFully(new byte[HANDSHAKE_SIZE]); } - // ── Phase 2: AMF0 Negotiation ────────────────────────────────────────── - /** * Sends server control messages then loops over incoming RTMP chunks, * processing AMF0 commands until a {@code publish} command is confirmed. @@ -231,8 +225,6 @@ public RtmpConnectionContext negotiate() throws IOException { streamKey); } - // ── Phase 3: FLV pump ────────────────────────────────────────────────── - /** * Starts a virtual thread that reads RTMP chunks and writes FLV-framed * bytes into a pipe. The returned {@link InputStream} is consumed by @@ -286,8 +278,9 @@ private void pumpFlv(OutputStream flvOut) throws IOException { } } - // ── FLV tag emission ─────────────────────────────────────────────────── - + /** + * Writes a media tag encapsulated in the FLV format to the specified output stream. + */ private void writeFlvMediaTag(OutputStream out, RtmpMessage msg) throws IOException { writeFlvTag(out, msg.type(), msg.data(), (int) msg.timestamp()); } @@ -325,8 +318,6 @@ private static void writeFlvTag(OutputStream out, int tagType, writeUInt32(out, tagTotal); // PreviousTagSize } - // ── Chunk reassembly ─────────────────────────────────────────────────── - /** * Reads and reassembles RTMP chunks until a complete RTMP message is ready. * @@ -420,7 +411,6 @@ private RtmpMessage readMessage() throws IOException { } } - // ── Server → Client: protocol control ───────────────────────────────── private void sendWindowAckSize(int size) throws IOException { sendMessage(CS_PROTOCOL, MSG_WINDOW_ACK_SIZE, 0, 0, encodeUInt32(size)); @@ -456,7 +446,7 @@ private void sendPingResponse(long token) throws IOException { sendMessage(CS_PROTOCOL, MSG_USER_CONTROL, 0, 0, payload); } - // ── Server → Client: AMF0 command responses ──────────────────────────── + private void sendConnectResult(double txId) throws IOException { ByteArrayOutputStream buf = new ByteArrayOutputStream(256); @@ -534,8 +524,6 @@ private void sendPublishStart(int streamId, String streamKey) throws IOException sendMessage(CS_STREAM, MSG_COMMAND_AMF0, streamId, 0, buf.toByteArray()); } - // ── Low-level RTMP message write ─────────────────────────────────────── - /** * Serialises one RTMP message. Uses fmt=0 for the first chunk (full * header) and fmt=3 (no header) for any continuation chunks. @@ -571,8 +559,14 @@ private void sendMessage(int csId, int msgType, int msgStreamId, out.flush(); } - // ── User control ─────────────────────────────────────────────────────── - + /** + * Handles user control messages by analyzing the provided byte array. + * If the data represents a ping request and contains a valid token, + * a ping response is sent back. + * + * @param data the byte array containing the user control message data + * @throws IOException if an error occurs during the processing of the message + */ private void handleUserControl(byte[] data) throws IOException { if (data.length < 2) return; int eventType = ((data[0] & 0xFF) << 8) | (data[1] & 0xFF); @@ -585,8 +579,15 @@ private void handleUserControl(byte[] data) throws IOException { } } - // ── Acknowledgement ──────────────────────────────────────────────────── - + /** + * Conditionally sends an acknowledgment message based on the number of bytes + * received and the configured acknowledgment window size. + *

+ * When the difference between the total bytes received and the bytes + * already acknowledged reaches or exceeds the window acknowledgment size, + * this method sends an acknowledgment for the total number of bytes received + * so far. + */ private void maybeAck() throws IOException { if (windowAckSize > 0 && (bytesReceived - bytesAcked) >= windowAckSize) { bytesAcked = bytesReceived; @@ -594,8 +595,17 @@ private void maybeAck() throws IOException { } } - // ── AMF0 read ────────────────────────────────────────────────────────── - + /** + * Reads and decodes an AMF0 (Action Message Format 0) value from the provided input stream. + * The AMF0 format is a binary serialization format used in Adobe Flash and other related technologies. + * The method interprets a type byte to identify the kind of data being read and processes it accordingly. + * + * @param src the input stream from which the AMF0 value is read. + * It must be properly positioned to read the type and value bytes. + * @return the deserialized object corresponding to the AMF0 type. + * May return {@code null} for certain AMF0 types such as null, undefined, object end, or date (skipped). + * @throws IOException if an I/O error occurs while reading from the input stream or if an unknown type is encountered. + */ private Object readAmf0Value(DataInputStream src) throws IOException { int type = src.readUnsignedByte(); return switch (type) { @@ -656,8 +666,6 @@ private String readOptionalString(DataInputStream src) throws IOException { return v instanceof String s ? s : ""; } - // ── AMF0 write ───────────────────────────────────────────────────────── - private static void writeAmf0Str(DataOutputStream d, String s) throws IOException { byte[] b = s.getBytes(StandardCharsets.UTF_8); d.writeByte(2); d.writeShort(b.length); d.write(b); @@ -704,8 +712,6 @@ private static void writeAmf0Field(DataOutputStream d, String key, Object value) } } - // ── Primitive I/O ────────────────────────────────────────────────────── - /** Reads one byte and increments the ACK counter. */ private int readByte() throws IOException { bytesReceived++; @@ -745,8 +751,6 @@ private static byte[] encodeUInt32(int v) { return new byte[]{ (byte)(v >> 24), (byte)(v >> 16), (byte)(v >> 8), (byte) v }; } - // ── Utilities ────────────────────────────────────────────────────────── - private static DataInputStream wrapBytes(byte[] data) { return new DataInputStream(new ByteArrayInputStream(data)); } @@ -796,8 +800,6 @@ private static String[] extractCredentials(String tcUrl) { } } - // ── Inner types ──────────────────────────────────────────────────────── - /** Per-chunk-stream reassembly state, persisted across chunk reads. */ private static final class ChunkStream { int messageType; diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java index e6aadce15..560341184 100644 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java @@ -7,6 +7,23 @@ import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; +/** + * Handles RTMP (Real-Time Messaging Protocol) connections on a specific port. + *

+ * This class provides functionality to listen for and manage incoming client connections + * on a specified port. Once started, it creates a server socket to accept connections + * and delegates each connection to a separate thread for handling. + *

+ * The class is designed to operate as a lightweight and efficient RTMP connection server, + * capable of handling multiple connections concurrently using a cached thread pool. + *

+ * Lifecycle: + * 1. The server is initialized with a port and a reference to an RtmpListenerManager. + * 2. Once started, it begins listening for incoming connections on the specified port. + * 3. Each accepted client connection is processed using an RtmpConnectionHandler. + * 4. The server can be gracefully stopped, releasing resources like the server socket + * and shutting down the thread pool. + */ class RtmpPortServer { private final int port; diff --git a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java deleted file mode 100644 index 298831c89..000000000 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/helpers/RtmpStringParser.java +++ /dev/null @@ -1,49 +0,0 @@ -package org.sensorhub.impl.sensor.rtmpcam.helpers; - -import java.net.MalformedURLException; -import java.net.URL; - -public class RtmpStringParser { - public static boolean isRtmp(String str) { - return str.startsWith("rtmp"); - } - - public static String getUsername(String str) { - URL url; - try { - url = new URL(str); - } catch (MalformedURLException e) { - return null; - } - - if (url.getUserInfo() == null) { - return null; - } else { - return url.getUserInfo().split(":")[0]; - } - } - - public static String getPassword(String str) { - URL url; - try { - url = new URL(str); - } catch (MalformedURLException e) { - return null; - } - if (url.getUserInfo() == null) { - return null; - } else { - return url.getUserInfo().split(":")[1]; - } - } - - public static String getPath(String str) { - URL url; - try { - url = new URL(str); - } catch (MalformedURLException e) { - return null; - } - return url.getPath(); - } -}