add Blocking_queue.size

src/bb_queue.ml

@@ -83,3 +83,9 @@ let try_push (self : _ t) x : bool =
     true
   ) else
     false
+
+let size (self : _ t) : int =
+  Mutex.lock self.mutex;
+  let n = Queue.length self.q in
+  Mutex.unlock self.mutex;
+  n

src/bb_queue.mli

@@ -10,6 +10,10 @@ val push : 'a t -> 'a -> unit
 (** [push q x] pushes [x] into [q], and returns [()].
     @raise Closed if [close q] was previously called.*)
 
+val size : _ t -> int
+(** Number of items currently in the queue.
+    @since 0.2 *)
+
 val pop : 'a t -> 'a
 (** [pop q] pops the next element in [q]. It might block until an element comes.
    @raise Closed if the queue was closed before a new element was available. *)

src/moonpool.mli

@@ -40,6 +40,12 @@ module Blocking_queue : sig
   val create : unit -> _ t
   (** Create a new unbounded queue. *)
 
+  val size : _ t -> int
+  (** Number of items currently in the queue. Note that [pop]
+      might still block if this returns a non-zero number, since another
+      thread might have consumed the items in the mean time.
+      @since 0.2 *)
+
   exception Closed
 
   val push : 'a t -> 'a -> unit