KafkaSenderConfig.scala

KafkaSenderConfig.scala

object KafkaSender {
  sealed trait Config
  object Config {
    case class WithPublishing(url: String, queueSize: Int) extends Config
    case object WithoutPublishing extends Config
  }
}

implicit val kafkaSenderConfigConvert: ConfigConvert[KafkaSender.Config] = {
  val key = "publishing"

  val withPublishing = implicitly[ConfigConvert[KafkaSender.Config.WithPublishing]]
  val withoutPublishing = implicitly[ConfigConvert[KafkaSender.Config.WithoutPublishing.type]]

  new ConfigConvert[KafkaSender.Config] {
    override def from(cv: ConfigValue): Either[ConfigReaderFailures, KafkaSender.Config] = {
      cv match {
        case co: ConfigObject =>
          Option(co.get(key)) match {
            case Some(publishingCv) =>
              val publishingOpt: Option[Boolean] = publishingCv.unwrapped() match {
                case x: java.lang.Boolean => Some(x)
                case x: String => Try(x.toBoolean).toOption
                case _ => None
              }
              publishingOpt match {
                case Some(publishing) =>
                  if (publishing) {
                    withPublishing.from(cv)
                  } else {
                    withoutPublishing.from(cv)
                  }
                case None =>
                  Left(ConfigReaderFailures(WrongType(
                    publishingCv.valueType, Set(ConfigValueType.BOOLEAN), ConfigValueLocation(publishingCv), key
                  )))
              }
            case None =>
              Left(ConfigReaderFailures(KeyNotFound(key, ConfigValueLocation(co))))
          }
        case _ =>
          Left(ConfigReaderFailures(WrongType(
            cv.valueType, Set(ConfigValueType.OBJECT), ConfigValueLocation(cv), ""
          )))
      }
    }

    override def to(config: KafkaSender.Config): ConfigValue = {
      config match {
        case x: KafkaSender.Config.WithPublishing =>
          withPublishing.to(x)
            .withFallback(ConfigFactory.empty().withValue(key, ConfigValueFactory.fromAnyRef(true)).root())

        case KafkaSender.Config.WithoutPublishing =>
          ConfigFactory.empty().withValue(key, ConfigValueFactory.fromAnyRef(false)).root()
      }
    }
  }
}

WithPublishing:

kafka {
  publishing = true
  queue-size = 999
  url = "ololo"
}

WithoutPublishing:

kafka {
  publishing = false
}