Invoke stopSync in NULL state transition (#1023)

* Add EOS specific handling

* New logging

* New logging 1

* New logging 2

* New logging 4

* New logging 5

* New logging 5

* Check with bool

* listen on NULL state transition

Author: disa6302

src/KinesisVideoStream.cpp

@@ -160,7 +160,6 @@ bool KinesisVideoStream::stop() {
 
 bool KinesisVideoStream::stopSync() {
     STATUS status;
-
     if (STATUS_FAILED(status = stopKinesisVideoStreamSync(stream_handle_))) {
         LOG_ERROR("Failed to stop the stream with: " << status << " for " << this->stream_name_);
         return false;

src/gstreamer/gstkvssink.cpp

@@ -1182,6 +1182,10 @@ gst_kvs_sink_handle_sink_event (GstCollectPads *pads,
             event = NULL;
             break;
         }
+        case GST_EVENT_EOS: {
+            LOG_INFO("EOS Event received in sink for " << kvssink->stream_name);
+            break;
+        }
         default:
             break;
     }
@@ -1245,12 +1249,19 @@ gst_kvs_sink_handle_buffer (GstCollectPads * pads,
     GstMapInfo info;
 
     info.data = NULL;
-
     // eos reached
     if (buf == NULL && track_data == NULL) {
         LOG_INFO("Received event for " << kvssink->stream_name);
-        data->kinesis_video_stream->stopSync();
-        LOG_INFO("Sending eos for " << kvssink->stream_name);
+        // Need this check in case pipeline is already being set to NULL and
+        // stream  is being or/already stopped. Although stopSync() is an idempotent call,
+        // we want to avoid an extra call. It is not possible for this callback to be invoked
+        // after stopSync() since we stop collecting on pads before invoking. But having this
+        // check anyways in case it happens
+        if(!data->streamingStopped.load()) {
+            data->kinesis_video_stream->stopSync();
+            data->streamingStopped.store(true);
+            LOG_INFO("Sending eos for " << kvssink->stream_name);
+        }
 
         // send out eos message to gstreamer bus
         message = gst_message_new_eos (GST_OBJECT_CAST (kvssink));
@@ -1599,7 +1610,22 @@ gst_kvs_sink_change_state(GstElement *element, GstStateChange transition) {
             gst_collect_pads_start (kvssink->collect);
             break;
         case GST_STATE_CHANGE_PAUSED_TO_READY:
+            LOG_INFO("Stopping kvssink for " << kvssink->stream_name);
             gst_collect_pads_stop (kvssink->collect);
+
+            // Need this check in case an EOS was received in the buffer handler and
+            // stream was already stopped. Although stopSync() is an idempotent call,
+            // we want to avoid an extra call
+            if(!data->streamingStopped.load()) {
+                data->kinesis_video_stream->stopSync();
+                data->streamingStopped.store(true);
+            } else {
+                LOG_INFO("Streaming already stopped for " << kvssink->stream_name);
+            }
+            LOG_INFO("Stopped kvssink for " << kvssink->stream_name);
+            break;
+        case GST_STATE_CHANGE_READY_TO_NULL:
+            LOG_INFO("Pipeline state changed to NULL in kvssink");
             break;
         default:
             break;

src/gstreamer/gstkvssink.h

@@ -134,6 +134,7 @@ struct _GstKvsSink {
     guint                       num_audio_streams;
     guint                       num_video_streams;
 
+
     std::unique_ptr<Credentials> credentials_;
     std::shared_ptr<KvsSinkCustomData> data;
 };
@@ -162,7 +163,8 @@ struct _KvsSinkCustomData {
             on_first_frame(true),
             frame_count(0),
             first_pts(GST_CLOCK_TIME_NONE),
-            producer_start_time(GST_CLOCK_TIME_NONE) {}
+            producer_start_time(GST_CLOCK_TIME_NONE),
+            streamingStopped(false) {}
     std::unique_ptr<KinesisVideoProducer> kinesis_video_producer;
     std::shared_ptr<KinesisVideoStream> kinesis_video_stream;
 
@@ -174,6 +176,7 @@ struct _KvsSinkCustomData {
     bool get_metrics;
     uint32_t frame_count;
     bool on_first_frame;
+    std::atomic<bool> streamingStopped;
     uint64_t frame_pts;
 
     std::atomic_uint stream_status;