mirror of
https://gitlab.freedesktop.org/gstreamer/gstreamer.git
synced 2024-12-16 13:26:36 +00:00
9eee5adb24
Useful when one wants to upload a video as `video/mp4` instead of 'video/quicktime` for example. Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/3371>
921 lines
30 KiB
C++
921 lines
30 KiB
C++
/* GStreamer
|
|
* Copyright (C) 2020 Julien Isorce <jisorce@oblong.com>
|
|
*
|
|
* gstgssink.cpp:
|
|
*
|
|
* This library is free software; you can redistribute it and/or
|
|
* modify it under the terms of the GNU Library General Public
|
|
* License as published by the Free Software Foundation; either
|
|
* version 2 of the License, or (at your option) any later version.
|
|
*
|
|
* This library is distributed in the hope that it will be useful,
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
* Library General Public License for more details.
|
|
*
|
|
* You should have received a copy of the GNU Library General Public
|
|
* License along with this library; if not, write to the
|
|
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
|
|
* Boston, MA 02110-1301, USA.
|
|
*/
|
|
/**
|
|
* SECTION:element-gssink
|
|
* @title: gssink
|
|
* @see_also: #GstGsSrc
|
|
*
|
|
* Write incoming data to a series of sequentially-named remote files on a
|
|
* Google Cloud Storage bucket.
|
|
*
|
|
* The object-name property should contain a string with a \%d placeholder
|
|
* that will be substituted with the index for each filename.
|
|
*
|
|
* If the #GstGsSink:post-messages property is %TRUE, it sends an application
|
|
* message named `GstGsSink` after writing each buffer.
|
|
*
|
|
* The message's structure contains these fields:
|
|
*
|
|
* * #gchararray `filename`: the filename where the buffer was written.
|
|
* * #gchararray `date`: the date of the current buffer, NULL if no start date
|
|
* is provided.
|
|
* * #gint `index`: index of the buffer.
|
|
* * #GstClockTime `timestamp`: the timestamp of the buffer.
|
|
* * #GstClockTime `stream-time`: the stream time of the buffer.
|
|
* * #GstClockTime `running-time`: the running_time of the buffer.
|
|
* * #GstClockTime `duration`: the duration of the buffer.
|
|
* * #guint64 `offset`: the offset of the buffer that triggered the message.
|
|
* * #guint64 `offset-end`: the offset-end of the buffer that triggered the
|
|
* message.
|
|
*
|
|
* ## Example launch line
|
|
* ```
|
|
* gst-launch-1.0 videotestsrc num-buffers=15 ! pngenc ! gssink
|
|
* object-name="mypath/myframes/frame%05d.png" bucket-name="mybucket"
|
|
* next-file=buffer post-messages=true
|
|
* ```
|
|
* ### Upload 15 png images into gs://mybucket/mypath/myframes/ where the file
|
|
* names are frame00000.png, frame00001.png, ..., frame00014.png
|
|
* ```
|
|
* gst-launch-1.0 videotestsrc num-buffers=6 ! video/x-raw, framerate=2/1 !
|
|
* pngenc ! gssink start-date="2020-04-16T08:55:03Z"
|
|
* object-name="mypath/myframes/im_%s_%03d.png" bucket-name="mybucket"
|
|
* next-file=buffer post-messages=true
|
|
* ```
|
|
* ### Upload png 6 images into gs://mybucket/mypath/myframes/ where the file
|
|
* names are im_2020-04-16T08:55:03Z_000.png, im_2020-04-16T08:55:03Z_001.png,
|
|
* im_2020-04-16T08:55:04Z_002.png, im_2020-04-16T08:55:04Z_003.png,
|
|
* im_2020-04-16T08:55:05Z_004.png, im_2020-04-16T08:55:05Z_005.png.
|
|
* ```
|
|
* gst-launch-1.0 filesrc location=some_video.mp4 ! gssink
|
|
* object-name="mypath/myvideos/video.mp4" bucket-name="mybucket" next-file=none
|
|
* ```
|
|
* ### Upload any stream as a single file into Google Cloud Storage. Similar as
|
|
* filesink in this case. The file is then accessible from:
|
|
* gs://mybucket/mypath/myvideos/video.mp4
|
|
*
|
|
* See also: #GstGsSrc
|
|
* Since: 1.20
|
|
*/
|
|
|
|
#ifdef HAVE_CONFIG_H
|
|
#include "config.h"
|
|
#endif
|
|
|
|
#include "gstgscommon.h"
|
|
#include "gstgssink.h"
|
|
|
|
#include <algorithm>
|
|
|
|
static GstStaticPadTemplate sinktemplate =
|
|
GST_STATIC_PAD_TEMPLATE("sink",
|
|
GST_PAD_SINK,
|
|
GST_PAD_ALWAYS,
|
|
GST_STATIC_CAPS_ANY);
|
|
|
|
GST_DEBUG_CATEGORY_STATIC(gst_gs_sink_debug);
|
|
#define GST_CAT_DEFAULT gst_gs_sink_debug
|
|
|
|
#define DEFAULT_INDEX 0
|
|
#define DEFAULT_NEXT_FILE GST_GS_SINK_NEXT_BUFFER
|
|
#define DEFAULT_OBJECT_NAME "%s_%05d"
|
|
#define DEFAULT_POST_MESSAGES FALSE
|
|
|
|
namespace gcs = google::cloud::storage;
|
|
|
|
enum {
|
|
PROP_0,
|
|
PROP_BUCKET_NAME,
|
|
PROP_OBJECT_NAME,
|
|
PROP_INDEX,
|
|
PROP_POST_MESSAGES,
|
|
PROP_NEXT_FILE,
|
|
PROP_SERVICE_ACCOUNT_EMAIL,
|
|
PROP_START_DATE,
|
|
PROP_SERVICE_ACCOUNT_CREDENTIALS,
|
|
PROP_METADATA,
|
|
PROP_CONTENT_TYPE,
|
|
};
|
|
|
|
class GSWriteStream;
|
|
|
|
struct _GstGsSink {
|
|
GstBaseSink parent;
|
|
|
|
std::unique_ptr<google::cloud::storage::Client> gcs_client;
|
|
std::unique_ptr<GSWriteStream> gcs_stream;
|
|
gchar* service_account_email;
|
|
gchar* service_account_credentials;
|
|
gchar* bucket_name;
|
|
gchar* object_name;
|
|
gchar* start_date_str;
|
|
GDateTime* start_date;
|
|
gint index;
|
|
gboolean post_messages;
|
|
GstGsSinkNext next_file;
|
|
const gchar* content_type;
|
|
gchar* content_type_prop;
|
|
size_t nb_percent_format;
|
|
gboolean percent_s_is_first;
|
|
GstStructure* metadata;
|
|
};
|
|
|
|
static void gst_gs_sink_finalize(GObject* object);
|
|
|
|
static void gst_gs_sink_set_property(GObject* object,
|
|
guint prop_id,
|
|
const GValue* value,
|
|
GParamSpec* pspec);
|
|
static void gst_gs_sink_get_property(GObject* object,
|
|
guint prop_id,
|
|
GValue* value,
|
|
GParamSpec* pspec);
|
|
|
|
static gboolean gst_gs_sink_start(GstBaseSink* bsink);
|
|
static gboolean gst_gs_sink_stop(GstBaseSink* sink);
|
|
static GstFlowReturn gst_gs_sink_render(GstBaseSink* sink, GstBuffer* buffer);
|
|
static GstFlowReturn gst_gs_sink_render_list(GstBaseSink* sink,
|
|
GstBufferList* buffer_list);
|
|
static gboolean gst_gs_sink_set_caps(GstBaseSink* sink, GstCaps* caps);
|
|
static gboolean gst_gs_sink_event(GstBaseSink* sink, GstEvent* event);
|
|
|
|
#define GST_TYPE_GS_SINK_NEXT (gst_gs_sink_next_get_type())
|
|
static GType gst_gs_sink_next_get_type(void) {
|
|
static GType gs_sink_next_type = 0;
|
|
static const GEnumValue next_types[] = {
|
|
{GST_GS_SINK_NEXT_BUFFER, "New file for each buffer", "buffer"},
|
|
{GST_GS_SINK_NEXT_NONE, "Only one file, no next file", "none"},
|
|
{0, NULL, NULL}};
|
|
|
|
if (!gs_sink_next_type) {
|
|
gs_sink_next_type = g_enum_register_static("GstGsSinkNext", next_types);
|
|
}
|
|
|
|
return gs_sink_next_type;
|
|
}
|
|
|
|
#define gst_gs_sink_parent_class parent_class
|
|
G_DEFINE_TYPE(GstGsSink, gst_gs_sink, GST_TYPE_BASE_SINK);
|
|
GST_ELEMENT_REGISTER_DEFINE(gssink, "gssink", GST_RANK_NONE, GST_TYPE_GS_SINK)
|
|
|
|
class GSWriteStream {
|
|
public:
|
|
GSWriteStream(google::cloud::storage::Client& client,
|
|
const char* bucket_name,
|
|
const char* object_name,
|
|
gcs::ObjectMetadata metadata)
|
|
: gcs_stream_(client.WriteObject(bucket_name,
|
|
object_name,
|
|
gcs::WithObjectMetadata(metadata))) {}
|
|
~GSWriteStream() { gcs_stream_.Close(); }
|
|
|
|
gcs::ObjectWriteStream& stream() { return gcs_stream_; }
|
|
|
|
private:
|
|
gcs::ObjectWriteStream gcs_stream_;
|
|
};
|
|
|
|
static void gst_gs_sink_class_init(GstGsSinkClass* klass) {
|
|
GObjectClass* gobject_class = G_OBJECT_CLASS(klass);
|
|
GstElementClass* gstelement_class = GST_ELEMENT_CLASS(klass);
|
|
GstBaseSinkClass* gstbasesink_class = GST_BASE_SINK_CLASS(klass);
|
|
|
|
gobject_class->set_property = gst_gs_sink_set_property;
|
|
gobject_class->get_property = gst_gs_sink_get_property;
|
|
|
|
/**
|
|
* GstGsSink:bucket-name:
|
|
*
|
|
* Name of the Google Cloud Storage bucket.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_BUCKET_NAME,
|
|
g_param_spec_string(
|
|
"bucket-name", "Bucket Name", "Google Cloud Storage Bucket Name",
|
|
NULL, (GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
|
|
|
|
/**
|
|
* GstGsSink:object-name:
|
|
*
|
|
* Full path name of the remote file.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_OBJECT_NAME,
|
|
g_param_spec_string(
|
|
"object-name", "Object Name", "Full path name of the remote file",
|
|
DEFAULT_OBJECT_NAME,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
|
|
|
|
/**
|
|
* GstGsSink:index:
|
|
*
|
|
* Index to use with location property to create file names.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_INDEX,
|
|
g_param_spec_int(
|
|
"index", "Index",
|
|
"Index to use with location property to create file names. The "
|
|
"index is incremented by one for each buffer written.",
|
|
0, G_MAXINT, DEFAULT_INDEX,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
|
|
|
|
/**
|
|
* GstGsSink:post-messages:
|
|
*
|
|
* Post a message on the GstBus for each file.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_POST_MESSAGES,
|
|
g_param_spec_boolean(
|
|
"post-messages", "Post Messages",
|
|
"Post a message for each file with information of the buffer",
|
|
DEFAULT_POST_MESSAGES,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
|
|
/**
|
|
* GstGsSink:next-file:
|
|
*
|
|
* A #GstGsSinkNext that specifies when to start a new file.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_NEXT_FILE,
|
|
g_param_spec_enum(
|
|
"next-file", "Next File", "When to start a new file",
|
|
GST_TYPE_GS_SINK_NEXT, DEFAULT_NEXT_FILE,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
|
|
|
|
/**
|
|
* GstGsSink:service-account-email:
|
|
*
|
|
* Service Account Email to use for credentials.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_SERVICE_ACCOUNT_EMAIL,
|
|
g_param_spec_string(
|
|
"service-account-email", "Service Account Email",
|
|
"Service Account Email to use for credentials", NULL,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
|
|
GST_PARAM_MUTABLE_READY)));
|
|
|
|
/**
|
|
* GstGsSink:service-account-credentials:
|
|
*
|
|
* Service Account Credentials as a JSON string to use for credentials.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_SERVICE_ACCOUNT_CREDENTIALS,
|
|
g_param_spec_string(
|
|
"service-account-credentials", "Service Account Credentials",
|
|
"Service Account Credentials as a JSON string to use for credentials",
|
|
NULL,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
|
|
GST_PARAM_MUTABLE_READY)));
|
|
|
|
/**
|
|
* GstGsSink:start-date:
|
|
*
|
|
* Start date in iso8601 format.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_START_DATE,
|
|
g_param_spec_string(
|
|
"start-date", "Start Date", "Start date in iso8601 format", NULL,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
|
|
GST_PARAM_MUTABLE_READY)));
|
|
|
|
/**
|
|
* GstGsSink:metadata:
|
|
*
|
|
* A map of metadata to store with the object; field values need to be
|
|
* convertible to strings.
|
|
*
|
|
* Since: 1.20
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_METADATA,
|
|
g_param_spec_boxed(
|
|
"metadata", "Metadata",
|
|
"A map of metadata to store with the object; field values need to be "
|
|
"convertible to strings.",
|
|
GST_TYPE_STRUCTURE,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
|
|
GST_PARAM_MUTABLE_READY)));
|
|
|
|
/**
|
|
* GstGsSink:content-type:
|
|
*
|
|
* The Content-Type of the object.
|
|
* If not set we use the name of the input caps.
|
|
*
|
|
* Since: 1.22
|
|
*/
|
|
g_object_class_install_property(
|
|
gobject_class, PROP_CONTENT_TYPE,
|
|
g_param_spec_string(
|
|
"content-type", "Content-Type",
|
|
"The Content-Type of the object",
|
|
NULL,
|
|
(GParamFlags)(G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
|
|
GST_PARAM_MUTABLE_READY)));
|
|
|
|
gobject_class->finalize = gst_gs_sink_finalize;
|
|
|
|
gstbasesink_class->start = GST_DEBUG_FUNCPTR(gst_gs_sink_start);
|
|
gstbasesink_class->stop = GST_DEBUG_FUNCPTR(gst_gs_sink_stop);
|
|
gstbasesink_class->render = GST_DEBUG_FUNCPTR(gst_gs_sink_render);
|
|
gstbasesink_class->render_list = GST_DEBUG_FUNCPTR(gst_gs_sink_render_list);
|
|
gstbasesink_class->set_caps = GST_DEBUG_FUNCPTR(gst_gs_sink_set_caps);
|
|
gstbasesink_class->event = GST_DEBUG_FUNCPTR(gst_gs_sink_event);
|
|
|
|
GST_DEBUG_CATEGORY_INIT(gst_gs_sink_debug, "gssink", 0, "gssink element");
|
|
|
|
gst_element_class_add_static_pad_template(gstelement_class, &sinktemplate);
|
|
gst_element_class_set_static_metadata(
|
|
gstelement_class, "Google Cloud Storage Sink", "Sink/File",
|
|
"Write buffers to a sequentially named set of files on Google Cloud "
|
|
"Storage",
|
|
"Julien Isorce <jisorce@oblong.com>");
|
|
}
|
|
|
|
static void gst_gs_sink_init(GstGsSink* sink) {
|
|
sink->gcs_client = nullptr;
|
|
sink->gcs_stream = nullptr;
|
|
sink->index = DEFAULT_INDEX;
|
|
sink->post_messages = DEFAULT_POST_MESSAGES;
|
|
sink->service_account_email = NULL;
|
|
sink->service_account_credentials = NULL;
|
|
sink->bucket_name = NULL;
|
|
sink->object_name = g_strdup(DEFAULT_OBJECT_NAME);
|
|
sink->start_date_str = NULL;
|
|
sink->start_date = NULL;
|
|
sink->next_file = DEFAULT_NEXT_FILE;
|
|
sink->content_type = NULL;
|
|
sink->content_type_prop = NULL;
|
|
sink->nb_percent_format = 0;
|
|
sink->percent_s_is_first = FALSE;
|
|
|
|
gst_base_sink_set_sync(GST_BASE_SINK(sink), FALSE);
|
|
}
|
|
|
|
static void gst_gs_sink_finalize(GObject* object) {
|
|
GstGsSink* sink = GST_GS_SINK(object);
|
|
|
|
sink->gcs_client = nullptr;
|
|
sink->gcs_stream = nullptr;
|
|
g_free(sink->service_account_email);
|
|
sink->service_account_email = NULL;
|
|
g_free(sink->service_account_credentials);
|
|
sink->service_account_credentials = NULL;
|
|
g_free(sink->bucket_name);
|
|
sink->bucket_name = NULL;
|
|
g_free(sink->object_name);
|
|
sink->object_name = NULL;
|
|
g_free(sink->start_date_str);
|
|
sink->start_date_str = NULL;
|
|
if (sink->start_date) {
|
|
g_date_time_unref(sink->start_date);
|
|
sink->start_date = NULL;
|
|
}
|
|
sink->content_type = NULL;
|
|
g_clear_pointer(&sink->content_type_prop, g_free);
|
|
g_clear_pointer(&sink->metadata, gst_structure_free);
|
|
|
|
G_OBJECT_CLASS(parent_class)->finalize(object);
|
|
}
|
|
|
|
static gboolean gst_gs_sink_set_object_name(GstGsSink* sink,
|
|
const gchar* object_name) {
|
|
g_free(sink->object_name);
|
|
sink->object_name = NULL;
|
|
sink->nb_percent_format = 0;
|
|
sink->percent_s_is_first = FALSE;
|
|
|
|
if (!object_name) {
|
|
GST_ERROR_OBJECT(sink, "Object name is null");
|
|
return FALSE;
|
|
}
|
|
|
|
const std::string name(object_name);
|
|
sink->nb_percent_format = std::count(name.begin(), name.end(), '%');
|
|
if (sink->nb_percent_format > 2) {
|
|
GST_ERROR_OBJECT(sink, "Object name has too many formats");
|
|
return FALSE;
|
|
}
|
|
|
|
const size_t delimiter_percent_s = name.find("%s");
|
|
if (delimiter_percent_s == std::string::npos) {
|
|
if (sink->nb_percent_format == 2) {
|
|
GST_ERROR_OBJECT(sink, "Object name must have just one number format");
|
|
return FALSE;
|
|
}
|
|
sink->object_name = g_strdup(object_name);
|
|
return TRUE;
|
|
}
|
|
|
|
const size_t delimiter_percent = name.find_first_of('%');
|
|
if (delimiter_percent_s == delimiter_percent) {
|
|
sink->percent_s_is_first = TRUE;
|
|
|
|
if (name.find("%s", delimiter_percent_s + 1) != std::string::npos) {
|
|
GST_ERROR_OBJECT(sink, "Object name expect max one string format");
|
|
return FALSE;
|
|
}
|
|
}
|
|
|
|
sink->object_name = g_strdup(object_name);
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static void gst_gs_sink_set_property(GObject* object,
|
|
guint prop_id,
|
|
const GValue* value,
|
|
GParamSpec* pspec) {
|
|
GstGsSink* sink = GST_GS_SINK(object);
|
|
|
|
switch (prop_id) {
|
|
case PROP_BUCKET_NAME:
|
|
g_free(sink->bucket_name);
|
|
sink->bucket_name = g_strdup(g_value_get_string(value));
|
|
break;
|
|
case PROP_OBJECT_NAME:
|
|
gst_gs_sink_set_object_name(sink, g_value_get_string(value));
|
|
break;
|
|
case PROP_INDEX:
|
|
sink->index = g_value_get_int(value);
|
|
break;
|
|
case PROP_POST_MESSAGES:
|
|
sink->post_messages = g_value_get_boolean(value);
|
|
break;
|
|
case PROP_NEXT_FILE:
|
|
sink->next_file = (GstGsSinkNext)g_value_get_enum(value);
|
|
break;
|
|
case PROP_SERVICE_ACCOUNT_EMAIL:
|
|
g_free(sink->service_account_email);
|
|
sink->service_account_email = g_strdup(g_value_get_string(value));
|
|
break;
|
|
case PROP_SERVICE_ACCOUNT_CREDENTIALS:
|
|
g_free(sink->service_account_credentials);
|
|
sink->service_account_credentials = g_strdup(g_value_get_string(value));
|
|
break;
|
|
case PROP_START_DATE:
|
|
g_free(sink->start_date_str);
|
|
if (sink->start_date)
|
|
g_date_time_unref(sink->start_date);
|
|
sink->start_date_str = g_strdup(g_value_get_string(value));
|
|
sink->start_date =
|
|
g_date_time_new_from_iso8601(sink->start_date_str, NULL);
|
|
if (!sink->start_date) {
|
|
GST_ERROR_OBJECT(sink, "Failed to parse start date %s",
|
|
sink->start_date_str);
|
|
g_free(sink->start_date_str);
|
|
sink->start_date_str = NULL;
|
|
}
|
|
break;
|
|
case PROP_METADATA:
|
|
g_clear_pointer(&sink->metadata, gst_structure_free);
|
|
sink->metadata = (GstStructure*)g_value_dup_boxed(value);
|
|
break;
|
|
case PROP_CONTENT_TYPE:
|
|
g_clear_pointer(&sink->content_type_prop, g_free);
|
|
sink->content_type_prop = g_value_dup_string(value);
|
|
break;
|
|
default:
|
|
G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
|
|
break;
|
|
}
|
|
}
|
|
|
|
static void gst_gs_sink_get_property(GObject* object,
|
|
guint prop_id,
|
|
GValue* value,
|
|
GParamSpec* pspec) {
|
|
GstGsSink* sink = GST_GS_SINK(object);
|
|
|
|
switch (prop_id) {
|
|
case PROP_BUCKET_NAME:
|
|
g_value_set_string(value, sink->bucket_name);
|
|
break;
|
|
case PROP_OBJECT_NAME:
|
|
g_value_set_string(value, sink->object_name);
|
|
break;
|
|
case PROP_INDEX:
|
|
g_value_set_int(value, sink->index);
|
|
break;
|
|
case PROP_POST_MESSAGES:
|
|
g_value_set_boolean(value, sink->post_messages);
|
|
break;
|
|
case PROP_NEXT_FILE:
|
|
g_value_set_enum(value, sink->next_file);
|
|
break;
|
|
case PROP_SERVICE_ACCOUNT_EMAIL:
|
|
g_value_set_string(value, sink->service_account_email);
|
|
break;
|
|
case PROP_SERVICE_ACCOUNT_CREDENTIALS:
|
|
g_value_set_string(value, sink->service_account_credentials);
|
|
break;
|
|
case PROP_START_DATE:
|
|
g_value_set_string(value, sink->start_date_str);
|
|
break;
|
|
case PROP_METADATA:
|
|
g_value_set_boxed(value, sink->metadata);
|
|
break;
|
|
case PROP_CONTENT_TYPE:
|
|
g_value_set_string(value, sink->content_type_prop);
|
|
break;
|
|
default:
|
|
G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
|
|
break;
|
|
}
|
|
}
|
|
|
|
static gboolean gst_gs_sink_start(GstBaseSink* bsink) {
|
|
GstGsSink* sink = GST_GS_SINK(bsink);
|
|
GError* err = NULL;
|
|
|
|
if (!sink->bucket_name) {
|
|
GST_ELEMENT_ERROR(sink, RESOURCE, SETTINGS, ("Bucket name is required"),
|
|
GST_ERROR_SYSTEM);
|
|
return FALSE;
|
|
}
|
|
|
|
if (!sink->object_name) {
|
|
GST_ELEMENT_ERROR(sink, RESOURCE, SETTINGS, ("Object name is required"),
|
|
GST_ERROR_SYSTEM);
|
|
return FALSE;
|
|
}
|
|
|
|
sink->content_type = "";
|
|
|
|
sink->gcs_client = gst_gs_create_client(
|
|
sink->service_account_email, sink->service_account_credentials, &err);
|
|
if (err) {
|
|
GST_ELEMENT_ERROR(sink, RESOURCE, OPEN_READ,
|
|
("Could not create client (%s)", err->message),
|
|
GST_ERROR_SYSTEM);
|
|
g_clear_error(&err);
|
|
return FALSE;
|
|
}
|
|
|
|
GST_INFO_OBJECT(sink, "Using bucket name (%s) and object name (%s)",
|
|
sink->bucket_name, sink->object_name);
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static gboolean gst_gs_sink_stop(GstBaseSink* bsink) {
|
|
GstGsSink* sink = GST_GS_SINK(bsink);
|
|
|
|
sink->gcs_client = nullptr;
|
|
sink->gcs_stream = nullptr;
|
|
sink->content_type = NULL;
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static void gst_gs_sink_post_message_full(GstGsSink* sink,
|
|
GstClockTime timestamp,
|
|
GstClockTime duration,
|
|
GstClockTime offset,
|
|
GstClockTime offset_end,
|
|
GstClockTime running_time,
|
|
GstClockTime stream_time,
|
|
const char* filename,
|
|
const gchar* date) {
|
|
GstStructure* s;
|
|
|
|
if (!sink->post_messages)
|
|
return;
|
|
|
|
s = gst_structure_new("GstGsSink", "filename", G_TYPE_STRING, filename,
|
|
"date", G_TYPE_STRING, date, "index", G_TYPE_INT,
|
|
sink->index, "timestamp", G_TYPE_UINT64, timestamp,
|
|
"stream-time", G_TYPE_UINT64, stream_time,
|
|
"running-time", G_TYPE_UINT64, running_time, "duration",
|
|
G_TYPE_UINT64, duration, "offset", G_TYPE_UINT64,
|
|
offset, "offset-end", G_TYPE_UINT64, offset_end, NULL);
|
|
|
|
gst_element_post_message(GST_ELEMENT_CAST(sink),
|
|
gst_message_new_element(GST_OBJECT_CAST(sink), s));
|
|
}
|
|
|
|
static void gst_gs_sink_post_message_from_time(GstGsSink* sink,
|
|
GstClockTime timestamp,
|
|
GstClockTime duration,
|
|
const char* filename) {
|
|
GstClockTime running_time, stream_time;
|
|
guint64 offset, offset_end;
|
|
GstSegment* segment;
|
|
GstFormat format;
|
|
|
|
if (!sink->post_messages)
|
|
return;
|
|
|
|
segment = &GST_BASE_SINK(sink)->segment;
|
|
format = segment->format;
|
|
|
|
offset = -1;
|
|
offset_end = -1;
|
|
|
|
running_time = gst_segment_to_running_time(segment, format, timestamp);
|
|
stream_time = gst_segment_to_stream_time(segment, format, timestamp);
|
|
|
|
gst_gs_sink_post_message_full(sink, timestamp, duration, offset, offset_end,
|
|
running_time, stream_time, filename, NULL);
|
|
}
|
|
|
|
static void gst_gs_sink_post_message(GstGsSink* sink,
|
|
GstBuffer* buffer,
|
|
const char* filename,
|
|
const char* date) {
|
|
GstClockTime duration, timestamp;
|
|
GstClockTime running_time, stream_time;
|
|
guint64 offset, offset_end;
|
|
GstSegment* segment;
|
|
GstFormat format;
|
|
|
|
if (!sink->post_messages)
|
|
return;
|
|
|
|
segment = &GST_BASE_SINK(sink)->segment;
|
|
format = segment->format;
|
|
|
|
timestamp = GST_BUFFER_PTS(buffer);
|
|
duration = GST_BUFFER_DURATION(buffer);
|
|
offset = GST_BUFFER_OFFSET(buffer);
|
|
offset_end = GST_BUFFER_OFFSET_END(buffer);
|
|
|
|
running_time = gst_segment_to_running_time(segment, format, timestamp);
|
|
stream_time = gst_segment_to_stream_time(segment, format, timestamp);
|
|
|
|
gst_gs_sink_post_message_full(sink, timestamp, duration, offset, offset_end,
|
|
running_time, stream_time, filename, date);
|
|
}
|
|
|
|
struct AddMetadataIter {
|
|
GstGsSink* sink;
|
|
gcs::ObjectMetadata* metadata;
|
|
};
|
|
|
|
static gboolean add_metadata_foreach(GQuark field_id,
|
|
const GValue* value,
|
|
gpointer user_data) {
|
|
struct AddMetadataIter* it = (struct AddMetadataIter*)user_data;
|
|
GValue svalue = G_VALUE_INIT;
|
|
|
|
g_value_init(&svalue, G_TYPE_STRING);
|
|
|
|
if (g_value_transform(value, &svalue)) {
|
|
const gchar* key = g_quark_to_string(field_id);
|
|
const gchar* value = g_value_get_string(&svalue);
|
|
|
|
GST_LOG_OBJECT(it->sink, "metadata '%s' -> '%s'", key, value);
|
|
it->metadata->upsert_metadata(key, value);
|
|
} else {
|
|
GST_WARNING_OBJECT(it->sink, "Failed to convert metadata '%s' to string",
|
|
g_quark_to_string(field_id));
|
|
}
|
|
|
|
g_value_unset(&svalue);
|
|
return TRUE;
|
|
}
|
|
|
|
static GstFlowReturn gst_gs_sink_write_buffer(GstGsSink* sink,
|
|
GstBuffer* buffer) {
|
|
GstMapInfo map = {0};
|
|
gchar* object_name = NULL;
|
|
gchar* buffer_date = NULL;
|
|
const gchar* content_type;
|
|
|
|
if (!gst_buffer_map(buffer, &map, GST_MAP_READ))
|
|
return GST_FLOW_ERROR;
|
|
|
|
if (sink->content_type_prop)
|
|
content_type = sink->content_type_prop;
|
|
else
|
|
content_type = sink->content_type;
|
|
|
|
gcs::ObjectMetadata metadata =
|
|
gcs::ObjectMetadata().set_content_type(content_type);
|
|
|
|
if (sink->metadata) {
|
|
struct AddMetadataIter it = {sink, &metadata};
|
|
|
|
gst_structure_foreach(sink->metadata, add_metadata_foreach, &it);
|
|
}
|
|
|
|
switch (sink->next_file) {
|
|
case GST_GS_SINK_NEXT_BUFFER: {
|
|
// Get buffer date if needed.
|
|
if (sink->start_date) {
|
|
if (sink->nb_percent_format != 2) {
|
|
GST_ERROR_OBJECT(sink, "Object name expects date and index");
|
|
gst_buffer_unmap(buffer, &map);
|
|
return GST_FLOW_ERROR;
|
|
}
|
|
|
|
if (!gst_gs_get_buffer_date(buffer, sink->start_date, &buffer_date)) {
|
|
GST_ERROR_OBJECT(sink, "Could not get buffer date %s", object_name);
|
|
gst_buffer_unmap(buffer, &map);
|
|
return GST_FLOW_ERROR;
|
|
}
|
|
|
|
if (sink->percent_s_is_first) {
|
|
object_name =
|
|
g_strdup_printf(sink->object_name, buffer_date, sink->index);
|
|
} else {
|
|
object_name =
|
|
g_strdup_printf(sink->object_name, sink->index, buffer_date);
|
|
}
|
|
} else {
|
|
if (sink->nb_percent_format != 1) {
|
|
GST_ERROR_OBJECT(sink, "Object name expects only an index");
|
|
gst_buffer_unmap(buffer, &map);
|
|
return GST_FLOW_ERROR;
|
|
}
|
|
|
|
object_name = g_strdup_printf(sink->object_name, sink->index);
|
|
}
|
|
|
|
GST_INFO_OBJECT(sink, "Writing %" G_GSIZE_FORMAT " bytes", map.size);
|
|
|
|
gcs::ObjectWriteStream gcs_stream = sink->gcs_client->WriteObject(
|
|
sink->bucket_name, object_name, gcs::WithObjectMetadata(metadata));
|
|
|
|
gcs_stream.write(reinterpret_cast<const char*>(map.data), map.size);
|
|
if (gcs_stream.fail()) {
|
|
GST_WARNING_OBJECT(sink, "Failed to write to %s", object_name);
|
|
}
|
|
gcs_stream.Close();
|
|
|
|
google::cloud::StatusOr<gcs::ObjectMetadata> object_metadata =
|
|
sink->gcs_client->GetObjectMetadata(sink->bucket_name, object_name);
|
|
if (!object_metadata) {
|
|
GST_ERROR_OBJECT(
|
|
sink, "Could not get object metadata for object %s (%s)",
|
|
object_name, object_metadata.status().message().c_str());
|
|
gst_buffer_unmap(buffer, &map);
|
|
g_free(object_name);
|
|
g_free(buffer_date);
|
|
return GST_FLOW_ERROR;
|
|
}
|
|
|
|
GST_INFO_OBJECT(sink, "Wrote object %s of size %" G_GUINT64_FORMAT "\n",
|
|
object_name, object_metadata->size());
|
|
|
|
gst_gs_sink_post_message(sink, buffer, object_name, buffer_date);
|
|
g_free(object_name);
|
|
g_free(buffer_date);
|
|
++sink->index;
|
|
break;
|
|
}
|
|
case GST_GS_SINK_NEXT_NONE: {
|
|
if (!sink->gcs_stream) {
|
|
GST_INFO_OBJECT(sink, "Opening %s", sink->object_name);
|
|
sink->gcs_stream = std::make_unique<GSWriteStream>(
|
|
*sink->gcs_client.get(), sink->bucket_name, sink->object_name,
|
|
metadata);
|
|
|
|
if (!sink->gcs_stream->stream().IsOpen()) {
|
|
GST_ELEMENT_ERROR(
|
|
sink, RESOURCE, OPEN_READ,
|
|
("Could not create write stream (%s)",
|
|
sink->gcs_stream->stream().last_status().message().c_str()),
|
|
GST_ERROR_SYSTEM);
|
|
gst_buffer_unmap(buffer, &map);
|
|
return GST_FLOW_OK;
|
|
}
|
|
}
|
|
|
|
GST_INFO_OBJECT(sink, "Writing %" G_GSIZE_FORMAT " bytes", map.size);
|
|
|
|
gcs::ObjectWriteStream& stream = sink->gcs_stream->stream();
|
|
stream.write(reinterpret_cast<const char*>(map.data), map.size);
|
|
if (stream.fail()) {
|
|
GST_WARNING_OBJECT(sink, "Failed to write to %s", object_name);
|
|
}
|
|
break;
|
|
}
|
|
default:
|
|
g_assert_not_reached();
|
|
}
|
|
|
|
gst_buffer_unmap(buffer, &map);
|
|
return GST_FLOW_OK;
|
|
}
|
|
|
|
static GstFlowReturn gst_gs_sink_render(GstBaseSink* bsink, GstBuffer* buffer) {
|
|
GstGsSink* sink = GST_GS_SINK(bsink);
|
|
GstFlowReturn flow = GST_FLOW_OK;
|
|
|
|
flow = gst_gs_sink_write_buffer(sink, buffer);
|
|
return flow;
|
|
}
|
|
|
|
static gboolean buffer_list_copy_data(GstBuffer** buf,
|
|
guint idx,
|
|
gpointer data) {
|
|
GstBuffer* dest = GST_BUFFER_CAST(data);
|
|
guint num, i;
|
|
|
|
if (idx == 0)
|
|
gst_buffer_copy_into(dest, *buf, GST_BUFFER_COPY_METADATA, 0, -1);
|
|
|
|
num = gst_buffer_n_memory(*buf);
|
|
for (i = 0; i < num; ++i) {
|
|
GstMemory* mem;
|
|
|
|
mem = gst_buffer_get_memory(*buf, i);
|
|
gst_buffer_append_memory(dest, mem);
|
|
}
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
/* Our assumption for now is that the buffers in a buffer list should always
|
|
* end up in the same file. If someone wants different behaviour, they'll just
|
|
* have to add a property for that. */
|
|
static GstFlowReturn gst_gs_sink_render_list(GstBaseSink* sink,
|
|
GstBufferList* list) {
|
|
GstBuffer* buf;
|
|
guint size;
|
|
|
|
size = gst_buffer_list_calculate_size(list);
|
|
GST_LOG_OBJECT(sink, "total size of buffer list %p: %u", list, size);
|
|
|
|
/* copy all buffers in the list into one single buffer, so we can use
|
|
* the normal render function (FIXME: optimise to avoid the memcpy) */
|
|
buf = gst_buffer_new();
|
|
gst_buffer_list_foreach(list, buffer_list_copy_data, buf);
|
|
g_assert(gst_buffer_get_size(buf) == size);
|
|
|
|
gst_gs_sink_render(sink, buf);
|
|
gst_buffer_unref(buf);
|
|
|
|
return GST_FLOW_OK;
|
|
}
|
|
|
|
static gboolean gst_gs_sink_set_caps(GstBaseSink* bsink, GstCaps* caps) {
|
|
GstGsSink* sink = GST_GS_SINK(bsink);
|
|
GstStructure* s = gst_caps_get_structure(caps, 0);
|
|
|
|
sink->content_type = gst_structure_get_name(s);
|
|
|
|
/* TODO: we could automatically convert some caps such as 'video/quicktime,variant=iso' to 'video/mp4' */
|
|
|
|
GST_INFO_OBJECT(sink, "Content-Type: caps: %s property: %s", sink->content_type, sink->content_type_prop);
|
|
|
|
return TRUE;
|
|
}
|
|
|
|
static gboolean gst_gs_sink_event(GstBaseSink* bsink, GstEvent* event) {
|
|
GstGsSink* sink = GST_GS_SINK(bsink);
|
|
|
|
switch (GST_EVENT_TYPE(event)) {
|
|
case GST_EVENT_EOS:
|
|
if (sink->gcs_stream) {
|
|
sink->gcs_stream = nullptr;
|
|
gst_gs_sink_post_message_from_time(
|
|
sink, GST_BASE_SINK(sink)->segment.position, -1, sink->object_name);
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
return GST_BASE_SINK_CLASS(parent_class)->event(bsink, event);
|
|
}
|