Change to support creation effect for serializers

[vlovgr]

In some cases, serializers need a creation effect, i.e. F[Deserializer[F, A]] or F[Serializer[F, A]]. One example is the Confluent Avro serializer, which allocates mutable state as part of initialization. Typically, we want one instance per consumer or producer, which means we end up with code similar to the following.

import cats.effect.IO
import fs2.kafka._

val keyDeserializer = 
  IO { println("key"); Deserializer[IO, Int] }

val valueDeserializer = 
  IO { println("value"); Deserializer[IO, String] }

val settings: IO[ConsumerSettings[IO, Int, String]] =
  keyDeserializer.flatMap { keyDeserializer =>
    valueDeserializer.map { valueDeserializer =>
      ConsumerSettings(
        keyDeserializer = keyDeserializer,
        valueDeserializer = valueDeserializer
      )
    }
  }

This pull request changes ConsumerSettings and ProducerSettings to accept serializers with creation effects, which will then be evaluated as part of consumer or producer creation. The above example can now be written as follows.

import cats.effect.IO
import fs2.kafka._

val keyDeserializer = 
  IO { println("key"); Deserializer[IO, Int] }

val valueDeserializer = 
  IO { println("value"); Deserializer[IO, String] }

val settings: ConsumerSettings[IO, Int, String] =
  ConsumerSettings(
    keyDeserializer = keyDeserializer,
    valueDeserializer = valueDeserializer
  )

Additionally, we use the OrElse construct from Monix to support implicit serializers with and without creation effects. This enables implicit generic derivation of serializers with creation effects. Essentially, this means we can write the example above using implicit serializers, as follows.

import cats.effect.IO
import fs2.kafka._

implicit val keyDeserializer: IO[Deserializer[IO, Int]] = 
  IO { println("key"); Deserializer[IO, Int] }

implicit val valueDeserializer: IO[Deserializer[IO, String]] = 
  IO { println("value"); Deserializer[IO, String] }

val settings =
  ConsumerSettings[IO, Int, String]

Note that settings can also be created for serializers without creation effects, just like before.

[codecov]

Codecov Report

Merging #142 into 0.20.x will decrease coverage by 0.07%.
The diff coverage is 92.59%.

@@            Coverage Diff             @@
##           0.20.x     #142      +/-   ##
==========================================
- Coverage   92.89%   92.81%   -0.08%     
==========================================
  Files          51       52       +1     
  Lines        1210     1225      +15     
  Branches       79       78       -1     
==========================================
+ Hits         1124     1137      +13     
- Misses         86       88       +2

Impacted Files	Coverage Δ
src/main/scala/fs2/kafka/KafkaConsumer.scala	100% <100%> (ø)	⬆️
src/main/scala/fs2/kafka/ProducerSettings.scala	100% <100%> (ø)	⬆️
src/main/scala/fs2/kafka/ConsumerSettings.scala	100% <100%> (ø)	⬆️
src/main/scala/fs2/kafka/KafkaProducer.scala	93.1% <100%> (+0.51%)	⬆️
.../scala/fs2/kafka/internal/KafkaConsumerActor.scala	85.02% <100%> (-0.15%)	⬇️
...n/scala/fs2/kafka/TransactionalKafkaProducer.scala	100% <100%> (ø)	⬆️
src/main/scala/fs2/kafka/internal/OrElse.scala	60% <60%> (ø)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f2e89bf...d79c61b. Read the comment docs.