<commit_before>package org.rgcase.reactivekafka

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.server.Directives._
import akka.kafka.ProducerMessage.Message
import akka.kafka.ProducerSettings
import akka.kafka.scaladsl.Producer
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.ByteArraySerializer

import scala.concurrent.Future

class ReactiveKafkaFromHttpScala {

  implicit val system = ActorSystem("fromhttp")
  implicit val mat = ActorMaterializer()

  val serverSource = Http().bind(interface = "localhost", port = 8080)

  val producerSettings = ProducerSettings(system, new ByteArraySerializer, new ByteArraySerializer)
    .withBootstrapServers("localhost:9092")

  def toProducerMessage(topic: String)(chunk: Array[Byte]) = {
    Message[Array[Byte], Array[Byte], NotUsed](new ProducerRecord(topic, chunk), NotUsed)
  }

  val handler: HttpRequest => Future[HttpResponse] = {

    // Chunked payload
    case HttpRequest(POST, Uri.Path("/chunks"), _, entity, _) =>
      entity
        .dataBytes
        .map(chunk => toProducerMessage("targetchunktopic")(chunk.toArray))
        .via(Producer.flow(producerSettings))
        .runWith(Sink.ignore)
        .map(_ => HttpResponse(StatusCodes.Created))

    // Full payload
    case HttpRequest(POST, Uri.Path("/"), _, entity, _) =>
      val fut =
        entity
          .dataBytes
          .runWith(Sink.fold(Array.empty[Byte]){ case (acc, bytes) => acc ++ bytes })

      Source
        .fromFuture(fut)
        .map(bytes => toProducerMessage("targetfulltopic")(bytes))
        .via(Producer.flow(producerSettings))
        .runWith(Sink.ignore)
        .map(_ => HttpResponse(StatusCodes.Created))

  }

  serverSource.to(Sink.foreach { connection =>
    connection.handleWithAsyncHandler(handler)
  }).run()

}<commit_msg>Change uri path<commit_after>package org.rgcase.reactivekafka

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.server.Directives._
import akka.kafka.ProducerMessage.Message
import akka.kafka.ProducerSettings
import akka.kafka.scaladsl.Producer
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.ByteArraySerializer

import scala.concurrent.Future

class ReactiveKafkaFromHttpScala {

  implicit val system = ActorSystem("fromhttp")
  implicit val mat = ActorMaterializer()

  val serverSource = Http().bind(interface = "localhost", port = 8080)

  val producerSettings = ProducerSettings(system, new ByteArraySerializer, new ByteArraySerializer)
    .withBootstrapServers("localhost:9092")

  def toProducerMessage(topic: String)(chunk: Array[Byte]) = {
    Message[Array[Byte], Array[Byte], NotUsed](new ProducerRecord(topic, chunk), NotUsed)
  }

  val handler: HttpRequest => Future[HttpResponse] = {

    // Chunked payload
    case HttpRequest(POST, Uri.Path("/chunkedpayload"), _, entity, _) =>
      entity
        .dataBytes
        .map(chunk => toProducerMessage("targetchunktopic")(chunk.toArray))
        .via(Producer.flow(producerSettings))
        .runWith(Sink.ignore)
        .map(_ => HttpResponse(StatusCodes.Created))

    // Full payload
    case HttpRequest(POST, Uri.Path("/fullpayload"), _, entity, _) =>
      val fut =
        entity
          .dataBytes
          .runWith(Sink.fold(Array.empty[Byte]){ case (acc, bytes) => acc ++ bytes })

      Source
        .fromFuture(fut)
        .map(bytes => toProducerMessage("targetfulltopic")(bytes))
        .via(Producer.flow(producerSettings))
        .runWith(Sink.ignore)
        .map(_ => HttpResponse(StatusCodes.Created))

  }

  serverSource.to(Sink.foreach { connection =>
    connection.handleWithAsyncHandler(handler)
  }).run()

}
{'fraction_non_alphanumeric': 0.11912225705329153, 'fraction_numerical': 0.003858210754762479, 'mean_word_length': 4.003618817852835, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4941889', 'n_tokens_mistral': 1284, 'n_tokens_neox': 1260, 'n_words': 224}