KafkaContext.scala

import java.util.concurrent.{ExecutorService, Executors}
import cats.effect.{ContextShift, IO, Resource}
import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}

class KafkaContext(cs: ContextShift[IO]) {
  // A thread pool with exactly 1 thread
  private val threadPool = Executors.newFixedThreadPool(1)
  protected val synchronousExecutionContext = ExecutionContext.fromExecutor(threadPool)

  def execute[A](f: => A): IO[A] = cs.evalOn(synchronousExecutionContext)(IO(f))

  // Alias for execute
  def ~>[A](f: => A): IO[A] = execute(f)

  def close(): IO[Unit] = IO(pool.shutdown())
}

object KafkaContext {
  def resource(cs: ContextShift[IO]): Resource[IO, KafkaContext] =
    Resource.make(IO(new KafkaContext(cs)))(_.close())
}