1.0 Release blocker? Suspicious code in ConnectionStream class

[mberndt123]

Hi,

I noticed some code in the ConnectionStream class today:

  private[fs2rabbit] lazy val connFactory: F[ConnectionFactory] =
    F.delay {
        …
    }

  private[fs2rabbit] def acquireConnection: F[(RabbitMQConnection, AMQPChannel)] =
    for {
      factory <- connFactory
      conn    <- F.delay(factory.newConnection)
      channel <- F.delay(conn.createChannel)
    } yield (conn, RabbitChannel(channel))

  override def createConnectionChannel: Stream[F, AMQPChannel] = …

connFactory is a lazy val, while acquireConnection and createConnectionChannel are defs without parameters. This seems odd to me. defs without parameters are equivalent to vals when you do functional programming (cf. John de Goes), so why not make everything a val and be more efficient?

To me, this looks like the intention was to lazily initialize the ConnectionFactory exactly once when a connection is first acquired. But what this code actually does is create a new ConnectionFactory every time a connection is acquired. I would guess that that's not what the intention was, right? You can easily check this by e. g. adding some print statements in connFactory's F.delay block.

I would like to point out that this issue is probably impossible to (correctly) fix this issue without changing the signature of Fs2Rabbit.apply. It currently returns Fs2Rabbit[F], but it will have to return F[Fs2Rabbit[F]]. This kind of initialization (whether lazy or non-lazy) ultimately requires some form of mutable state, and creating a mutable variable is in itself a side effect, as can be seen e. g. from the signature of cats.effect.concurrent.Ref.of, which returns F[Ref[F, A]] rather than Ref[F, A]. So this is something that needs to be taken care of before the 1.0 release which I understand is imminent.

[gvolpe]

Hi @mberndt123 ,

Thanks for raising your concerns here. You are right about that lazy val and def, they can be made a val since they are referentially transparent (sending in a PR as I reply).

However, your other assumptions are wrong, let me explain why.

You can easily check this by e. g. adding some print statements in connFactory's F.delay block.

Have you tried doing this? It doesn't happen because the functions are RT.

I would like to point out that this issue is probably impossible to (correctly) fix this issue without changing the signature of Fs2Rabbit.apply. It currently returns Fs2Rabbit[F], but it will have to return F[Fs2Rabbit[F]]. This kind of initialization (whether lazy or non-lazy) ultimately requires some form of mutable state, and creating a mutable variable is in itself a side effect, as can be seen e. g. from the signature of cats.effect.concurrent.Ref.of, which returns F[Ref[F, A]] rather than Ref[F, A]. So this is something that needs to be taken care of before the 1.0 release which I understand is imminent.

F[Fs2Rabbit[F]] used to be the shape of the creation but it has changed since 1.0-RC2 (see #105) because the creation of Fs2Rabbit[F] is no longer effectful as it used to be. The only effect that happened before was the creation of the internal fs2 Queue to interact with the Java consumers and publishers but that is now in a different interpreter.

You can try creating as many interpreters of Fs2abbit[F] as you want and introduce a println in connFactory if you want to be sure that nothing is happening.

[mberndt123]

Hi @gvolpe, thanks for responding. However I don't believe this is fixed.

Have you tried doing this? It doesn't happen because the functions are RT.

I have tried it just now, and it does happen as I describe it. Here's a quick and dirty test I've written:

  it should "foobar" in {
    val cfg = Fs2RabbitConfig(
      "localhost",
      5672,
      "/",
      1000,
      false,
      None,
      Some("guest"),
      Some("guest"),
      false
    )
    val s = new ConnectionStream[IO](cfg)
    (for {
      _ <- s.acquireConnection
      _ <- s.acquireConnection
    } yield ()).unsafeRunSync()
  }

I have also inserted a println("making ConnectionFactory") statement just before the new ConnectionFactory() line inside the F.delay block in connFactory. And I see the "making ConnectionFactory" line twice when running the test. Is that really the intended behaviour? Because I don't see the point in creating two connection factories.

[gvolpe]

Well, you are calling acquireConnection twice so the effect runs... twice as one would expect.

Anyway the acquireConnection method is private[fs2rabbit](not exposed to the users of the library) so I believe you are just writing a test in the library itself?

Users shouldn't be creating a ConnectionStream themselves but if they do then they should know what they are doing there.

The library only exposes the createConnectionChannel which (as you might have guessed) creates a new connection and channel every time is invoked.

Does that answer your question?

[mberndt123]

Well, you are calling acquireConnection twice so the effect runs... twice as one would expect.

Sure, I do expect two connections to be acquired. However that doesn't require creating two different ConnectionFactory objects, yet this is what the code does.

Anyway the acquireConnection method is private[fs2rabbit](not exposed to the users of the library) so I believe you are just writing a test in the library itself?

Well, that was for simplicity, but as a user I could also do s.createConnectionChannel.repeat.take(2). I wouldn't expect two ConnectionFactory objects to be created in this case.

The library only exposes the createConnectionChannel which (as you might have guessed) creates a new connection and channel every time is invoked.

Right, but it also create a new ConnectionFactory every time, and I would argue that's unnecessary.

[gvolpe]

Right, but it also create a new ConnectionFactory every time, and I would argue that's unnecessary.

Oh that's what you really meant. Sure, I think you're right here. Looking more into ConnectionFactory itself it seems a bit side-effecty so I might change the way a connection is being acquire now to make sure only one instance of ConnectionFactory is created. Thanks! :)