Unify TCP/UDP and UnixSocket APIs

.github/workflows/ci.yml

@@ -339,7 +339,7 @@ jobs:
     name: Test I/O on macOS
     strategy:
       matrix:
-        os: [macos-latest]
+        os: [macos-14]
         java: [temurin@17]
         project: [ioJS, ioJVM, ioNative]
     runs-on: ${{ matrix.os }}

build.sbt

@@ -34,7 +34,9 @@ ThisBuild / githubWorkflowAddedJobs +=
     scalas = Nil,
     sbtStepPreamble = Nil,
     javas = List(githubWorkflowJavaVersions.value.head),
-    oses = List("macos-latest"),
+    oses = List(
+      "macos-14"
+    ), // FIXME: macos-15 breaks sending multicast to local network - https://github.com/actions/runner-images/issues/10924
     matrixAdds = Map("project" -> List("ioJS", "ioJVM", "ioNative")),
     steps = githubWorkflowJobSetup.value.toList ++ List(
       WorkflowStep.Run(List("brew install s2n"), cond = Some("matrix.project == 'ioNative'")),
@@ -272,6 +274,93 @@ 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"),
+  ProblemFilters.exclude[IncompatibleResultTypeProblem](
+    "fs2.io.net.AsynchronousDatagramSocketGroup#WriterDatagram.remote"
+  ),
+  ProblemFilters.exclude[IncompatibleMethTypeProblem](
+    "fs2.io.net.AsynchronousDatagramSocketGroup#WriterDatagram.this"
+  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.address"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem](
+    "fs2.io.net.DatagramSocket.supportedOptions"
+  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.getOption"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.setOption"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.readGen"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.connect"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.disconnect"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.write"),
+  ProblemFilters.exclude[MissingClassProblem](
+    "fs2.io.net.DatagramSocketGroupCompanionPlatform$AsyncDatagramSocketGroup"
+  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Network.bindDatagramSocket"),
+  ProblemFilters.exclude[MissingClassProblem]("fs2.io.net.SocketGroup$"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem](
+    "fs2.io.net.SocketOptionCompanionPlatform#Key.fs2$io$net$SocketOptionCompanionPlatform$Key$$$outer"
+  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem](
+    "fs2.io.net.DatagramSocketOption#Key.toSocketOption"
+  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.DatagramSocket.join"),
+  ProblemFilters.exclude[IncompatibleMethTypeProblem](
+    "fs2.io.net.DatagramSocketOption.multicastInterface"
+  ),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Network.dns"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem]("fs2.io.net.Network.interfaces"),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem](
+    "fs2.io.net.tls.TLSContext#Builder.fromKeyStoreFile"
+  ),
+  ProblemFilters.exclude[InheritedNewAbstractMethodProblem](
+    "fs2.io.net.tls.TLSContext#Builder.fs2$io$net$tls$TLSContextCompanionPlatform$BuilderPlatform$$$outer"
   )
 )
 

io/js-jvm/src/test/scala/fs2/io/net/udp/UdpSuite.scala

@@ -31,25 +31,29 @@ import com.comcast.ip4s._
 
 import scala.concurrent.duration._
 
-class UdpSuite extends Fs2Suite with UdpSuitePlatform {
-  def sendAndReceive(socket: DatagramSocket[IO], toSend: Datagram): IO[Datagram] =
+class UdpSuite extends Fs2Suite {
+  private def sendAndReceive(socket: DatagramSocket[IO], toSend: Datagram): IO[Datagram] =
     socket
       .write(toSend) >> socket.read.timeoutTo(1.second, IO.defer(sendAndReceive(socket, toSend)))
 
+  private def sendAndReceiveBytes(socket: DatagramSocket[IO], bytes: Chunk[Byte]): IO[Datagram] =
+    socket
+      .write(bytes) >> socket.read.timeoutTo(1.second, IO.defer(sendAndReceiveBytes(socket, bytes)))
+
   group("udp") {
     test("echo one") {
       val msg = Chunk.array("Hello, world!".getBytes)
       Stream
-        .resource(Network[IO].openDatagramSocket())
+        .resource(Network[IO].bindDatagramSocket(SocketAddress.Wildcard))
         .flatMap { serverSocket =>
-          Stream.eval(serverSocket.localAddress).map(_.port).flatMap { serverPort =>
-            val serverAddress = SocketAddress(ip"127.0.0.1", serverPort)
-            val server = serverSocket.reads.foreach(packet => serverSocket.write(packet))
-            val client = Stream.resource(Network[IO].openDatagramSocket()).evalMap { clientSocket =>
-              sendAndReceive(clientSocket, Datagram(serverAddress, msg))
+          val serverAddress = SocketAddress(ip"127.0.0.1", serverSocket.address.asIpUnsafe.port)
+          val server = serverSocket.reads.foreach(packet => serverSocket.write(packet))
+          val client =
+            Stream.resource(Network[IO].bindDatagramSocket(SocketAddress.Wildcard)).evalMap {
+              clientSocket =>
+                sendAndReceive(clientSocket, Datagram(serverAddress, msg))
             }
-            client.concurrently(server)
-          }
+          client.concurrently(server)
         }
         .compile
         .lastOrError
@@ -67,55 +71,88 @@ class UdpSuite extends Fs2Suite with UdpSuitePlatform {
         .sorted
 
       Stream
-        .resource(Network[IO].openDatagramSocket())
+        .resource(Network[IO].bindDatagramSocket())
         .flatMap { serverSocket =>
-          Stream.eval(serverSocket.localAddress).map(_.port).flatMap { serverPort =>
-            val serverAddress = SocketAddress(ip"127.0.0.1", serverPort)
-            val server = serverSocket.reads.foreach(packet => serverSocket.write(packet))
-            val client = Stream.resource(Network[IO].openDatagramSocket()).flatMap { clientSocket =>
-              Stream
-                .emits(msgs.map(msg => Datagram(serverAddress, msg)))
-                .evalMap(msg => sendAndReceive(clientSocket, msg))
-            }
-            val clients = Stream
-              .constant(client)
-              .take(numClients.toLong)
-              .parJoin(numParallelClients)
-            clients.concurrently(server)
+          val serverAddress = SocketAddress(ip"127.0.0.1", serverSocket.address.asIpUnsafe.port)
+          val server = serverSocket.reads.foreach(packet => serverSocket.write(packet))
+          val client = Stream.resource(Network[IO].bindDatagramSocket()).flatMap { clientSocket =>
+            Stream
+              .emits(msgs.map(msg => Datagram(serverAddress, msg)))
+              .evalMap(msg => sendAndReceive(clientSocket, msg))
           }
+          val clients = Stream
+            .constant(client)
+            .take(numClients.toLong)
+            .parJoin(numParallelClients)
+          clients.concurrently(server)
         }
         .compile
         .toVector
         .map(_.map(p => new String(p.bytes.toArray)).sorted)
         .assertEquals(expected)
     }
 
-    test("multicast".ignore) {
-      // Fails often based on routing table of host machine
-      val group = mip"232.10.10.10"
-      val groupJoin = MulticastJoin.asm(group)
+    test("echo connected") {
       val msg = Chunk.array("Hello, world!".getBytes)
       Stream
-        .resource(
-          Network[IO].openDatagramSocket(
-            options = List(DatagramSocketOption.multicastTtl(1)),
-            protocolFamily = Some(v4ProtocolFamily)
-          )
-        )
+        .resource(Network[IO].bindDatagramSocket())
         .flatMap { serverSocket =>
-          Stream.eval(serverSocket.localAddress).map(_.port).flatMap { serverPort =>
-            val server = Stream
-              .exec(
-                v4Interfaces.traverse_(interface => serverSocket.join(groupJoin, interface))
-              ) ++
-              serverSocket.reads.foreach(packet => serverSocket.write(packet))
-            val client = Stream.resource(Network[IO].openDatagramSocket()).flatMap { clientSocket =>
-              Stream(Datagram(SocketAddress(group.address, serverPort), msg))
-                .through(clientSocket.writes)
-                .drain ++ Stream.eval(clientSocket.read)
-            }
-            client.concurrently(server)
+          val serverAddress = serverSocket.address.asIpUnsafe
+          val server = serverSocket.reads.foreach(packet => serverSocket.write(packet))
+          val client = Stream.resource(Network[IO].bindDatagramSocket()).evalMap { clientSocket =>
+            clientSocket.connect(serverAddress) >> sendAndReceiveBytes(clientSocket, msg)
           }
+          client.concurrently(server)
+        }
+        .compile
+        .lastOrError
+        .map(_.bytes)
+        .assertEquals(msg)
+    }
+
+    test("multicast") {
+      val group = mip"239.10.10.10"
+      val groupJoin = MulticastJoin.asm(group)
+      val msg = Chunk.array("Hello, world!".getBytes)
+      val outgoingInterface =
+        // Get first non-loopback interface with an IPv4 address
+        Network[IO].interfaces.getAll.map { interfaces =>
+          interfaces.values
+            .filterNot(_.isLoopback)
+            .flatMap(iface =>
+              iface.addresses.filter(_.address.fold(_ => true, _ => false)).as(iface)
+            )
+            .head
+        }
+      Stream
+        .eval(outgoingInterface)
+        .flatMap { out =>
+          Stream
+            .resource(
+              Network[IO]
+                .bindDatagramSocket(
+                  options = List(SocketOption.multicastTtl(1), SocketOption.multicastInterface(out))
+                )
+                .evalMap { serverSocket =>
+                  Network[IO].interfaces.getAll.flatMap { interfaces =>
+                    interfaces.values.toList
+                      .filter(iface =>
+                        iface.addresses.exists(_.address.fold(_ => true, _ => false))
+                      )
+                      .traverse_(iface => serverSocket.join(groupJoin, iface))
+                      .as(serverSocket)
+                  }
+                }
+            )
+            .flatMap { serverSocket =>
+              val server = serverSocket.reads.foreach(packet => serverSocket.write(packet))
+              val client =
+                Stream.resource(Network[IO].bindDatagramSocket()).flatMap { clientSocket =>
+                  val to = SocketAddress(group.address, serverSocket.address.asIpUnsafe.port)
+                  Stream.eval(clientSocket.write(msg, to) >> clientSocket.read)
+                }
+              client.concurrently(server)
+            }
         }
         .compile
         .lastOrError

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

@@ -45,6 +45,11 @@ private[io] object dgram {
 
     def bind(options: BindOptions, cb: js.Function0[Unit]): Unit = js.native
 
+    def connect(port: Int, address: String, cb: js.Function1[js.UndefOr[js.Error], Unit]): Unit =
+      js.native
+
+    def disconnect(): Unit = js.native
+
     def addMembership(multicastAddress: String, multicastInterface: String): Unit = js.native
 
     def dropMembership(multicastAddress: String, multicastInterface: String): Unit = js.native
@@ -63,6 +68,9 @@ private[io] object dgram {
 
     def close(cb: js.Function0[Unit]): Unit = js.native
 
+    def send(msg: Uint8Array, cb: js.Function1[js.Error, Unit]): Unit =
+      js.native
+
     def send(msg: Uint8Array, port: Int, address: String, cb: js.Function1[js.Error, Unit]): Unit =
       js.native
 
@@ -74,8 +82,12 @@ private[io] object dgram {
 
     def setMulticastTTL(ttl: Int): Unit = js.native
 
+    def getRecvBufferSize: Int = js.native
+
     def setRecvBufferSize(size: Int): Unit = js.native
 
+    def getSendBufferSize: Int = js.native
+
     def setSendBufferSize(size: Int): Unit = js.native
 
     def setTTL(ttl: Int): Unit = js.native
@@ -85,7 +97,7 @@ private[io] object dgram {
   @js.native
   trait AddressInfo extends js.Object {
     def address: String = js.native
-    def family: Int = js.native
+    def family: String = js.native
     def port: Int = js.native
   }
 
@@ -97,7 +109,7 @@ private[io] object dgram {
   @js.native
   trait RemoteInfo extends js.Object {
     def address: String = js.native
-    def family: Int = js.native
+    def family: String = js.native
     def port: Int = js.native
     def size: Int = js.native
   }