libraryDependencies += "com.megafarad" % "play-kafka-utils" % "1.0.0"

libraryDependencies += "io.micrometer" % "micrometer-registry-prometheus" % "1.13.2"

libraryDependencies +=   "io.micrometer" % "micrometer-registry-graphite" % "1.13.2"

libraryDependencies += "io.micrometer" % "micrometer-registry-datadog" % "1.13.2"

import com.megafarad.play.kafka.services.KafkaMessageHandlerService
import com.megafarad.play.kafka.model.KafkaMessage
import play.api.Logging

import javax.inject.Inject
import scala.concurrent.Future

class SampleKafkaMessageHandlerService @Inject() extends KafkaMessageHandlerService[String, String] with Logging {
  def processMessage(message: KafkaMessage[String, String]): Future[Unit] = Future.successful({
    logger.info(s"Processed message ${message.key} -> ${message.value}")
  })
}

import javax.inject.{Inject, Named}

//Be sure to inject all KafkaConsumerService instances with their named annotations here...
class StartupService @Inject()(@Named("consumer") consumerService: KafkaConsumerService[String, String])
  extends KafkaConsumerStartupService {

  val services: Seq[KafkaConsumerService[_, _]] = Seq(consumerService) //...and be sure you include all class parameters here.

  //Be sure to call startAll() in the constructor.
  startAll()
}

class KafkaModule extends AbstractModule with ScalaModule {
  override def configure(): Unit = {
    //Binds a named KafkaMessageHandlerService. You can have multiple instances of the service... 
    bind[KafkaMessageHandlerService[String, String]]
      .annotatedWith(Names.named("handler")) // ...just be sure to name them uniquely here.
      .to[SampleKafkaMessageHandlerService]
    //Binds your startup service as an eager singleton - so that it starts polling.
    bind[KafkaConsumerStartupService].to[StartupService].asEagerSingleton()
    //Binds the MeterRegistry. You should choose a backend. Let's say you chose Prometheus. Your binding should look 
    //something like:
    bind[PrometheusMeterRegistry].toInstance(new PrometheusMeterRegistry(PrometheusConfig.DEFAULT))
    //You'd then inject an instance of PrometheusMeterRegistry in the following Provides method.
    //OR: If you don't care about metrics, just bind MetricsRegistry to a SimpleMeterRegistry instance:
    bind[MeterRegistry].toInstance(new SimpleMeterRegistry())
    
  }

  //Because KafkaConsumerService and KafkaProducerService require both general configuration and instance-specific 
  //configuration, you need to use @Provides methods to inject the correct settings for each Kafka consumer and 
  //producer.
  
  //This binds KafkaConsumerService. You can bind multiple instances with multiple @Provides methods... 
  @Provides
  @Singleton
  @Named("consumer")//...just be sure to name them uniquely...
  def provideConsumerService(@Named("handler") messageHandlerService: KafkaMessageHandlerService[String, String],
                              config: Configuration,
                              metrics: MeterRegistry, //If using Prometheus (or other backend), inject PrometheusMeterRegistry (or some other MeterRegistry) instead
                              lifecycle: ApplicationLifecycle)
                             (implicit system: ActorSystem,
                              ec: ExecutionContext): KafkaConsumerService[String, String] = {
    val consumerConfig = config.get[Configuration]("kafka.consumer") //...and be sure to extract the right configuration key for the consumer.
    new KafkaConsumerService[String, String](messageHandlerService, config, consumerConfig, metrics, lifecycle)
  }

  //This binds KafkaProducerService. You can bind multiple instances with multiple @Provides methods...
  @Provides
  @Singleton
  @Named("producer")//...just be sure to name them uniquely...
  def provideProducerService(config: Configuration,
                              metrics: MeterRegistry,//If using Prometheus (or other backend), inject PrometheusMeterRegistry (or some other MeterRegistry) instead
                              lifecycle: ApplicationLifecycle)
                             (implicit ec: ExecutionContext): KafkaProducerService[String, String] = {
    val producerConfig = config.get[Configuration]("kafka.producer")//...and be sure to extract the right configuration key for the producer.
    new KafkaProducerService[String, String](config, producerConfig, metrics, lifecycle)

  }
}

libraryDependencies += "io.micrometer" % "micrometer-registry-prometheus" % "1.13.2"

class KafkaModule extends AbstractModule with ScalaModule {
  override def configure(): Unit = {
    //Other bindings
    bind[PrometheusMeterRegistry].toInstance(new PrometheusMeterRegistry(PrometheusConfig.DEFAULT))
    //Other bindings
  }
  //Provides methods
}

import javax.inject.Inject
import io.micrometer.prometheus.PrometheusMeterRegistry
import play.api.mvc._

class MetricsController @Inject()(cc: ControllerComponents, registry: PrometheusMeterRegistry) extends AbstractController(cc) {
  def metrics() = Action {
    Ok(registry.scrape()).as("text/plain")
  }
}