mirror of
https://gitlab.freedesktop.org/gstreamer/gstreamer.git
synced 2024-11-29 21:21:12 +00:00
Merge commit 'http-launch/merge'
This commit is contained in:
Sebastian Dröge
commit
e9c4fd17a7
1 changed files with 548 additions and 0 deletions
548
network/http-launch/http-launch.c
Normal file
548
network/http-launch/http-launch.c
Normal file
|
|
@ -0,0 +1,548 @@
|
|||
/*
|
||||
* Copyright (C) 2013 Sebastian Dröge <slomo@circular-chaos.org>
|
||||
*
|
||||
* 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.
|
||||
*/
|
||||
#ifdef HAVE_CONFIG_H
|
||||
#include "config.h"
|
||||
#endif
|
||||
|
||||
#include <string.h>
|
||||
#include <gst/gst.h>
|
||||
#include <gio/gio.h>
|
||||
|
||||
typedef struct
|
||||
{
|
||||
gchar *name;
|
||||
GSocketConnection *connection;
|
||||
GSocket *socket;
|
||||
GInputStream *istream;
|
||||
GOutputStream *ostream;
|
||||
GSource *isource, *tosource;
|
||||
GByteArray *current_message;
|
||||
gchar *http_version;
|
||||
gboolean waiting_200_ok;
|
||||
} Client;
|
||||
|
||||
static const char *known_mimetypes[] = {
|
||||
"video/webm",
|
||||
"multipart/x-mixed-replace",
|
||||
NULL
|
||||
};
|
||||
|
||||
static GMainLoop *loop = NULL;
|
||||
G_LOCK_DEFINE_STATIC (clients);
|
||||
static GList *clients = NULL;
|
||||
static GstElement *pipeline = NULL;
|
||||
static GstElement *multisocketsink = NULL;
|
||||
static gboolean started = FALSE;
|
||||
static gchar *content_type;
|
||||
G_LOCK_DEFINE_STATIC (caps);
|
||||
static gboolean caps_resolved = FALSE;
|
||||
|
||||
static void
|
||||
remove_client (Client * client)
|
||||
{
|
||||
g_print ("Removing connection %s\n", client->name);
|
||||
|
||||
G_LOCK (clients);
|
||||
clients = g_list_remove (clients, client);
|
||||
G_UNLOCK (clients);
|
||||
|
||||
g_free (client->name);
|
||||
g_free (client->http_version);
|
||||
|
||||
if (client->isource) {
|
||||
g_source_destroy (client->isource);
|
||||
g_source_unref (client->isource);
|
||||
}
|
||||
if (client->tosource) {
|
||||
g_source_destroy (client->tosource);
|
||||
g_source_unref (client->tosource);
|
||||
}
|
||||
g_object_unref (client->connection);
|
||||
g_byte_array_unref (client->current_message);
|
||||
|
||||
g_slice_free (Client, client);
|
||||
}
|
||||
|
||||
static void
|
||||
write_bytes (Client * client, const gchar * data, guint len)
|
||||
{
|
||||
gssize w;
|
||||
GError *err = NULL;
|
||||
|
||||
/* TODO: We assume this never blocks */
|
||||
do {
|
||||
w = g_output_stream_write (client->ostream, data, len, NULL, &err);
|
||||
if (w > 0) {
|
||||
len -= w;
|
||||
data += w;
|
||||
}
|
||||
} while (w > 0 && len > 0);
|
||||
|
||||
if (w <= 0) {
|
||||
if (err) {
|
||||
g_print ("Write error %s\n", err->message);
|
||||
g_clear_error (&err);
|
||||
}
|
||||
remove_client (client);
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
send_response_200_ok (Client * client)
|
||||
{
|
||||
gchar *response;
|
||||
response = g_strdup_printf ("%s 200 OK\r\n%s\r\n", client->http_version,
|
||||
content_type);
|
||||
write_bytes (client, response, strlen (response));
|
||||
g_free (response);
|
||||
}
|
||||
|
||||
static void
|
||||
send_response_404_not_found (Client * client)
|
||||
{
|
||||
gchar *response;
|
||||
response = g_strdup_printf ("%s 404 Not Found\r\n\r\n", client->http_version);
|
||||
write_bytes (client, response, strlen (response));
|
||||
g_free (response);
|
||||
}
|
||||
|
||||
static void
|
||||
client_message (Client * client, const gchar * data, guint len)
|
||||
{
|
||||
gboolean http_head_request = FALSE;
|
||||
gboolean http_get_request = FALSE;
|
||||
gchar **lines = g_strsplit_set (data, "\r\n", -1);
|
||||
|
||||
if (g_str_has_prefix (lines[0], "HEAD"))
|
||||
http_head_request = TRUE;
|
||||
else if (g_str_has_prefix (lines[0], "GET"))
|
||||
http_get_request = TRUE;
|
||||
|
||||
if (http_head_request || http_get_request) {
|
||||
gchar **parts = g_strsplit (lines[0], " ", -1);
|
||||
gboolean ok = FALSE;
|
||||
|
||||
g_free (client->http_version);
|
||||
|
||||
if (parts[1] && parts[2] && *parts[2] != '\0')
|
||||
client->http_version = g_strdup (parts[2]);
|
||||
else
|
||||
client->http_version = g_strdup ("HTTP/1.0");
|
||||
|
||||
if (parts[1] && strcmp (parts[1], "/") == 0) {
|
||||
G_LOCK (caps);
|
||||
if (caps_resolved)
|
||||
send_response_200_ok (client);
|
||||
else
|
||||
client->waiting_200_ok = TRUE;
|
||||
G_UNLOCK (caps);
|
||||
ok = TRUE;
|
||||
} else {
|
||||
send_response_404_not_found (client);
|
||||
}
|
||||
g_strfreev (parts);
|
||||
|
||||
if (ok) {
|
||||
if (http_get_request) {
|
||||
/* Start streaming to client socket */
|
||||
g_source_destroy (client->isource);
|
||||
g_source_unref (client->isource);
|
||||
client->isource = NULL;
|
||||
g_source_destroy (client->tosource);
|
||||
g_source_unref (client->tosource);
|
||||
client->tosource = NULL;
|
||||
g_print ("Starting to stream to %s\n", client->name);
|
||||
g_signal_emit_by_name (multisocketsink, "add", client->socket);
|
||||
}
|
||||
|
||||
if (!started) {
|
||||
g_print ("Starting pipeline\n");
|
||||
if (gst_element_set_state (pipeline,
|
||||
GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) {
|
||||
g_print ("Failed to start pipeline\n");
|
||||
g_main_loop_quit (loop);
|
||||
}
|
||||
started = TRUE;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
gchar **parts = g_strsplit (lines[0], " ", -1);
|
||||
gchar *response;
|
||||
const gchar *http_version;
|
||||
|
||||
if (parts[1] && parts[2] && *parts[2] != '\0')
|
||||
http_version = parts[2];
|
||||
else
|
||||
http_version = "HTTP/1.0";
|
||||
|
||||
response = g_strdup_printf ("%s 400 Bad Request\r\n\r\n", http_version);
|
||||
write_bytes (client, response, strlen (response));
|
||||
g_free (response);
|
||||
g_strfreev (parts);
|
||||
remove_client (client);
|
||||
}
|
||||
|
||||
g_strfreev (lines);
|
||||
}
|
||||
|
||||
static gboolean
|
||||
on_timeout (Client * client)
|
||||
{
|
||||
g_print ("Timeout\n");
|
||||
remove_client (client);
|
||||
|
||||
return FALSE;
|
||||
}
|
||||
|
||||
static gboolean
|
||||
on_read_bytes (GPollableInputStream * stream, Client * client)
|
||||
{
|
||||
gssize r;
|
||||
gchar data[4096];
|
||||
GError *err = NULL;
|
||||
|
||||
do {
|
||||
r = g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM
|
||||
(client->istream), data, sizeof (data), NULL, &err);
|
||||
if (r > 0)
|
||||
g_byte_array_append (client->current_message, (guint8 *) data, r);
|
||||
} while (r > 0);
|
||||
|
||||
if (r == 0) {
|
||||
remove_client (client);
|
||||
return FALSE;
|
||||
} else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
|
||||
guint8 *tmp = client->current_message->data;
|
||||
guint tmp_len = client->current_message->len;
|
||||
|
||||
g_clear_error (&err);
|
||||
|
||||
while (tmp_len > 3) {
|
||||
if (tmp[0] == 0x0d && tmp[1] == 0x0a && tmp[2] == 0x0d && tmp[3] == 0x0a) {
|
||||
guint len;
|
||||
|
||||
g_byte_array_append (client->current_message, (const guint8 *) "\0", 1);
|
||||
len = tmp - client->current_message->data + 5;
|
||||
client_message (client, (gchar *) client->current_message->data, len);
|
||||
g_byte_array_remove_range (client->current_message, 0, len);
|
||||
tmp = client->current_message->data;
|
||||
tmp_len = client->current_message->len;
|
||||
} else {
|
||||
tmp++;
|
||||
tmp_len--;
|
||||
}
|
||||
}
|
||||
|
||||
if (client->current_message->len >= 1024 * 1024) {
|
||||
g_print ("No complete request after 1MB of data\n");
|
||||
remove_client (client);
|
||||
return FALSE;
|
||||
}
|
||||
|
||||
return TRUE;
|
||||
} else {
|
||||
g_print ("Read error %s\n", err->message);
|
||||
g_clear_error (&err);
|
||||
remove_client (client);
|
||||
return FALSE;
|
||||
}
|
||||
|
||||
return FALSE;
|
||||
}
|
||||
|
||||
static gboolean
|
||||
on_new_connection (GSocketService * service, GSocketConnection * connection,
|
||||
GObject * source_object, gpointer user_data)
|
||||
{
|
||||
Client *client = g_slice_new0 (Client);
|
||||
GSocketAddress *addr;
|
||||
GInetAddress *iaddr;
|
||||
gchar *ip;
|
||||
guint16 port;
|
||||
|
||||
addr = g_socket_connection_get_remote_address (connection, NULL);
|
||||
iaddr = g_inet_socket_address_get_address (G_INET_SOCKET_ADDRESS (addr));
|
||||
port = g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (addr));
|
||||
ip = g_inet_address_to_string (iaddr);
|
||||
client->name = g_strdup_printf ("%s:%u", ip, port);
|
||||
g_free (ip);
|
||||
g_object_unref (addr);
|
||||
|
||||
g_print ("New connection %s\n", client->name);
|
||||
|
||||
client->waiting_200_ok = FALSE;
|
||||
client->http_version = g_strdup ("");
|
||||
client->connection = g_object_ref (connection);
|
||||
client->socket = g_socket_connection_get_socket (connection);
|
||||
client->istream =
|
||||
g_io_stream_get_input_stream (G_IO_STREAM (client->connection));
|
||||
client->ostream =
|
||||
g_io_stream_get_output_stream (G_IO_STREAM (client->connection));
|
||||
client->current_message = g_byte_array_sized_new (1024);
|
||||
|
||||
client->tosource = g_timeout_source_new_seconds (5);
|
||||
g_source_set_callback (client->tosource, (GSourceFunc) on_timeout, client,
|
||||
NULL);
|
||||
g_source_attach (client->tosource, NULL);
|
||||
|
||||
client->isource =
|
||||
g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM
|
||||
(client->istream), NULL);
|
||||
g_source_set_callback (client->isource, (GSourceFunc) on_read_bytes, client,
|
||||
NULL);
|
||||
g_source_attach (client->isource, NULL);
|
||||
|
||||
G_LOCK (clients);
|
||||
clients = g_list_prepend (clients, client);
|
||||
G_UNLOCK (clients);
|
||||
|
||||
return TRUE;
|
||||
}
|
||||
|
||||
static gboolean
|
||||
on_message (GstBus * bus, GstMessage * message, gpointer user_data)
|
||||
{
|
||||
switch (GST_MESSAGE_TYPE (message)) {
|
||||
case GST_MESSAGE_ERROR:{
|
||||
gchar *debug;
|
||||
GError *err;
|
||||
|
||||
gst_message_parse_error (message, &err, &debug);
|
||||
g_print ("Error %s\n", err->message);
|
||||
g_error_free (err);
|
||||
g_free (debug);
|
||||
g_main_loop_quit (loop);
|
||||
break;
|
||||
}
|
||||
case GST_MESSAGE_WARNING:{
|
||||
gchar *debug;
|
||||
GError *err;
|
||||
|
||||
gst_message_parse_warning (message, &err, &debug);
|
||||
g_print ("Warning %s\n", err->message);
|
||||
g_error_free (err);
|
||||
g_free (debug);
|
||||
break;
|
||||
}
|
||||
case GST_MESSAGE_EOS:{
|
||||
g_print ("EOS\n");
|
||||
g_main_loop_quit (loop);
|
||||
}
|
||||
default:
|
||||
break;
|
||||
}
|
||||
|
||||
return TRUE;
|
||||
}
|
||||
|
||||
static void
|
||||
on_client_socket_removed (GstElement * element, GSocket * socket,
|
||||
gpointer user_data)
|
||||
{
|
||||
GList *l;
|
||||
Client *client = NULL;
|
||||
|
||||
G_LOCK (clients);
|
||||
for (l = clients; l; l = l->next) {
|
||||
Client *tmp = l->data;
|
||||
if (socket == tmp->socket) {
|
||||
client = tmp;
|
||||
break;
|
||||
}
|
||||
}
|
||||
G_UNLOCK (clients);
|
||||
|
||||
if (client)
|
||||
remove_client (client);
|
||||
}
|
||||
|
||||
static void on_stream_caps_changed (GObject *obj, GParamSpec *pspec,
|
||||
gpointer user_data)
|
||||
{
|
||||
GstPad *src_pad;
|
||||
GstCaps *src_caps;
|
||||
GstStructure *gstrc;
|
||||
GList *l;
|
||||
|
||||
src_pad = (GstPad *) obj;
|
||||
src_caps = gst_pad_get_current_caps (src_pad);
|
||||
gstrc = gst_caps_get_structure (src_caps, 0);
|
||||
|
||||
/*
|
||||
* Include a Content-type header in the case we know the mime
|
||||
* type is OK in HTTP. Required for MJPEG streams.
|
||||
*/
|
||||
int i = 0;
|
||||
const gchar *mimetype = gst_structure_get_name(gstrc);
|
||||
while (known_mimetypes[i] != NULL)
|
||||
{
|
||||
if (strcmp(mimetype, known_mimetypes[i]) == 0)
|
||||
{
|
||||
if (content_type)
|
||||
g_free(content_type);
|
||||
|
||||
/* Handle the (maybe not so) especial case of multipart to add boundary */
|
||||
if (strcmp(mimetype, "multipart/x-mixed-replace") == 0 &&
|
||||
gst_structure_has_field_typed(gstrc, "boundary", G_TYPE_STRING))
|
||||
{
|
||||
const gchar *boundary = gst_structure_get_string(gstrc, "boundary");
|
||||
content_type = g_strdup_printf ("Content-Type: "
|
||||
"multipart/x-mixed-replace;boundary=--%s\r\n", boundary);
|
||||
}
|
||||
else
|
||||
{
|
||||
content_type = g_strdup_printf ("Content-Type: %s\r\n", mimetype);
|
||||
}
|
||||
g_print("%s", content_type);
|
||||
break;
|
||||
}
|
||||
i++;
|
||||
}
|
||||
|
||||
gst_caps_unref (src_caps);
|
||||
|
||||
/* Send 200 OK to those clients waiting for it */
|
||||
G_LOCK (caps);
|
||||
|
||||
caps_resolved = TRUE;
|
||||
|
||||
G_LOCK (clients);
|
||||
for (l = clients; l; l = l->next) {
|
||||
Client *cl = l->data;
|
||||
if (cl->waiting_200_ok) {
|
||||
send_response_200_ok (cl);
|
||||
cl->waiting_200_ok = FALSE;
|
||||
break;
|
||||
}
|
||||
}
|
||||
G_UNLOCK (clients);
|
||||
|
||||
G_UNLOCK (caps);
|
||||
}
|
||||
|
||||
int
|
||||
main (gint argc, gchar ** argv)
|
||||
{
|
||||
GSocketService *service;
|
||||
GstElement *bin, *stream;
|
||||
GstPad *srcpad, *ghostpad, *sinkpad;
|
||||
GError *err = NULL;
|
||||
GstBus *bus;
|
||||
|
||||
gst_init (&argc, &argv);
|
||||
|
||||
if (argc < 4) {
|
||||
g_print ("usage: %s PORT <launch line>\n"
|
||||
"example: %s 8080 ( videotestsrc ! theoraenc ! oggmux name=stream )\n",
|
||||
argv[0], argv[0]);
|
||||
return -1;
|
||||
}
|
||||
|
||||
const gchar *port_str = argv[1];
|
||||
const int port = (int) g_ascii_strtoll(port_str, NULL, 10);
|
||||
|
||||
bin = gst_parse_launchv ((const gchar **) argv + 2, &err);
|
||||
if (!bin) {
|
||||
g_print ("invalid pipeline: %s\n", err->message);
|
||||
g_clear_error (&err);
|
||||
return -2;
|
||||
}
|
||||
|
||||
stream = gst_bin_get_by_name (GST_BIN (bin), "stream");
|
||||
if (!stream) {
|
||||
g_print ("no element with name \"stream\" found\n");
|
||||
gst_object_unref (bin);
|
||||
return -3;
|
||||
}
|
||||
|
||||
srcpad = gst_element_get_static_pad (stream, "src");
|
||||
if (!srcpad) {
|
||||
g_print ("no \"src\" pad in element \"stream\" found\n");
|
||||
gst_object_unref (stream);
|
||||
gst_object_unref (bin);
|
||||
return -4;
|
||||
}
|
||||
|
||||
content_type = g_strdup ("");
|
||||
g_signal_connect (srcpad, "notify::caps",
|
||||
G_CALLBACK (on_stream_caps_changed),
|
||||
NULL);
|
||||
|
||||
ghostpad = gst_ghost_pad_new ("src", srcpad);
|
||||
gst_element_add_pad (GST_ELEMENT (bin), ghostpad);
|
||||
gst_object_unref (srcpad);
|
||||
|
||||
pipeline = gst_pipeline_new (NULL);
|
||||
|
||||
multisocketsink = gst_element_factory_make ("multisocketsink", NULL);
|
||||
g_object_set (multisocketsink,
|
||||
"unit-format", GST_FORMAT_TIME,
|
||||
"units-max", (gint64) 7 * GST_SECOND,
|
||||
"units-soft-max", (gint64) 3 * GST_SECOND,
|
||||
"recover-policy", 3 /* keyframe */ ,
|
||||
"timeout", (guint64) 10 * GST_SECOND,
|
||||
"sync-method", 1 /* next-keyframe */ ,
|
||||
NULL);
|
||||
|
||||
gst_bin_add_many (GST_BIN (pipeline), bin, multisocketsink, NULL);
|
||||
|
||||
sinkpad = gst_element_get_static_pad (multisocketsink, "sink");
|
||||
gst_pad_link (ghostpad, sinkpad);
|
||||
gst_object_unref (sinkpad);
|
||||
|
||||
bus = gst_element_get_bus (pipeline);
|
||||
gst_bus_add_signal_watch (bus);
|
||||
g_signal_connect (bus, "message", G_CALLBACK (on_message), NULL);
|
||||
gst_object_unref (bus);
|
||||
|
||||
g_signal_connect (multisocketsink, "client-socket-removed",
|
||||
G_CALLBACK (on_client_socket_removed), NULL);
|
||||
|
||||
loop = g_main_loop_new (NULL, FALSE);
|
||||
|
||||
if (gst_element_set_state (pipeline,
|
||||
GST_STATE_READY) == GST_STATE_CHANGE_FAILURE) {
|
||||
gst_object_unref (pipeline);
|
||||
g_main_loop_unref (loop);
|
||||
g_print ("Failed to set pipeline to ready\n");
|
||||
return -5;
|
||||
}
|
||||
|
||||
service = g_socket_service_new ();
|
||||
g_socket_listener_add_inet_port (G_SOCKET_LISTENER (service), port, NULL,
|
||||
NULL);
|
||||
|
||||
g_signal_connect (service, "incoming", G_CALLBACK (on_new_connection), NULL);
|
||||
|
||||
g_socket_service_start (service);
|
||||
|
||||
g_print ("Listening on http://127.0.0.1:%d/\n", port);
|
||||
|
||||
g_main_loop_run (loop);
|
||||
|
||||
g_socket_service_stop (service);
|
||||
g_object_unref (service);
|
||||
|
||||
gst_element_set_state (pipeline, GST_STATE_NULL);
|
||||
gst_object_unref (pipeline);
|
||||
|
||||
g_main_loop_unref (loop);
|
||||
|
||||
return 0;
|
||||
}
|
||||
Loading…
Reference in a new issue