Proper handling end of streams (both producer and produced)

Libs/Hopac/Stream.fs

@@ -281,12 +281,25 @@ module Stream =
 
   let ambAll (xxs: Stream<#Stream<_>>) = joinWith amb' xxs
 
-  let rec mergeAll' combined xss =
-    Alt.choose [ xss      ^=> function Nil -> mergeAll' combined never
-                                     | Cons (s, xss) -> mergeAll' (merge' combined s) xss
-                 combined ^=> function Nil -> nilj
-                                     | Cons (x, tail) -> consj x (mergeAll' tail xss |> memo) ]
-  and mergeAll (xxs: Stream<#Stream<'x>>) : Stream<'x> = mergeAll' never xxs |> memo
+  let rec mergeAll' (streams: Stream<'x> option) (producer: Stream<Stream<'x>> option) : Stream<'x> =
+    match streams, producer with
+    | None, Some p ->
+        p ^=> function Nil -> mergeAll' None None
+                     | Cons (x, xs) -> mergeAll' (Some x) (Some xs)
+        |> memo
+    | Some s, None ->
+        s ^=> function Cons (x, xs) -> consj x (mergeAll' (Some xs) producer)
+                     | Nil -> mergeAll' None producer :> _
+        |> memo
+    | Some s, Some p ->
+        Alt.choose [ s ^=> function Cons (x, xs) -> consj x (mergeAll' (Some xs) producer)
+                                  | Nil -> mergeAll' None producer :> _
+                     p ^=> function Cons (x, xs) -> mergeAll' (merge s x |> Some) (Some xs)
+                                    | Nil -> mergeAll' streams None ] |> memo
+    | None, None -> memo nilj
+
+  let mergeAll producer = mergeAll' None (Some producer)
+
 
   let appendAll (xxs: Stream<#Stream<_>>) = joinWith append' xxs
   let switchAll (xxs: Stream<#Stream<_>>) = joinWith switch' xxs

Libs/Hopac/Stream.fsi

@@ -1161,7 +1161,7 @@ module Stream =
   val    ambAll: Stream<#Stream<'x>> -> Stream<'x>
 
   /// Joins all the streams together with `merge`.
-  val  mergeAll: Stream<#Stream<'x>> -> Stream<'x>
+  val  mergeAll: Stream<Stream<'x>> -> Stream<'x>
 
   /// Joins all the streams together with `switch`.
   val switchAll: Stream<#Stream<'x>> -> Stream<'x>

Tests/AdHocTests/StreamTests.fs

@@ -174,3 +174,16 @@ let run () =
     <|> timeOutMillis 100 ^-> Choice2Of2
     |> run = Choice1Of2 xs
 
+  quick <| fun xs ->
+    let values =
+        [ Stream.ofList xs; Stream.once (timeOutMillis 50) ]
+        |> Stream.ofSeq
+        |> Stream.mergeAll
+        |> Stream.toList
+        |> memo
+        :> Alt<_>
+
+    values ^-> Choice1Of2
+    <|> timeOutMillis 100 ^-> Choice2Of2
+    |> run = Choice1Of2 (xs @ [ () ])
+