Add Scala 2.13.1 to crossScalaVersions and update the dependencies

.travis.yml

@@ -2,15 +2,17 @@ language : scala
 
 scala:
   - 2.11.12
-  - 2.12.4
+  - 2.12.10
+  - 2.13.1
 
 cache:
   directories:
   - $HOME/.ivy2
   - $HOME/.sbt
 
 jdk:
-  - oraclejdk8
+  - openjdk8
+  - openjdk11
 
 script:
   - sbt ++$TRAVIS_SCALA_VERSION -Dfile.encoding=UTF8 "project fs2-zk" test

build.sbt

@@ -9,7 +9,7 @@ lazy val contributors = Seq(
 lazy val commonSettings = Seq(
   organization := "com.spinoco",
   scalaVersion := "2.11.12",
-  crossScalaVersions := Seq("2.11.12", "2.12.4"),
+  crossScalaVersions := Seq("2.11.12", "2.12.10", "2.13.1"),
   scalacOptions ++= Seq(
     "-feature",
     "-deprecation",
@@ -18,20 +18,26 @@ lazy val commonSettings = Seq(
     "-language:existentials",
     "-language:postfixOps",
     "-Xfatal-warnings",
-    "-Yno-adapted-args",
-    "-Ywarn-value-discard",
-    "-Ywarn-unused-import"
-  ),
+    "-Ywarn-value-discard"
+  ) ++ {
+    if (scalaVersion.value.startsWith("2.13."))
+      Seq.empty
+    else
+      Seq(
+        "-Yno-adapted-args",
+        "-Ywarn-unused-import"
+      )
+  },
   scalacOptions in (Compile, console) ~= {_.filterNot("-Ywarn-unused-import" == _)},
   scalacOptions in (Test, console) := (scalacOptions in (Compile, console)).value,
   libraryDependencies ++= Seq(
-    "org.scalatest" %% "scalatest" % "3.0.0" % "test"
-    , "org.scalacheck" %% "scalacheck" % "1.13.4" % "test"
-    , "org.slf4j" % "slf4j-simple" % "1.6.1" % "test" // uncomment this for logs when testing
+    "org.scalatest" %% "scalatest" % "3.0.8" % "test"
+    , "org.scalacheck" %% "scalacheck" % "1.14.1" % "test"
+    , "org.slf4j" % "slf4j-simple" % "1.7.28" % "test" // uncomment this for logs when testing
 
-    , "co.fs2" %% "fs2-core" % "1.0.0"
-    , "co.fs2" %% "fs2-io" % "1.0.0"
-    , "org.apache.zookeeper" % "zookeeper" % "3.4.10"
+    , "co.fs2" %% "fs2-core" % "2.0.1"
+    , "co.fs2" %% "fs2-io" % "2.0.1"
+    , "org.apache.zookeeper" % "zookeeper" % "3.5.5"
 
   ),
   scmInfo := Some(ScmInfo(url("https://github.com/Spinoco/fs2-zk"), "[email protected]:Spinoco/fs2-zk.git")),
@@ -109,7 +115,7 @@ lazy val `fs2-zk` =
   .settings(commonSettings)
   .settings(
    name := "fs2-zk"
-  ) 
- 
- 
+  )
+
+
 

project/build.properties

@@ -1 +1 @@
-sbt.version=1.1.6
+sbt.version=1.3.2

src/main/scala-2.11/spinoco/fs2/zk/ListConverters.scala

@@ -0,0 +1,11 @@
+package spinoco.fs2.zk
+
+import scala.collection.JavaConverters._
+
+private[zk] object ListConverters {
+  def toScalaList[T](list: java.util.List[T]): List[T] =
+    list.asScala.toList
+
+  def toJavaList[T](list: List[T]): java.util.List[T] =
+    list.asJava
+}

src/main/scala-2.12/spinoco/fs2/zk/ListConverters.scala

@@ -0,0 +1,11 @@
+package spinoco.fs2.zk
+
+import scala.collection.JavaConverters._
+
+private[zk] object ListConverters {
+  def toScalaList[T](list: java.util.List[T]): List[T] =
+    list.asScala.toList
+
+  def toJavaList[T](list: List[T]): java.util.List[T] =
+    list.asJava
+}

src/main/scala-2.13/spinoco/fs2/zk/ListConverters.scala

@@ -0,0 +1,11 @@
+package spinoco.fs2.zk
+
+import scala.jdk.CollectionConverters._
+
+private[zk] object ListConverters {
+  def toScalaList[T](list: java.util.List[T]): List[T] =
+    list.asScala.toList
+
+  def toJavaList[T](list: List[T]): java.util.List[T] =
+    list.asJava
+}

src/main/scala/spinoco/fs2/zk/ZkClient.scala

@@ -15,9 +15,9 @@ import org.apache.zookeeper.AsyncCallback._
 import org.apache.zookeeper.Watcher.Event.EventType
 import org.apache.zookeeper._
 import org.apache.zookeeper.data.{ACL, Id, Stat}
+import spinoco.fs2.zk.ListConverters.{toScalaList, toJavaList}
 import spinoco.fs2.zk.ZkACL.Permission
 
-import scala.collection.JavaConverters._
 import scala.concurrent.duration.FiniteDuration
 import scala.util.Success
 
@@ -333,7 +333,7 @@ object ZkClient {
           def result:Option[(List[ZkNode], ZkStat)] = {
             if (children == null) Some(List.empty -> zkStats(stat))
             else {
-              val result = children.asScala.toList.map(parent / _).collect { case Success(zkn) => zkn}
+              val result = toScalaList(children).map(parent / _).collect { case Success(zkn) => zkn }
               Some(result -> zkStats(stat))
             }
           }
@@ -395,7 +395,7 @@ object ZkClient {
       def readF: F[(O, F[Unit])] = {
         Monad[F].flatMap(mkWatcher[F]){ case (watcher, awaitWatch) =>
         Async[F].async[(O, F[Unit])] { cb =>
-          register({ r => cb(r.right.map(_ -> awaitWatch))}, watcher)
+          register({ r => cb(r.map(_ -> awaitWatch))}, watcher)
         }}
       }
 
@@ -417,14 +417,14 @@ object ZkClient {
       }
     }
 
-    def fromZkACL(acls: List[ZkACL]): JList[ACL] = {
+    def fromZkACL(acls: List[ZkACL]): JList[ACL] = toJavaList {
       acls.map { zkAcl =>
         new ACL(zkAcl.permission.value,new Id(zkAcl.scheme, zkAcl.entity) )
-      }.asJava
+      }
     }
 
     def toZkACL(acls: JList[ACL]): List[ZkACL] = {
-      acls.asScala.toList.map { acl =>
+      toScalaList(acls).map { acl =>
         ZkACL(Permission(acl.getPerms),acl.getId.getScheme, acl.getId.getId)
       }
     }
@@ -445,7 +445,7 @@ object ZkClient {
     }
 
     def fromOpResult(results: JList[OpResult]): List[ZkOpResult] = {
-      results.asScala.toList.map {
+      toScalaList(results).map {
         case create: OpResult.CreateResult => ZkOpResult.CreateResult(create.getPath)
         case _: OpResult.DeleteResult => ZkOpResult.DeleteResult
         case data: OpResult.SetDataResult => ZkOpResult.SetDataResult(zkStats(data.getStat))
@@ -454,13 +454,13 @@ object ZkClient {
       }
     }
 
-    def toOp(ops: List[ZkOp]): JList[Op] = {
+    def toOp(ops: List[ZkOp]): JList[Op] = toJavaList {
       ops.map {
         case ZkOp.Create(node, mode, data, acl) => Op.create(node.path,data.map(_.toArray).orNull,fromZkACL(acl),zkCreateMode(mode))
         case ZkOp.Delete(node, version) => Op.delete(node.path, version.getOrElse(-1))
         case ZkOp.SetData(node, data, version) => Op.setData(node.path, data.map(_.toArray).orNull, version.getOrElse(-1))
         case ZkOp.Check(node, version) => Op.check(node.path, version.getOrElse(-1))
-      }.asJava
+      }
     }
 
   }

src/main/scala/spinoco/fs2/zk/ZkClientState.scala

@@ -12,6 +12,7 @@ object ZkClientState extends Enumeration {
       , ConnectedReadOnly
       , SASLAuthenticated
       , Expired
+      , Closed
       = Value
 
   def fromZk(s:KeeperState):ZkClientState.Value = {
@@ -22,6 +23,7 @@ object ZkClientState extends Enumeration {
       case KeeperState.ConnectedReadOnly => ConnectedReadOnly
       case KeeperState.SaslAuthenticated => SASLAuthenticated
       case KeeperState.Expired => Expired
+      case KeeperState.Closed => Closed
     }
   }
-}
+}

src/test/scala/spinoco/fs2/zk/Fs2ZkClientSpec.scala

@@ -4,17 +4,17 @@ import cats.effect.{ContextShift, IO, Timer}
 import fs2.Stream._
 import fs2._
 import org.scalatest.concurrent.{Eventually, TimeLimitedTests}
-import org.scalatest.prop.GeneratorDrivenPropertyChecks
 import org.scalatest.time.SpanSugar._
 import org.scalatest.{FreeSpec, Matchers}
+import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks
 
 import scala.concurrent.ExecutionContext
 
 /**
   * Created by pach on 14/05/16.
   */
 class Fs2ZkClientSpec extends FreeSpec
-  with GeneratorDrivenPropertyChecks
+  with ScalaCheckDrivenPropertyChecks
   with Matchers
   with TimeLimitedTests
   with Eventually {

src/test/scala/spinoco/fs2/zk/ZkClientSpec.scala

@@ -46,7 +46,7 @@ class ZkClientSpec extends Fs2ZkClientSpec {
           observe concurrently modify
         }
         .map { _.map(_.dataLength) }
-        .take(4).compile.toVector.unsafeRunTimed(5.seconds)
+        .take(4).compile.toVector.unsafeRunTimed(1.minute)
 
       result shouldBe Some(Vector(
         None
@@ -81,7 +81,7 @@ class ZkClientSpec extends Fs2ZkClientSpec {
            observe concurrently modify
          }
          . map { _.map(_._1) }
-         .take(9).compile.toVector.unsafeRunTimed(10.seconds)
+         .take(9).compile.toVector.unsafeRunTimed(1.minute)
 
       result shouldBe Some(Vector(
         None
@@ -107,7 +107,7 @@ class ZkClientSpec extends Fs2ZkClientSpec {
           observe concurrently (shutdown ++ startup)
         }
         .take(3)
-        .compile.toVector.unsafeRunTimed(10.seconds)
+        .compile.toVector.unsafeRunTimed(1.minute)
 
       result shouldBe Some(Vector(
         ZkClientState.SyncConnected

src/test/scala/spinoco/fs2/zk/ZkSpecServer.scala

@@ -1,6 +1,5 @@
 package spinoco.fs2.zk
 
-import java.io.File
 import java.net.InetSocketAddress
 import java.nio.file.{Files, Path}
 import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
@@ -119,7 +118,13 @@ object ZkSpecServer {
         def shutdown: F[Unit] = F.suspend { F.pure {
           val server = runningServer.get()
           runningServer.set(null)
-          if (server != null)  { server.getServerCnxnFactory.shutdown(); server.shutdown() }
+          if (server != null) {
+            val connectionsFactory = server.getServerCnxnFactory
+            if (connectionsFactory != null)
+              connectionsFactory.shutdown()
+
+            server.shutdown()
+          }
         }}
       }
     }
@@ -133,7 +138,7 @@ object ZkSpecServer {
     def configureServer[F[_]](config:ServerConfig)(implicit F:Effect[F]): F[ZooKeeperServer] = F.delay {
       val zkServer: ZooKeeperServer = new ZooKeeperServer
 
-      val txnLog = new FileTxnSnapLog(new File(config.getDataDir), new File(config.getDataLogDir))
+      val txnLog = new FileTxnSnapLog(config.getDataDir, config.getDataLogDir)
       zkServer.setTxnLogFactory(txnLog)
       zkServer.setTickTime(config.getTickTime)
       zkServer.setMinSessionTimeout(config.getMinSessionTimeout)