Add utils/Watcher

src/Makefile

@@ -164,6 +164,7 @@ UTILS_LIB_INTFS := \
   utils/IO.mli \
   utils/moving_average.mli \
   utils/ring.mli \
+  utils/watcher.mli \
 
 UTILS_LIB_IMPLS := \
   utils/base58.ml \
@@ -181,6 +182,7 @@ UTILS_LIB_IMPLS := \
   utils/IO.ml \
   utils/moving_average.ml \
   utils/ring.ml \
+  utils/watcher.ml \
 
 UTILS_PACKAGES := \
   ${MINUTILS_PACKAGES} \

src/node/shell/node.mli

@@ -35,9 +35,9 @@ module RPC : sig
   val raw_block_info:
     t -> Block_hash.t -> block_info Lwt.t
   val block_watcher:
-    t -> block_info Lwt_stream.t * (unit -> unit)
+    t -> block_info Lwt_stream.t * Watcher.stopper
   val valid_block_watcher:
-    t -> (block_info Lwt_stream.t * (unit -> unit)) Lwt.t
+    t -> (block_info Lwt_stream.t * Watcher.stopper) Lwt.t
   val heads: t -> block_info Block_hash_map.t Lwt.t
 
   val list:
@@ -51,7 +51,7 @@ module RPC : sig
   val operation_content:
     t -> Operation_hash.t -> Store.operation tzresult Time.timed_data option Lwt.t
   val operation_watcher:
-    t -> (Operation_hash.t * Store.operation) Lwt_stream.t * (unit -> unit)
+    t -> (Operation_hash.t * Store.operation) Lwt_stream.t * Watcher.stopper
 
   val pending_operations:
     t -> block -> (error Updater.preapply_result * Operation_hash_set.t) Lwt.t
@@ -61,7 +61,7 @@ module RPC : sig
   val protocol_content:
     t -> Protocol_hash.t -> Store.protocol tzresult Time.timed_data option Lwt.t
   val protocol_watcher:
-    t -> (Protocol_hash.t * Store.protocol) Lwt_stream.t * (unit -> unit)
+    t -> (Protocol_hash.t * Store.protocol) Lwt_stream.t * Watcher.stopper
 
   val context_dir:
     t -> block -> 'a RPC.directory option Lwt.t

src/node/shell/node_rpc.ml

@@ -271,18 +271,16 @@ let list_blocks
         requested_blocks in
     RPC.Answer.return infos
   else begin
-    Node.RPC.valid_block_watcher node >>= fun (bi_stream, shutdown) ->
+    Node.RPC.valid_block_watcher node >>= fun (bi_stream, stopper) ->
-    let stream, shutdown =
+    let stream =
       match delay with
       | None ->
-          Lwt_stream.map (fun bi -> [[filter_bi include_ops bi]]) bi_stream,
+          Lwt_stream.map (fun bi -> [[filter_bi include_ops bi]]) bi_stream
-          shutdown
       | Some delay ->
           let filtering = heads <> None in
           create_delayed_stream
-            ~filtering ~include_ops requested_heads bi_stream delay,
+            ~filtering ~include_ops requested_heads bi_stream delay in
-          shutdown
+    let shutdown () = Watcher.shutdown stopper in
-    in
     let first_request = ref true in
     let next () =
       if not !first_request then begin
@@ -313,7 +311,8 @@ let list_operations node {Services.Operations.monitor; contents} =
   if not monitor then
     RPC.Answer.return operations
   else
-    let stream, shutdown = Node.RPC.operation_watcher node in
+    let stream, stopper = Node.RPC.operation_watcher node in
+    let shutdown () = Watcher.shutdown stopper in
     let first_request = ref true in
     let next () =
       if not !first_request then
@@ -349,7 +348,8 @@ let list_protocols node {Services.Protocols.monitor; contents} =
   if not monitor then
     RPC.Answer.return protocols
   else
-    let stream, shutdown = Node.RPC.protocol_watcher node in
+    let stream, stopper = Node.RPC.protocol_watcher node in
+    let shutdown () = Watcher.shutdown stopper in
     let first_request = ref true in
     let next () =
       if not !first_request then

src/node/shell/state.ml

@@ -46,29 +46,6 @@ let () =
     (function Unknown_network x -> Some x | _ -> None)
     (fun x -> Unknown_network x) ;
 
-module Watcher = struct
-
-  type 'a t = {
-    id: int ;
-    push: ('a option -> unit) ;
-  }
-
-  let notify watchers info =
-    List.iter (fun w -> w.push (Some info)) watchers
-
-  let create_stream watchers =
-    let cpt = ref 0 in
-    fun () ->
-      let id = incr cpt ; !cpt in
-      let stream, push = Lwt_stream.create () in
-      watchers := { id ; push } :: !watchers ;
-      let unregister () =
-        push None ;
-        watchers := List.filter (fun w -> w.id <> id) !watchers in
-      stream, unregister
-
-end
-
 (** *)
 
 type net_id = Store.net_id = Net of Block_hash.t
