Merge branch 'main' into topic/net-metrics

Author: mpilquist

core/shared/src/main/scala-2.12/fs2/CollectorPlatform.scala

@@ -25,9 +25,11 @@ import scala.collection.generic.{
   GenericTraversableTemplate,
   MapFactory,
   SetFactory,
+  SortedMapFactory,
+  SortedSetFactory,
   TraversableFactory
 }
-import scala.collection.{MapLike, SetLike, Traversable}
+import scala.collection.{MapLike, SetLike, SortedMapLike, SortedSetLike, Traversable}
 
 import fs2.internal._
 
@@ -52,11 +54,31 @@ private[fs2] trait CollectorPlatform { self: Collector.type =>
   ): Collector.Aux[(K, V), C[K, V]] =
     make(Builder.fromMapFactory(f))
 
+  implicit def supportsSortedMapFactory[
+      K: Ordering,
+      V,
+      C[a, b] <: collection.SortedMap[a, b] with SortedMapLike[
+        a,
+        b,
+        C[a, b]
+      ]
+  ](
+      f: SortedMapFactory[C]
+  ): Collector.Aux[(K, V), C[K, V]] =
+    make(Builder.fromSortedMapFactory(f))
+
   implicit def supportsSetFactory[A, C[x] <: Set[x] with SetLike[x, C[x]]](
       f: SetFactory[C]
   ): Collector.Aux[A, C[A]] =
     make(Builder.fromSetFactory(f))
 
+  implicit def supportsSortedSetFactory[A: Ordering, C[x] <: collection.SortedSet[
+    x
+  ] with SortedSetLike[x, C[x]]](
+      f: SortedSetFactory[C]
+  ): Collector.Aux[A, C[A]] =
+    make(Builder.fromSortedSetFactory(f))
+
   private[fs2] trait BuilderPlatform { self: Collector.Builder.type =>
     def fromFactory[A, C[_], B](f: Factory[A, C[B]]): Builder[A, C[B]] =
       fromBuilder(f())
@@ -71,9 +93,25 @@ private[fs2] trait CollectorPlatform { self: Collector.type =>
     ): Builder[(K, V), C[K, V]] =
       fromBuilder(f.newBuilder)
 
+    def fromSortedMapFactory[
+        K: Ordering,
+        V,
+        C[a, b] <: collection.SortedMap[a, b] with SortedMapLike[a, b, C[a, b]]
+    ](
+        f: SortedMapFactory[C]
+    ): Builder[(K, V), C[K, V]] =
+      fromBuilder(f.newBuilder)
+
     def fromSetFactory[A, C[x] <: collection.Set[x] with SetLike[x, C[x]]](
         f: SetFactory[C]
     ): Builder[A, C[A]] =
       fromBuilder(f.newBuilder)
+
+    def fromSortedSetFactory[A: Ordering, C[x] <: collection.SortedSet[x] with SortedSetLike[x, C[
+      x
+    ]]](
+        f: SortedSetFactory[C]
+    ): Builder[A, C[A]] =
+      fromBuilder(f.newBuilder)
   }
 }

core/shared/src/main/scala-2.13/fs2/CollectorPlatform.scala

@@ -22,7 +22,13 @@
 package fs2
 
 import scala.collection.immutable.ArraySeq
