feat bb_queue: add transfer_into

src/bb_queue.ml

@@ -111,6 +111,21 @@ let transfer (self : 'a t) q2 : unit =
     )
   done
 
+let transfer_into q (self : _ t) : unit =
+  if not (Queue.is_empty q) then (
+    Mutex.lock self.mutex;
+
+    if self.closed then (
+      Mutex.unlock self.mutex;
+      raise Closed
+    );
+
+    let was_empty = Queue.is_empty self.q in
+    Queue.transfer q self.q;
+    if was_empty then Condition.broadcast self.cond;
+    Mutex.unlock self.mutex
+  )
+
 type 'a gen = unit -> 'a option
 type 'a iter = ('a -> unit) -> unit
 

src/bb_queue.mli

@@ -62,6 +62,11 @@ val transfer : 'a t -> 'a Queue.t -> unit
 
     @since 0.4 *)
 
+val transfer_into : 'a Queue.t -> 'a t -> unit
+(** [transfer q bq] transfers all items from [q] to [bq], atomically.
+    @raise Closed if [bq] is closed and [q] is not empty.
+    @since NEXT_RELEASE *)
+
 val close : _ t -> unit
 (** Close the queue, meaning there won't be any more [push] allowed. *)