add refresh message

Add a new refresh message to request a keyframe from the pinos server.
pinospay: pass the refresh-request message upstream
pinossink: turn refresh-request messages into events
pinossrc: turn a keyframe event into a refresh-request message

pinos/gst/gstpinospay.c

@@ -46,6 +46,8 @@
 #include "gsttmpfileallocator.h"
 
 #include <gst/net/gstnetcontrolmessagemeta.h>
+#include <gst/video/video.h>
+
 #include <gio/gunixfdmessage.h>
 
 #include <fcntl.h>
@@ -198,13 +200,19 @@ client_buffer_received (GstPinosPay *pay, GstBuffer *buffer,
 {
   PinosBuffer pbuf;
   PinosBufferIter it;
+  PinosBufferBuilder b;
   GstMapInfo info;
   const gchar *client_path;
+  gboolean have_out = FALSE;
 
   client_path = g_object_get_data (obj, "pinos-client-path");
   if (client_path == NULL)
     return;
 
+  if (pay->pinos_input) {
+    pinos_buffer_builder_init (&b);
+  }
+
   gst_buffer_map (buffer, &info, GST_MAP_READ);
   pinos_buffer_init_data (&pbuf, info.data, info.size, NULL);
   pinos_buffer_iter_init (&it, &pbuf);
@@ -224,12 +232,54 @@ client_buffer_received (GstPinosPay *pay, GstBuffer *buffer,
         pinos_fd_manager_remove (pay->fdmanager, client_path, id);
         break;
       }
+      case PINOS_PACKET_TYPE_REFRESH_REQUEST:
+      {
+        PinosPacketRefreshRequest p;
+
+        if (!pinos_buffer_iter_parse_refresh_request (&it, &p))
+          continue;
+
+        GST_LOG ("refresh request");
+        if (!pay->pinos_input) {
+          gst_pad_push_event (pay->sinkpad,
+              gst_video_event_new_upstream_force_key_unit (p.pts,
+              p.request_type == 1, 0));
+        } else {
+          pinos_buffer_builder_add_refresh_request (&b, &p);
+          have_out = TRUE;
+        }
+        break;
+      }
       default:
         break;
     }
   }
   gst_buffer_unmap (buffer, &info);
   pinos_buffer_clear (&pbuf);
+
+  if (pay->pinos_input) {
+    GstBuffer *outbuf;
+    GstEvent *ev;
+    gsize size;
+    gpointer data;
+
+    if (have_out) {
+      pinos_buffer_builder_end (&b, &pbuf);
+
+      data = pinos_buffer_steal (&pbuf, &size, NULL);
+
+      outbuf = gst_buffer_new_wrapped (data, size);
+      ev = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
+              gst_structure_new ("GstNetworkMessage",
+                  "object", G_TYPE_OBJECT, pay,
+                  "buffer", GST_TYPE_BUFFER, outbuf, NULL));
+      gst_buffer_unref (outbuf);
+
+      gst_pad_push_event (pay->sinkpad, ev);
+    } else {
+      pinos_buffer_builder_clear (&b);
+    }
+  }
 }
 
 static gboolean
@@ -264,9 +314,10 @@ gst_pinos_pay_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
         client_buffer_received (pay, buf, obj);
         gst_buffer_unref (buf);
         g_object_unref (obj);
+
       }
-      gst_event_unref (event);
       res = TRUE;
+      gst_event_unref (event);
       break;
     }
     default:

pinos/gst/gstpinossink.c

@@ -41,6 +41,7 @@
 
 #include <gio/gunixfdmessage.h>
 #include <gst/allocators/gstfdmemory.h>
+#include <gst/video/video.h>
 
 #include "gsttmpfileallocator.h"
 
@@ -290,7 +291,19 @@ on_new_buffer (GObject    *gobject,
         g_hash_table_remove (pinossink->fdids, GINT_TO_POINTER (p.id));
         break;
       }
