Initial implementation of ADGStreamer by guirodrigueslima · Pull Request #1 · cnpem/ADGStreamer

guirodrigueslima · 2026-08-06T20:31:00Z

Hi everyone,

I'm currently developing an AreaDetector driver based on GStreamer to acquire images from network video streams. The current version supports RTSP communication only, and it has been working very well.

My goal is to extend the driver to support additional streaming protocols, such as HTTP, RTMP, and SRT. To make future development and maintenance easier, I introduced a PipelineBuilder class that encapsulates the pipeline construction and provides a scalable foundation for adding new protocols.

gustavosr8

Partial review, for now.

Overall, I think that a great improve for this project could be to add a .clang-format rule set.

For references on how to do it, you can refer to lnls-dig/uhal#67.

gustavosr8 · 2026-08-07T16:34:14Z

+               1,
+               1,
+               maxBuffers,
+               maxMemory,
+               0,
+               0,
+               ASYN_CANBLOCK,
+               1,
+               priority,
+               stackSize)


It would be nice to document which parameter the literal values refer to, similar to what is being done in ADSpecsPEEM

gustavosr8 · 2026-08-07T16:40:36Z

    ADGStreamer(const char *portName, int maxBuffers=0, size_t maxMemory=0, int priority=0, int stackSize=0);
    virtual ~ADGStreamer();

+    virtual asynStatus writeInt32(asynUser *pasynUser, epicsInt32 value);


It would be nice to add the override identifier to explicit the function purpose.

gustavosr8 · 2026-08-07T16:45:10Z

+public:
+    ADGStreamer(const char *portName, int maxBuffers=0, size_t maxMemory=0, int priority=0, int stackSize=0);
+    virtual ~ADGStreamer();
+
+protected:
+
+private:
+    bool initializeGStreamer();


Since classes have private as default, you can declare all private members first, and the members which require specifiers like public and protected after that. This way, it is not necessary to specify private.

Since private has no members, I think that it is also not necessary to be specified.

gustavosr8 · 2026-08-07T17:07:54Z

+        if (error)
+        {   
+
+        }


Is this really needed?

gustavosr8 · 2026-08-07T17:33:40Z

+    if (!pipeline_)
+    {
+        if (error)
+        {   
+
+        }
+        return false;
+    }
+
+    if (error)
+    {
+        return false;
+    }


What about unifying this checks in a single if statement?

gustavosr8 · 2026-08-07T17:51:38Z

+    return true;
+}
+
+bool ADGStreamer::createPipeline()


I'm missing some explanation in the commit message about the required steps when creating a new pipeline

Same for creating/stopping the pipeline

gustavosr8 · 2026-08-07T18:20:07Z

+    pPvt->acquisitionTask();
+}
+
+void ADGStreamer::acquisitionTask()


I see no reason on creating an empty acquisition task. You can create the task without it

gustavosr8 · 2026-08-07T18:24:28Z

+    if (!bus_)
+    {
+        return false;
+    }


Is this checking really needed once you are already checking for bus_ in createPipeline()?

gustavosr8 · 2026-08-07T18:28:26Z