@@ -78,13 +55,11 @@ type t = {
   nets: net Block_hash_table.t ;
   store: Store.store ;
   block_db: Db_proxy.Block.t ;
-  block_watchers: (Block_hash.t * Store.block) Watcher.t list ref ;
+  block_watchers: (Block_hash.t * Store.block) Watcher.input ;
   operation_db: Db_proxy.Operation.t ;
-  operation_watchers:
+  operation_watchers: (Operation_hash.t * Store.operation) Watcher.input ;
-    (Operation_hash.t * Store.operation) Watcher.t list ref ;
   protocol_db: Db_proxy.Protocol.t ;
-  protocol_watchers:
+  protocol_watchers: (Protocol_hash.t * Store.protocol) Watcher.input ;
-    (Protocol_hash.t * Store.protocol) Watcher.t list ref ;
   valid_block_state: valid_block_state Persist.shared_ref ;
 }
 
@@ -101,7 +76,7 @@ and valid_block_state = {
   ttl: Int64.t ;
   index: Context.index ;
   block_db: Db_proxy.Block.t ;
-  watchers: valid_block Watcher.t list ref ;
+  watchers: valid_block Watcher.input ;
 }
 
 and blockchain_state = {
@@ -227,7 +202,7 @@ module Operation = struct
           Db_proxy.Operation.store t.operation_db h (Time.make_timed (Ok op))
           >>= function
           | true ->
-              Watcher.notify !(t.operation_watchers) (h, op) ;
+              Watcher.notify t.operation_watchers (h, op) ;
               return (Some (h, op))
           | false ->
               return None
@@ -244,7 +219,7 @@ module Operation = struct
   let invalid state =
     Persist.use state.store.global_store InvalidOperations.read
 
-  let create_watcher t = Watcher.create_stream t.operation_watchers ()
+  let create_watcher t = Watcher.create_stream t.operation_watchers
 
 end
 
@@ -285,7 +260,7 @@ module Protocol = struct
         Db_proxy.Protocol.store t.protocol_db h (Time.make_timed (Ok proto))
         >>= function
         | true ->
-            Watcher.notify !(t.protocol_watchers) (h, proto) ;
+            Watcher.notify t.protocol_watchers (h, proto) ;
             return (Some (h, proto))
         | false ->
             return None
@@ -302,7 +277,7 @@ module Protocol = struct
   let invalid state =
     Persist.use state.store.global_store InvalidProtocols.read
 
-  let create_watcher t = Watcher.create_stream t.protocol_watchers ()
+  let create_watcher t = Watcher.create_stream t.protocol_watchers
 
   let keys { protocol_db } = Db_proxy.Protocol.keys protocol_db
 
@@ -429,10 +404,10 @@ module Block = struct
               Persist.update t.store.global_store (fun store ->
                   PostponedBlocks.set store h >>= fun store ->
                   Lwt.return (Some store)) >>= fun _ ->
-              Watcher.notify !(t.block_watchers) (h, b) ;
+              Watcher.notify t.block_watchers (h, b) ;
               return (Some (h, b))
           | false -> return None
-  let create_watcher t = Watcher.create_stream t.block_watchers ()
+  let create_watcher t = Watcher.create_stream t.block_watchers
 
   let check_block state h =
     known state h >>= function
@@ -604,8 +579,8 @@ module Valid_block = struct
     let ttl = Int64.of_int ttl in
     Lwt.return
       (Persist.share { global_store = store ;
-                       block_db ; index ;
+                       block_db ; index ; ttl ;
-                       ttl ; watchers = ref []  })
+                       watchers = Watcher.create_input () })
 
   let locked_valid vstate h =
     Context.checkout vstate.index h >>= function
@@ -688,7 +663,7 @@ module Valid_block = struct
                       Store.Block_valid_succs.set
                         store block.shell.predecessor successors >>= fun () ->
                       Lwt.return (Some store)) >>= fun _ ->
-                  Watcher.notify !(vstate.watchers) valid_block ;
+                  Watcher.notify vstate.watchers valid_block ;
                   Lwt.return (Ok valid_block)
                 end
 
@@ -837,7 +812,7 @@ module Valid_block = struct
 
   let create_watcher state =
     use state (fun vstate ->
-        Lwt.return (Watcher.create_stream vstate.watchers ()))
+        Lwt.return (Watcher.create_stream vstate.watchers))
 
   module Store = struct
     type t = valid_block_state
