merge in fallback to file, override also works

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

doc/icecast2_config_file.html

@@ -434,6 +434,17 @@ mountpoint that is just not available, then those clients will be disconnected.
 If clients are falling back to a mountpoint and the fallback-mount is not actively streaming
 but defines a fallback-mount itself then those clients may be moved there instead.
 This multi-level fallback allows clients to cascade several mountpoints. 
+    <p>A fallback mount can also state a file that is located in webroot.  This is useful for
+    playing a pre-recorded file in the case of a stream going down. It will repeat until either
+    the listener disconnects or a stream comes back available and takes the listeners back.
+    As per usual, the file format should match the stream format, failing to do so may cause
+    problems with playback.
+    </p>
+    <p>Note that the fallback file is not timed so be careful if you intend to relay this.
+    They are fine on slave streams but don't use them on master streams, if you do then the
+    relay will consume stream data at a faster rate and the listeners on the relay would
+    eventually get kicked off.
+    </p>
 </div>
 <h4>fallback-override</h4>
 <div class="indentedbox">

src/client.c

@@ -75,10 +75,11 @@ void client_destroy(client_t *client)
     /* write log entry if ip is set (some things don't set it, like outgoing 
      * slave requests
      */
-    if(client->con->ip)
+    if (client->con && client->con->ip)
         logging_access(client);
     
-    connection_close(client->con);
+    if (client->con)
+        connection_close(client->con);
     httpp_destroy(client->parser);
 
     global_lock ();

src/format.c

@@ -150,8 +150,16 @@ int format_check_file_buffer (source_t *source, client_t *client)
     if (refbuf == NULL)
     {
         /* client refers to no data, must be from a move */
-        find_client_start (source, client);
+        if (source->client->con)
-        return -1;
+        {
+            find_client_start (source, client);
+            return -1;
+        }
+        /* source -> file fallback, need a refbuf for data */
+        refbuf = refbuf_new (4096);
+        client->refbuf = refbuf;
+        client->pos = refbuf->len;
+        client->intro_offset = 0;
     }
     if (client->pos == refbuf->len)
     {

src/fserve.c

@@ -317,7 +317,7 @@ static void *fserv_thread_function(void *arg)
     return NULL;
 }
 
