typelevel / typelevel/fs2

Topic#subscribe on closed topics has surprising (non-terminating) semantics

Open
#2,558 2 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

bug
Dominant language
Scala
Stars
2.5k
Forks
636
Avg merge
2d 4h
Merged PRs (30d)
7

Description

using fs2 3.0.6 and 3.1.0. Here's a scatsie. All code pasted in this issue is also in the scatsie.

We have as an example of subscriber to a topic subscribing after the topic gets closed:

def topicSubscribeClosedTopicNonTerminating(topic: Topic[IO, Int]): IO[Unit] = {
  val subscriber1: Stream[IO, Unit] =
    topic
      .subscribe(maxQueued = 1)
      .evalMap(i => Console[IO].println(s"subscriber 1 received: $i"))

  for {
    fib1 <- subscriber1.compile.drain.start
    _ <- Console[IO].println(s"closing topic")
    _ <- topic.close
    _ <- fib1.join
  } yield ()
}

This consistently ends up non-terminating, and it's a bit surprising given that other methods seem to signal operating on closed topics better. I can think of 3 possibilities here:

  1. write in the scaladoc of subscribe a warning about this behaviour
  2. have the method return a Stream.empty
  3. have the method return a Stream.raiseError(...)

Additionally, you'll see in the scatsie the methods topicAwaitInStream*, which could be included as an example in the scaladoc on how to avoid surprising semantics.

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with Topic#subscribe and the topicAwaitInStream* examples in the linked Scastie; reproduce the closed-topic case on fs2 3.0.6 or 3.1.0. Compare the proposed empty-stream, error, and documentation outcomes, then establish the expected behavior for subscriptions made after close. Done means the chosen behavior is implemented or clearly documented and the non-terminating case is addressed.

Written by the indexing model from the issue text.

Assessment

Tech stack
scala
Domain
stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.