Added xan EmbeddedKafka object.

Author: manub

README.md

@@ -23,6 +23,20 @@ scalatest-embedded-kafka is available on Bintray and Maven Central, compiled for
 
 * In-memory Zookeeper and Kafka will be instantiated respectively on port 6000 and 6001 and automatically shutdown at the end of the test.
 
+### Use without the `withRunningKafka` method
+
+A `EmbeddedKafka` companion object is provided for usage without the `EmbeddedKafka` trait. Zookeeper and Kafka can be started an stopped in a programmatic way.
+
+        class MySpec extends WordSpec {
+    
+        "runs with embedded kafka" should {
+
+            withRunningKafka {
+                // ... code goes here
+            }
+        
+        }
+
 ## Configuration
 
 It's possible to change the ports on which Zookeeper and Kafka are started by providing an implicit `EmbeddedKafkaConfig`
@@ -39,6 +53,8 @@ It's possible to change the ports on which Zookeeper and Kafka are started by pr
         
         }
         
+This works for both `withRunningKafka` and `EmbeddedKafka.start()`
+        
 ## Utility methods
 
 The `EmbeddedKafka` trait provides also some utility methods to interact with the embedded kafka, in order to set preconditions or verifications in your specs:
@@ -47,8 +63,14 @@ The `EmbeddedKafka` trait provides also some utility methods to interact with th
         
         def consumeFirstMessageFrom(topic: String): String
         
+## Custom producers
+
+It is possible to create producers for custom types in two ways:
+
+* Using the syntax `aKafkaProducer thatSerializesValuesWith classOf[Serializer[V]]`. This will return a `KafkaProducer[String, V]`
+* Using the syntax `aKafkaProducer[V]`. This will return a `KafkaProducer[String, V]`, using an implicit `Serializer[V]`.
 
-For more information about how to use those method, you can either look at the Scaladocs or at the tests of this project.
+For more information about how to use the utility methods, you can either look at the Scaladocs or at the tests of this project.
 
 ## Badges 
 

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

@@ -19,19 +19,51 @@ import scala.language.{higherKinds, postfixOps}
 import scala.reflect.io.Directory
 import scala.util.Try
 
