Refactor merge implementation

[zainab-ali]

To give some context, @djspiewak and I were looking through merge and found the implementation difficult to understand, then tried to rewrite it to use new cats-effect primitives.

I'm not sure that the code I've settled on is actually any better than the old implementation, but think it's worth getting a second opinion on.

The main differences are:

• The use of SingallingRef over two Deferreds for managing the upstream state.
• The use of evalMap(guard.aquire) instead of an explicit pull for pulling from upstream.

[zainab-ali]

The most recent builds failed due to failures in the following tests:

• StreamCombinatorsSuite - awakeEvery - basic on MacOS
• UnixSocketsSuite.echoes - jdk on MacOS

Neither of these failures are related to the merge implementation. I assume these tests are flaky and should be marked as such, or corrected?

[armanbilge]

or corrected

most time-based tests can be run on the Cats Effect test runtime with executeEmbed, this should fix the flakiness

fs2/core/shared/src/test/scala/fs2/StreamCombinatorsSuite.scala

Lines 42 to 45 in d7f1b71

	group("awakeEvery") {
	test("basic") {
	Stream
	.awakeEvery[IO](500.millis)

[zainab-ali]

I've used TestControl for most viable cases in the StreamCombinatorsSuite. I've also added a pretty printer for the scalacheck failures so we can see the elements of a stream on test failure.

Looking at the size of the changeset, it might be worth pulling it out into a different PR - if so, let me know.

[zainab-ali]

On adding the pretty printer implicit conversion, Stream.empty resolved to the val empty instead of def empty[A] on the monoidK instance. I'm at a loss as to why.

[zainab-ali]

See the comment below

[armanbilge]

Out of curiosity, why this change?

[armanbilge]

Oh, sorry, I missed #3242 (comment). Thanks!

[zainab-ali]

There's a compile error in the build (locally too), which I'm still figuring out:

[error] /home/runner/work/fs2/fs2/io/js/src/test/scala/fs2/io/IoPlatformSuite.scala:111:13: value empty of type fs2.Stream[fs2.Pure,Nothing] does not take type parameters.
[error]       .empty[IO]
[error]             ^
[error] one error found

It's caused by the implicit added for the pretty printer.

[zainab-ali]

I think I know why:

Stream.empty[IO] uses an apply, so is desugared to Stream.empty.apply[IO]:

The apply[F[_]] function is added by PureOps:

  implicit final class PureOps[O](private val self: Stream[Pure, O]) extends AnyVal {
    def apply[F[_]]: Stream[F, O] = covary
}

This conflicts with the apply function in the scalacheck pretty printer.