Unify TCP/UDP and UnixSocket APIs

build.sbt

@@ -353,7 +353,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.6.0-91-51bd018-SNAPSHOT",
     tlJdkRelease := None
   )
   .jvmSettings(

io/jvm-native/src/main/scala/fs2/io/net/SocketGroupPlatform.scala

@@ -23,140 +23,20 @@ package fs2
 package io
 package net
 
-import java.net.InetSocketAddress
-import java.nio.channels.{
-  AsynchronousCloseException,
-  AsynchronousServerSocketChannel,
-  AsynchronousSocketChannel,
-  CompletionHandler
-}
-import java.nio.channels.AsynchronousChannelGroup
 import cats.syntax.all._
 import cats.effect.kernel.{Async, Resource}
-import com.comcast.ip4s.{Dns, Host, IpAddress, Port, SocketAddress}
+import com.comcast.ip4s.{Host, IpAddress, Ipv4Address, Port, SocketAddress}
 
 private[net] trait SocketGroupCompanionPlatform { self: SocketGroup.type =>
-  private[fs2] def unsafe[F[_]: Async: Dns](
-      channelGroup: AsynchronousChannelGroup
-  ): SocketGroup[F] =
-    new AsyncSocketGroup[F](channelGroup)
-
-  private final class AsyncSocketGroup[F[_]: Async: Dns](channelGroup: AsynchronousChannelGroup)
-      extends AbstractAsyncSocketGroup[F] {
-
-    def client(
-        to: SocketAddress[Host],
-        options: List[SocketOption]
-    ): Resource[F, Socket[F]] = {
-      def setup: Resource[F, AsynchronousSocketChannel] =
-        Resource
-          .make(
-            Async[F].delay(
-              AsynchronousSocketChannel.open(channelGroup)
-            )
-          )(ch => Async[F].delay(if (ch.isOpen) ch.close else ()))
-          .evalTap(ch => Async[F].delay(options.foreach(opt => ch.setOption(opt.key, opt.value))))
-
-      def connect(ch: AsynchronousSocketChannel): F[AsynchronousSocketChannel] =
-        to.resolve[F].flatMap { ip =>
-          Async[F].async[AsynchronousSocketChannel] { cb =>
-            Async[F]
-              .delay {
-                ch.connect(
-                  ip.toInetSocketAddress,
-                  null,
-                  new CompletionHandler[Void, Void] {
-                    def completed(result: Void, attachment: Void): Unit =
-                      cb(Right(ch))
-                    def failed(rsn: Throwable, attachment: Void): Unit =
-                      cb(Left(rsn))
-                  }
-                )
-              }
-              .as(Some(Async[F].delay(ch.close())))
-          }
-        }
-
-      setup.evalMap(ch => connect(ch) *> Socket.forAsync(ch))
-    }
 
-    def serverResource(
-        address: Option[Host],
-        port: Option[Port],
-        options: List[SocketOption]
-    ): Resource[F, (SocketAddress[IpAddress], Stream[F, Socket[F]])] = {
+  def fromIpSockets[F[_]: Async](ipSockets: IpSocketsProvider[F]): SocketGroup[F] = new SocketGroup[F] {
+    def client(to: SocketAddress[Host], options: List[SocketOption]) =
+      ipSockets.connect(to, options)
 
-      val setup: Resource[F, AsynchronousServerSocketChannel] =
-        Resource.eval(address.traverse(_.resolve[F])).flatMap { addr =>
-          Resource
-            .make(
-              Async[F].delay(
-                AsynchronousServerSocketChannel.open(channelGroup)
-              )
-            )(sch => Async[F].delay(if (sch.isOpen) sch.close()))
-            .evalTap(ch =>
-              Async[F].delay(
-                ch.bind(
-                  new InetSocketAddress(
-                    addr.map(_.toInetAddress).orNull,
-                    port.map(_.value).getOrElse(0)
-                  )
-                )
-              )
-            )
-        }
+    def server(address: Option[Host], port: Option[Port], options: List[SocketOption]): Stream[F, Socket[F]] =
+      Stream.resource(serverResource(address, port, options)).flatMap(_._2)
 
-      def acceptIncoming(
-          sch: AsynchronousServerSocketChannel
-      ): Stream[F, Socket[F]] = {
-        def go: Stream[F, Socket[F]] = {
-          def acceptChannel = Resource.makeFull[F, AsynchronousSocketChannel] { poll =>
-            poll {
-              Async[F].async[AsynchronousSocketChannel] { cb =>
-                Async[F]
-                  .delay {
-                    sch.accept(
-                      null,
-                      new CompletionHandler[AsynchronousSocketChannel, Void] {
-                        def completed(ch: AsynchronousSocketChannel, attachment: Void): Unit =
-                          cb(Right(ch))
-                        def failed(rsn: Throwable, attachment: Void): Unit =
-                          cb(Left(rsn))
-                      }
-                    )
-                  }
-                  .as(Some(Async[F].delay(sch.close())))
-              }
-            }
-          }(ch => Async[F].delay(if (ch.isOpen) ch.close else ()))
-
-          def setOpts(ch: AsynchronousSocketChannel) =
-            Async[F].delay {
-              options.foreach(o => ch.setOption(o.key, o.value))
-            }
-
-          Stream.resource(acceptChannel.attempt).flatMap {
-            case Left(_)         => Stream.empty[F]
-            case Right(accepted) => Stream.eval(setOpts(accepted) *> Socket.forAsync(accepted))
-          } ++ go
-        }
-
-        go.handleErrorWith {
-          case err: AsynchronousCloseException =>
-            Stream.eval(Async[F].delay(sch.isOpen)).flatMap { isOpen =>
-              if (isOpen) Stream.raiseError[F](err)
-              else Stream.empty
-            }
-          case err => Stream.raiseError[F](err)
-        }
-      }
-
-      setup.map { sch =>
-        val jLocalAddress = sch.getLocalAddress.asInstanceOf[java.net.InetSocketAddress]
-        val localAddress = SocketAddress.fromInetSocketAddress(jLocalAddress)
-        (localAddress, acceptIncoming(sch))
-      }
-    }
+    def serverResource(address: Option[Host], port: Option[Port], options: List[SocketOption]): Resource[F, (SocketAddress[IpAddress], Stream[F, Socket[F]])] =
+      ipSockets.bind(SocketAddress(address.getOrElse(Ipv4Address.Wildcard), port.getOrElse(Port.Wildcard)), options).evalMap(b => b.socketInfo.localAddress.tupleRight(b.accept))
   }
-
 }

io/jvm-native/src/main/scala/fs2/io/net/SocketInfoPlatform.scala

@@ -0,0 +1,83 @@
+/*
+ * 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 com.comcast.ip4s.{GenSocketAddress, IpAddress, SocketAddress}
+import cats.effect.Async
+
+import java.net.InetSocketAddress
+import java.nio.channels.NetworkChannel
+
+import scala.jdk.CollectionConverters.*
+
+private[net] trait SocketInfoCompanionPlatform {
+  private[net] def forAsync[F[_]](ch: NetworkChannel)(implicit F: Async[F]): SocketInfo[F] =
+    new AsyncSocketInfo[F] {
+      def asyncInstance = F
+      def channel = ch
+    }
+
+  private[net] trait AsyncSocketInfo[F[_]] extends SocketInfo[F] {
+
+    implicit protected def asyncInstance: Async[F]
+    protected def channel: NetworkChannel
+
+    override def localAddress: F[SocketAddress[IpAddress]] =
+      asyncInstance.delay(
+        SocketAddress.fromInetSocketAddress(
+          channel.getLocalAddress.asInstanceOf[InetSocketAddress]
+        )
+      )
+
+    override def localAddressGen: F[GenSocketAddress] =
+      asyncInstance.delay(
+        channel.getLocalAddress match {
+          case addr: InetSocketAddress => SocketAddress.fromInetSocketAddress(addr)
+          // TODO handle unix sockets
+        }
+      )
+
+    override def supportedOptions: F[Set[SocketOption.Key[_]]] =
+      asyncInstance.delay {
+        channel.supportedOptions.asScala.toSet
+      }
+
+    override def getOption[A](key: SocketOption.Key[A]): F[Option[A]] =
+      asyncInstance.delay {
+        try {
+          Some(channel.getOption(key))
+        } catch {
+          case _: UnsupportedOperationException => None
+        }
+      }
+
+    override def setOption[A](key: SocketOption.Key[A], value: A): F[Unit] =
+      asyncInstance.delay {
+        channel.setOption(key, value)
+        ()
+      }
+  }
+
+}
+

io/jvm-native/src/main/scala/fs2/io/net/SocketOptionPlatform.scala

@@ -71,4 +71,21 @@ private[net] trait SocketOptionCompanionPlatform {
 
   def noDelay(value: Boolean): SocketOption =
     boolean(StandardSocketOptions.TCP_NODELAY, value)
+
+  val UnixServerSocketDeleteIfExists: Key[JBoolean] = new Key[JBoolean] {
+    def name() = "FS2_UNIX_DELETE_IF_EXISTS"
+    def `type`() = classOf[JBoolean]
+  }
+
+  def unixServerSocketDeleteIfExists(value: JBoolean): SocketOption =
+    boolean(UnixServerSocketDeleteIfExists, value)
+
+  val UnixServerSocketDeleteOnClose: Key[JBoolean] = new Key[JBoolean] {
+    def name() = "FS2_UNIX_DELETE_ON_CLOSE"
+    def `type`() = classOf[JBoolean]
+  }
+
+  def unixServerSocketDeleteOnClose(value: Boolean): SocketOption =
+    boolean(UnixServerSocketDeleteOnClose, value)
+
 }

io/jvm-native/src/main/scala/fs2/io/net/SocketPlatform.scala

@@ -23,7 +23,7 @@ package fs2
 package io
 package net
 
-import com.comcast.ip4s.{IpAddress, SocketAddress}
+import com.comcast.ip4s.{GenSocketAddress, IpAddress, SocketAddress}
 import cats.effect.Async
 import cats.effect.std.Mutex
 import cats.syntax.all._
@@ -108,7 +108,10 @@ private[net] trait SocketCompanionPlatform {
       readMutex: Mutex[F],
       writeMutex: Mutex[F]
   )(implicit F: Async[F])
-      extends BufferedReads[F](readMutex) {
+      extends BufferedReads[F](readMutex) with SocketInfo.AsyncSocketInfo[F] {
+
+    protected def asyncInstance = F
+    protected def channel = ch
 
     protected def readChunk(buffer: ByteBuffer): F[Int] =
       F.async[Int] { cb =>
@@ -139,20 +142,15 @@ private[net] trait SocketCompanionPlatform {
       }
     }
 
-    def localAddress: F[SocketAddress[IpAddress]] =
-      F.delay(
-        SocketAddress.fromInetSocketAddress(
-          ch.getLocalAddress.asInstanceOf[InetSocketAddress]
-        )
-      )
-
     def remoteAddress: F[SocketAddress[IpAddress]] =
       F.delay(
         SocketAddress.fromInetSocketAddress(
           ch.getRemoteAddress.asInstanceOf[InetSocketAddress]
         )
       )
 
+    override def remoteAddressGen: F[GenSocketAddress] = ???
+
     def isOpen: F[Boolean] = F.delay(ch.isOpen)
 
     def endOfOutput: F[Unit] =