+      case PINOS_PACKET_TYPE_REFRESH_REQUEST:
+      {
+        PinosPacketRefreshRequest p;
 
+        if (!pinos_buffer_iter_parse_refresh_request (&it, &p))
+          continue;
+
+        GST_LOG ("refresh request");
+        gst_pad_push_event (GST_BASE_SINK_PAD (pinossink),
+            gst_video_event_new_upstream_force_key_unit (p.pts,
+            p.request_type == 1, 0));
+        break;
+      }
       default:
         break;
     }

pinos/gst/gstpinossrc.c

@@ -42,6 +42,7 @@
 #include <gio/gunixfdmessage.h>
 #include <gst/net/gstnetclientclock.h>
 #include <gst/allocators/gstfdmemory.h>
+#include <gst/video/video.h>
 
 
 static GQuark fdpayload_data_quark;
@@ -84,6 +85,7 @@ static gboolean gst_pinos_src_unlock (GstBaseSrc * basesrc);
 static gboolean gst_pinos_src_unlock_stop (GstBaseSrc * basesrc);
 static gboolean gst_pinos_src_start (GstBaseSrc * basesrc);
 static gboolean gst_pinos_src_stop (GstBaseSrc * basesrc);
+static gboolean gst_pinos_src_event (GstBaseSrc * src, GstEvent * event);
 
 static void
 gst_pinos_src_set_property (GObject * object, guint prop_id,
@@ -245,7 +247,7 @@ gst_pinos_src_class_init (GstPinosSrcClass * klass)
   gstbasesrc_class->unlock_stop = gst_pinos_src_unlock_stop;
   gstbasesrc_class->start = gst_pinos_src_start;
   gstbasesrc_class->stop = gst_pinos_src_stop;
-
+  gstbasesrc_class->event = gst_pinos_src_event;
   gstpushsrc_class->create = gst_pinos_src_create;
 
   GST_DEBUG_CATEGORY_INIT (pinos_src_debug, "pinossrc", 0,
@@ -724,6 +726,55 @@ gst_pinos_src_unlock_stop (GstBaseSrc * basesrc)
   return TRUE;
 }
 
+static gboolean
+gst_pinos_src_event (GstBaseSrc * src, GstEvent * event)
+{
+  gboolean res = FALSE;
+  GstPinosSrc *pinossrc;
+
+  pinossrc = GST_PINOS_SRC (src);
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_CUSTOM_UPSTREAM:
+      if (gst_video_event_is_force_key_unit (event)) {
+        GstClockTime running_time;
+        gboolean all_headers;
+        guint count;
+        PinosPacketRefreshRequest refresh;
+        PinosBufferBuilder b;
+        PinosBuffer pbuf;
+
+        gst_video_event_parse_upstream_force_key_unit (event,
+                &running_time, &all_headers, &count);
+
+        refresh.last_id = 0;
+        refresh.request_type = all_headers ? 1 : 0;
+        refresh.pts = running_time;
+
+        pinos_buffer_builder_init (&b);
+        pinos_buffer_builder_add_refresh_request (&b, &refresh);
+        pinos_buffer_builder_end (&b, &pbuf);
+
+        GST_OBJECT_LOCK (pinossrc);
+        if (pinossrc->stream_state == PINOS_STREAM_STATE_STREAMING) {
+          GST_DEBUG_OBJECT (pinossrc, "send refresh request");
+          pinos_stream_send_buffer (pinossrc->stream, &pbuf);
+        }
+        GST_OBJECT_UNLOCK (pinossrc);
+
+        pinos_buffer_clear (&pbuf);
+        res = TRUE;
+      } else {
+        res = GST_BASE_SRC_CLASS (parent_class)->event (src, event);
+      }
+      break;
+    default:
+      res = GST_BASE_SRC_CLASS (parent_class)->event (src, event);
+      break;
+  }
+  return res;
+}
+
 static GstFlowReturn
 gst_pinos_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
 {