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/rtmp/RtmpDriver.java deleted file mode 100644 index f28a7f409..000000000 --- a/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDriver.java +++ /dev/null @@ -1,504 +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 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.rtmp.config.RtmpConfig; -import org.sensorhub.mpegts.MpegTsProcessor; -import org.sensorhub.utils.Async; - -import java.security.SecureRandom; -import java.util.HexFormat; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicReference; - -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 RtmpPortSingleton}. - *

- */ -public class RtmpDriver extends AbstractSensorModule { - 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 final RtmpPortSingleton portSingleton = RtmpPortSingleton.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<>(); - - int connectionPort = -1; - String connectionUrl = ""; - //String path = ""; - - /** - * Indicates whether the driver has successfully connected to an RTMP stream at least once since starting. - */ - volatile boolean hasConnected = false; - - /** - * Indicates whether the driver is currently connected to an RTMP stream. - */ - volatile boolean isConnected = false; - - /** - * Initializes the driver configuration and generated identifiers. - *

- * If no unique identifier has been assigned, this method generates both the - * OpenSensorHub unique identifier and XML identifier from the configured - * serial number. It also releases any previously tracked port and rebuilds - * the RTMP listener URL from the current configuration. - *

- * - * @throws SensorHubException if initialization fails - */ - @Override - protected void doInit() throws SensorHubException { - super.doInit(); - - if (getUniqueIdentifier() == null) { - generateUniqueID("urn:osh:sensor:rtmp:", config.serialNumber); - generateXmlID("RTMP_", config.serialNumber); - } - - portSingleton.removeConnection(connectionPort); - - setConnectionUrl(); - - //createMpegTsProcessor(); - } - - private static String generateStreamKey() { - byte[] bytes = new byte[16]; - new SecureRandom().nextBytes(bytes); - 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); - } - - /** - * 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. - * - * @throws SensorHubException if shutdown fails - */ - @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(); - } - - /** - * Indicates whether the driver is currently started and has an open RTMP stream. - * - * @return {@code true} if the module is started and the RTMP stream is open; - * otherwise {@code false} - */ - @Override - public boolean isConnected() { - return isConnected; - } -} \ No newline at end of file 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/rtmpcam/RtmpDriver.java b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java new file mode 100644 index 000000000..c0539a4ec --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/RtmpDriver.java @@ -0,0 +1,199 @@ +/***************************** 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; + +import org.sensorhub.api.common.SensorHubException; +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.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 java.security.SecureRandom; +import java.util.HexFormat; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicReference; + +import static org.bytedeco.ffmpeg.global.avutil.av_log_set_callback; + + +/** + * 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 volatile boolean doStreamProcessing = false; + private final RtmpListenerManager rtmpListenerManager = RtmpListenerManager.getInstance(); + private ExecutorService videoExecutorService = Executors.newSingleThreadExecutor(); + private ExecutorService audioExecutorService = Executors.newSingleThreadExecutor(); + + // TODO: Create a DataOutput + final AtomicReference> videoOutput = new AtomicReference<>(); + final AtomicReference> audioOutput = new AtomicReference<>(); + + String connectionUrl = ""; + + /** + * Indicates whether the driver has successfully connected to an RTMP stream at least once since starting. + */ + volatile boolean hasConnected = false; + + /** + * Indicates whether the driver is currently connected to an RTMP stream. + */ + volatile boolean isConnected = false; + + /** + * Initializes the driver configuration and generated identifiers. + *

+ * If no unique identifier has been assigned, this method generates both the + * OpenSensorHub unique identifier and XML identifier from the configured + * serial number. It also releases any previously tracked port and rebuilds + * the RTMP listener URL from the current configuration. + *

+ * + * @throws SensorHubException if initialization fails + */ + @Override + protected void doInit() throws SensorHubException { + super.doInit(); + + if (getUniqueIdentifier() == null) { + 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); + } + + @Override + protected void doStart() throws SensorHubException { + super.doStart(); + reportStatus("RTMP: Listening for connection"); + rtmpListenerManager.addListener(this); + doStreamProcessing = true; + } + + private static String generateStreamKey() { + byte[] bytes = new byte[16]; + new SecureRandom().nextBytes(bytes); + return HexFormat.of().formatHex(bytes); + } + + public boolean doStreamProcessing() { + return doStreamProcessing; + } + + + /** + * Stops the driver and releases all RTMP stream resources. + * + * @throws SensorHubException if shutdown fails + */ + @Override + protected void doStop() throws SensorHubException { + super.doStop(); + doStreamProcessing = false; + rtmpListenerManager.removeListener(this); + } + + /** + * Indicates whether the driver is currently started and has an open RTMP stream. + * + * @return {@code true} if the module is started and the RTMP stream is open; + * otherwise {@code false} + */ + @Override + public boolean isConnected() { + return isConnected; + } + + @Override + public ConnectionConfig config() { + return config.connectionConfig; + } + + @Override + 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(); + + 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); + } + reportStatus("RTMP: Connected"); + } + + @Override + public void onDisconnected(RtmpDisconnectEvent event) { + removeAllOutputs(); + reportStatus("RTMP: Disconnected"); + } + + @Override + public void onReconnected(RtmpReconnectEvent event) { + reportStatus("RTMP: Connected"); + } + + @Override + 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 new file mode 100644 index 000000000..fa7fe3671 --- /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; +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpConnectionContext; + +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(label = "Username") + public String username = ""; + + @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(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 : ""); + } + + 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/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..00d9c6046 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpConnectionHandler.java @@ -0,0 +1,268 @@ +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; +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.impl.sensor.rtmpcam.stream.StreamInfo; +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 java.util.stream.Stream; + +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 + */ +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; + + 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; + 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 + logger.warn("[RTMP:{}] No listener for path={} key={}", + port, ctx.path(), ctx.streamKey()); + return; + } + + RtmpListener listener = match.get(); + RtmpConnectEvent connectEvent = new RtmpConnectEvent(ctx); + listener.onConnected(connectEvent); + + try { + pipeToFfmpeg(negotiator.buildFlvStream(), listener); + } finally { + listener.onDisconnected(new RtmpDisconnectEvent(ctx)); + } + + } catch (Exception e) { + logger.error("[RTMP:{}] Error handling RTMP connection", port, e); + } + } + + + /** + * 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: + // 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; } + + 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); + + avformat_close_input(fmtCtx); + freeAVIO(avioCtx); + // readCb and avioBuf are now safe to collect + } + + 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) { + 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 == 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 == 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 == AVMEDIA_TYPE_DATA) { + logger.debug("Data stream present with id: {}", streamId); + + dataStreamContext.setStreamId(streamId); + dataStreamContext.setStreamTimeBase(timeBaseUnits); + streamContextMap.put(streamId, dataStreamContext); + } + } + + return new StreamInfo(videoDimensions, videoCodec, audioSampleRate, audioCodec); + } + + private void streamContextSetup(AVFormatContext avFormatContext, RtmpListener listener) { + var streamInfo = queryEmbeddedStreams(avFormatContext); + + videoStreamContext.setInjectingExtradata(true); + videoStreamContext.openCodecContext(avFormatContext); + audioStreamContext.openCodecContext(avFormatContext); + dataStreamContext.openCodecContext(avFormatContext); + + RtmpStreamEvent streamEvent = new RtmpStreamEvent(streamInfo); + listener.onStreamConnected(streamEvent); + + if (listener.getVideoOutput() != null) + videoStreamContext.setDataBufferListener(listener.getVideoOutput()); + if (listener.getAudioOutput() != null) + audioStreamContext.setDataBufferListener(listener.getAudioOutput()); + } + + private void packetLoop(AVFormatContext fmtCtx, RtmpListener listener) { + AVPacket pkt = av_packet_alloc(); + try { + int ret; + 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); + } + 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; + 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..1e7de3f44 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListener.java @@ -0,0 +1,38 @@ +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; +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(); + + /** 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. */ + public void onDisconnected(RtmpDisconnectEvent event); + + public void onReconnected(RtmpReconnectEvent 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/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..61798a488 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpListenerManager.java @@ -0,0 +1,64 @@ +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; + +/** + * 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; } + + private final ConcurrentHashMap listeners = + new ConcurrentHashMap<>(); + + private final ConcurrentHashMap portServers = + new ConcurrentHashMap<>(); + + private final Object portLock = new Object(); + + 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(); + } + } + } + + Optional route(RtmpConnectionContext ctx) { + return Optional.of(listeners.get(ConnectionConfig.compositeKey(ctx))); + } +} \ 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..d626d1d87 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpNegotiator.java @@ -0,0 +1,817 @@ +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; + } + + /** + * 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]); + } + + /** + * 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", "_checkbw" -> { /* 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); + } + + /** + * 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 */ } + } + } + } + } + } + } + + /** + * 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()); + } + + /** + * 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 + } + + /** + * 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 + } + } + + + 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); + } + + + + 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()); + } + + /** + * 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(); + } + + /** + * 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); + 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); + } + } + + /** + * 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; + sendAck(bytesReceived); + } + } + + /** + * 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) { + 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 : ""; + } + + 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()); + } + } + + /** 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 }; + } + + 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}; + } + } + + /** 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..560341184 --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/connection/RtmpPortServer.java @@ -0,0 +1,77 @@ +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; + +/** + * 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; + 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..83ffcb1aa --- /dev/null +++ b/sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmpcam/event/RtmpDisconnectEvent.java @@ -0,0 +1,16 @@ +package org.sensorhub.impl.sensor.rtmpcam.event; + +import org.sensorhub.impl.sensor.rtmpcam.connection.RtmpConnectionContext; + +public class RtmpDisconnectEvent implements RtmpEvent { + RtmpConnectionContext payload; + + public RtmpDisconnectEvent(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/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/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