Refactoring for aKafkaProducerThat object

Author: manub

build.sbt

@@ -1,3 +1,5 @@
+import sbtrelease.Version
+
 lazy val commonSettings = Seq(
   name := "scalatest-embedded-kafka",
   organization := "net.manub",
@@ -34,7 +36,7 @@ lazy val publishSettings = Seq(
 )
 
 lazy val releaseSettings = Seq(
-  releaseVersionBump := sbtrelease.Version.Bump.Minor,
+  releaseVersionBump := Version.Bump.Minor,
   releaseCrossBuild := true
 )
 

src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala

@@ -15,8 +15,9 @@ import org.scalatest.Suite
 import scala.collection.JavaConversions.mapAsJavaMap
 import scala.concurrent._
 import scala.concurrent.duration._
-import scala.language.postfixOps
+import scala.language.{higherKinds, postfixOps}
 import scala.reflect.io.Directory
+import scala.reflect.runtime.universe._
 import scala.util.Try
 
 trait EmbeddedKafka {
@@ -110,24 +111,18 @@ trait EmbeddedKafka {
     }
   }
 
-  def aKafkaProducerThat(): KafkaProducerConfiguration = {
-    new KafkaProducerConfiguration()
-  }
-
-  sealed class KafkaProducerConfiguration {
-
-    def serializesValuesWith[T <: Serializer[_]](serializer: Class[T])(implicit config: EmbeddedKafkaConfig) = {
-      new KafkaProducer[String, T](Map(
-        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG       -> s"localhost:${config.kafkaPort}",
-        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG    -> classOf[StringSerializer].getName,
-        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG  -> serializer.getName,
-        ProducerConfig.METADATA_FETCH_TIMEOUT_CONFIG  -> 3000.toString,
-        ProducerConfig.RETRY_BACKOFF_MS_CONFIG        -> 1000.toString
+  object aKafkaProducer {
+    def thatSerializesValuesWith[V](serializer: Class[_ <: Serializer[V]])(implicit config: EmbeddedKafkaConfig) = {
+      new KafkaProducer[String, V](Map(
+        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> s"localhost:${config.kafkaPort}",
+        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG -> classOf[StringSerializer].getName,
+        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG -> serializer.getName,
+        ProducerConfig.METADATA_FETCH_TIMEOUT_CONFIG -> 3000.toString,
+        ProducerConfig.RETRY_BACKOFF_MS_CONFIG -> 1000.toString
       ))
     }
   }
 
-
   private def startZooKeeper(zooKeeperPort: Int): ServerCnxnFactory = {
     val zkLogsDir = Directory.makeTemp("zookeeper-logs")
     val tickTime = 2000

src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaSpec.scala

@@ -1,6 +1,5 @@
 package net.manub.embeddedkafka
 
-import java.lang.Class
 import java.net.InetSocketAddress
 import java.util.Properties
 import java.util.concurrent.TimeoutException
@@ -23,7 +22,6 @@ import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.concurrent.duration._
 import scala.language.postfixOps
-import scala.reflect.ClassTag
 
 class EmbeddedKafkaSpec
   extends TestKit(ActorSystem("embedded-kafka-spec")) with WordSpecLike with EmbeddedKafka with Matchers
@@ -194,9 +192,12 @@ class EmbeddedKafkaSpec
   "the aKafkaProducerThat method" should {
 
     "return a producer that encodes messages for the given encoder" in {
-        val producer = aKafkaProducerThat serializesValuesWith classOf[ByteArraySerializer]
 
-        producer.isInstanceOf[KafkaProducer[String, ByteArraySerializer]] shouldBe true
+      withRunningKafka {
+        val producer = aKafkaProducer thatSerializesValuesWith classOf[ByteArraySerializer]
+        producer.send(new ProducerRecord[String, Array[Byte]]("a topic", "a message".getBytes))
+      }
+
     }
   }