<commit_before>Reactive Streams for Kafka
====

[Reactive Streams](http://www.reactive-streams.org) wrapper for [Apache Kafka](https://kafka.apache.org/).

Initiated by [SoftwareMill](https://softwaremill.com)

Supports Kafka 0.8.2-beta

Available at Maven Central for Scala 2.10 and 2.11:

    libraryDependencies += "com.softwaremill" %% "reactive-kafka" % "0.1.0"

Testing
----
Tests require Apache Kafka and Zookeeper to be available on localhost:9092 and localhost:2181

Example usage
----

```Scala
import akka.actor.ActorSystem
import akka.stream.FlowMaterializer
import akka.stream.scaladsl.{Sink, Source}
import com.softwaremill.react.kafka.ReactiveKafka

implicit val materializer = FlowMaterializer()
implicit  val actorSystem = ActorSystem("ReactiveKafka")

val kafka = new ReactiveKafka(host = "localhost:9092", zooKeeperHost = "localhost:2181")
val publisher = kafka.consume("lowercaseStrings", "groupName")
val subscriber = kafka.publish("uppercaseStrings", "groupName")


Source(publisher).map(_.toUpperCase).to(Sink(subscriber)).run()
```

Tuning
----

KafkaActorSubscriber and KafkaActorPublisher have their own thread pools, configured in `application.conf`.
You can tune them by overriding `kafka-publisher-dispatcher.thread-pool-executor` and
`kafka-subscriber-dispatcher.thread-pool-executor`
<commit_msg>application.conf renamed to reference.conf also in README.md<commit_after>Reactive Streams for Kafka
====

[Reactive Streams](http://www.reactive-streams.org) wrapper for [Apache Kafka](https://kafka.apache.org/).

Initiated by [SoftwareMill](https://softwaremill.com)

Supports Kafka 0.8.2-beta

Available at Maven Central for Scala 2.10 and 2.11:

    libraryDependencies += "com.softwaremill" %% "reactive-kafka" % "0.1.0"

Testing
----
Tests require Apache Kafka and Zookeeper to be available on localhost:9092 and localhost:2181

Example usage
----

```Scala
import akka.actor.ActorSystem
import akka.stream.FlowMaterializer
import akka.stream.scaladsl.{Sink, Source}
import com.softwaremill.react.kafka.ReactiveKafka

implicit val materializer = FlowMaterializer()
implicit  val actorSystem = ActorSystem("ReactiveKafka")

val kafka = new ReactiveKafka(host = "localhost:9092", zooKeeperHost = "localhost:2181")
val publisher = kafka.consume("lowercaseStrings", "groupName")
val subscriber = kafka.publish("uppercaseStrings", "groupName")


Source(publisher).map(_.toUpperCase).to(Sink(subscriber)).run()
```

Tuning
----

KafkaActorSubscriber and KafkaActorPublisher have their own thread pools, configured in `reference.conf`.
You can tune them by overriding `kafka-publisher-dispatcher.thread-pool-executor` and
`kafka-subscriber-dispatcher.thread-pool-executor` in your `application.conf` file.

{'fraction_non_alphanumeric': 0.12115666178623719, 'fraction_numerical': 0.020497803806734993, 'mean_word_length': 5.633495145631068, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, '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': '1104822', 'n_tokens_mistral': 919, 'n_tokens_neox': 848, 'n_words': 210}