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/src/main/scala/fs2/io/net/NetworkPlatform.scala

@@ -28,7 +28,7 @@ import cats.effect.LiftIO
 import cats.effect.Selector
 import cats.effect.kernel.{Async, Resource}
 
-import com.comcast.ip4s.{Dns, Host, IpAddress, Port, SocketAddress}
+import com.comcast.ip4s.{Dns, GenSocketAddress, Host, IpAddress, Port, SocketAddress, UnixSocketAddress}
 
 import fs2.internal.ThreadFactories
 import fs2.io.net.tls.TLSContext
@@ -80,7 +80,7 @@ private[net] trait NetworkCompanionPlatform extends NetworkLowPriority { self: N
   def forIO: Network[IO] = forLiftIO
 
   implicit def forLiftIO[F[_]: Async: LiftIO]: Network[F] =
-    new UnsealedNetwork[F] {
+    new AsyncNetwork[F] {
       private lazy val fallback = forAsync[F]
 
       private def tryGetSelector =
@@ -132,15 +132,13 @@ private[net] trait NetworkCompanionPlatform extends NetworkLowPriority { self: N
           protocolFamily: Option[ProtocolFamily]
       ): Resource[F, DatagramSocket[F]] =
         fallback.openDatagramSocket(address, port, options, protocolFamily)
-
-      def tlsContext: TLSContext.Builder[F] = TLSContext.Builder.forAsync[F]
     }
 
   def forAsync[F[_]](implicit F: Async[F]): Network[F] =
     forAsyncAndDns(F, Dns.forAsync(F))
 
   def forAsyncAndDns[F[_]](implicit F: Async[F], dns: Dns[F]): Network[F] =
-    new UnsealedNetwork[F] {
+    new AsyncNetwork[F] {
       private lazy val globalSocketGroup = SocketGroup.unsafe[F](globalAcg)
       private lazy val globalDatagramSocketGroup = DatagramSocketGroup.unsafe[F](globalAdsg)
 
@@ -185,8 +183,23 @@ private[net] trait NetworkCompanionPlatform extends NetworkLowPriority { self: N
           protocolFamily: Option[ProtocolFamily]
       ): Resource[F, DatagramSocket[F]] =
         globalDatagramSocketGroup.openDatagramSocket(address, port, options, protocolFamily)
+    }
+
+
+  private abstract class AsyncNetwork[F[_]](implicit F: Async[F]) extends UnsealedNetwork[F] {
 
-      def tlsContext: TLSContext.Builder[F] = TLSContext.Builder.forAsync[F]
+    def tlsContext: TLSContext.Builder[F] = TLSContext.Builder.forAsync[F]
+
+    private def unixSockets = fs2.io.net.unixsocket.UnixSockets.forAsync[F]
+
+    def tcp: TcpBuilder.NeedAddress[F] = new TcpBuilder.UnsealedNeedAddress[F] {
+      def mkClient(address: GenSocketAddress, options: List[SocketOption]) =
+        address match {
+          case sa: SocketAddress[_] => client(sa, options)
+          case ua: UnixSocketAddress => unixSockets.client(fs2.io.net.unixsocket.UnixSocketAddress(ua.path))
+          case _ => ???
+        }
     }
+  }
 
 }

io/native/src/main/scala/fs2/io/net/Network.scala → io/native/src/main/scala/fs2/io/net/DatagramSocketGroup.scala

@@ -23,14 +23,6 @@ package fs2
 package io
 package net
 
-import fs2.io.net.tls.TLSContext
-
-sealed trait Network[F[_]] extends NetworkPlatform[F] with SocketGroup[F] {
-  def tlsContext: TLSContext.Builder[F]
+trait DatagramSocketGroup[F[_]] {
 }
 
-object Network extends NetworkCompanionPlatform {
-  private[fs2] trait UnsealedNetwork[F[_]] extends Network[F]
-
-  def apply[F[_]](implicit F: Network[F]): F.type = F
-}

io/js-jvm/src/main/scala/fs2/io/net/Network.scala → io/shared/src/main/scala/fs2/io/net/Network.scala

@@ -51,6 +51,8 @@ sealed trait Network[F[_]]
     with SocketGroup[F]
     with DatagramSocketGroup[F] {
 
+  def tcp: TcpBuilder.NeedAddress[F]
+
   /** Returns a builder for `TLSContext[F]` values.
     *
     * For example, `Network[IO].tlsContext.system` returns a `F[TLSContext[F]]`.
@@ -63,3 +65,68 @@ object Network extends NetworkCompanionPlatform {
 
   def apply[F[_]](implicit F: Network[F]): F.type = F
 }
+
+import cats.effect.Resource
+import com.comcast.ip4s.*
+
+sealed trait BoundServer[F[_]] {
+  def serverSocket: Socket[F]
+  def clients: Stream[F, Socket[F]]
+}
+private[net] trait UnsealedBoundServer[F[_]] extends BoundServer[F]
+
+
+sealed trait TcpBuilder[F[_]] {
+  def withAddress(address: GenSocketAddress): TcpBuilder[F]
+  def withSocketOptions(options: List[SocketOption]): TcpBuilder[F]
+  def connect: Resource[F, Socket[F]]
+  def bind: Resource[F, BoundServer[F]]
+  def bindAndServe: Stream[F, Socket[F]]
+}
+
+object TcpBuilder {
+  sealed trait NeedAddress[F[_]] {
+    protected def mkClient(address: GenSocketAddress, options: List[SocketOption]): Resource[F, Socket[F]]
+
+    def address(address: GenSocketAddress): TcpBuilder[F] =
+      TcpBuilder[F](mkClient, address, Nil)
+
+    def hostAndPort(host: Host, port: Port): TcpBuilder[F] =
+      address(SocketAddress(host, port))
+
+    def port(port: Port): TcpBuilder[F] =
+      hostAndPort(Ipv4Address.Wildcard, port)
+
+    def port(p: Int): TcpBuilder[F] =
+      Port.fromInt(p) match {
+        case Some(pp) => port(pp)
+        case None => ??? // TODO
+      }
+
+    def hostAndEphemeralPort(host: Host): TcpBuilder[F] =
+      hostAndPort(host, Port.Wildcard)
+
+    def ephemeralPort: TcpBuilder[F] =
+      address(SocketAddress.Wildcard)
+
+    def unixSocket(path: fs2.io.file.Path): TcpBuilder[F] =
+      address(UnixSocketAddress(path.toString))
+  }
+
+  private[net] trait UnsealedNeedAddress[F[_]] extends NeedAddress[F]
+
+  private[net] def apply[F[_]](
+    mkClient: (GenSocketAddress, List[SocketOption]) => Resource[F, Socket[F]],
+    address: GenSocketAddress,
+    options: List[SocketOption]
+  ): TcpBuilder[F] = new TcpBuilder[F] {
+    private def copy(address: GenSocketAddress = address, options: List[SocketOption] = options): TcpBuilder[F] =
+      apply[F](mkClient, address, options)
+    def withSocketOptions(options: List[SocketOption]) = copy(options = options)
+    def withAddress(address: GenSocketAddress) = copy(address = address)
+    def connect: Resource[F, Socket[F]] = mkClient(address, options)
+    def bind: Resource[F, BoundServer[F]]= ???
+    def bindAndServe: Stream[F, Socket[F]] = ??? //Stream.resource(bind).flatMap(_.clients) 
+  }
+
+}