functions to allow for reserving a source_t with a mountpoint

svn path=/trunk/icecast/; revision=5844

src/connection.c

@@ -426,9 +426,99 @@ void connection_inject_event(int eventnum, void *event_data) {
     _signal_pool();
 }
 
-/* TODO: Make this return an appropriate error code so that we can use HTTP
- * codes where appropriate
+
+/* Called when activating a source. Verifies that the source count is not
+ * exceeded and applies any initial parameters.
  */
+int connection_complete_source (source_t *source)
+{
+    ice_config_t *config = config_get_config();
+
+    global_lock ();
+    DEBUG1 ("sources count is %d", global.sources);
+
+    if (global.sources < config->source_limit)
+    {
+        char *contenttype;
+        mount_proxy *mountproxy = config->mounts;
+        format_type_t format_type;
+
+        /* setup format handler */
+        contenttype = httpp_getvar (source->parser, "content-type");
+        if (contenttype != NULL)
+        {
+            format_type = format_get_type (contenttype);
+
+            if (format_type == FORMAT_ERROR)
+            {
+                global_unlock();
+                config_release_config();
+                if (source->client)
+                    client_send_404 (source->client, "Content-type not supported");
+                WARN1("Content-type \"%s\" not supported, dropping source", contenttype);
+                return -1;
+            }
+        }
+        else
+        {
+            WARN0("No content-type header, falling back to backwards compatibility mode"
+                    "for icecast 1.x relays. Assuming content is mp3.");
+            format_type = FORMAT_TYPE_MP3;
+        }
+        source->format = format_get_plugin (format_type, source->mount, source->parser);
+
+        if (source->format == NULL)
+        {
+            global_unlock();
+            config_release_config();
+            if (source->client)
+                client_send_404 (source->client, "internal format allocation problem");
+            WARN1 ("plugin format failed for \"%s\"", source->mount);
+            return -1;
+        }
+
+        global.sources++;
+        global_unlock();
+
+        /* for relays, we don't yet have a client, however we do require one
+         * to retrieve the stream from.  This is created here, quite late,
+         * because we can't use this client to return an error code/message,
+         * so we only do this once we know we're going to accept the source.
+         */
+        if (source->client == NULL)
+            source->client = client_create (source->con, source->parser);
+
+        sock_set_blocking (source->con->sock, SOCK_NONBLOCK);
+
+        while (mountproxy)
+        {
+            if (strcmp (mountproxy->mountname, source->mount) == 0)
+            {
+                source_apply_mount (source, mountproxy);
+                break;
+            }
+            mountproxy = mountproxy->next;
+        }
+        config_release_config();
+
+        source->shutdown_rwlock = &_source_shutdown_rwlock;
+        DEBUG0 ("source is ready to start");
+
+        return 0;
+    }
+    WARN1("Request to add source when maximum source limit"
+            "reached %d", global.sources);
+
+    global_unlock();
+    config_release_config();
+
+    if (source->client)
+        client_send_404 (source->client, "too many sources connected");
+
+    return -1;
+}
+
+
 int connection_create_source(client_t *client, connection_t *con, http_parser_t *parser, char *mount) {
     source_t *source;
     char *contenttype;

src/connection.h

@@ -21,6 +21,7 @@
 #include "net/sock.h"
 
 struct _client_tag;
+struct source_tag;
 
 typedef struct connection_tag
 {
@@ -49,6 +50,7 @@ void connection_close(connection_t *con);
 connection_t *create_connection(sock_t sock, sock_t serversock, char *ip);
 int connection_create_source(struct _client_tag *client, connection_t *con, 
         http_parser_t *parser, char *mount);
+int connection_complete_source (struct source_tag *source);
 
 void connection_inject_event(int eventnum, void *event_data);
 

src/source.c

@@ -63,6 +63,44 @@ static int _compare_clients(void *compare_arg, void *a, void *b);
 static int _free_client(void *key);
 static int _parse_audio_info(source_t *source, char *s);
 
+/* Allocate a new source with the stated mountpoint, if one already
+ * exists with that mountpoint in the global source tree then return
+ * NULL.
+ */
+source_t *source_reserve (const char *mount)
+{
+    source_t *src = NULL;
+
+    do
+    {
+        avl_tree_wlock (global.source_tree);
+        src = source_find_mount_raw (mount);
+        if (src)
+        {
+            src = NULL;
+            break;
+        }
+
+        src = calloc (1, sizeof(source_t));
+        if (src == NULL)
+            break;
+
+        src->client_tree = avl_tree_new(_compare_clients, NULL);
+        src->pending_tree = avl_tree_new(_compare_clients, NULL);
+
+        /* make duplicates for strings or similar */
+        src->mount = strdup (mount);
+        src->max_listeners = -1;
+
+        avl_insert (global.source_tree, src);
+
+    } while (0);
+
+    avl_tree_unlock (global.source_tree);
+    return src;
+}
+
+
 source_t *source_create(client_t *client, connection_t *con, 
     http_parser_t *parser, const char *mount, format_type_t type, 
     mount_proxy *mountinfo)
@@ -204,6 +242,62 @@ int source_compare_sources(void *arg, void *a, void *b)
     return strcmp(srca->mount, srcb->mount);
 }
 
+
+void source_clear_source (source_t *source)
+{
+#ifdef USE_YP
+     int i;
+#endif
+    DEBUG1 ("clearing source \"%s\"", source->mount);
+    client_destroy(source->client);
+    source->client = NULL;
+
+    /* lets kick off any clients that are left on here */
+    avl_tree_rlock (source->client_tree);
+    while (avl_get_first (source->client_tree))
+    {
+        avl_delete (source->client_tree,
+                avl_get_first (source->client_tree)->key, _free_client);
+    }
+    avl_tree_unlock (source->client_tree);
+
+    avl_tree_rlock (source->pending_tree);
+    while (avl_get_first (source->pending_tree))
+    {
+        avl_delete (source->pending_tree,
+                avl_get_first(source->pending_tree)->key, _free_client);
+    }
+    avl_tree_unlock (source->pending_tree);
+
+    if (source->format && source->format->free_plugin)
+    {
+        source->format->free_plugin (source->format);
+    }
+    source->format = NULL;
+#ifdef USE_YP
+    for (i=0; i<source->num_yp_directories; i++)
+    {
+        yp_destroy_ypdata(source->ypdata[i]);
+        source->ypdata[i] = NULL;
+    }
+    source->num_yp_directories = 0;
+#endif
+    source->listeners = 0;
+    source->no_mount = 0;
+    source->max_listeners = -1;
+    source->yp_public = 0;
+
+    util_dict_free(source->audio_info);
+    source->audio_info = NULL;
+
+    free(source->fallback_mount);
+    source->fallback_mount = NULL;
+
+    free(source->dumpfilename);
+    source->dumpfilename = NULL;
+}
+
+
 int source_free_source(void *key)
 {
     source_t *source = key;
@@ -495,6 +589,7 @@ void *source_main(void *arg)
     }
 
     DEBUG0("Source creation complete");
+    source->running = 1;
 
     /*
     ** Now, if we have a fallback source and override is on, we want
@@ -857,3 +952,39 @@ static int _parse_audio_info(source_t *source, char *s)
     }
     return 1;
 }
+
+
+void source_apply_mount (source_t *source, mount_proxy *mountinfo)
+{
+    DEBUG1("Applying mount information for \"%s\"", source->mount);
+    source->max_listeners = mountinfo->max_listeners;
+    source->fallback_override = mountinfo->fallback_override;
+    source->no_mount = mountinfo->no_mount;
+    if (mountinfo->fallback_mount)
+    {
+        source->fallback_mount = strdup (mountinfo->fallback_mount);
+        DEBUG1 ("fallback %s", mountinfo->fallback_mount);
+    }
+    if (mountinfo->auth_type != NULL)
+    {
+        source->authenticator = auth_get_authenticator(
+                mountinfo->auth_type, mountinfo->auth_options);
+    }
+    if (mountinfo->dumpfile)
+    {
+        DEBUG1("Dumping stream to %s", mountinfo->dumpfile);
+        source->dumpfilename = strdup (mountinfo->dumpfile);
+    }
+}
+
+
+void *source_client_thread (void *arg)
+{
+    source_t *source = arg;
+
+    source->send_return = 1;
+    source_main (source);
+    source_free_source (source);
+    return NULL;
+}
+

src/source.h

@@ -62,6 +62,10 @@ typedef struct source_tag
 source_t *source_create(client_t *client, connection_t *con, 
         http_parser_t *parser, const char *mount, format_type_t type,
         mount_proxy *mountinfo);
+source_t *source_reserve (const char *mount);
+void *source_client_thread (void *arg);
+void source_apply_mount (source_t *source, mount_proxy *mountinfo);
+void source_clear_source (source_t *source);
 source_t *source_find_mount(const char *mount);
 source_t *source_find_mount_raw(const char *mount);
 client_t *source_find_client(source_t *source, int id);