Блог

Я использую Play Framework (Scala) для микросервиса и использую Kafka в качестве шины событий. У меня есть потребитель событий, который сопоставляется с классом событий, который выглядит как:

 case class MovieEvent[T] (
                          mediaId: String,
                          config: T
                        )

object MovieEvent {
  implicit def movieEventFormat[T: Format]: Format[MovieEvent[T]] =
    ((__  "mediaId").format[String] ~
      (__  "config").format[T]
      )(MovieEvent.apply _, unlift(MovieEvent.unapply))
}

object MovieProvider extends SerializableEnumeration {
  implicit val providerReads: Reads[MovieProvider.Value] = SerializableEnumeration.jsonReader(MovieProvider)
  implicit val providerWrites: Writes[MovieProvider.Value] = SerializableEnumeration.jsonWrites
  val  Dreamworks, Disney, Paramount = Value
}
  

Потребитель выглядит так:

 class MovieEventConsumer @Inject()(movieService: MovieService
                                    ) extends ConsumerRecordProcessor with LazyLogging {
  override def process(record: IncomingRecord): Unit = {

    val movieEventJson = Json.parse(record.valueString).validate[MovieEvent[DreamworksConfiguration]]
    movieEventJson match {
      case event: JsSuccess[MovieEvent[DreamworksJobOptions]] => processMovieEvent(event.get)
      case er: JsError =>
        logger.error("Unrecognized MovieEvent, attempting to parse as MovieUploadEvent: "   JsError.toJson(er).toString())
        try {
          val data = (Json.parse(record.valueString)  "upload").as[MovieUploadEvent]
          processUploadEvent(data)
        } catch {
          case er: Exception => logger.error("Unrecognized kafka event", er)
        }
    }
  }

  def processMovieEvent[T](event: MovieEvent[T]): Unit = {
    logger.debug(s"Received movie event: ${event}")
    movieService.createMovieJob(event)
  }

  def processUploadEvent(event: MovieUploadEvent): Unit = {
    logger.debug(s"Received upload event: ${event}")
    movieService.addToCollection(event)
  }

}
  

Прямо сейчас я могу проверить только одну из трех разных конфигураций MovieEvent (Dreamwork, Disney и Paramount). Я могу поменять местами, какой из них я проверяю с помощью кода, но это не главное. Тем не менее, я хотел бы проверить любую из трех без необходимости создавать дополнительных потребителей. Я пробовал играть с несколькими разными идеями, но ни одна из них не компилировалась. Я довольно новичок в Play и Kafka и задаюсь вопросом, есть ли хороший способ сделать это.

Заранее спасибо!

Я собираюсь предположить, что число возможных конфигураций конечно и все они известны во время компиляции (в вашем примере 3).

Одна из возможностей — создать MovieEvent запечатанный признак с универсальным типом T . Вот минимальный пример:

 case class DreamWorksJobOptions(anOption: String, anotherOption: String)
case class DisneyJobOptions(anOption: String)

sealed trait MovieEvent[T] {
  def mediaId: String
  def config: T
}
case class DreamWorksEvent(mediaId: String, config: DreamWorksJobOptions) extends MovieEvent[DreamWorksJobOptions]
case class DisneyEvent(mediaId: String, config: DisneyJobOptions) extends MovieEvent[DisneyJobOptions]

def tryParse(jsonString: String): MovieEvent[_] = {
  // ... parsing logic goes here
  DreamWorksEvent("dw", DreamWorksJobOptions("some option", "another option"))
}

val parseResult = tryParse("asdfasdf")

parseResult match {
  case DreamWorksEvent(mediaId, config) => println(mediaId   " : "   config.anOption   " : "   config.anotherOption)
  case DisneyEvent(mediaId, config) => println(mediaId   config)
}
  

который выводит

 dw : some option : another option
  

Я опустил часть синтаксического анализа, потому что у меня нет доступа к Play Json atm. Но поскольку у вас закрытая иерархия, вы можете попробовать каждый из ваших вариантов один за другим. (И вам в значительной степени придется, поскольку мы не можем гарантировать статически, что DreamWorksEvent у него нет той же структуры Json, DisneyEvent что и — вам нужно решить, какой из них будет опробован первым, и вернуться к разбору JSON как другого типа, когда первый не удается проанализировать).

Теперь ваш другой код очень общий. Чтобы добавить новый тип события, вам просто нужно добавить еще один подкласс MovieEvent , а также убедиться, что ваша логика синтаксического анализа обрабатывает этот новый случай. Волшебство здесь в том, что вам не нужно указывать свой T при обращении MovieEvent , поскольку вы знаете, что у вас закрытая иерархия, и, таким образом, можете восстановить T ее с помощью сопоставления с образцом.