Make listen return Resource[F, Stream] instead of Stream to handle readiness

[mbaechler]

In ChannelTest, we have:

for {
      // There is a race here. If this fiber doesn't start running quickly enough all the data will
      // be written to the channel before we execute LISTEN. We can't add a latch to `listen` that
      // indicates LISTEN has completed because it makes it impossible to implement `mapK` for
      // `Channel` and thus for `Session`. So for now we're just going to sleep a while. I'm not
      // sure it's a problem in real life but it makes this test hard to write.
      f <- ch.listen(42).map(_.value).takeThrough(_ != data.last).compile.toList.start
      _ <- IO.sleep(1.second) // sigh
      _ <- data.traverse_(ch.notify)
      o <- f.join
      Succeeded(fa) = o
      d <- fa
      _ <- assert(s"channel data $d $data", data.endsWith(d)) // we may miss the first few
    } yield "ok"

If we change listen to return a Resource, we could write something like that:

      ch.listenR(42).use { r =>
        for {
          f <- r.map(_.value).takeThrough(_ != data.last).compile.toList.start
          _ <- data.traverse_(ch.notify)
          o <- f.join
          Succeeded(fa) = o
          d <- fa
          _ <- assert(s"channel data $d $data", data == d)
        } yield "ok"
      }

Because once we acquire the Resource, we know that we started listening, so we can't miss any event.

[mpilquist]

Sounds good to me, want to create a PR?

[mbaechler]

I tried but I can't make it working. Would you mind looking at it?

[mbaechler]

I managed to make it work thanks to me peers @trobert and @rlecomte

[mbaechler]

It has been merged