Odtnhj

Hi every one how ca i get data with http-akka ad send to kafka with Alpakka Kafka connector and this is my source :

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")
 

There are so many different things that you can do with Akka Streams

Here is just one sample using a Producer sink, per the documentation https://doc.akka.io/docs/akka-stream-kafka/current/producer.html. Be sure to check out passThrough and some of things that it can do:

 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>"))