@@ -1309,10 +1284,10 @@ let read
     active_net = [] ;
     nets = Block_hash_table.create 7 ;
     operation_db ;
-    operation_watchers = ref [] ;
+    operation_watchers = Watcher.create_input () ;
     protocol_db ;
-    protocol_watchers = ref [] ;
+    protocol_watchers = Watcher.create_input () ;
-    block_db ; block_watchers = ref [] ;
+    block_db ; block_watchers = Watcher.create_input () ;
     valid_block_state ;
   }
   in

src/node/shell/state.mli

@@ -124,7 +124,7 @@ module Operation : sig
   (** Create a stream of all the newly locally-stored operations.
       The returned function allows to terminate the stream. *)
   val create_watcher:
-      state -> (key * operation) Lwt_stream.t * (unit -> unit)
+      state -> (key * operation) Lwt_stream.t * Watcher.stopper
 
 end
 
@@ -189,7 +189,7 @@ module Block : sig
   (** Create a stream of all the newly locally-stored blocks.
       The returned function allows to terminate the stream. *)
   val create_watcher:
-    state -> (Block_hash.t * block) Lwt_stream.t * (unit -> unit)
+    state -> (Block_hash.t * block) Lwt_stream.t * Watcher.stopper
 
   (** If [h1] is an ancestor of [h2] in the current [state],
       then [path state h1 h2] returns the chain of block from
@@ -297,7 +297,7 @@ module Valid_block : sig
 
   (** Create a stream of all the newly validated blocks.
       The returned function allows to terminate the stream. *)
-  val create_watcher: state -> (valid_block Lwt_stream.t * (unit -> unit)) Lwt.t
+  val create_watcher: state -> (valid_block Lwt_stream.t * Watcher.stopper) Lwt.t
 
   (** If [h1] is an ancestor of [h2] in the current [state],
       then [path state h1 h2] returns the chain of block from
@@ -410,7 +410,7 @@ module Protocol : sig
   (** Create a stream of all the newly locally-stored protocols.
       The returned function allows to terminate the stream. *)
   val create_watcher:
-    state -> (key * protocol) Lwt_stream.t * (unit -> unit)
+    state -> (key * protocol) Lwt_stream.t * Watcher.stopper
 
   val keys: state -> key list Lwt.t
 end

src/utils/watcher.ml

@@ -0,0 +1,50 @@
+(**************************************************************************)
+(*                                                                        *)
+(*    Copyright (c) 2014 - 2017.                                          *)
+(*    Dynamic Ledger Solutions, Inc. <contact@tezos.com>                  *)
+(*                                                                        *)
+(*    All rights reserved. No warranty, explicit or implicit, provided.   *)
+(*                                                                        *)
+(**************************************************************************)
+
+type 'a inner_stopper = {
+  id: int ;
+  push: ('a option -> unit) ;
+  mutable active : bool;
+  input : 'a input;
+}
+
+and 'a input =
+  { mutable watchers : 'a inner_stopper list;
+    mutable cpt : int; }
+
+type stopper = unit -> unit
+
+let create_input () =
+  { watchers = [];
+    cpt = 0 }
+
+let create_fake_stream () =
+  let str, push = Lwt_stream.create () in
+  str, (fun () -> push None)
+
+let notify input info =
+  List.iter (fun w -> w.push (Some info)) input.watchers
+
+let shutdown_output output =
+  if output.active then begin
+    output.active <- false;
+    output.push None;
+    output.input.watchers <-
+      List.filter (fun w -> w.id <> output.id) output.input.watchers;
+  end
+
+let create_stream input =
+  input.cpt <- input.cpt + 1;
+  let id = input.cpt in
+  let stream, push = Lwt_stream.create () in
+  let output = { id; push; input; active = true } in
+  input.watchers <- output :: input.watchers;
+  stream, (fun () -> shutdown_output output)
+
+let shutdown f = f ()

src/utils/watcher.mli

@@ -0,0 +1,19 @@
+(**************************************************************************)
+(*                                                                        *)
+(*    Copyright (c) 2014 - 2017.                                          *)
+(*    Dynamic Ledger Solutions, Inc. <contact@tezos.com>                  *)
+(*                                                                        *)
+(*    All rights reserved. No warranty, explicit or implicit, provided.   *)
+(*                                                                        *)
+(**************************************************************************)
+
+(** {1 Notification callbacks} *)
+
+type 'a input
+type stopper
+
+val create_input : unit -> 'a input
+val notify : 'a input -> 'a -> unit
+val create_stream : 'a input -> 'a Lwt_stream.t * stopper
+val create_fake_stream : unit -> 'a Lwt_stream.t * stopper
+val shutdown : stopper -> unit