Apply formatting

Author: joan38

build.sbt

@@ -15,22 +15,26 @@ developers := List(
 organizationName := "lightbend"
 organizationHomepage := Option(url("http://lightbend.com/"))
 homepage := scmInfo.value map (_.browseUrl)
-scmInfo := Option(ScmInfo(url("https://github.com/lightbend/kafka-streams-scala"), "[email protected]:lightbend/kafka-streams-scala.git"))
+scmInfo := Option(
+  ScmInfo(url("https://github.com/lightbend/kafka-streams-scala"), "[email protected]:lightbend/kafka-streams-scala.git")
+)
 
 parallelExecution in Test := false
 testFrameworks += new TestFramework("minitest.runner.Framework")
 
 libraryDependencies ++= Seq(
-  kafkaStreams excludeAll(ExclusionRule("org.slf4j", "slf4j-log4j12"), ExclusionRule("org.apache.zookeeper", "zookeeper")),
+  kafkaStreams excludeAll (ExclusionRule("org.slf4j", "slf4j-log4j12"), ExclusionRule("org.apache.zookeeper",
+                                                                                      "zookeeper")),
   scalaLogging % "test",
-  logback % "test",
-  kafka % "test" excludeAll(ExclusionRule("org.slf4j", "slf4j-log4j12"), ExclusionRule("org.apache.zookeeper", "zookeeper")),
-  curator % "test",
-  minitest % "test",
+  logback      % "test",
+  kafka        % "test" excludeAll (ExclusionRule("org.slf4j", "slf4j-log4j12"), ExclusionRule("org.apache.zookeeper",
+                                                                                        "zookeeper")),
+  curator      % "test",
+  minitest     % "test",
   minitestLaws % "test",
-  algebird % "test",
-  chill % "test",
-  avro4s % "test"
+  algebird     % "test",
+  chill        % "test",
+  avro4s       % "test"
 )
 
 credentials += Credentials(Path.userHome / ".ivy2" / ".credentials")

project/Dependencies.scala

@@ -5,21 +5,22 @@ object Dependencies {
 
   implicit class Exclude(module: ModuleID) {
     def log4jExclude: ModuleID =
-      module excludeAll(ExclusionRule("log4j"))
+      module.excludeAll(ExclusionRule("log4j"))
 
     def driverExclusions: ModuleID =
-      module.log4jExclude.exclude("com.google.guava", "guava")
+      module.log4jExclude
+        .exclude("com.google.guava", "guava")
         .excludeAll(ExclusionRule("org.slf4j"))
   }
 
-  val kafkaStreams = "org.apache.kafka"                % "kafka-streams"     % KafkaVersion
-  val scalaLogging = "com.typesafe.scala-logging"     %% "scala-logging"     % ScalaLoggingVersion
-  val logback = "ch.qos.logback"                       % "logback-classic"   % LogbackVersion
-  val kafka = "org.apache.kafka"                      %% "kafka"             % KafkaVersion
-  val curator = "org.apache.curator"                   % "curator-test"      % CuratorVersion
-  val minitest = "io.monix"                           %% "minitest"          % MinitestVersion
-  val minitestLaws = "io.monix"                       %% "minitest-laws"     % MinitestVersion
-  val algebird = "com.twitter"                        %% "algebird-core"     % AlgebirdVersion
-  val chill = "com.twitter"                           %% "chill"             % ChillVersion
-  val avro4s = "com.sksamuel.avro4s"                  %% "avro4s-core"       % Avro4sVersion
+  val kafkaStreams = "org.apache.kafka"           % "kafka-streams"   % KafkaVersion
+  val scalaLogging = "com.typesafe.scala-logging" %% "scala-logging"  % ScalaLoggingVersion
+  val logback      = "ch.qos.logback"             % "logback-classic" % LogbackVersion
+  val kafka        = "org.apache.kafka"           %% "kafka"          % KafkaVersion
+  val curator      = "org.apache.curator"         % "curator-test"    % CuratorVersion
+  val minitest     = "io.monix"                   %% "minitest"       % MinitestVersion
+  val minitestLaws = "io.monix"                   %% "minitest-laws"  % MinitestVersion
+  val algebird     = "com.twitter"                %% "algebird-core"  % AlgebirdVersion
+  val chill        = "com.twitter"                %% "chill"          % ChillVersion
+  val avro4s       = "com.sksamuel.avro4s"        %% "avro4s-core"    % Avro4sVersion
 }

project/Versions.scala

@@ -1,14 +1,14 @@
 object Versions {
-  val AlgebirdVersion = "0.13.0"
-  val ChillVersion = "0.9.2"
-  val LogbackVersion = "1.2.3"
-  val KafkaVersion = "1.0.0"
+  val AlgebirdVersion     = "0.13.0"
+  val ChillVersion        = "0.9.2"
+  val LogbackVersion      = "1.2.3"
+  val KafkaVersion        = "1.0.0"
   val ScalaLoggingVersion = "3.5.0"
-  val CuratorVersion = "4.0.0"
-  val MinitestVersion = "2.0.0"
-  val JDKVersion = "1.8"
-  val Scala_2_12_Version = "2.12.5"
-  val Scala_2_11_Version = "2.11.11"
-  val Avro4sVersion = "1.8.3"
-  val CrossScalaVersions = Seq(Scala_2_12_Version, Scala_2_11_Version )
+  val CuratorVersion      = "4.0.0"
+  val MinitestVersion     = "2.0.0"
+  val JDKVersion          = "1.8"
+  val Scala_2_12_Version  = "2.12.5"
+  val Scala_2_11_Version  = "2.11.11"
+  val Avro4sVersion       = "1.8.3"
+  val CrossScalaVersions  = Seq(Scala_2_12_Version, Scala_2_11_Version)
 }

project/plugins.sbt

@@ -1,2 +1,2 @@
 addSbtPlugin("org.scalastyle" %% "scalastyle-sbt-plugin" % "1.0.0")
-addSbtPlugin("com.geirsson" % "sbt-scalafmt" % "1.5.0")
+addSbtPlugin("com.geirsson"   % "sbt-scalafmt"           % "1.5.0")

src/main/scala/com/lightbend/kafka/scala/streams/DefaultSerdes.scala

@@ -1,22 +1,20 @@
 /**
- * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
- * Copyright 2017-2018 Alexis Seigneurin.
- */
-
+  * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+  * Copyright 2017-2018 Alexis Seigneurin.
+  */
 package com.lightbend.kafka.scala.streams
 
 import org.apache.kafka.common.serialization.{Serde, Serdes}
 
-
 /**
- * Implicit values for default serdes
- */
+  * Implicit values for default serdes
+  */
 object DefaultSerdes {
-  implicit val stringSerde: Serde[String] = Serdes.String()
-  implicit val longSerde: Serde[Long] = Serdes.Long().asInstanceOf[Serde[Long]]
-  implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.ByteArray()
+  implicit val stringSerde: Serde[String]                             = Serdes.String()
+  implicit val longSerde: Serde[Long]                                 = Serdes.Long().asInstanceOf[Serde[Long]]
+  implicit val byteArraySerde: Serde[Array[Byte]]                     = Serdes.ByteArray()
   implicit val bytesSerde: Serde[org.apache.kafka.common.utils.Bytes] = Serdes.Bytes()
-  implicit val floatSerde: Serde[Float] = Serdes.Float().asInstanceOf[Serde[Float]]
-  implicit val doubleSerde: Serde[Double] = Serdes.Double().asInstanceOf[Serde[Double]]
-  implicit val integerSerde: Serde[Int] = Serdes.Integer().asInstanceOf[Serde[Int]]
+  implicit val floatSerde: Serde[Float]                               = Serdes.Float().asInstanceOf[Serde[Float]]
+  implicit val doubleSerde: Serde[Double]                             = Serdes.Double().asInstanceOf[Serde[Double]]
+  implicit val integerSerde: Serde[Int]                               = Serdes.Integer().asInstanceOf[Serde[Int]]
 }

src/main/scala/com/lightbend/kafka/scala/streams/FunctionConversions.scala

@@ -1,30 +1,29 @@
 /**
- * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
- * Copyright 2017-2018 Alexis Seigneurin.
- */
-
+  * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+  * Copyright 2017-2018 Alexis Seigneurin.
+  */
 package com.lightbend.kafka.scala.streams
 
 import org.apache.kafka.streams.KeyValue
 import org.apache.kafka.streams.kstream._
 
 /**
- * Implicit classes that offer conversions of Scala function literals to
- * SAM (Single Abstract Method) objects in Java. These make the Scala APIs much
- * more expressive, with less boilerplate and more succinct.
- */
+  * Implicit classes that offer conversions of Scala function literals to
+  * SAM (Single Abstract Method) objects in Java. These make the Scala APIs much
+  * more expressive, with less boilerplate and more succinct.
+  */
 object FunctionConversions {
 
   implicit class PredicateFromFunction[K, V](val test: (K, V) => Boolean) extends AnyVal {
-    def asPredicate: Predicate[K,V] = test(_,_)
+    def asPredicate: Predicate[K, V] = test(_, _)
   }
 
-  implicit class MapperFromFunction[T, U, V](val f:(T,U) => V) extends AnyVal {
+  implicit class MapperFromFunction[T, U, V](val f: (T, U) => V) extends AnyVal {
     def asKeyValueMapper: KeyValueMapper[T, U, V] = (k: T, v: U) => f(k, v)
-    def asValueJoiner: ValueJoiner[T,U,V] = (v1, v2) => f(v1, v2)
+    def asValueJoiner: ValueJoiner[T, U, V]       = (v1, v2) => f(v1, v2)
   }
 
-  implicit class KeyValueMapperFromFunction[K, V, KR, VR](val f:(K,V) => (KR, VR)) extends AnyVal {
+  implicit class KeyValueMapperFromFunction[K, V, KR, VR](val f: (K, V) => (KR, VR)) extends AnyVal {
     def asKeyValueMapper: KeyValueMapper[K, V, KeyValue[KR, VR]] = (k, v) => {
       val (kr, vr) = f(k, v)
       KeyValue.pair(kr, vr)
@@ -36,10 +35,10 @@ object FunctionConversions {
   }
 
   implicit class AggregatorFromFunction[K, V, VR](val f: (K, V, VR) => VR) extends AnyVal {
-    def asAggregator: Aggregator[K, V, VR] = (k,v,r) => f(k,v,r)
+    def asAggregator: Aggregator[K, V, VR] = (k, v, r) => f(k, v, r)
   }
 
-  implicit class MergerFromFunction[K,VR](val f: (K, VR, VR) => VR) extends  AnyVal {
+  implicit class MergerFromFunction[K, VR](val f: (K, VR, VR) => VR) extends AnyVal {
     def asMerger: Merger[K, VR] = (k, v1, v2) => f(k, v1, v2)
   }
 

src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala

@@ -1,20 +1,19 @@
 /**
- * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
- * Copyright 2017-2018 Alexis Seigneurin.
- */
-
+  * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+  * Copyright 2017-2018 Alexis Seigneurin.
+  */
 package com.lightbend.kafka.scala.streams
 
 import org.apache.kafka.streams.kstream._
-import org.apache.kafka.streams.{ KeyValue, Consumed }
+import org.apache.kafka.streams.{Consumed, KeyValue}
 import org.apache.kafka.common.serialization.Serde
 
 import scala.language.implicitConversions
 
 /**
- * Implicit conversions between the Scala wrapper objects and the underlying Java
- * objects.
- */
+  * Implicit conversions between the Scala wrapper objects and the underlying Java
+  * objects.
+  */
 object ImplicitConversions {
 
   implicit def wrapKStream[K, V](inner: KStream[K, V]): KStreamS[K, V] =
@@ -50,7 +49,8 @@ object ImplicitConversions {
   implicit def producedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Produced[K, V] =
     Produced.`with`(keySerde, valueSerde)
 
-  implicit def joinedFromKVOSerde[K, V, VO](implicit keySerde: Serde[K], valueSerde: Serde[V],
+  implicit def joinedFromKVOSerde[K, V, VO](implicit keySerde: Serde[K],
+                                            valueSerde: Serde[V],
                                             otherValueSerde: Serde[VO]): Joined[K, V, VO] =
     Joined.`with`(keySerde, valueSerde, otherValueSerde)
 }

src/main/scala/com/lightbend/kafka/scala/streams/KGroupedStreamS.scala

@@ -1,8 +1,7 @@
 /**
- * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
- * Copyright 2017-2018 Alexis Seigneurin.
- */
-
+  * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+  * Copyright 2017-2018 Alexis Seigneurin.
+  */
 package com.lightbend.kafka.scala.streams
 
 import org.apache.kafka.streams.kstream._
@@ -12,10 +11,9 @@ import org.apache.kafka.common.serialization.Serde
 import ImplicitConversions._
 import FunctionConversions._
 
-
 /**
- * Wraps the Java class KGroupedStream and delegates method calls to the underlying Java object.
- */
+  * Wraps the Java class KGroupedStream and delegates method calls to the underlying Java object.
+  */
 class KGroupedStreamS[K, V](inner: KGroupedStream[K, V]) {
 
   def count(): KTableS[K, Long] = {
@@ -24,47 +22,41 @@ class KGroupedStreamS[K, V](inner: KGroupedStream[K, V]) {
   }
 
   def count(store: String, keySerde: Option[Serde[K]] = None): KTableS[K, Long] = {
-    val materialized = keySerde.foldLeft(Materialized.as[K, java.lang.Long, KeyValueStore[Bytes, Array[Byte]]](store))((m,serde)=> m.withKeySerde(serde))
+    val materialized = keySerde.foldLeft(Materialized.as[K, java.lang.Long, KeyValueStore[Bytes, Array[Byte]]](store))(
+      (m, serde) => m.withKeySerde(serde)
+    )
 
     val c: KTableS[K, java.lang.Long] = inner.count(materialized)
     c.mapValues[Long](Long2long _)
   }
 
-  def reduce(reducer: (V, V) => V): KTableS[K, V] = {
+  def reduce(reducer: (V, V) => V): KTableS[K, V] =
     inner.reduce((v1, v2) => reducer(v1, v2))
-  }
-
-  def reduce(reducer: (V, V) => V,
-    materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] = {
 
+  def reduce(reducer: (V, V) => V, materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] =
     // need this explicit asReducer for Scala 2.11 or else the SAM conversion doesn't take place
     // works perfectly with Scala 2.12 though
     inner.reduce(((v1: V, v2: V) => reducer(v1, v2)).asReducer, materialized)
-  }
-
-  def reduce(reducer: (V, V) => V,
-    storeName: String)(implicit keySerde: Serde[K], valueSerde: Serde[V]): KTableS[K, V] = {
 
+  def reduce(reducer: (V, V) => V, storeName: String)(implicit keySerde: Serde[K],
+                                                      valueSerde: Serde[V]): KTableS[K, V] =
     // need this explicit asReducer for Scala 2.11 or else the SAM conversion doesn't take place
     // works perfectly with Scala 2.12 though
-    inner.reduce(((v1: V, v2: V) =>
-      reducer(v1, v2)).asReducer,
-      Materialized.as[K, V, KeyValueStore[Bytes, Array[Byte]]](storeName)
+    inner.reduce(
+      ((v1: V, v2: V) => reducer(v1, v2)).asReducer,
+      Materialized
+        .as[K, V, KeyValueStore[Bytes, Array[Byte]]](storeName)
         .withKeySerde(keySerde)
         .withValueSerde(valueSerde)
     )
-  }
 
-  def aggregate[VR](initializer: () => VR,
-    aggregator: (K, V, VR) => VR): KTableS[K, VR] = {
+  def aggregate[VR](initializer: () => VR, aggregator: (K, V, VR) => VR): KTableS[K, VR] =
     inner.aggregate(initializer.asInitializer, aggregator.asAggregator)
-  }
 
   def aggregate[VR](initializer: () => VR,
-    aggregator: (K, V, VR) => VR,
-    materialized: Materialized[K, VR, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, VR] = {
+                    aggregator: (K, V, VR) => VR,
+                    materialized: Materialized[K, VR, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, VR] =
     inner.aggregate(initializer.asInitializer, aggregator.asAggregator, materialized)
-  }
 
   def windowedBy(windows: SessionWindows): SessionWindowedKStreamS[K, V] =
     inner.windowedBy(windows)

src/main/scala/com/lightbend/kafka/scala/streams/KGroupedTableS.scala

@@ -1,8 +1,7 @@
 /**
- * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
- * Copyright 2017-2018 Alexis Seigneurin.
- */
-
+  * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+  * Copyright 2017-2018 Alexis Seigneurin.
+  */
 package com.lightbend.kafka.scala.streams
 
 import ImplicitConversions._
@@ -12,8 +11,8 @@ import org.apache.kafka.common.utils.Bytes
 import FunctionConversions._
 
 /**
- * Wraps the Java class KGroupedTable and delegates method calls to the underlying Java object.
- */
+  * Wraps the Java class KGroupedTable and delegates method calls to the underlying Java object.
+  */
 class KGroupedTableS[K, V](inner: KGroupedTable[K, V]) {
 
   type ByteArrayKVStore = KeyValueStore[Bytes, Array[Byte]]
@@ -26,35 +25,24 @@ class KGroupedTableS[K, V](inner: KGroupedTable[K, V]) {
   def count(materialized: Materialized[K, Long, ByteArrayKVStore]): KTableS[K, Long] =
     inner.count(materialized)
 
-  def reduce(adder: (V, V) => V,
-             subTractor: (V, V) => V): KTableS[K, V] = {
-
+  def reduce(adder: (V, V) => V, subTractor: (V, V) => V): KTableS[K, V] =
     // need this explicit asReducer for Scala 2.11 or else the SAM conversion doesn't take place
     // works perfectly with Scala 2.12 though
     inner.reduce(((v1, v2) => adder(v1, v2)).asReducer, ((v1, v2) => subTractor(v1, v2)).asReducer)
-  }
 
   def reduce(adder: (V, V) => V,
              subtractor: (V, V) => V,
-             materialized: Materialized[K, V, ByteArrayKVStore]): KTableS[K, V] = {
-
+             materialized: Materialized[K, V, ByteArrayKVStore]): KTableS[K, V] =
     // need this explicit asReducer for Scala 2.11 or else the SAM conversion doesn't take place
     // works perfectly with Scala 2.12 though
     inner.reduce(((v1, v2) => adder(v1, v2)).asReducer, ((v1, v2) => subtractor(v1, v2)).asReducer, materialized)
-  }
-
-  def aggregate[VR](initializer: () => VR,
-                    adder: (K, V, VR) => VR,
-                    subtractor: (K, V, VR) => VR): KTableS[K, VR] = {
 
+  def aggregate[VR](initializer: () => VR, adder: (K, V, VR) => VR, subtractor: (K, V, VR) => VR): KTableS[K, VR] =
     inner.aggregate(initializer.asInitializer, adder.asAggregator, subtractor.asAggregator)
-  }
 
   def aggregate[VR](initializer: () => VR,
                     adder: (K, V, VR) => VR,
                     subtractor: (K, V, VR) => VR,
-                    materialized: Materialized[K, VR, ByteArrayKVStore]): KTableS[K, VR] = {
-
+                    materialized: Materialized[K, VR, ByteArrayKVStore]): KTableS[K, VR] =
     inner.aggregate(initializer.asInitializer, adder.asAggregator, subtractor.asAggregator, materialized)
-  }
 }