@@ -609,9 +609,50 @@ let ``ofAsyncEnumerableResolved should preserve a resolution failure over Dispos
609609 AsyncVal.wrap n
610610 let observedFailure = ref " "
611611 use sub =
612- Observable.ofAsyncEnumerableResolved 2 resolve ( fun ex ->
613- observedFailure.Value <- ex.Message
614- - 1 )
612+ Observable.ofAsyncEnumerableResolved
613+ 2
614+ resolve
615+ ( fun ex ->
616+ observedFailure.Value <- ex.Message
617+ - 1 )
618+ source
619+ |> Observer.create
620+ sub.WaitCompleted ( timeout = ms 10 )
621+ sub.Received |> seqEquals [ - 1 ]
622+ Assert.Equal ( " Boom resolving" , observedFailure.Value)
623+
624+ [<Fact>]
625+ let ``ofAsyncEnumerableResolved should preserve a resolution failure over a later enumeration exception after canceling MoveNextAsync`` () =
626+ // Regression test: once a resolution failure has already stopped the stream, a source that reacts to the linked
627+ // cancellation by throwing a different exception from MoveNextAsync must not replace that original failure.
628+ let moveFailed = TaskCompletionSource ()
629+ let source =
630+ SuspendingAsyncEnumerable< int>( fun cancellationToken index -> task {
631+ match index with
632+ | 0 -> return ValueSome 1
633+ | 1 ->
634+ use _ = cancellationToken.Register ( fun () -> moveFailed.TrySetResult () |> ignore)
635+ do ! moveFailed.Task
636+ return failwith " Boom during enumeration"
637+ | _ -> return ValueNone
638+ })
639+ let resolve _ ( n : int ) =
640+ if n = 1 then
641+ async {
642+ do ! Async.Sleep ( ms 50 )
643+ return failwith " Boom resolving"
644+ }
645+ |> AsyncVal.ofAsync
646+ else
647+ AsyncVal.wrap n
648+ let observedFailure = ref " "
649+ use sub =
650+ Observable.ofAsyncEnumerableResolved
651+ 2
652+ resolve
653+ ( fun ex ->
654+ observedFailure.Value <- ex.Message
655+ - 1 )
615656 source
616657 |> Observer.create
617658 sub.WaitCompleted ( timeout = ms 10 )
0 commit comments