Rtmp driver

sensors/video/sensorhub-driver-ffmpeg/src/main/java/org/sensorhub/mpegts/MpegTsProcessor.java

@@ -14,12 +14,14 @@
 import org.bytedeco.ffmpeg.avcodec.AVCodecParameters;
 import org.bytedeco.ffmpeg.avcodec.AVPacket;
 import org.bytedeco.ffmpeg.avformat.AVFormatContext;
+import org.bytedeco.ffmpeg.avformat.AVIOInterruptCB;
 import org.bytedeco.ffmpeg.avformat.AVInputFormat;
 import org.bytedeco.ffmpeg.avutil.AVDictionary;
 import org.bytedeco.ffmpeg.avutil.AVRational;
 import org.bytedeco.ffmpeg.global.avcodec;
 import org.bytedeco.ffmpeg.global.avformat;
 import org.bytedeco.ffmpeg.global.avutil;
+import org.bytedeco.javacpp.Pointer;
 import org.bytedeco.javacpp.PointerPointer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,8 +33,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.bytedeco.ffmpeg.global.avdevice.avdevice_register_all;
-import static org.bytedeco.ffmpeg.global.avformat.av_find_input_format;
-import static org.bytedeco.ffmpeg.global.avformat.av_read_frame;
+import static org.bytedeco.ffmpeg.global.avformat.*;
 
 /**
  * The class provides a wrapper to bytedeco.org JavaCpp-Platform.
@@ -98,6 +99,13 @@ public class MpegTsProcessor extends Thread {
      */
     private AVFormatContext avFormatContext;
 
+    /**
+     * Callbacks allow some blocking AVFormatContext operations to terminate early on
+     * stream close. These are global for GC purposes; do not directly modify.
+     */
+    private AVIOInterruptCB.Callback_Pointer callbackPointer;
+    private AVIOInterruptCB interruptCallback;
+
     /**
      * Flag indicating if processing of the transport stream should be terminated.
      */
@@ -186,7 +194,17 @@ public boolean openStream() {
         avformat.avformat_network_init();
 
         // Create a new AV Format Context for I/O
-        avFormatContext = new AVFormatContext(null);
+        avFormatContext = avformat_alloc_context();
+
+        // Allow for context to be interrupted after attempt to stop stream
+        interruptCallback = avFormatContext.interrupt_callback();
+        callbackPointer = new AVIOInterruptCB.Callback_Pointer() {
+            @Override
+            public int call(Pointer opaque) {
+                return terminateProcessing.get() ? 1 : 0;
+            }
+        };
+        interruptCallback.callback(callbackPointer);
 
         AVInputFormat inputFormat = null;
 
@@ -526,11 +544,8 @@ public void closeStream() {
      */
     public void stopProcessingStream() {
         logger.debug("stopProcessingStream");
-
-        if (streamOpened) {
-            loop = false;
-            terminateProcessing.set(true);
-        }
+        loop = false;
+        terminateProcessing.set(true);
     }
 
     /**

sensors/video/sensorhub-driver-rtmp/README.md

@@ -0,0 +1,38 @@
+# FFmpeg RTMP Driver
+
+OSH sensor driver using FFmpeg to listen to RTMP streams.
+
+This driver depends on the following modules at runtime:
+* sensorhub-driver-ffmpeg
+
+## Config
+
+### Connection
+
+* **Host**
+  - RTMP server host 
+    - **UNSPECIFIED**: listen for remote connections.
+    - **LOCALHOST**: listen for local connections.
+    - **DOCKER_INTERNAL**: listen for connections inside a docker container.
+* **Port**
+  - RTMP server port
+    - 1935 is the default port for RTMP.
+    - Any port may be used as long as it is not in use by another listener (or any other application).
+      - RTMP sensor driver modules track ports in use. To release ownership of a port, STOP the module.
+
+## Usage
+
+### Initialization
+
+<p>To initialize the driver, provide a unique serial number in the configuration.
+In connection, set the host and provide an unused port. Once initialized, other RTMP sensor driver modules cannot
+use the same port. 
+<br><br><strong>Stop</strong> the module to free the port.</p>
+
+### Listening
+
+<p>Once initialized, start the module to begin listening for RTMP connections. <strong>After</strong> the driver
+is started, begin publishing the RTMP stream. 
+<br><br><strong>Note</strong>: The module does NOT attempt to validate the 
+username, password, RTMP app, or RTMP playpath. Only the host and port need to match.</p>
+

sensors/video/sensorhub-driver-rtmp/build.gradle

@@ -0,0 +1,28 @@
+description = 'RTMP Video Driver'
+ext.details = 'Driver to listen for and demux RTMP video streams'
+version = '1.0.0'
+
+dependencies {
+  implementation 'org.sensorhub:sensorhub-core:' + oshCoreVersion
+  implementation project(':sensorhub-driver-ffmpeg')
+}
+
+// add info to OSGi manifest
+osgi {
+  manifest {
+    attributes('Bundle-Vendor': 'GeoRobotix Innovative Research')
+    attributes('Bundle-Activator': 'org.sensorhub.impl.sensor.rtmp.Activator')
+  }
+}
+
+// add info to maven pom
+ext.pom >>= {
+  developers {
+    developer {
+      id 'kyle-fitzp'
+      name 'Kyle Fitzpatrick'
+      organization 'GeoRobotix Innovative Research'
+      organizationUrl 'https://georobotix.us'
+    }
+  }
+}

sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/Activator.java

@@ -0,0 +1,25 @@
+/***************************** 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.osgi.framework.BundleActivator;
+import org.sensorhub.utils.OshBundleActivator;
+
+
+/*
+ * Needed to expose java services as OSGi services
+ */
+public class Activator extends OshBundleActivator implements BundleActivator
+{
+
+}

sensors/video/sensorhub-driver-rtmp/src/main/java/org/sensorhub/impl/sensor/rtmp/RtmpDescriptor.java

@@ -0,0 +1,38 @@
+/***************************** 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.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;
+
+
+public class RtmpDescriptor extends JarModuleProvider implements IModuleProvider
+{
+
+	@Override
+	public Class<? extends IModule<?>> getModuleClass()
+	{
+		return RtmpDriver.class;
+	}
+
+
+	@Override
+	public Class<? extends ModuleConfig> getModuleConfigClass()
+	{
+	       return RtmpConfig.class;
+	}
+
+}