Merge pull request #141 from bikallem/stream-is_empty

lib_eio: implement Stream.is_empty

lib_eio/eio.mli

@@ -308,7 +308,13 @@ module Stream : sig
       it returns [None] if the stream is empty rather than waiting.
       Note that if another domain may add to the stream then a [None]
       result may already be out-of-date by the time this returns. *)
-end  
+
+  val length : 'a t -> int
+  (** [length t] returns the number of items currently in [t]. *)
+
+  val is_empty : 'a t -> bool
+  (** [is_empty t] is [length t = 0]. *)
+end
 
 (** Cancelling other fibres when an exception occurs. *)
 module Cancel : sig
@@ -734,7 +740,7 @@ module Private : sig
     type 'a enqueue = ('a, exn) result -> unit
     (** A function provided by the scheduler to reschedule a previously-suspended thread. *)
 
-    type _ eff += 
+    type _ eff +=
       | Suspend : (Fibre_context.t -> 'a enqueue -> unit) -> 'a eff
       (** [Suspend fn] is performed when a fibre must be suspended
           (e.g. because it called {!Promise.await} on an unresolved promise).

lib_eio/stream.ml

@@ -118,3 +118,11 @@ let take_nonblocking t =
     end;
     Mutex.unlock t.mutex;
     Some v
+
+let length t =
+  Mutex.lock t.mutex;
+  let len = Queue.length t.items in
+  Mutex.unlock t.mutex;
+  len
+
+let is_empty t = (length t = 0)