Stream created from reactive publisher after completing or failing doesn't cancel subscription.

Question

Stream created from reactive publisher after completing or failing doesn't cancel subscription.

Zuchos opened this issue 5 years ago · 12 comments

    val publisher: Publisher[Int]  = ??? //(publisher created with akka-streams with fanout = true)
    env.unsafeRunAsync(
      publisher
        .toStream(2)
        .takeWhile(_ < 300)
        .map { e =>
          println(s"ZIO 1: $e")
          e
        }
        .run(zio.stream.Sink.drain)
    )(e => println(s"1 EXITING! $e"))

After running that stream, ZStream will finish consuming from publisher after 300 element, but it won't cancel subscription and in that way will prevent publisher from publishing to other subscribers. I have other subscriptions from that publisher and they also stops receiving new elements when this streams stops.

I made my own decorators on Subscriber, Publisher and Subscription and I verified that subscription.cancel is never called.

Answer 1 · 2020-05-26T13:15:24.000Z

I created gist to run that case:
https://gist.github.com/Zuchos/be215e25e78066db0cddfafe83d3cddd

Answer 2 · 2020-05-26T15:24:58.000Z

Thanks for reporting, @Zuchos. Unfortunately, I can't reproduce the issue you describe. Which version do you use? I have added tests for cancellation in #126. If these tests do not cover what you mean, please comment on the MR.

Answer 3 · 2020-05-26T15:41:36.000Z

@runtologist did you try running gist that I pasted? (sorry for thread sleeps). In console println should not reach 500, it should stop around ~316.
Akka 2.6.5
Zio interop 1.0.3.5-RC (Caliban 0.8.0 dependency)
I will try to check lib's test it the mean time.

Answer 4 · 2020-05-26T15:43:22.000Z

Which RC? There are 9 of them. ;-)

Answer 5 · 2020-05-26T16:22:13.000Z

sorry, RC8.
I'm looking in a test suite and I guess that those PublisherToStreamSpec will always pass
Task(probe.expectCancelling()) since this one is not throwing exceptions or errors. So I guess it should be asserted with testEnv.verifyNoAsyncErrors().
But I'm not familiar with this reactive test kit, so maybe I'm missing the point.

Answer 6 · 2020-05-26T16:25:59.000Z

    testM("test that is always passing") {
        assertM((for {
          probe <- makeProbe
          _     <- URIO(probe.expectCancelling())
        } yield ()).run)(
          succeeds(isUnit)
        )
      }

It always passes and apparently it should not.

Answer 7 · 2020-05-26T17:12:42.000Z

Interesting. Good catch! I'll look into it.

Answer 8 · 2020-05-26T18:13:37.000Z

I can confirm that the existing tests fail when rewritten as you proposed. The PR has been updated with the adjusted tests and a fix.

Answer 9 · 2020-05-26T20:05:53.000Z

fixed in v1.0.3.5-RC10

Answer 10 · 2020-05-26T20:06:27.000Z

Thanks again @Zuchos for your help in analyzing the issue.

Answer 11 · 2020-05-27T08:08:11.000Z

It looks like artifact was only published for Scala 2.12 and it's missing for Scala 2.13.

Answer 12 · 2020-05-27T09:30:52.000Z

All cross versions should be published now.