media: add lock to protect state changes
[platform/upstream/gstreamer.git] / gst / rtsp-server / rtsp-server.c
index 81c59c7..0edf635 100644 (file)
@@ -13,8 +13,8 @@
  *
  * 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., 59 Temple Place - Suite 330,
- * Boston, MA 02111-1307, USA.
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
  */
 
 #include <stdlib.h>
 #include "rtsp-client.h"
 
 #define DEFAULT_ADDRESS         "0.0.0.0"
+#define DEFAULT_BOUND_PORT      -1
 /* #define DEFAULT_ADDRESS         "::0" */
 #define DEFAULT_SERVICE         "8554"
 #define DEFAULT_BACKLOG         5
+#define DEFAULT_MAX_THREADS     0
 
 /* Define to use the SO_LINGER option so that the server sockets can be resused
  * sooner. Disabled for now because it is not very well implemented by various
@@ -38,10 +40,12 @@ enum
   PROP_0,
   PROP_ADDRESS,
   PROP_SERVICE,
+  PROP_BOUND_PORT,
   PROP_BACKLOG,
 
   PROP_SESSION_POOL,
   PROP_MEDIA_MAPPING,
+  PROP_MAX_THREADS,
   PROP_LAST
 };
 
@@ -56,6 +60,8 @@ G_DEFINE_TYPE (GstRTSPServer, gst_rtsp_server, G_TYPE_OBJECT);
 GST_DEBUG_CATEGORY_STATIC (rtsp_server_debug);
 #define GST_CAT_DEFAULT rtsp_server_debug
 
+typedef struct _ClientContext ClientContext;
+
 static guint gst_rtsp_server_signals[SIGNAL_LAST] = { 0 };
 
 static void gst_rtsp_server_get_property (GObject * object, guint propid,
@@ -64,6 +70,7 @@ static void gst_rtsp_server_set_property (GObject * object, guint propid,
     const GValue * value, GParamSpec * pspec);
 static void gst_rtsp_server_finalize (GObject * object);
 
+static gpointer do_loop (ClientContext * ctx);
 static GstRTSPClient *default_create_client (GstRTSPServer * server);
 static gboolean default_accept_client (GstRTSPServer * server,
     GstRTSPClient * client, GSocket * socket, GError ** error);
@@ -80,7 +87,7 @@ gst_rtsp_server_class_init (GstRTSPServerClass * klass)
   gobject_class->finalize = gst_rtsp_server_finalize;
 
   /**
-   * GstRTSPServer::address
+   * GstRTSPServer::address:
    *
    * The address of the server. This is the address where the server will
    * listen on.
@@ -90,7 +97,7 @@ gst_rtsp_server_class_init (GstRTSPServerClass * klass)
           "The address the server uses to listen on", DEFAULT_ADDRESS,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
   /**
-   * GstRTSPServer::service
+   * GstRTSPServer::service:
    *
    * The service of the server. This is either a string with the service name or
    * a port number (as a string) the server will listen on.
@@ -100,7 +107,19 @@ gst_rtsp_server_class_init (GstRTSPServerClass * klass)
           "The service or port number the server uses to listen on",
           DEFAULT_SERVICE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
   /**
-   * GstRTSPServer::backlog
+   * GstRTSPServer::bound-port:
+   *
+   * The actual port the server is listening on. Can be used to retrieve the
+   * port number when the server is started on port 0, which means bind to a
+   * random port. Set to -1 if the server has not been bound yet.
+   */
+  g_object_class_install_property (gobject_class, PROP_BOUND_PORT,
+      g_param_spec_int ("bound-port", "Bound port",
+          "The port number the server is listening on",
+          -1, G_MAXUINT16, DEFAULT_BOUND_PORT,
+          G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
+  /**
+   * GstRTSPServer::backlog:
    *
    * The backlog argument defines the maximum length to which the queue of
    * pending connections for the server may grow. If a connection request arrives
@@ -114,7 +133,7 @@ gst_rtsp_server_class_init (GstRTSPServerClass * klass)
           "of pending connections may grow", 0, G_MAXINT, DEFAULT_BACKLOG,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
   /**
-   * GstRTSPServer::session-pool
+   * GstRTSPServer::session-pool:
    *
    * The session pool of the server. By default each server has a separate
    * session pool but sessions can be shared between servers by setting the same
@@ -126,7 +145,7 @@ gst_rtsp_server_class_init (GstRTSPServerClass * klass)
           GST_TYPE_RTSP_SESSION_POOL,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
   /**
-   * GstRTSPServer::media-mapping
+   * GstRTSPServer::media-mapping:
    *
    * The media mapping to use for this server. By default the server has no
    * media mapping and thus cannot map urls to media streams.
@@ -136,6 +155,18 @@ gst_rtsp_server_class_init (GstRTSPServerClass * klass)
           "The media mapping to use for client session",
           GST_TYPE_RTSP_MEDIA_MAPPING,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+  /**
+   * GstRTSPServer::max-threads:
+   *
+   * The maximum amount of threads to use for client connections. A value of
+   * 0 means to use only the mainloop, -1 means an unlimited amount of
+   * threads.
+   */
+  g_object_class_install_property (gobject_class, PROP_MAX_THREADS,
+      g_param_spec_int ("max-threads", "Max Threads",
+          "The maximum amount of threads to use for client connections "
+          "(0 = only mainloop, -1 = unlimited)", -1, G_MAXINT,
+          DEFAULT_MAX_THREADS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   gst_rtsp_server_signals[SIGNAL_CLIENT_CONNECTED] =
       g_signal_new ("client-connected", G_TYPE_FROM_CLASS (gobject_class),
@@ -146,6 +177,8 @@ gst_rtsp_server_class_init (GstRTSPServerClass * klass)
   klass->create_client = default_create_client;
   klass->accept_client = default_accept_client;
 
+  klass->pool = g_thread_pool_new ((GFunc) do_loop, klass, -1, FALSE, NULL);
+
   GST_DEBUG_CATEGORY_INIT (rtsp_server_debug, "rtspserver", 0, "GstRTSPServer");
 }
 
@@ -155,6 +188,7 @@ gst_rtsp_server_init (GstRTSPServer * server)
   g_mutex_init (&server->lock);
   server->address = g_strdup (DEFAULT_ADDRESS);
   server->service = g_strdup (DEFAULT_SERVICE);
+  server->socket = NULL;
   server->backlog = DEFAULT_BACKLOG;
   server->session_pool = gst_rtsp_session_pool_new ();
   server->media_mapping = gst_rtsp_media_mapping_new ();
@@ -169,6 +203,8 @@ gst_rtsp_server_finalize (GObject * object)
 
   g_free (server->address);
   g_free (server->service);
+  if (server->socket)
+    g_object_unref (server->socket);
 
   g_object_unref (server->session_pool);
   g_object_unref (server->media_mapping);
@@ -239,6 +275,36 @@ gst_rtsp_server_get_address (GstRTSPServer * server)
 }
 
 /**
+ * gst_rtsp_server_get_bound_port:
+ * @server: a #GstRTSPServer
+ *
+ * Get the port number where the server was bound to.
+ *
+ * Returns: the port number
+ */
+int
+gst_rtsp_server_get_bound_port (GstRTSPServer * server)
+{
+  GSocketAddress *address;
+  int result = -1;
+
+  g_return_val_if_fail (GST_IS_RTSP_SERVER (server), result);
+
+  GST_RTSP_SERVER_LOCK (server);
+  if (server->socket == NULL)
+    goto out;
+
+  address = g_socket_get_local_address (server->socket, NULL);
+  result = g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (address));
+  g_object_unref (address);
+
+out:
+  GST_RTSP_SERVER_UNLOCK (server);
+
+  return result;
+}
+
+/**
  * gst_rtsp_server_set_service:
  * @server: a #GstRTSPServer
  * @service: the service
@@ -358,7 +424,7 @@ gst_rtsp_server_set_session_pool (GstRTSPServer * server,
  *
  * Get the #GstRTSPSessionPool used as the session pool of @server.
  *
- * Returns: the #GstRTSPSessionPool used for sessions. g_object_unref() after
+ * Returns: (transfer full): the #GstRTSPSessionPool used for sessions. g_object_unref() after
  * usage.
  */
 GstRTSPSessionPool *
@@ -410,7 +476,7 @@ gst_rtsp_server_set_media_mapping (GstRTSPServer * server,
  *
  * Get the #GstRTSPMediaMapping used as the media mapping of @server.
  *
- * Returns: the #GstRTSPMediaMapping of @server. g_object_unref() after
+ * Returns: (transfer full): the #GstRTSPMediaMapping of @server. g_object_unref() after
  * usage.
  */
 GstRTSPMediaMapping *
@@ -461,7 +527,7 @@ gst_rtsp_server_set_auth (GstRTSPServer * server, GstRTSPAuth * auth)
  *
  * Get the #GstRTSPAuth used as the authentication manager of @server.
  *
- * Returns: the #GstRTSPAuth of @server. g_object_unref() after
+ * Returns: (transfer full): the #GstRTSPAuth of @server. g_object_unref() after
  * usage.
  */
 GstRTSPAuth *
@@ -479,6 +545,49 @@ gst_rtsp_server_get_auth (GstRTSPServer * server)
   return result;
 }
 
+/**
+ * gst_rtsp_server_set_max_threads:
+ * @server: a #GstRTSPServer
+ * @max_threads: maximum threads
+ *
+ * Set the maximum threads used by the server to handle client requests.
+ * A value of 0 will use the server mainloop, a value of -1 will use an
+ * unlimited number of threads.
+ */
+void
+gst_rtsp_server_set_max_threads (GstRTSPServer * server, gint max_threads)
+{
+  g_return_if_fail (GST_IS_RTSP_SERVER (server));
+
+  GST_RTSP_SERVER_LOCK (server);
+  server->max_threads = max_threads;
+  GST_RTSP_SERVER_UNLOCK (server);
+}
+
+/**
+ * gst_rtsp_server_get_max_threads:
+ * @server: a #GstRTSPServer
+ *
+ * Get the maximum number of threads used for client connections.
+ * See gst_rtsp_server_set_max_threads().
+ *
+ * Returns: the maximum number of threads.
+ */
+gint
+gst_rtsp_server_get_max_threads (GstRTSPServer * server)
+{
+  gint res;
+
+  g_return_val_if_fail (GST_IS_RTSP_SERVER (server), -1);
+
+  GST_RTSP_SERVER_LOCK (server);
+  res = server->max_threads;
+  GST_RTSP_SERVER_UNLOCK (server);
+
+  return res;
+}
+
+
 static void
 gst_rtsp_server_get_property (GObject * object, guint propid,
     GValue * value, GParamSpec * pspec)
@@ -492,6 +601,9 @@ gst_rtsp_server_get_property (GObject * object, guint propid,
     case PROP_SERVICE:
       g_value_take_string (value, gst_rtsp_server_get_service (server));
       break;
+    case PROP_BOUND_PORT:
+      g_value_set_int (value, gst_rtsp_server_get_bound_port (server));
+      break;
     case PROP_BACKLOG:
       g_value_set_int (value, gst_rtsp_server_get_backlog (server));
       break;
@@ -501,6 +613,9 @@ gst_rtsp_server_get_property (GObject * object, guint propid,
     case PROP_MEDIA_MAPPING:
       g_value_take_object (value, gst_rtsp_server_get_media_mapping (server));
       break;
+    case PROP_MAX_THREADS:
+      g_value_set_int (value, gst_rtsp_server_get_max_threads (server));
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
   }
@@ -528,6 +643,9 @@ gst_rtsp_server_set_property (GObject * object, guint propid,
     case PROP_MEDIA_MAPPING:
       gst_rtsp_server_set_media_mapping (server, g_value_get_object (value));
       break;
+    case PROP_MAX_THREADS:
+      gst_rtsp_server_set_max_threads (server, g_value_get_int (value));
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
   }
@@ -542,7 +660,7 @@ gst_rtsp_server_set_property (GObject * object, guint propid,
  * Create a #GSocket for @server. The socket will listen on the
  * configured service.
  *
- * Returns: the #GSocket for @server or NULL when an error occured.
+ * Returns: (transfer full): the #GSocket for @server or NULL when an error occured.
  */
 GSocket *
 gst_rtsp_server_create_socket (GstRTSPServer * server,
@@ -566,7 +684,7 @@ gst_rtsp_server_create_socket (GstRTSPServer * server,
 
   /* resolve the server IP address */
   port = atoi (server->service);
-  if (port != 0)
+  if (port != 0 || !strcmp (server->service, "0"))
     conn = g_network_address_new (server->address, port);
   else
     conn = g_network_service_new (server->service, "tcp", server->address);
@@ -699,34 +817,95 @@ close_error:
   }
 }
 
+struct _ClientContext
+{
+  GstRTSPServer *server;
+  GMainLoop *loop;
+  GMainContext *context;
+  GstRTSPClient *client;
+};
+
+static void
+free_client_context (ClientContext * ctx)
+{
+  g_main_context_unref (ctx->context);
+  if (ctx->loop)
+    g_main_loop_unref (ctx->loop);
+  g_object_unref (ctx->client);
+  g_slice_free (ClientContext, ctx);
+}
+
+static gpointer
+do_loop (ClientContext * ctx)
+{
+  GST_INFO ("enter mainloop");
+  g_main_loop_run (ctx->loop);
+  GST_INFO ("exit mainloop");
+
+  free_client_context (ctx);
+
+  return NULL;
+}
+
 static void
-unmanage_client (GstRTSPClient * client, GstRTSPServer * server)
+unmanage_client (GstRTSPClient * client, ClientContext * ctx)
 {
+  GstRTSPServer *server = ctx->server;
+
   GST_DEBUG_OBJECT (server, "unmanage client %p", client);
 
   g_object_ref (server);
   gst_rtsp_client_set_server (client, NULL);
 
   GST_RTSP_SERVER_LOCK (server);
-  server->clients = g_list_remove (server->clients, client);
+  server->clients = g_list_remove (server->clients, ctx);
   GST_RTSP_SERVER_UNLOCK (server);
-  g_object_unref (server);
 
-  g_object_unref (client);
+  if (ctx->loop)
+    g_main_loop_quit (ctx->loop);
+  else
+    free_client_context (ctx);
+
+  g_object_unref (server);
 }
 
-/* add the client to the active list of clients, takes ownership of
- * the client */
+/* add the client context to the active list of clients, takes ownership
+ * of client */
 static void
 manage_client (GstRTSPServer * server, GstRTSPClient * client)
 {
+  ClientContext *ctx;
+
   GST_DEBUG_OBJECT (server, "manage client %p", client);
   gst_rtsp_client_set_server (client, server);
 
+  ctx = g_slice_new0 (ClientContext);
+  ctx->server = server;
+  ctx->client = client;
+  if (server->max_threads == 0) {
+    GSource *source;
+
+    /* find the context to add the watch */
+    if ((source = g_main_current_source ()))
+      ctx->context = g_main_context_ref (g_source_get_context (source));
+    else
+      ctx->context = NULL;
+  } else {
+    ctx->context = g_main_context_new ();
+    ctx->loop = g_main_loop_new (ctx->context, TRUE);
+  }
+  gst_rtsp_client_attach (client, ctx->context);
+
   GST_RTSP_SERVER_LOCK (server);
-  g_signal_connect (client, "closed", (GCallback) unmanage_client, server);
-  server->clients = g_list_prepend (server->clients, client);
+  g_signal_connect (client, "closed", (GCallback) unmanage_client, ctx);
+  server->clients = g_list_prepend (server->clients, ctx);
   GST_RTSP_SERVER_UNLOCK (server);
+
+  if (ctx->loop) {
+    GstRTSPServerClass *klass = GST_RTSP_SERVER_GET_CLASS (server);
+
+    g_thread_pool_push (klass->pool, ctx, NULL);
+  }
 }
 
 static GstRTSPClient *
@@ -803,8 +982,8 @@ gst_rtsp_server_transfer_connection (GstRTSPServer * server, GSocket * socket,
     goto client_failed;
 
   /* a new client connected, create a client object to handle the client. */
-  if (!gst_rtsp_client_create_from_socket (client, socket, ip, port,
-          initial_buffer, &error)) {
+  if (!gst_rtsp_client_use_socket (client, socket, ip,
+          port, initial_buffer, &error)) {
     goto transfer_failed;
   }
 
@@ -826,7 +1005,7 @@ transfer_failed:
   {
     GST_ERROR_OBJECT (server, "failed to accept client: %s", error->message);
     g_error_free (error);
-    gst_object_unref (client);
+    g_object_unref (client);
     return FALSE;
   }
 }
@@ -835,6 +1014,7 @@ transfer_failed:
  * gst_rtsp_server_io_func:
  * @socket: a #GSocket
  * @condition: the condition on @source
+ * @server: a #GstRTSPServer
  *
  * A default #GSocketSourceFunc that creates a new #GstRTSPClient to accept and handle a
  * new connection on @socket or @server.
@@ -884,7 +1064,7 @@ accept_failed:
   {
     GST_ERROR_OBJECT (server, "failed to accept client: %s", error->message);
     g_error_free (error);
-    gst_object_unref (client);
+    g_object_unref (client);
     return FALSE;
   }
 }
@@ -917,7 +1097,7 @@ GSource *
 gst_rtsp_server_create_source (GstRTSPServer * server,
     GCancellable * cancellable, GError ** error)
 {
-  GSocket *socket;
+  GSocket *socket, *old;
   GSource *source;
 
   g_return_val_if_fail (GST_IS_RTSP_SERVER (server), NULL);
@@ -926,6 +1106,14 @@ gst_rtsp_server_create_source (GstRTSPServer * server,
   if (socket == NULL)
     goto no_socket;
 
+  GST_RTSP_SERVER_LOCK (server);
+  old = server->socket;
+  server->socket = g_object_ref (socket);
+  GST_RTSP_SERVER_UNLOCK (server);
+
+  if (old)
+    g_object_unref (old);
+
   /* create a watch for reads (new connections) and possible errors */
   source = g_socket_create_source (socket, G_IO_IN |
       G_IO_ERR | G_IO_HUP | G_IO_NVAL, cancellable);
@@ -948,8 +1136,7 @@ no_socket:
 /**
  * gst_rtsp_server_attach:
  * @server: a #GstRTSPServer
- * @context: a #GMainContext
- * @error: a #GError
+ * @context: (allow-none): a #GMainContext
  *
  * Attaches @server to @context. When the mainloop for @context is run, the
  * server will be dispatched. When @context is NULL, the default context will be