Merge pull request #276 from outr/cats

Cats

app/jvm/src/main/scala/io/youi/app/OfflineGenerator.scala

@@ -25,8 +25,8 @@ class OfflineGenerator(application: ServerApplication,
                 file.getParentFile.mkdirs()
                 scribe.info(s"Writing $path to ${file.getAbsolutePath}..")
                 content match {
-                  case c: StringContent => IO.stream(c.value, file)
-                  case c: URLContent => IO.stream(c.url, file)
+                  case c: StringContent => Stream.apply(c.value, file)
+                  case c: URLContent => Stream.apply(c.url, file)
                   case _ => throw new RuntimeException(s"Unsupported Content-Type: $content")
                 }
               }

app/jvm/src/main/scala/io/youi/app/ServerApplication.scala

@@ -254,7 +254,7 @@ trait ServerApplication extends YouIApplication with Server {
     val directory = cacheDirectory()
     val file = new File(directory, path)
     file.getParentFile.mkdirs()
-    IO.stream(new java.net.URL(url.toString), file)
+    Stream.apply(new java.net.URL(url.toString), file)
     val content = Content.file(file)
     handler.matcher(http.path.exact(path)).resource(content)
     path

app/jvm/src/main/scala/io/youi/upload/UploadManager.scala

@@ -6,7 +6,7 @@ import io.youi.http.content.{Content, FormDataContent}
 import io.youi.http.{HttpConnection, HttpStatus}
 import io.youi.net._
 import io.youi.server.handler.HttpHandler
-import io.youi.stream.IO
+import io.youi.stream.Stream
 import reactify.Channel
 
 import scala.concurrent.Future
@@ -58,7 +58,7 @@ case class UploadManager(path: Path = path"/upload",
                 val slices = content.string("slices").value.split(',').toList
                 val sources = slices.map(fn => new File(directory, fn))
                 val destination = File.createTempFile(fileName, s".$ext", directory)
-                IO.merge(sources, destination)
+                Stream.merge(sources, destination)
                 received @= UploadedFile(destination, fileName)
                 destination.getName
               }

build.sbt

@@ -6,15 +6,11 @@ ThisBuild / organization := "io.youi"
 ThisBuild / version := "0.15.0-SNAPSHOT"
 ThisBuild / scalaVersion := "2.13.8"
 ThisBuild / crossScalaVersions := List("2.13.8", "2.12.16")
-ThisBuild / resolvers ++= Seq(
-  Resolver.sonatypeRepo("releases"),
-  Resolver.sonatypeRepo("snapshots")
-)
 ThisBuild / scalacOptions ++= Seq("-unchecked", "-deprecation", "-feature")
 
 ThisBuild / publishTo := sonatypePublishToBundle.value
 ThisBuild / sonatypeProfileName := "io.youi"
-ThisBuild / publishMavenStyle := true
+//ThisBuild / publishMavenStyle := true
 ThisBuild / licenses := Seq("MIT" -> url("https://github.com/outr/youi/blob/master/LICENSE"))
 ThisBuild / sonatypeProjectHosting := Some(xerial.sbt.Sonatype.GitHubHosting("outr", "youi", "matt@outr.com"))
 ThisBuild / homepage := Some(url("https://github.com/outr/youi"))
@@ -70,6 +66,8 @@ val fs2Version: String = "3.2.12"
 
 val scalaTestVersion: String = "3.2.13"
 
+val catsEffectTestVersion: String = "1.4.0"
+
 ThisBuild / evictionErrorLevel := Level.Info
 
 lazy val root = project.in(file("."))
@@ -93,7 +91,8 @@ lazy val core = crossProject(JSPlatform, JVMPlatform).in(file("core"))
       "com.outr" %%% "reactify" % reactifyVersion,
       "org.typelevel" %%% "cats-effect" % catsVersion,
       "co.fs2" %% "fs2-core" % fs2Version,
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .jsSettings(
@@ -109,7 +108,8 @@ lazy val client = crossProject(JSPlatform, JVMPlatform).in(file("client"))
   .settings(
     name := "youi-client",
     libraryDependencies ++= Seq(
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .jvmSettings(
@@ -126,7 +126,8 @@ lazy val spatial = crossProject(JSPlatform, JVMPlatform).in(file("spatial"))
   .settings(
     name := "youi-spatial",
     libraryDependencies ++= Seq(
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .dependsOn(core)
@@ -138,7 +139,8 @@ lazy val stream = project.in(file("stream"))
   .settings(
     name := "youi-stream",
     libraryDependencies ++= Seq(
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .dependsOn(coreJVM)
@@ -149,7 +151,8 @@ lazy val dom = project.in(file("dom"))
     name := "youi-dom",
     libraryDependencies ++= Seq(
       "com.outr" %%% "profig" % profigVersion,
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     ),
     test := {},     // TODO: figure out why this no longer works
     jsEnv := new org.scalajs.jsenv.jsdomnodejs.JSDOMNodeJSEnv()
@@ -164,7 +167,8 @@ lazy val communication = crossProject(JSPlatform, JVMPlatform)
     name := "youi-communication",
     libraryDependencies ++= Seq(
       "org.scala-lang" % "scala-reflect" % scalaVersion.value,
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .dependsOn(core)
@@ -177,7 +181,9 @@ lazy val server = project.in(file("server"))
     name := "youi-server",
     libraryDependencies ++= Seq(
       "net.sf.uadetector" % "uadetector-resources" % uaDetectorVersion,
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.typelevel" %% "cats-effect" % catsVersion,
+      "org.scalatest" %% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .dependsOn(communicationJVM, stream)
@@ -188,7 +194,8 @@ lazy val serverUndertow = project.in(file("serverUndertow"))
     fork := true,
     libraryDependencies ++= Seq(
       "io.undertow" % "undertow-core" % undertowVersion,
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .dependsOn(server, clientJVM % "test->test")
@@ -228,7 +235,8 @@ lazy val app = crossProject(JSPlatform, JVMPlatform).in(file("app"))
   .settings(
     name := "youi-app",
     libraryDependencies ++= Seq(
-      "org.scalatest" %%% "scalatest" % scalaTestVersion % "test"
+      "org.scalatest" %%% "scalatest" % scalaTestVersion % Test,
+      "org.typelevel" %%% "cats-effect-testing-scalatest" % catsEffectTestVersion % Test
     )
   )
   .dependsOn(core, communication)
@@ -244,7 +252,6 @@ lazy val example = crossApplication.in(file("example"))
     jsEnv := new org.scalajs.jsenv.jsdomnodejs.JSDOMNodeJSEnv()
   )
   .jvmSettings(
-    scalaJSUseMainModuleInitializer := true,
     libraryDependencies ++= Seq(
       "org.scala-lang" % "scala-reflect" % scalaVersion.value,
       "org.scala-lang.modules" %% "scala-xml" % scalaXMLVersion

client/js/src/main/scala/io/youi/client/JSHttpClientImplementation.scala

@@ -1,17 +1,17 @@
 package io.youi.client
 
+import cats.effect.IO
 import io.youi.ajax.{AjaxAction, AjaxRequest}
 import io.youi.http.content._
 import io.youi.http.{Headers, HttpRequest, HttpResponse, HttpStatus}
 import io.youi.net.ContentType
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.util.{Failure, Success, Try}
 
 class JSHttpClientImplementation(config: HttpClientConfig) extends HttpClientImplementation(config) {
   private val HeaderRegex = """(.+)[:](.+)""".r
 
-  override def send(request: HttpRequest, executionContext: ExecutionContext): Future[HttpResponse] = {
-    implicit val implicitContext: ExecutionContext = executionContext
+  override def send(request: HttpRequest): IO[Try[HttpResponse]] = {
     val manager = config.connectionPool.asInstanceOf[JSConnectionPool].manager
     val ajaxRequest = new AjaxRequest(
       url = request.url,
@@ -22,25 +22,27 @@ class JSHttpClientImplementation(config: HttpClientConfig) extends HttpClientImp
       responseType = ""
     )
     val action = new AjaxAction(ajaxRequest)
-    manager.enqueue(action).map { xmlHttpRequest =>
-      val headers: Map[String, List[String]] = xmlHttpRequest.getAllResponseHeaders().split('\n').map(_.trim).map {
-        case HeaderRegex(key, value) => key.trim -> value.trim
-        case s => throw new RuntimeException(s"Invalid Header: [$s]")
-      }.groupBy(_._1).map {
-        case (key, array) => key -> array.toList.map(_._2)
-      }
-      val content = xmlHttpRequest.responseType match {
-        case null => None
-        case _ => {
-          val `type` = if (xmlHttpRequest.responseType == "") ContentType.`text/plain` else ContentType.parse(xmlHttpRequest.responseType)
-          Some(Content.string(xmlHttpRequest.responseText, `type`))
+    manager.enqueue(action).map {
+      case Failure(err) => Failure(err)
+      case Success(xmlHttpRequest) =>
+        val headers: Map[String, List[String]] = xmlHttpRequest.getAllResponseHeaders().split('\n').map(_.trim).map {
+          case HeaderRegex(key, value) => key.trim -> value.trim
+          case s => throw new RuntimeException(s"Invalid Header: [$s]")
+        }.groupBy(_._1).map {
+          case (key, array) => key -> array.toList.map(_._2)
         }
-      }
-      HttpResponse(
-        status = HttpStatus(xmlHttpRequest.status, xmlHttpRequest.statusText),
-        headers = Headers(headers),
-        content = content
-      )
+        val content = xmlHttpRequest.responseType match {
+          case null => None
+          case _ => {
+            val `type` = if (xmlHttpRequest.responseType == "") ContentType.`text/plain` else ContentType.parse(xmlHttpRequest.responseType)
+            Some(Content.string(xmlHttpRequest.responseText, `type`))
+          }
+        }
+        Success(HttpResponse(
+          status = HttpStatus(xmlHttpRequest.status, xmlHttpRequest.statusText),
+          headers = Headers(headers),
+          content = content
+        ))
     }
   }
 

client/jvm/src/main/scala/io/youi/client/JVMHttpClientImplementation.scala

@@ -1,23 +1,26 @@
 package io.youi.client
 
+import cats.effect.{Deferred, IO}
+
 import java.io.{File, IOException}
 import java.net.{InetAddress, Socket}
 import java.security.SecureRandom
 import java.security.cert.X509Certificate
 import java.util
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicLong
-
 import io.youi.http._
 import io.youi.http.content._
 import io.youi.net.ContentType
+import io.youi.stream
 import io.youi.stream._
+
 import javax.net.ssl._
 import okhttp3.Dns
 
-import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.collection.mutable
 import scala.jdk.CollectionConverters._
-import scala.util.{Failure, Success}
+import scala.util.{Failure, Success, Try}
 
 /**
   * Asynchronous HttpClient for simple request response support.
@@ -118,19 +121,17 @@ class JVMHttpClientImplementation(config: HttpClientConfig) extends HttpClientIm
     HttpsURLConnection.setDefaultHostnameVerifier(allHostsValid)
   }*/
 
-  override def send(request: HttpRequest,
-                    executionContext: ExecutionContext): Future[HttpResponse] = {
+  override def send(request: HttpRequest): IO[Try[HttpResponse]] = Deferred[IO, Try[HttpResponse]].flatMap { deferred =>
     val req = requestToOk(request)
-    val promise = Promise[HttpResponse]()
     client.newCall(req).enqueue(new okhttp3.Callback {
       override def onResponse(call: okhttp3.Call, res: okhttp3.Response): Unit = {
         val response = responseFromOk(res)
-        promise.success(response)
+        deferred.complete(Success(response))
       }
 
-      override def onFailure(call: okhttp3.Call, exc: IOException): Unit = promise.failure(exc)
+      override def onFailure(call: okhttp3.Call, exc: IOException): Unit = deferred.complete(Failure(exc))
     })
-    JVMHttpClientImplementation.process(promise.future)(executionContext)
+    JVMHttpClientImplementation.process(deferred.get)
   }
 
   private def requestToOk(request: HttpRequest): okhttp3.Request = {
@@ -198,7 +199,7 @@ class JVMHttpClientImplementation(config: HttpClientConfig) extends HttpClientIm
       } else {
         val suffix = contentType.extension.getOrElse("client")
         val file = File.createTempFile("youi", s".$suffix", new File(config.saveDirectory))
-        IO.stream(responseBody.byteStream(), file)
+        stream.Stream.apply(responseBody.byteStream(), file)
         Content.file(file, contentType)
       }
     }
@@ -218,7 +219,7 @@ class JVMHttpClientImplementation(config: HttpClientConfig) extends HttpClientIm
   override def content2String(content: Content): String = content match {
     case c: StringContent => c.value
     case c: BytesContent => String.valueOf(c.value)
-    case c: FileContent => IO.stream(c.file, new StringBuilder).toString
+    case c: FileContent => stream.Stream.apply(c.file, new mutable.StringBuilder).toString
     case _ => throw new RuntimeException(s"$content not supported")
   }
 
@@ -229,7 +230,7 @@ class JVMHttpClientImplementation(config: HttpClientConfig) extends HttpClientIm
   protected def content2Bytes(content: Content): Array[Byte] = content match {
     case c: StringContent => c.value.getBytes("UTF-8")
     case c: BytesContent => c.value
-    case c: FileContent => IO.stream(c.file, new StringBuilder).toString.getBytes("UTF-*")
+    case c: FileContent => stream.Stream.apply(c.file, new mutable.StringBuilder).toString.getBytes("UTF-*")
     case _ => throw new RuntimeException(s"$content not supported")
   }
 
@@ -245,20 +246,21 @@ object JVMHttpClientImplementation {
   private[client] val _successful = new AtomicLong(0L)
   private[client] val _failure = new AtomicLong(0L)
 
-  private[client] def process(future: Future[HttpResponse])(implicit executionContext: ExecutionContext): Future[HttpResponse] = {
+  private[client] def process(io: IO[Try[HttpResponse]]): IO[Try[HttpResponse]] = {
     _total.incrementAndGet()
     _active.incrementAndGet()
-    future.onComplete {
-      case Success(_) => {
-        _successful.incrementAndGet()
-        _active.decrementAndGet()
-      }
-      case Failure(_) => {
-        _failure.incrementAndGet()
-        _active.decrementAndGet()
-      }
+    io.flatMap { t =>
+      IO {
+        t match {
+          case Success(_) =>
+            _successful.incrementAndGet()
+            _active.decrementAndGet()
+          case Failure(_) =>
+            _failure.incrementAndGet()
+            _active.decrementAndGet()
+        }
+      }.map(_ => t)
     }
-    future
   }
 
   def total: Long = _total.get()