Added implicit conversion for filtering servers down to EmbeddedK or EmbeddedZ.

Author: Thomas Heslin

embedded-kafka/src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala

@@ -97,9 +97,7 @@ object EmbeddedKafka extends EmbeddedKafkaSupport {
     * Stops all in memory Zookeeper instances, preserving the logs directories.
     */
   def stopZooKeeper(): Unit = {
-    val factories = servers
-      .filter(_.isInstanceOf[EmbeddedZ])
-      .asInstanceOf[Seq[EmbeddedZ]]
+    val factories: Seq[EmbeddedZ] = servers.toFilteredSeq(isEmbeddedZ)
 
     factories
       .foreach(_.stop(false))
@@ -111,9 +109,7 @@ object EmbeddedKafka extends EmbeddedKafkaSupport {
     * Stops all in memory Kafka instances, preserving the logs directories.
     */
   def stopKafka(): Unit = {
-    val brokers = servers
-      .filter(_.isInstanceOf[EmbeddedK])
-      .asInstanceOf[Seq[EmbeddedK]]
+    val brokers: Seq[EmbeddedK] = servers.toFilteredSeq(isEmbeddedK)
 
     brokers
       .foreach(_.stop(false))
@@ -123,12 +119,21 @@ object EmbeddedKafka extends EmbeddedKafkaSupport {
 
   /**
     * Returns whether the in memory Kafka and Zookeeper are running.
+    *
+    * `brokers` val is required for type information.
     */
-  def isRunning: Boolean =
-    servers
-      .filter(_.isInstanceOf[EmbeddedK])
-      .asInstanceOf[Seq[EmbeddedK]]
-      .exists(_.factory.isDefined)
+  def isRunning: Boolean = {
+    val brokers: Seq[EmbeddedK] = servers.toFilteredSeq(isEmbeddedK)
+    brokers.exists(_.factory.isDefined)
+  }
+
+  private def isEmbeddedK(server: EmbeddedServer): Boolean = server.isInstanceOf[EmbeddedK]
+  private def isEmbeddedZ(server: EmbeddedServer): Boolean = server.isInstanceOf[EmbeddedZ]
+
+  implicit class ServerOps[T <: EmbeddedServer](servers: Seq[EmbeddedServer]) {
+    def toFilteredSeq(implicit filter: EmbeddedServer => Boolean): Seq[T] =
+      servers.filter(filter).asInstanceOf[Seq[T]]
+  }
 }
 
 sealed trait EmbeddedKafkaSupport {