-trait EmbeddedKafka {
-
+trait EmbeddedKafka extends EmbeddedKafkaSupport {
   this: Suite =>
+}
+
+object EmbeddedKafka extends EmbeddedKafkaSupport {
+
+  private[this] var factory: Option[ServerCnxnFactory] = None
+  private[this] var broker: Option[KafkaServer] = None
 
+  /**
+    * Starts a ZooKeeper instance and a Kafka broker in memory.
+    *
+    * @param config an implicit [[EmbeddedKafkaConfig]]
+    */
+  def start()(implicit config: EmbeddedKafkaConfig) = {
+    factory = Option(startZooKeeper(config.zooKeeperPort))
+    broker = Option(startKafka(config))
+  }
+
+  /**
+    * Stops the in memory ZooKeeper instance and Kafka broker.
+    */
+  def stop(): Unit = {
+    broker.foreach(_.shutdown)
+    factory.foreach(_.shutdown())
+    broker = None
+    factory = None
+  }
+
+  /**
+    * Returns whether the in memory Kafka and Zookeeper are running.
+    */
+  def isRunning: Boolean = factory.nonEmpty && broker.nonEmpty
+}
+
+sealed trait EmbeddedKafkaSupport {
   val executorService = Executors.newFixedThreadPool(2)
   implicit val executionContext = ExecutionContext.fromExecutorService(executorService)
 
   /**
-   * Starts a ZooKeeper instance and a Kafka broker, then executes the body passed as a parameter.
-   *
-   * @param body the function to execute
-   * @param config an implicit [[EmbeddedKafkaConfig]]
-   */
+    * Starts a ZooKeeper instance and a Kafka broker, then executes the body passed as a parameter.
+    *
+    * @param body the function to execute
+    * @param config an implicit [[EmbeddedKafkaConfig]]
+    */
   def withRunningKafka(body: => Unit)(implicit config: EmbeddedKafkaConfig) = {
 
     val factory = startZooKeeper(config.zooKeeperPort)
@@ -45,28 +77,27 @@ trait EmbeddedKafka {
     }
   }
 
-
   /**
-   * Publishes synchronously a message of type [[String]] to the running Kafka broker.
-   *
-   * @see [[EmbeddedKafka#publishToKafka]]
-   * @param topic the topic to which publish the message (it will be auto-created)
-   * @param message the [[String]] message to publish
-   * @param config an implicit [[EmbeddedKafkaConfig]]
-   * @throws KafkaUnavailableException if unable to connect to Kafka
-   */
+    * Publishes synchronously a message of type [[String]] to the running Kafka broker.
+    *
+    * @see [[EmbeddedKafka#publishToKafka]]
+    * @param topic the topic to which publish the message (it will be auto-created)
+    * @param message the [[String]] message to publish
+    * @param config an implicit [[EmbeddedKafkaConfig]]
+    * @throws KafkaUnavailableException if unable to connect to Kafka
+    */
   def publishStringMessageToKafka(topic: String, message: String)(implicit config: EmbeddedKafkaConfig): Unit =
     publishToKafka(topic, message)(config, new StringSerializer)
 
   /**
-   * Publishes synchronously a message to the running Kafka broker.
-   *
-   * @param topic the topic to which publish the message (it will be auto-created)
-   * @param message the message of type [[T]] to publish
-   * @param config an implicit [[EmbeddedKafkaConfig]]
-   * @param serializer an implicit [[Serializer]] for the type [[T]]
-   * @throws KafkaUnavailableException if unable to connect to Kafka
-   */
+    * Publishes synchronously a message to the running Kafka broker.
+    *
+    * @param topic the topic to which publish the message (it will be auto-created)
+    * @param message the message of type [[T]] to publish
+    * @param config an implicit [[EmbeddedKafkaConfig]]
+    * @param serializer an implicit [[Serializer]] for the type [[T]]
+    * @throws KafkaUnavailableException if unable to connect to Kafka
+    */
   @throws(classOf[KafkaUnavailableException])
   def publishToKafka[T](topic: String, message: T)
                        (implicit config: EmbeddedKafkaConfig, serializer: Serializer[T]): Unit = {
@@ -92,20 +123,20 @@ trait EmbeddedKafka {
 
 
   /**
-   * Consumes the first message available in a given topic, deserializing it as a String.
-   *
-   * @param topic the topic to consume a message from
-   * @param config an implicit [[EmbeddedKafkaConfig]]
-   * @param decoder an implicit [[Decoder]] for the type [[T]]
-   * @return the first message consumed from the given topic, with a type [[T]]
-   * @throws TimeoutException if unable to consume a message within 3 seconds
-   * @throws KafkaUnavailableException if unable to connect to Kafka
-   */
+    * Consumes the first message available in a given topic, deserializing it as a String.
+    *
+    * @param topic the topic to consume a message from
+    * @param config an implicit [[EmbeddedKafkaConfig]]
+    * @param decoder an implicit [[Decoder]] for the type [[T]]
+    * @return the first message consumed from the given topic, with a type [[T]]
+    * @throws TimeoutException if unable to consume a message within 3 seconds
+    * @throws KafkaUnavailableException if unable to connect to Kafka
+    */
   @throws(classOf[TimeoutException])
   @throws(classOf[KafkaUnavailableException])
   def consumeFirstMessageFrom[T](topic: String)(implicit config: EmbeddedKafkaConfig, decoder: Decoder[T]): T = {
     val props = new Properties()
-    props.put("group.id", s"embedded-kafka-spec-$suiteId")
+    props.put("group.id", s"embedded-kafka-spec")
     props.put("zookeeper.connect", s"localhost:${config.zooKeeperPort}")
     props.put("auto.offset.reset", "smallest")
     props.put("zookeeper.connection.timeout.ms", "6000")
@@ -146,7 +177,7 @@ trait EmbeddedKafka {
     )
   }
 
-  private def startZooKeeper(zooKeeperPort: Int): ServerCnxnFactory = {
+  def startZooKeeper(zooKeeperPort: Int): ServerCnxnFactory = {
     val zkLogsDir = Directory.makeTemp("zookeeper-logs")
     val tickTime = 2000
 
@@ -158,7 +189,7 @@ trait EmbeddedKafka {
     factory
   }
 
-  private def startKafka(config: EmbeddedKafkaConfig): KafkaServer = {
+  def startKafka(config: EmbeddedKafkaConfig): KafkaServer = {
     val kafkaLogDir = Directory.makeTemp("kafka")
 
     val zkAddress = s"localhost:${config.zooKeeperPort}"
@@ -176,4 +207,4 @@ trait EmbeddedKafka {
     broker.startup()
     broker
   }
-}
+}

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

@@ -0,0 +1,39 @@
+package net.manub.embeddedkafka
+
+class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
+
+  "the EmbeddedKafka object" when {
+    "invoking the start and stop methods" should {
+      "start and stop Kafka and Zookeeper on the default ports" in {
+        EmbeddedKafka.start()
+
+        kafkaIsAvailable()
+        zookeeperIsAvailable()
+
+        EmbeddedKafka.stop()
+
+        kafkaIsNotAvailable()
+        zookeeperIsNotAvailable()
+      }
+
+      "start and stop Kafka and Zookeeper on different specified ports using an implicit configuration" in {
+        implicit val config = EmbeddedKafkaConfig(kafkaPort = 12345, zooKeeperPort = 54321)
+        EmbeddedKafka.start()
+
+        kafkaIsAvailable(12345)
+        zookeeperIsAvailable(54321)
+
+        EmbeddedKafka.stop()
+      }
+    }
+
+    "invoking the isRunnning method" should {
+      "return whether both Kafka and Zookeeper are running" in {
+        EmbeddedKafka.start()
+        EmbeddedKafka.isRunning shouldBe true
+        EmbeddedKafka.stop()
+        EmbeddedKafka.isRunning shouldBe false
+      }
+    }
+  }
+}