Unify TCP and UnixSocket APIs

build.sbt

@@ -272,7 +272,55 @@ ThisBuild / mimaBinaryIssueFilters ++= Seq(
   ),
   ProblemFilters.exclude[MissingTypesProblem](
     "fs2.interop.flow.StreamSubscriber$State$WaitingOnUpstream$"
-  )
+  ),
+  // Network refactor: #3563
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Network.connect"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Network.bind"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Network.bindAndAccept"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Socket.address"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Socket.peerAddress"),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("fs2.io.net.Socket.address"),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("fs2.io.net.Socket.supportedOptions"),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("fs2.io.net.Socket.getOption"),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("fs2.io.net.Socket.setOption"),
+  ProblemFilters.exclude[DirectMissingMethodProblem](
+    "fs2.io.net.SocketCompanionPlatform#AsyncSocket.this"
+  ),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.SocketGroup$AbstractAsyncSocketGroup"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.SocketGroupCompanionPlatform"),
+  ProblemFilters.exclude[MissingClassProblem](
+    "fs2.io.net.SocketGroupCompanionPlatform$AsyncSocketGroup"
+  ),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("fs2.io.net.tls.TLSSocket.address"),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem](
+    "fs2.io.net.tls.TLSSocket.supportedOptions"
+  ),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("fs2.io.net.tls.TLSSocket.getOption"),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("fs2.io.net.tls.TLSSocket.setOption"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.unixsocket.JdkUnixSockets"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.unixsocket.JdkUnixSockets$"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.unixsocket.JdkUnixSocketsImpl"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.unixsocket.JnrUnixSockets"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.unixsocket.JnrUnixSockets$"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.unixsocket.JnrUnixSocketsImpl"),
+  ProblemFilters.exclude[MissingClassProblem](
+    "fs2.io.net.unixsocket.UnixSocketsCompanionPlatform$AsyncSocket"
+  ),
+  ProblemFilters.exclude[MissingClassProblem](
+    "fs2.io.net.unixsocket.UnixSocketsCompanionPlatform$AsyncUnixSockets"
+  ),
+  ProblemFilters.exclude[IncompatibleMethTypeProblem]("fs2.io.net.SelectingSocket.apply"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.SelectingSocketGroup"),
+  ProblemFilters.exclude[DirectMissingMethodProblem]("fs2.io.net.Socket.forAsync"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem](
+    "fs2.io.net.SocketOptionCompanionPlatform#Key.get"
+  ),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem](
+    "fs2.io.net.Network.openDatagramSocket"
+  ),
+  ProblemFilters.exclude[IncompatibleMethTypeProblem]("fs2.io.net.FdPollingSocket.apply"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.FdPollingSocketGroup"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.unixsocket.FdPollingUnixSockets")
 )
 
 lazy val root = tlCrossRootProject