-        }
-        else
-        {
+            processBusMessage();


If I get it right, the bus messages warns us about pipelines states. If it is the case, the message processing shouldn't be a standalone thread instead of being attached to the acquisition one?

Implemented the pipeline start and stop logic through the `startPipeline()` and `stopPipeline()` routines. When an acquisition starts, samples are received by the `onNewSampleCallback()` callback function. At this stage, the pipeline string configuration is hardcoded.

Implemented the `processSample()` routine to process samples received by the GStreamer callback. The function extracts the image data from the `GstSample` and prepares it for delivery to the AreaDetector driver and plugins.

An epicsThread has been added to support the future implementation of a GstBus monitoring.

Added `GstBus` monitoring for pipeline events such as state changes, errors, and End Of Stream (EOS), keeping the driver updated with the current pipeline execution state.

construction Introduce the PipelineBuilder architecture to separate GStreamer pipeline construction from the ADGStreamer driver implementation. Common pipeline stages, such as queue, decoder, converter, and sink, are implemented in PipelineBuilder, allowing them to be shared across different communication protocols. Protocol-specific implementations, such as PipelineBuilderRTSP, are responsible only for configuring protocol-related pipeline elements, making it easier to support additional protocols such as HTTP, RTMP, and SRT without requiring changes to the driver implementation.

Add the reportStatus() helper method to centralize driver status updates and status message reporting.

gustavosr8 · 2026-08-20T13:12:11Z

Just another overall suggestion, I've notice you are using a lot of booleans to handling the status of different operations. I suggest changing it to use asynStatus interface, or throw exceptions where it fits.

gustavosr8

Commit suggestions:

08dea92: s/clang format/clang-format

5f8d88d: configure: include areaDetector CONFIG_SITE files

8b55858: ADGStreamerApp: add ADGStreamer constructor and iocsh functions

gustavosr8 · 2026-08-20T12:21:43Z

+-include $(TOP)/../configure/RELEASE_LIBS_INCLUDE
+-include $(TOP)/RELEASE.local


Are those related to AREA_DETECTOR?

gustavosr8 · 2026-08-20T12:32:53Z

+ADGStreamer::ADGStreamer(const char *portName, int maxBuffers, size_t maxMemory,
+    int priority, int stackSize)
+    : ADDriver(portName, 1, 1, maxBuffers, maxMemory, 0, 0, ASYN_CANBLOCK, 1,
+          priority, stackSize)
+{
+    initializeGStreamer();
+}
+
+ADGStreamer::~ADGStreamer() { }
+
+bool ADGStreamer::initializeGStreamer()
+{
+    gst_init(nullptr, nullptr);
+    return true;
+}


Why not simply call gst_init() directly in the constructor?

gustavosr8 · 2026-08-20T15:44:59Z

+    GstStateChangeReturn ret;
+    ret = gst_element_set_state(pipeline_, GST_STATE_PLAYING);
+
+    if (ret == GST_STATE_CHANGE_FAILURE) {
+        return false;
+    }


What about

if (gst_element_set_state(pipeline_, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) { ... }

gustavosr8 · 2026-08-20T15:48:10Z

+    GstState state;
+    gst_element_get_state(pipeline_, &state, nullptr, GST_CLOCK_TIME_NONE);
+
+    if (state != GST_STATE_PLAYING) {
+        return false;
+    }
+
+    return true;


if (gst_element_get_state(pipeline_, &state, nullptr, GST_CLOCK_TIME_NONE) != GST_STATE_PLAYING) { ... }

gustavosr8 · 2026-08-20T15:50:07Z

+        GstStateChangeReturn ret;
+        ret = gst_element_set_state(pipeline_, GST_STATE_NULL);
+
+        if (ret == GST_STATE_CHANGE_FAILURE) {
+            return false;
+        }
+
+        GstState state;
+        gst_element_get_state(pipeline_, &state, nullptr, GST_CLOCK_TIME_NONE);
+
+        if (state != GST_STATE_NULL) {
+            return false;
+        }


Same as commented in startPipeline()

gustavosr8 · 2026-08-20T15:55:17Z

+        = std::string("rtspsrc "
+                      "location=rtsp://admin:r00tr00t@10.20.21.40:554/cam/"
+                      "realmonitor?channel=1&subtype=0 protocols=udp latency=0 "
+                      "drop-on-latency=true")
+        + std::string(" ! rtph264depay ! h264parse config-interval=-1")
+        + std::string(
+            " ! queue max-size-buffers=1 leaky=downstream silent=true")
+        + std::string(" ! avdec_h264 ! videoconvert ! video/x-raw,format=RGB ! "
+                      "appsink name=appsink sync=false");


Missing some explanation in commit message about the reasoning around this string.

gustavosr8 · 2026-08-20T15:57:24Z

+    if (!pipeline_) {
+        if (error) { }
+        return false;
+    }
+
+    if (error) {
+        return false;
+    }


Why not simply

if ( !pipeline_ || error)

gustavosr8 · 2026-08-20T16:04:27Z

+    if (!sink_) {
+        return GST_FLOW_ERROR;
+    }


As far as I understand, this function in registered as the callback of sink when receiving a new sample. Is there any way to get here without sink_ being defined? Is this check really needed?

gustavosr8

I think that 15fdbff and b519d2d should be squashed together

gustavosr8 · 2026-08-20T16:18:55Z

+    int imageCounter_;
+    int numImagesCounter_;


What is the difference between those two?

gustavosr8 · 2026-08-20T16:20:21Z

+    getIntegerParam(NDArrayCounter, &imageCounter_);
+    getIntegerParam(ADNumImagesCounter, &numImagesCounter_);
+    getIntegerParam(NDArrayCallbacks, &arrayCallbacks_);
+    imageCounter_++;
+    numImagesCounter_++;
+    setIntegerParam(NDArrayCounter, imageCounter_);
+    setIntegerParam(ADNumImagesCounter, numImagesCounter_);
+    pImage_->uniqueId = imageCounter_;
+    pImage_->timeStamp = startTime.secPastEpoch + startTime.nsec / 1.e9;
+    updateTimeStamp(&pImage_->epicsTS);


Isn't missing a callParamCallbacks() after those operations?

guirodrigueslima requested review from GuPecz, ericonr, gustavosr8 and henriquesimoes August 6, 2026 20:31

guirodrigueslima force-pushed the dev branch 2 times, most recently from e568679 to 61c27fe Compare August 10, 2026 17:44

gustavosr8 reviewed Aug 10, 2026

View reviewed changes

guirodrigueslima force-pushed the dev branch from 61c27fe to d89e641 Compare August 11, 2026 19:10

guirodrigueslima added 14 commits August 19, 2026 18:04

Add clang format file

08dea92

configure: use areaDetector configure

5f8d88d

ADGStreamerApp: add gstreamer constructor and iocsh function

8b55858

ADGStreamerApp: add sample processing

f8b1bec

Implemented the `processSample()` routine to process samples received by the GStreamer callback. The function extracts the image data from the `GstSample` and prepares it for delivery to the AreaDetector driver and plugins.

ADGStreamerApp: add epicsThread

15fdbff

An epicsThread has been added to support the future implementation of a GstBus monitoring.

ADGStreamerApp: add GstBus monitoring

b519d2d

Added `GstBus` monitoring for pipeline events such as state changes, errors, and End Of Stream (EOS), keeping the driver updated with the current pipeline execution state.

ADGStreamerApp: include record read pipeline string

b1ce8e9

ADGStreamerApp: centralize driver status updates

eba3322

Add the reportStatus() helper method to centralize driver status updates and status message reporting.

iocs: link ioc to support module

38ad0f5

iocs: configure st.cmd

f9f8010

readme: include readme file

3717679

docs: add initial ADGStreamer documentation

8368c61

guirodrigueslima force-pushed the dev branch from d89e641 to 8368c61 Compare August 19, 2026 21:12

gustavosr8 reviewed Aug 20, 2026

View reviewed changes

		-include $(TOP)/../configure/RELEASE_LIBS_INCLUDE
		-include $(TOP)/RELEASE.local

Uh oh!

Conversation

guirodrigueslima commented Aug 6, 2026

Uh oh!

gustavosr8 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gustavosr8 commented Aug 20, 2026

Uh oh!

gustavosr8 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gustavosr8 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants