Add NettyFuture spec (#247)

- improve NettyFuture in when it comes to resolved Futures
- add NettyFuture Spec

build.sbt

@@ -100,6 +100,6 @@ lazy val root = (project in file("."))
           scalaStm,
           slf4jApi,
           slf4jSimple,
-          akkaHttp(scalaV.value).cross(CrossVersion.for3Use2_13),
-          akkaStream(scalaV.value).cross(CrossVersion.for3Use2_13)))
+          akkaHttp(scalaVersion.value).cross(CrossVersion.binary),
+          akkaStream(scalaVersion.value).cross(CrossVersion.binary)))
 

project/Dependencies.scala

@@ -1,4 +1,4 @@
-import sbt._
+import sbt.*
 
 object Dependencies {
   lazy val Examples = config("example") extend Compile
@@ -14,10 +14,10 @@ object Dependencies {
 
   object akkaHttp {
 
-    def apply(scalaVer: ScalaVer) = {
-      val v = scalaVer match {
-        case ScalaVer._211 => "10.1.9"
-        case _             => "10.2.6"
+    def apply(scalaVer: String): ModuleID = {
+      val v = ScalaVer.fromString(scalaVer) match {
+        case Some(ScalaVer._211) => "10.1.9"
+        case _                   => "10.5.3"
 
       }
 
@@ -26,10 +26,10 @@ object Dependencies {
   }
 
   object akkaStream {
-    def apply(scalaVer: ScalaVer) = {
-      val v = scalaVer match {
-        case ScalaVer._211 => "2.5.32"
-        case _             => "2.6.16"
+    def apply(scalaVer: String): ModuleID = {
+      val v = ScalaVer.fromString(scalaVer) match {
+        case Some(ScalaVer._211) => "2.5.32"
+        case _                   => "2.8.5"
       }
 
       "com.typesafe.akka" %% "akka-stream" % v % Test

project/ScalaVer.scala

@@ -1,5 +1,6 @@
 import sbt.Keys.{crossScalaVersions, scalaVersion}
-import sbt.settingKey
+import sbt.librarymanagement.CrossVersion
+import sbt.{Def, settingKey}
 
 sealed abstract class ScalaVer(val full: String)
 
@@ -15,21 +16,21 @@ object ScalaVer {
 
   val values: Seq[ScalaVer] = Set(_211, _212, _213, _300).toSeq
 
-  val default: ScalaVer = _300
+  val default: ScalaVer = _211
 
   def fromEnv: Option[ScalaVer] = sys.env.get("SCALA_VER") flatMap fromString
 
-  def fromString(full: String): Option[ScalaVer] = full match {
-    case x if x startsWith "2.11" => Some(_211)
-    case x if x startsWith "2.12" => Some(_212)
-    case x if x startsWith "2.13" => Some(_213)
-    case x if x startsWith "3.0"  => Some(_300)
-    case _                        => None
+  def fromString(full: String): Option[ScalaVer] = CrossVersion.partialVersion(full) match {
+    case Some((2, 11)) => Some(_211)
+    case Some((2, 12)) => Some(_212)
+    case Some((2, 13)) => Some(_213)
+    case Some((3, _))  => Some(_300)
+    case _             => None
   }
 
   lazy val scalaV = settingKey[ScalaVer]("Current Scala Version")
 
-  def settings = Seq(
+  def settings: Seq[Def.Setting[? >: String & Seq[String] & ScalaVer <: Object]] = Seq(
     scalaVersion        := (ScalaVer.fromEnv getOrElse ScalaVer.default).full,
     crossScalaVersions  := ScalaVer.values.map(_.full),
     scalaV              := ScalaVer.fromString(scalaVersion.value) getOrElse ScalaVer.default)

src/main/scala/com/github/andyglow/websocket/util/NettyFuture.scala

@@ -10,14 +10,18 @@ import scala.concurrent.Promise
 object NettyFuture {
 
   def apply[T](f: NFuture[T]): Future[T] = {
-    val promise = Promise[T]()
-    val futureListener = new GenericFutureListener[NFuture[T]]() {
-      override def operationComplete(future: NFuture[T]): Unit = {
-        if (future.isSuccess) promise.success(future.getNow) else promise.failure(future.cause())
-        ()
+    if (f.isDone) {
+      if (f.isSuccess) Future.successful(f.getNow) else Future.failed(f.cause())
+    } else {
+      val promise = Promise[T]()
+      val futureListener = new GenericFutureListener[NFuture[T]]() {
+        override def operationComplete(future: NFuture[T]): Unit = {
+          if (future.isSuccess) promise.success(future.getNow) else promise.failure(future.cause())
+          ()
+        }
       }
+      f.addListener(futureListener)
+      promise.future
     }
-    f.addListener(futureListener)
-    promise.future
   }
 }

src/test/scala-2.11/com/github/andyglow/websocket/testserver/CrossScalaVersionAkkaApi.scala → src/test/scala-2.11/com/github/andyglow/websocket/testserver/PlatformDependent.scala

@@ -6,9 +6,10 @@ import akka.http.scaladsl.model.ws.UpgradeToWebSocket
 import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
 import akka.stream.Materializer
 
+import java.util.concurrent.Callable
 import scala.concurrent.Future
 
-trait CrossScalaVersionAkkaApi {
+trait PlatformDependent {
 
   @inline def launchServer(
     interface: String,
@@ -20,3 +21,10 @@ trait CrossScalaVersionAkkaApi {
 
   @inline def websocketAttribution(x: HttpRequest): Option[UpgradeToWebSocket] = x.header[UpgradeToWebSocket]
 }
+
+object PlatformDependent {
+
+  @inline def callable[T](fn: () => T): Callable[T] = new Callable[T] {
+    def call(): T = fn()
+  }
+}

src/test/scala-2.12/com/github/andyglow/websocket/testserver/CrossScalaVersionAkkaApi.scala → src/test/scala-2.12/com/github/andyglow/websocket/testserver/PlatformDependent.scala

@@ -6,9 +6,10 @@ import akka.http.scaladsl.model.ws.WebSocketUpgrade
 import akka.http.scaladsl.model.{AttributeKeys, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 
+import java.util.concurrent.Callable
 import scala.concurrent.Future
 
-trait CrossScalaVersionAkkaApi {
+trait PlatformDependent {
 
   @inline def launchServer(
     interface: String,
@@ -20,3 +21,8 @@ trait CrossScalaVersionAkkaApi {
 
   @inline def websocketAttribution(x: HttpRequest): Option[WebSocketUpgrade] = x.attribute(AttributeKeys.webSocketUpgrade)
 }
+
+object PlatformDependent {
+
+  @inline def callable[T](fn: () => T): Callable[T] = () => fn()
+}

src/test/scala-3/com/github/andyglow/websocket/testserver/CrossScalaVersionAkkaApi.scala → src/test/scala-2.13/com/github/andyglow/websocket/testserver/PlatformDependent.scala

@@ -6,9 +6,10 @@ import akka.http.scaladsl.model.ws.WebSocketUpgrade
 import akka.http.scaladsl.model.{AttributeKeys, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 
+import java.util.concurrent.Callable
 import scala.concurrent.Future
 
-trait CrossScalaVersionAkkaApi {
+trait PlatformDependent {
 
   @inline def launchServer(
     interface: String,
@@ -20,3 +21,8 @@ trait CrossScalaVersionAkkaApi {
 
   @inline def websocketAttribution(x: HttpRequest): Option[WebSocketUpgrade] = x.attribute(AttributeKeys.webSocketUpgrade)
 }
+
+object PlatformDependent {
+
+  @inline def callable[T](fn: () => T): Callable[T] = () => fn()
+}

src/test/scala-2.13/com/github/andyglow/websocket/testserver/CrossScalaVersionAkkaApi.scala → src/test/scala-3/com/github/andyglow/websocket/testserver/PlatformDependent.scala

@@ -6,11 +6,12 @@ import akka.http.scaladsl.model.ws.WebSocketUpgrade
 import akka.http.scaladsl.model.{AttributeKeys, HttpRequest, HttpResponse}
 import akka.stream.Materializer
 
+import java.util.concurrent.Callable
 import scala.concurrent.Future
 
-trait CrossScalaVersionAkkaApi {
+trait PlatformDependent {
 
-  def launchServer(
+  @inline def launchServer(
     interface: String,
     port: Int,
     handler: HttpRequest => HttpResponse)(implicit system: ActorSystem, mat: Materializer): Future[Http.ServerBinding] = {
@@ -20,3 +21,8 @@ trait CrossScalaVersionAkkaApi {
 
   @inline def websocketAttribution(x: HttpRequest): Option[WebSocketUpgrade] = x.attribute(AttributeKeys.webSocketUpgrade)
 }
+
+object PlatformDependent {
+
+  @inline def callable[T](fn: () => T): Callable[T] = () => fn()
+}

src/test/scala/com/github/andyglow/websocket/testserver/TestServer.scala

@@ -7,7 +7,7 @@ import akka.stream._
 import akka.stream.scaladsl.Flow
 
 
-object TestServer extends CrossScalaVersionAkkaApi {
+object TestServer extends PlatformDependent {
 
   def main(args: Array[String]): Unit = {
     implicit val system: ActorSystem = ActorSystem()

src/test/scala/com/github/andyglow/websocket/util/NettyFutureSpec.scala

@@ -0,0 +1,57 @@
+package com.github.andyglow.websocket.util
+
+import io.netty.channel.nio.NioEventLoopGroup
+import io.netty.util.concurrent.EventExecutorGroup
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.OptionValues._
+import org.scalatest.TryValues._
+import org.scalatest.matchers.should.Matchers._
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import com.github.andyglow.websocket.testserver.PlatformDependent._
+
+class NettyFutureSpec extends AnyWordSpec with BeforeAndAfterAll {
+
+  private var executors: EventExecutorGroup = _
+
+  override def beforeAll(): Unit = {
+    executors = new NioEventLoopGroup()
+  }
+
+  override def afterAll(): Unit = {
+    executors.shutdownGracefully().syncUninterruptibly()
+    ()
+  }
+
+  "NettyFuture" should {
+
+    "handled resolved successful futures" in {
+      val fut = NettyFuture(executors.submit(callable(() => "foo")).await())
+      fut.value.isDefined shouldBe true
+      fut.value.value.isSuccess shouldBe true
+      fut.value.value.get shouldBe "foo"
+    }
+
+    "handled un-resolved successful futures" in {
+      val fut = NettyFuture(executors.submit(callable(() => "foo")))
+      val v = Await.result(fut, 1.second)
+      v shouldBe "foo"
+    }
+
+    "handled resolved failed futures" in {
+      val fut = NettyFuture(executors.submit(callable[String](() => throw new IllegalStateException("err"))).await())
+      fut.value.isDefined shouldBe true
+      fut.value.value.isSuccess shouldBe false
+      fut.value.value.failure.exception shouldBe a [IllegalStateException]
+      fut.value.value.failure.exception.getMessage shouldBe "err"
+    }
+
+    "handled un-resolved failed futures" in {
+      val fut = NettyFuture(executors.submit(callable[String](() => throw new IllegalStateException("err"))))
+      val v = the [IllegalStateException] thrownBy { Await.result(fut, 1.second) }
+      v.getMessage shouldBe "err"
+    }
+  }
+}