@@ -353,7 +401,7 @@ lazy val io = crossProject(JVMPlatform, JSPlatform, NativePlatform)
   .settings(
     name := "fs2-io",
     tlVersionIntroduced ~= { _.updated("3", "3.1.0") },
-    libraryDependencies += "com.comcast" %%% "ip4s-core" % "3.6.0",
+    libraryDependencies += "com.comcast" %%% "ip4s-core" % "3.7.0",
     tlJdkRelease := None
   )
   .jvmSettings(

io/js/src/main/scala/fs2/io/internal/facade/net.scala

@@ -41,7 +41,7 @@ private[io] object net {
   @js.native
   trait Server extends EventEmitter {
 
-    def address(): ServerAddress = js.native
+    def address(): js.UndefOr[ServerAddress] = js.native
 
     def listening: Boolean = js.native
 
@@ -110,6 +110,7 @@ private[io] object net {
 
     def setTimeout(timeout: Double): Socket = js.native
 
+    def timeout: Double = js.native
   }
 
 }

io/js/src/main/scala/fs2/io/net/AsyncSocketsProvider.scala

@@ -0,0 +1,181 @@
+/*
+ * Copyright (c) 2013 Functional Streams for Scala
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ * the Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package fs2
+package io
+package net
+
+import cats.effect.{Async, Resource}
+import cats.effect.std.Dispatcher
+import cats.effect.syntax.all._
+import cats.syntax.all._
+import com.comcast.ip4s.{Host, IpAddress, Port, SocketAddress, UnixSocketAddress}
+import fs2.concurrent.Channel
+import fs2.io.internal.facade
+
+import scala.scalajs.js
+
+private[net] abstract class AsyncSocketsProvider[F[_]](implicit F: Async[F]) {
+
+  private def setSocketOptions(options: List[SocketOption])(socket: facade.net.Socket): F[Unit] =
+    options.traverse_(option => option.key.set(socket, option.value))
+
+  protected def connectIpOrUnix(
+      to: Either[SocketAddress[Host], UnixSocketAddress],
+      options: List[SocketOption]
+  ): Resource[F, Socket[F]] =
+    (for {
+      sock <- Resource
+        .make(
+          F.delay(
+            new facade.net.Socket(new facade.net.SocketOptions { allowHalfOpen = true })
+          )
+        )(sock =>
+          F.delay {
+            if (!sock.destroyed)
+              sock.destroy()
+          }
+        )
+        .evalTap(setSocketOptions(options))
+
+      _ <- F
+        .async[Unit] { cb =>
+          sock
+            .registerOneTimeListener[F, js.Error]("error") { error =>
+              cb(Left(js.JavaScriptException(error)))
+            } <* F.delay {
+            to match {
+              case Left(addr) =>
+                sock.connect(addr.port.value, addr.host.toString, () => cb(Right(())))
+              case Right(addr) =>
+                sock.connect(addr.path, () => cb(Right(())))
+            }
+          }
+        }
+        .toResource
+
+      address = to match {
+        case Left(_) =>
+          SocketAddress(
+            IpAddress.fromString(sock.localAddress.get).get,
+            Port.fromInt(sock.localPort.get).get
+          )
+        case Right(_) => UnixSocketAddress("")
+      }
+      peerAddress = to match {
+        case Left(_) =>
+          SocketAddress(
+            IpAddress.fromString(sock.remoteAddress.get).get,
+            Port.fromInt(sock.remotePort.get).get
+          )
+        case Right(addr) => addr
+      }
+      socket <- Socket.forAsync(sock, address, peerAddress)
+    } yield socket).adaptError { case IOException(ex) => ex }
+
+  protected def bindIpOrUnix(
+      address: Either[SocketAddress[Host], UnixSocketAddress],
+      options: List[SocketOption]
+  ): Resource[F, ServerSocket[F]] =
+    (for {
+      dispatcher <- Dispatcher.sequential[F]
+      channel <- Channel.unbounded[F, facade.net.Socket].toResource
+      server <- Resource.make(
+        F
+          .delay(
+            facade.net.createServer(
+              new facade.net.ServerOptions {
+                pauseOnConnect = true
+                allowHalfOpen = true
+              },
+              sock => dispatcher.unsafeRunAndForget(channel.send(sock))
+            )
+          )
+      )(server =>
+        F.async[Unit] { cb =>
+          if (server.listening)
+            F.delay(server.close(e => cb(e.toLeft(()).leftMap(js.JavaScriptException)))) *>
+              channel.close.as(None)
+          else
+            F.delay(cb(Right(()))).as(None)
+        }
+      )
+      _ <- F
+        .async[Unit] { cb =>
+          server.registerOneTimeListener[F, js.Error]("error") { e =>
+            cb(Left(js.JavaScriptException(e)))
+          } <* F.delay {
+            address match {
+              case Left(addr) =>
+                if (
+                  addr.host.isInstanceOf[IpAddress] && addr.host.asInstanceOf[IpAddress].isWildcard
+                )
+                  server.listen(addr.port.value, () => cb(Right(())))
+                else
+                  server.listen(addr.port.value, addr.host.toString, () => cb(Right(())))
+              case Right(addr) =>
+                server.listen(addr.path, () => cb(Right(())))
+            }
+          }
+        }
+        .toResource
+      serverSocketAddress = address match {
+        case Left(_) =>
+          val addr = server.address().get
+          SocketAddress(IpAddress.fromString(addr.address).get, Port.fromInt(addr.port).get)
+        case Right(addr) => addr
+      }
+      info = new SocketInfo[F] {
+        val address = serverSocketAddress
+        private def raiseOptionError[A]: F[A] =
+          F.raiseError(
+            new UnsupportedOperationException(
+              "Node.js server sockets do not support socket options"
+            )
+          )
+        def getOption[A](key: SocketOption.Key[A]) = raiseOptionError
+        def setOption[A](key: SocketOption.Key[A], value: A) = raiseOptionError
+        def supportedOptions = F.pure(Set.empty)
+      }
+      address0 = address
+      sockets = channel.stream
+        .evalTap(setSocketOptions(options))
+        .flatMap { sock =>
+          val address = address0 match {
+            case Left(_) =>
+              SocketAddress(
+                IpAddress.fromString(sock.localAddress.get).get,
+                Port.fromInt(sock.localPort.get).get
+              )
+            case Right(addr) => addr
+          }
+          val peerAddress = address0 match {
+            case Left(_) =>
+              SocketAddress(
+                IpAddress.fromString(sock.remoteAddress.get).get,
+                Port.fromInt(sock.remotePort.get).get
+              )
+            case Right(_) => UnixSocketAddress("")
+          }
+          Stream.resource(Socket.forAsync(sock, address, peerAddress))
+        }
+    } yield ServerSocket(info, sockets)).adaptError { case IOException(ex) => ex }
+}

io/js/src/main/scala/fs2/io/net/IpSocketsProviderPlatform.scala

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2013 Functional Streams for Scala
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ * the Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package fs2
+package io
+package net
+
+import cats.effect.{Async, Resource}
+import com.comcast.ip4s.{Host, SocketAddress}
+
+private[net] trait IpSocketsProviderCompanionPlatform { self: IpSocketsProvider.type =>
+
+  private[net] def forAsync[F[_]: Async]: IpSocketsProvider[F] =
+    new AsyncSocketsProvider[F] with IpSocketsProvider[F] {
+
+      override def connect(
+          address: SocketAddress[Host],
+          options: List[SocketOption]
+      ): Resource[F, Socket[F]] =
+        connectIpOrUnix(Left(address), options)
+
+      override def bind(
+          address: SocketAddress[Host],
+          options: List[SocketOption]
+      ): Resource[F, ServerSocket[F]] =
+        bindIpOrUnix(Left(address), options)
+    }
+}