small rework for switchover handling

the source clienttrigger was not working in all cases, as the existing client was
getting replaced before the trigger via the format_init routine. This meant the
switchover failed to work properly and ended up crashing with null deref.

split the format_init routine into 2, wrapper and internal, are call the internal
one when going through the switchover route.

src/slave.c

@@ -1775,10 +1775,10 @@ static int relay_switchover (client_t *client, relay_server *relay, source_t *so
     source->flags &= ~SOURCE_SWITCHOVER;
     INFO1 ("Detected switch over to another client for %s", source->mount);
     source->client = (client_t *)client->aux_data;
+    DEBUG3 ("switchover client %p for %p on relay %p", client, source->client, relay);
     format_apply_client (source->format, source->client);
     source->linger_time = 0;
     thread_rwlock_unlock (&source->lock);
-    DEBUG2 ("switchover client %p on relay %p", client, relay);
     if (relay->flags & RELAY_IN_LIST)
     {   // source client on relay mount, so reset this relay for later use, this client goes
         DEBUG1 ("old client for relay %s going away", relay->localmount);
@@ -2130,7 +2130,6 @@ static int relay_startup (client_t *client)
 
         start_relay = source->listeners ? 1 : 0;
         source->flags |= SOURCE_ON_DEMAND;
-        thread_rwlock_unlock (&source->lock);
 
         mountinfo = config_lock_mount (NULL, source->mount);
 
@@ -2151,12 +2150,12 @@ static int relay_startup (client_t *client)
                 slave_update_mounts();
             }
             client->schedule_ms = (worker->time_ms + 5000);
+            thread_rwlock_unlock (&source->lock);
             return 0;
         }
         INFO1 ("starting on-demand relay %s", relay->localmount);
     }
-    else
-        thread_rwlock_unlock (&source->lock);
+    thread_rwlock_unlock (&source->lock);
 
     /* limit the number of relays starting up at the same time */
     thread_spin_lock (&relay_start_lock);

src/source.c

@@ -2988,7 +2988,7 @@ static int source_client_http_send (client_t *client)
 }
 
 
-int source_format_init (source_t *source, client_t *client)
+static int _source_format_init (source_t *source, client_t *client)
 {
     format_plugin_t *format = source->format;
 
@@ -3024,9 +3024,17 @@ int source_format_init (source_t *source, client_t *client)
             return -1;
         }
     }
+    return 0;
+}
+
+
+int source_format_init (source_t *source, client_t *client)
+{
+    format_plugin_t *format = source->format;
+    int ret = _source_format_init (source, client);
     source->client = client;
     format_apply_client (format, client);
-    return 0;
+    return ret;
 }
 
 
@@ -3147,7 +3155,7 @@ static int  source_client_startup (client_t *client)
                 if (rc == 1) source_free_source (source);
                 return client_send_403 (client, "too many streams connected");
             }
-            if (source_format_init (source, client) < 0)
+            if (rc == 1 && source_format_init (source, client) < 0)
             {
                 global_unlock();
                 thread_rwlock_unlock (&source->lock);
@@ -3163,14 +3171,15 @@ static int  source_client_startup (client_t *client)
         }
         if (rc == 2 && (hijack || source->linger_time))
         {
+            _source_format_init (source, client);
             client_t *sc = source->client;
             if (sc->connection.discon.time)
                 sc->connection.discon.time = 0;
-            DEBUG2 ("client %p flag for switchover on %s", client, source->mount);
-            source->flags |= SOURCE_SWITCHOVER;
+            DEBUG3 ("old %p, new %p,  flagged for switchover on %s", sc, client, source->mount);
             client->queue_pos = sc->queue_pos;
             source->format->parser = client->parser;
             sc->aux_data = (uintptr_t)client;  // ask the original client to make us live on source
+            source->flags |= SOURCE_SWITCHOVER;
 
             worker_t *worker = sc->worker;
             thread_spin_lock (&worker->lock);