Как может Akka Alppaka Producer интегрироваться с Akka HTTP - PullRequest
Новая
       10

Как может Akka Alppaka Producer интегрироваться с Akka HTTP

0 голосов
/ 13 ноября 2018

Привет всем, как я могу получить данные с помощью http-akka ad, отправив их на kafka с разъемом Alpakka Kafka, и это мой источник: 

object WebTrack extends App  with  Directives with LazyLogging {
val host = "localhost"
val port = 7070

val authorization = "Authorization"

val route = withSizeLimit(96239727) {
  post {
    headerValueByName("Authorization") { auth =>
      entity(as[data]) { trans =>
        //************************** I need send data to Alpakka Kafka connector ********************************

        complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, "<h1>Say hello to akka-http</h1>"))
      }
    }
  }
}

val bindingFuture = Http().bindAndHandleAsync(Route.asyncHandler(route), host, port)
.onComplete {
  case Success(_) => {
    logger.debug("This is very convenient ;-)")
    println(s"Server online at http://localhost:7070 \\nPress RETURN to stop...")
  }
  case Failure(e) => {
    println("Error Bind Http().bindAndHandleAsync")
  }
}

}

1 Ответ

0 голосов
/ 14 ноября 2018

Есть так много разных вещей, которые вы можете сделать с помощью Akka Streams

Вот только один пример с использованием Producer приемника, согласно документации https://doc.akka.io/docs/akka-stream-kafka/current/producer.html. Обязательно ознакомьтесь с passThrough и некоторыми вещами, которые он может сделать: 

  implicit val system: ActorSystem = ActorSystem("my-system")
  implicit val materializer: ActorMaterializer = ActorMaterializer()
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher

  case class User(firstName:String, lastName:String)
  object UserJsonSupport extends DefaultJsonProtocol with SprayJsonSupport {
    implicit val PortofolioFormats = jsonFormat2(User)
  }

  import UserJsonSupport._
  val config = system.settings.config.getConfig("akka.kafka.producer")
  val producerSettings = ProducerSettings(config, new StringSerializer, new StringSerializer)

  val route = withSizeLimit(96239727) {
    post {
      headerValueByName("Authorization") { auth =>
        entity(as[User]) { user: User =>
          Source.single(user)
            .map(user => new ProducerRecord[String, String]("last_names_topic", user.lastName, user.firstName))
            .runWith(Producer.plainSink(producerSettings))
          complete(HttpEntity(ContentTypes.`text/html(UTF-8)`,
            "<h1>Say hello to akka-http</h1>"))
        }
      }
    }
  }
...