Better source telnet commands.

src/core/clock.ml

@@ -259,9 +259,9 @@ let unify =
       (Queue.push clock'.pending_activations);
     Queue.flush_iter clock.sub_clocks (Queue.push clock'.sub_clocks);
     Queue.flush_iter clock.on_error (Queue.push clock'.on_error);
-    Queue.filter clocks (fun el -> el != c);
     Unifier.(clock.id <-- clock'.id);
-    Unifier.(c <-- c')
+    Unifier.(c <-- c');
+    Queue.filter clocks (fun el -> sync el <> `Passive && el != c)
   in
   fun ~pos c c' ->
     let _c = Unifier.deref c in

src/core/lang_source.ml

@@ -188,6 +188,37 @@ let source_methods =
       fun s ->
         val_fun [] (fun _ ->
             match s#last_metadata with None -> null | Some m -> metadata m) );
+    ( "register_command",
+      ( [],
+        fun_t
+          [
+            (true, "usage", Lang.nullable_t Lang.string_t);
+            (false, "description", Lang.string_t);
+            (false, "", Lang.string_t);
+            (false, "", Lang.fun_t [(false, "", Lang.string_t)] Lang.string_t);
+          ]
+          unit_t ),
+      "Register a server command for this source. Command is registered under \
+       the source's id namespace when it gets up and de-registered when it \
+       gets down.",
+      fun s ->
+        val_fun
+          [
+            ("usage", "usage", Some Lang.null);
+            ("description", "description", None);
+            ("", "", None);
+            ("", "", None);
+          ]
+          (fun p ->
+            let usage =
+              Lang.to_valued_option Lang.to_string (List.assoc "usage" p)
+            in
+            let descr = Lang.to_string (List.assoc "description" p) in
+            let command = Lang.to_string (Lang.assoc "" 1 p) in
+            let f = Lang.assoc "" 2 p in
+            let f x = Lang.to_string (Lang.apply f [("", Lang.string x)]) in
+            s#register_command ?usage ~descr command f;
+            unit) );
     ( "on_metadata",
       ([], fun_t [(false, "", fun_t [(false, "", metadata_t)] unit_t)] unit_t),
       "Call a given handler on metadata packets.",

src/core/source.ml

@@ -24,6 +24,8 @@ open Mm
 
 exception Unavailable
 
+module Queue = Liquidsoap_lang.Queues.Queue
+
 type streaming_state =
   [ `Pending | `Unavailable | `Ready of unit -> unit | `Done of Frame.t ]
 
@@ -65,8 +67,8 @@ type watcher = {
 
 let source_log = Log.make ["source"]
 
-let sleep s =
-  source_log#info "Source %s gets down." s#id;
+let finalise s =
+  source_log#debug "Source %s is collected." s#id;
   try s#sleep
   with e ->
     let bt = Printexc.get_backtrace () in
@@ -132,6 +134,18 @@ class virtual operator ?(stack = []) ?clock ?(name = "src") sources =
 
     method virtual fallible : bool
     method source_type : source_type = `Passive
+    val mutable registered_commands = Queue.create ()
+
+    method register_command ?usage ~descr name cmd =
+      self#on_wake_up (fun () ->
+          let ns = [self#id] in
+          Server.add ~ns ?usage ~descr name cmd;
+          Queue.push registered_commands (ns, name))
+
+    initializer
+      self#on_sleep (fun () ->
+          Queue.flush_iter registered_commands (fun (ns, name) ->
+              Server.remove ~ns name))
 
     method active =
       match self#source_type with
@@ -246,7 +260,7 @@ class virtual operator ?(stack = []) ?clock ?(name = "src") sources =
         | _ -> self#force_sleep
 
     initializer
-      Gc.finalise sleep self;
+      Gc.finalise finalise self;
       self#on_sleep (fun () -> self#iter_watchers (fun w -> w.sleep ()))
 
     (** Streaming *)

src/core/source.mli

@@ -219,6 +219,11 @@ class virtual source :
        method reset_last_metadata_on_track : bool
        method set_reset_last_metadata_on_track : bool -> unit
 
+       (** Register a server command. The command is registered when the source
+           wakes up under its own id as namespace and deregistered when it goes down. *)
+       method register_command :
+         ?usage:string -> descr:string -> string -> (string -> string) -> unit
+
        (** Register a callback to be called on new metadata *)
        method on_metadata : (Frame.metadata -> unit) -> unit
 

src/libs/extra/audio.liq

@@ -401,8 +401,7 @@ def mix(~id=null(), ~register_server_commands=true, sources) =
        remaining=#{source.remaining(input.source)}"
     end
 
-    server.register(
-      namespace=source.id(s),
+    s.register_command(
       description=
         "Skip current track on all enabled sources.",
       "skip",
@@ -418,8 +417,7 @@ def mix(~id=null(), ~register_server_commands=true, sources) =
         end
     )
 
-    server.register(
-      namespace=source.id(s),
+    s.register_command(
       description=
         "Set volume for a given source.",
       usage=
@@ -438,8 +436,7 @@ def mix(~id=null(), ~register_server_commands=true, sources) =
         end
     )
 
-    server.register(
-      namespace=source.id(s),
+    s.register_command(
       description=
         "Enable/disable a source.",
       usage=
@@ -458,8 +455,7 @@ def mix(~id=null(), ~register_server_commands=true, sources) =
         end
     )
 
-    server.register(
-      namespace=source.id(s),
+    s.register_command(
       description=
         "Enable/disable automatic stop at the end of track.",
       usage=
@@ -478,8 +474,7 @@ def mix(~id=null(), ~register_server_commands=true, sources) =
         end
     )
 
-    server.register(
-      namespace=source.id(s),
+    s.register_command(
       description=
         "Display current status.",
       "status",
@@ -493,8 +488,7 @@ def mix(~id=null(), ~register_server_commands=true, sources) =
         end
     )
 
-    server.register(
-      namespace=source.id(s),
+    s.register_command(
       description=
         "Print the list of input sources.",
       "inputs",

src/libs/extra/server.liq

@@ -10,19 +10,12 @@ def server.rms(~id=null(), s) =
     "#{rms}"
   end
 
-  s.on_wake_up(
-    memoize(
-      {
-        server.register(
-          namespace="#{source.id(s)}",
-          description=
-            "Return the current RMS of the source.",
-          usage="rms",
-          "rms",
-          rms
-        )
-      }
-    )
+  s.register_command(
+    description=
+      "Return the current RMS of the source.",
+    usage="rms",
+    "rms",
+    rms
   )
 
   s
@@ -50,20 +43,13 @@ def server.insert_metadata(~id=null(), s) =
     end
   end
 
-  s.on_wake_up(
-    memoize(
-      {
-        server.register(
-          namespace="#{source.id(s)}",
-          description=
-            "Insert a metadata chunk.",
-          usage=
-            "insert key1=\"val1\",key2=\"val2\",..",
-          "insert",
-          insert
-        )
-      }
-    )
+  s.register_command(
+    description=
+      "Insert a metadata chunk.",
+    usage=
+      "insert key1=\"val1\",key2=\"val2\",..",
+    "insert",
+    insert
   )
 
   s