-import scala.collection.{Factory, IterableFactory, MapFactory}
+import scala.collection.{
+  Factory,
+  EvidenceIterableFactory,
+  IterableFactory,
+  MapFactory,
+  SortedMapFactory
+}
 import scala.reflect.ClassTag
 
 private[fs2] trait CollectorPlatform { self: Collector.type =>
@@ -38,6 +44,16 @@ private[fs2] trait CollectorPlatform { self: Collector.type =>
   implicit def supportsMapFactory[K, V, C[_, _]](f: MapFactory[C]): Collector.Aux[(K, V), C[K, V]] =
     make(Builder.fromMapFactory(f))
 
+  implicit def supportsSortedMapFactory[K: Ordering, V, C[_, _]](
+      f: SortedMapFactory[C]
+  ): Collector.Aux[(K, V), C[K, V]] =
+    make(Builder.fromSortedMapFactory(f))
+
+  implicit def supportsEvidenceIterableFactory[A, C[_], E[_]](f: EvidenceIterableFactory[C, E])(
+      implicit ev: E[A]
+  ): Collector.Aux[A, C[A]] =
+    make(Builder.fromEvidenceIterableFactory(f))
+
   /** Use `ArraySeq.untagged` to build a `Collector` where a `ClassTag` is not available.
     */
   implicit def supportsTaggedArraySeq[A: ClassTag](
@@ -64,6 +80,16 @@ private[fs2] trait CollectorPlatform { self: Collector.type =>
     def fromMapFactory[K, V, C[_, _]](f: MapFactory[C]): Builder[(K, V), C[K, V]] =
       fromBuilder(f.newBuilder)
 
+    def fromSortedMapFactory[K: Ordering, V, C[_, _]](
+        f: SortedMapFactory[C]
+    ): Builder[(K, V), C[K, V]] =
+      fromBuilder(f.newBuilder)
+
+    def fromEvidenceIterableFactory[A, C[_], E[_]](f: EvidenceIterableFactory[C, E])(implicit
+        ev: E[A]
+    ): Builder[A, C[A]] =
+      fromBuilder(f.newBuilder)
+
     def taggedArraySeq[A: ClassTag]: Builder[A, ArraySeq[A]] =
       array[A].mapResult(ArraySeq.unsafeWrapArray)
   }

core/shared/src/main/scala-3/fs2/CollectorPlatform.scala

@@ -22,7 +22,13 @@
 package fs2
 
 import scala.collection.immutable.ArraySeq
-import scala.collection.{Factory, IterableFactory, MapFactory}
+import scala.collection.{
+  Factory,
+  EvidenceIterableFactory,
+  IterableFactory,
+  MapFactory,
+  SortedMapFactory
+}
 import scala.reflect.ClassTag
 
 private[fs2] trait CollectorPlatform { self: Collector.type =>
@@ -38,6 +44,16 @@ private[fs2] trait CollectorPlatform { self: Collector.type =>
   implicit def supportsMapFactory[K, V, C[_, _]](f: MapFactory[C]): Collector.Aux[(K, V), C[K, V]] =
     make(Builder.fromMapFactory(f))
 
+  implicit def supportsSortedMapFactory[K: Ordering, V, C[_, _]](
+      f: SortedMapFactory[C]
+  ): Collector.Aux[(K, V), C[K, V]] =
+    make(Builder.fromSortedMapFactory(f))
+
+  implicit def supportsEvidenceIterableFactory[A, C[_], E[_]](f: EvidenceIterableFactory[C, E])(
+      implicit ev: E[A]
+  ): Collector.Aux[A, C[A]] =
+    make(Builder.fromEvidenceIterableFactory(f))
+
   /** Use `ArraySeq.untagged` to build a `Collector` where a `ClassTag` is not available.
     */
   implicit def supportsTaggedArraySeq[A: ClassTag](
@@ -72,6 +88,16 @@ private[fs2] trait CollectorPlatform { self: Collector.type =>
     def fromMapFactory[K, V, C[_, _]](f: MapFactory[C]): Builder[(K, V), C[K, V]] =
       fromBuilder(f.newBuilder)
 
+    def fromSortedMapFactory[K: Ordering, V, C[_, _]](
+        f: SortedMapFactory[C]
+    ): Builder[(K, V), C[K, V]] =
+      fromBuilder(f.newBuilder)
+
+    def fromEvidenceIterableFactory[A, C[_], E[_]](f: EvidenceIterableFactory[C, E])(implicit
+        ev: E[A]
+    ): Builder[A, C[A]] =
+      fromBuilder(f.newBuilder)
+
     def taggedArraySeq[A: ClassTag]: Builder[A, ArraySeq[A]] =
       array[A].mapResult(ArraySeq.unsafeWrapArray)
   }

core/shared/src/test/scala/fs2/CompilationTest.scala

@@ -120,4 +120,8 @@ object ThisModuleShouldCompile {
   Stream(1, 2, 3).compile.to(Set)
   Stream(1, 2, 3).to(List)
   Stream(1, 2, 3).covary[Fallible].to(List)
+  Stream(1, 2, 3).to(Set)
+  Stream(1, 2, 3).to(collection.immutable.SortedSet)
+  Stream(1 -> 1, 2 -> 2, 3 -> 3).to(Map)
+  Stream(1 -> 1, 2 -> 2, 3 -> 3).to(collection.immutable.SortedMap)
 }

io/jvm/src/main/scala/fs2/io/net/NetworkPlatform.scala

@@ -94,6 +94,15 @@ private[net] trait NetworkCompanionPlatform extends NetworkLowPriority { self: N
           case None           => orElse
         }
 
+      private def selectingDatagram[A](
+          ifSelecting: SelectingIpDatagramSocketsProvider[F] => Resource[F, A],
+          orElse: => Resource[F, A]
+      ): Resource[F, A] =
+        Resource.eval(tryGetSelector).flatMap {
+          case Some(selector) => ifSelecting(new SelectingIpDatagramSocketsProvider(selector))
+          case None           => orElse
+        }
+
       override def connect(
           address: GenSocketAddress,
           options: List[SocketOption]
@@ -118,7 +127,15 @@ private[net] trait NetworkCompanionPlatform extends NetworkLowPriority { self: N
           address: GenSocketAddress,
           options: List[SocketOption]
       ): Resource[F, DatagramSocket[F]] =
-        fallback.bindDatagramSocket(address, options)
+        matchAddress(
+          address,
+          sa =>
+            selectingDatagram(
+              _.bindDatagramSocket(sa, options),
+              fallback.bindDatagramSocket(sa, options)
+            ),
+          ua => fallback.bindDatagramSocket(ua, options)
+        )
 
       // Implementations of deprecated operations
 

io/jvm/src/main/scala/fs2/io/net/SelectingDatagramSocket.scala

@@ -0,0 +1,165 @@
+/*
+ * 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.{LiftIO, Selector}
+import cats.effect.kernel.Async
+import cats.effect.std.Mutex
+import cats.syntax.all._
+import com.comcast.ip4s.{IpAddress, SocketAddress, GenSocketAddress, NetworkInterface}
+
+import java.net.InetSocketAddress
+import java.nio.ByteBuffer
+import java.nio.channels.{DatagramChannel, SelectionKey}
+import java.net.{NetworkInterface => JNetworkInterface}
+import com.comcast.ip4s.MulticastJoin
+import CollectionCompat.*
+
+private final class SelectingDatagramSocket[F[_]: LiftIO] private (
+    selector: Selector,
+    ch: DatagramChannel,
+    readMutex: Mutex[F],
+    writeMutex: Mutex[F],
+    override val address: SocketAddress[IpAddress]
+)(implicit F: Async[F])
+    extends DatagramSocket[F] {
+
+  private[this] val bufferSize = 1 << 16
+
+  def localAddress: F[SocketAddress[IpAddress]] =
+    F.pure(address)
+
+  def read: F[Datagram] =
+    readMutex.lock.surround {
+      val buf = ByteBuffer.allocate(bufferSize)
+
+      def go: F[Datagram] =
+        F.delay(ch.receive(buf)).flatMap {
+          case null =>
+            selector.select(ch, SelectionKey.OP_READ).to *> go
+          case src =>
+            F.delay {
+              buf.flip()
+              val bytes = new Array[Byte](buf.remaining())
+              buf.get(bytes)
+              buf.clear()
+              Datagram(
+                SocketAddress.fromInetSocketAddress(src.asInstanceOf[InetSocketAddress]),
+                Chunk.array(bytes)
+              )
+            }
+        }
+      go
+    }
+
+  def readGen: F[GenDatagram] =
+    read.map(_.toGenDatagram)
+
+  def reads: Stream[F, Datagram] =
+    Stream.repeatEval(read)
+
+  private def write0(bytes: Chunk[Byte], addr: Option[InetSocketAddress]): F[Unit] =
+    writeMutex.lock.surround {
+      val buf = bytes.toByteBuffer
+
+      def go: F[Unit] =
+        F.delay {
+          addr match {
+            case Some(a) => ch.send(buf, a)
+            case None    => ch.write(buf)
+          }
+        }.flatMap { _ =>
+          if (buf.hasRemaining) selector.select(ch, SelectionKey.OP_WRITE).to[F] *> go
+          else F.unit
+        }
+
+      go
+    }
+
+  def write(bytes: Chunk[Byte], address: GenSocketAddress): F[Unit] =
+    write0(bytes, Some(address.asIpUnsafe.toInetSocketAddress))
+
+  def write(bytes: Chunk[Byte]): F[Unit] =
+    write0(bytes, None)
+
+  def writes: Pipe[F, Datagram, Nothing] =
+    _.evalMap(write).drain
+
+  def connect(addr: GenSocketAddress): F[Unit] =
+    F.delay(ch.connect(addr.asIpUnsafe.toInetSocketAddress)).void
+
+  def disconnect: F[Unit] =
+    F.delay(ch.disconnect()).void
+
+  def getOption[A](key: java.net.SocketOption[A]): F[Option[A]] =
+    F.delay(Option(ch.getOption(key)))
+
+  def setOption[A](key: java.net.SocketOption[A], value: A): F[Unit] =
+    F.delay(ch.setOption(key, value)).void
+
+  override def join(
+      join: MulticastJoin[IpAddress],
+      interface: NetworkInterface
+  ): F[GroupMembership] =
+    F.delay {
+      val jinterface = JNetworkInterface.getByName(interface.name)
+      val membership = join.fold(
+        j => ch.join(j.group.address.toInetAddress, jinterface),
+        j => ch.join(j.group.address.toInetAddress, jinterface, j.source.toInetAddress)
+      )
+      new GroupMembership {
+        def drop = F.delay(membership.drop)
+        def block(source: IpAddress) =
+          F.delay { membership.block(source.toInetAddress); () }
+        def unblock(source: IpAddress) =
+          F.delay { membership.unblock(source.toInetAddress); () }
+        override def toString = "GroupMembership"
+      }
+    }
+
+  override def join(
+      j: MulticastJoin[IpAddress],
+      interface: JNetworkInterface
+  ): F[GroupMembership] =
+    join(j, NetworkInterface.fromJava(interface))
+
+  override def supportedOptions: F[Set[SocketOption.Key[?]]] =
+    F.delay {
+      ch.supportedOptions.asScala.toSet
+    }
+
+}
+
+private object SelectingDatagramSocket {
+  def apply[F[_]: LiftIO](
+      selector: Selector,
+      ch: DatagramChannel,
+      local: SocketAddress[IpAddress]
+  )(implicit F: Async[F]): F[DatagramSocket[F]] =
+    (Mutex[F], Mutex[F]).flatMapN { (readM, writeM) =>
+      F.delay {
+        new SelectingDatagramSocket[F](selector, ch, readM, writeM, local)
+      }
+    }
+}