-static const char *fserve_content_type(const char *path)
+char *fserve_content_type (const char *path)
 {
     char *ext = util_get_extension(path);
     mime_type exttype = {ext, NULL};

src/fserve.h

@@ -32,6 +32,7 @@ typedef struct _fserve_t
 void fserve_initialize(void);
 void fserve_shutdown(void);
 int fserve_client_create(client_t *httpclient, const char *path);
+char *fserve_content_type (const char *path);
 
 
 #endif

src/source.c

@@ -48,6 +48,7 @@
 #include "util.h"
 #include "source.h"
 #include "format.h"
+#include "fserve.h"
 #include "auth.h"
 #include "os.h"
 
@@ -373,6 +374,8 @@ void source_move_clients (source_t *source, source_t *dest)
             {
                 client_set_queue (client, NULL);
                 client->check_buffer = format_check_file_buffer;
+                if (source->con == NULL)
+                    client->intro_offset = -1;
             }
 
             avl_insert (dest->pending_tree, (void *)client);
@@ -397,6 +400,8 @@ void source_move_clients (source_t *source, source_t *dest)
             {
                 client_set_queue (client, NULL);
                 client->check_buffer = format_check_file_buffer;
+                if (source->con == NULL)
+                    client->intro_offset = -1;
             }
             avl_insert (dest->pending_tree, (void *)client);
             count++;
@@ -435,10 +440,16 @@ static refbuf_t *get_next_buffer (source_t *source)
         delay = 0;
     while (global.running == ICE_RUNNING && source->running)
     {
-        int fds;
+        int fds = 0;
         time_t current = time (NULL);
 
-        fds = util_timed_wait_for_fd (source->con->sock, delay);
+        if (source->client->con)
+            fds = util_timed_wait_for_fd (source->con->sock, delay);
+        else
+        {
+            thread_sleep (delay*1000);
+            source->last_read = current;
+        }
 
         if (current >= source->client_stats_update)
         {
@@ -470,7 +481,7 @@ static refbuf_t *get_next_buffer (source_t *source)
         }
         source->last_read = current;
         refbuf = source->format->get_buffer (source);
-        if (source->client->con->error)
+        if (source->client->con && source->client->con->error)
         {
             INFO1 ("End of Stream %s", source->mount);
             source->running = 0;
@@ -585,7 +596,8 @@ static void source_init (source_t *source)
     stats_event_inc (NULL, "source_total_connections");
     stats_event (source->mount, "slow_listeners", "0");
 
-    sock_set_blocking (source->con->sock, SOCK_NONBLOCK);
+    if (source->client->con)
+        sock_set_blocking (source->con->sock, SOCK_NONBLOCK);
 
     DEBUG0("Source creation complete");
     source->last_read = time (NULL);
@@ -1095,6 +1107,9 @@ static void source_apply_mount (source_t *source, mount_proxy *mountinfo)
  */
 void source_update_settings (ice_config_t *config, source_t *source, mount_proxy *mountinfo)
 {
+    /*  skip if source is a fallback to file */
+    if (source->running && source->client->con == NULL)
+        return;
     /* set global settings first */
     source->queue_size_limit = config->queue_size_limit;
     source->timeout = config->source_timeout;
@@ -1147,26 +1162,29 @@ void source_update_settings (ice_config_t *config, source_t *source, mount_proxy
 void *source_client_thread (void *arg)
 {
     source_t *source = arg;
-    const char ok_msg[] = "HTTP/1.0 200 OK\r\n\r\n";
-    int bytes;
-    const char *agent;
 
-    source->client->respcode = 200;
+    if (source->client && source->client->con)
-    bytes = sock_write_bytes (source->client->con->sock, ok_msg, sizeof (ok_msg)-1);
-    if (bytes < (int)(sizeof (ok_msg)-1))
     {
-        global_lock();
+        const char ok_msg[] = "HTTP/1.0 200 OK\r\n\r\n";
-        global.sources--;
+        int bytes;
-        global_unlock();
+        const char *agent;
-        WARN0 ("Error writing 200 OK message to source client");
-        source_free_source (source);
-        return NULL;
-    }
-    stats_event (source->mount, "source_ip", source->client->con->ip);
-    agent = httpp_getvar (source->client->parser, "user-agent");
-    if (agent)
-        stats_event (source->mount, "user_agent", agent);
 
+        source->client->respcode = 200;
+        bytes = sock_write_bytes (source->client->con->sock, ok_msg, sizeof (ok_msg)-1);
+        if (bytes < (int)(sizeof (ok_msg)-1))
+        {
+            global_lock();
+            global.sources--;
+            global_unlock();
+            WARN0 ("Error writing 200 OK message to source client");
+            source_free_source (source);
+            return NULL;
+        }
+        stats_event (source->mount, "source_ip", source->client->con->ip);
+        agent = httpp_getvar (source->client->parser, "user-agent");
+        if (agent)
+            stats_event (source->mount, "user_agent", agent);
+    }
     stats_event_inc(NULL, "source_client_connections");
     stats_event (source->mount, "listeners", "0");
 
@@ -1214,6 +1232,67 @@ static void source_run_script (char *command, char *mountpoint)
 #endif
 
 
+static void *source_fallback_file (void *arg)
+{
+    char *mount = arg;
+    char *type;
+    char *path;
+    unsigned int len;
+    FILE *file = NULL;
+    source_t *source = NULL;
+    ice_config_t *config;
+    http_parser_t *parser;
+
+    do
+    {
+        if (mount == NULL || mount[0] != '/')
+            break;
+        config = config_get_config();
+        len  = strlen (config->webroot_dir) + strlen (mount) + 1;
+        path = malloc (len);
+        if (path)
+            snprintf (path, len, "%s%s", config->webroot_dir, mount);
+        
+        config_release_config ();
+        if (path == NULL)
+            break;
+
+        file = fopen (path, "rb");
+        if (file == NULL)
+        {
+            DEBUG1 ("unable to open file \"%s\"", path);
+            free (path);
+            break;
+        }
+        free (path);
+        source = source_reserve (mount);
+        if (source == NULL)
+        {
+            DEBUG1 ("mountpoint \"%s\" already reserved", mount);
+            break;
+        }
+        type = fserve_content_type (mount);
+        parser = httpp_create_parser();
+        httpp_initialize (parser, NULL);
+        httpp_setvar (parser, "content-type", type);
+
+        source->hidden = 1;
+        source->yp_public = 0;
+        source->intro_file = file;
+        source->parser = parser;
+        file = NULL;
+
+        if (connection_complete_source (source) < 0)
+            break;
+        source_client_thread (source);
+    } while (0);
+    if (file)
+        fclose (file);
+    free (mount);
+    return NULL;
+}
+
+
 /* rescan the mount list, so that xsl files are updated to show
  * unconnected but active fallback mountpoints
  */
@@ -1249,6 +1328,17 @@ void source_recheck_mounts (void)
         else
             stats_event (mount->mountname, NULL, NULL);
 
+        /* check for fallback to file */
+        if (global.running == ICE_RUNNING && mount->fallback_mount)
+        {
+            source_t *fallback = source_find_mount (mount->fallback_mount);
+            if (fallback == NULL)
+            {
+                thread_create ("Fallback file thread", source_fallback_file,
+                        strdup (mount->fallback_mount), THREAD_DETACHED);
+            }
+        }
+
         mount = mount->next;
     }
     avl_tree_unlock (global.source_tree);