Merge branch 'main' into flow-processor

Author: BalmungSan

.git-blame-ignore-revs

@@ -21,3 +21,6 @@ e5525d3f0da44052fdcfbe844993260bdc044270
 
 # Scala Steward: Reformat with scalafmt 3.8.2
 a0a37ece16ee55056270b4d9ba5c1505ead8af17
+
+# Scala Steward: Reformat with scalafmt 3.8.6
+52e52b013db077ecb5b5a8f5b6e6113f912556d8

.scalafmt.conf

@@ -1,4 +1,4 @@
-version = "3.8.2"
+version = "3.9.2"
 
 style = default
 

build.sbt

@@ -8,10 +8,10 @@ ThisBuild / organization := "co.fs2"
 ThisBuild / organizationName := "Functional Streams for Scala"
 ThisBuild / startYear := Some(2013)
 
-val Scala213 = "2.13.15"
+val Scala213 = "2.13.16"
 
 ThisBuild / scalaVersion := Scala213
-ThisBuild / crossScalaVersions := Seq("2.12.20", Scala213, "3.3.4")
+ThisBuild / crossScalaVersions := Seq("2.12.20", Scala213, "3.3.5")
 ThisBuild / tlVersionIntroduced := Map("3" -> "3.0.3")
 
 ThisBuild / githubWorkflowOSes := Seq("ubuntu-latest")

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

@@ -86,7 +86,8 @@ private[fs2] trait ChunkCompanionPlatform extends ChunkCompanion213And3Compat {
       private[fs2] val ct: ClassTag[O]
   ) extends Chunk[O] {
     require(
-      offset >= 0 && offset <= values.size && length >= 0 && length <= values.size && offset + length <= values.size
+      offset >= 0 && offset <= values.size && length >= 0 && length <= values.size && offset + length <= values.size,
+      "IArraySlice out of bounds"
     )
 
     def size = length

core/shared/src/main/scala/fs2/Chunk.scala

@@ -863,7 +863,8 @@ object Chunk
     // ClassTag(values.getClass.getComponentType) -- we only keep it for bincompat
 
     require(
-      offset >= 0 && offset <= values.size && length >= 0 && length <= values.size && offset + length <= values.size
+      offset >= 0 && offset <= values.size && length >= 0 && length <= values.size && offset + length <= values.size,
+      "ArraySlice out of bounds"
     )
 
     override protected def thisClassTag: ClassTag[Any] = ct.asInstanceOf[ClassTag[Any]]

core/shared/src/main/scala/fs2/concurrent/Signal.scala

@@ -481,8 +481,11 @@ object SignallingMapRef {
       .map { case (state, ids) =>
         def newId = ids.getAndUpdate(_ + 1)
 
-        def updateAndNotify[U](state: State, k: K, f: Option[V] => (Option[V], U))
-            : (State, F[U]) = {
+        def updateAndNotify[U](
+            state: State,
+            k: K,
+            f: Option[V] => (Option[V], U)
+        ): (State, F[U]) = {
 
           val keyState = state.keys.get(k)
 

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

@@ -198,9 +198,8 @@ class StreamPerformanceSuite extends Fs2Suite {
                 val s: Stream[SyncIO, Int] =
                   List
                     .fill(N)(bracketed)
-                    .foldLeft(Stream.raiseError[SyncIO](new Err): Stream[SyncIO, Int]) {
-                      (acc, hd) =>
-                        acc.handleErrorWith(_ => hd)
+                    .foldLeft(Stream.raiseError[SyncIO](new Err): Stream[SyncIO, Int]) { (acc, hd) =>
+                      acc.handleErrorWith(_ => hd)
                     }
                 s.compile.toList.attempt
                   .flatMap(_ => (ok.get, open.get).tupled)

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

@@ -177,9 +177,8 @@ class StreamZipSuite extends Fs2Suite {
         Logger[IO]
           .flatMap { logger =>
             def s(tag: String) =
-              logger.logLifecycle(tag) >> {
+              logger.logLifecycle(tag) >>
                 logger.logLifecycle(s"$tag - 1") ++ logger.logLifecycle(s"$tag - 2")
-              }
 
             s("a").zip(s("b")).compile.drain >>
               logger.get.assertEquals(

flake.lock

@@ -46,7 +46,7 @@ class MemoryLeakSpec extends FunSuite {
       warmupIterations: Int = 3,
       samplePeriod: FiniteDuration = 1.seconds,
       monitorPeriod: FiniteDuration = 10.seconds,
-      limitTotalBytesIncreasePerSecond: Long = 700000,
+      limitTotalBytesIncreasePerSecond: Long = 1400000,
       limitConsecutiveIncreases: Int = 10
   )
 

integration/src/test/scala/fs2/MemoryLeakSpec.scala

@@ -64,12 +64,11 @@ class IoPlatformSuite extends Fs2Suite {
                 (bs1.length != (o1 + l1)) &&
                 // we expect that next slice will wrap same buffer
                 ((bs2 eq bs1) && (o2 == o1 + l1))
-              } || {
-                // if first slice buffer is 'full'
-                (bs2.length == (o1 + l1)) &&
-                // we expect new buffer allocated for next slice
-                ((bs2 ne bs1) && (o2 == 0))
-              }
+              } ||
+              // if first slice buffer is 'full'
+              (bs2.length == (o1 + l1)) &&
+              // we expect new buffer allocated for next slice
+              ((bs2 ne bs1) && (o2 == 0))
             case _ => false // unexpected chunk subtype
           }
       }

io/jvm/src/test/scala/fs2/io/IoPlatformSuite.scala

@@ -1 +1 @@
-sbt.version=1.10.7
+sbt.version=1.10.10

project/build.properties

@@ -1,7 +1,7 @@
-val sbtTypelevelVersion = "0.7.5"
+val sbtTypelevelVersion = "0.7.7"
 addSbtPlugin("org.typelevel" % "sbt-typelevel" % sbtTypelevelVersion)
 addSbtPlugin("org.typelevel" % "sbt-typelevel-site" % sbtTypelevelVersion)
-addSbtPlugin("org.scala-js" % "sbt-scalajs" % "1.17.0")
+addSbtPlugin("org.scala-js" % "sbt-scalajs" % "1.18.2")
 addSbtPlugin("org.scala-native" % "sbt-scala-native" % "0.4.17")
 addSbtPlugin("com.armanbilge" % "sbt-scala-native-config-brew-github-actions" % "0.3.0")
 addSbtPlugin("io.github.sbt-doctest" % "sbt-doctest" % "0.11.1")

project/plugins.sbt

@@ -44,8 +44,9 @@ object BlockCodec {
     ("Block Total Length"     | constant(length.bv)                 )}
   // format: on
 
-  def unknownByteOrder[L <: HList, LB <: HList](hexConstant: ByteVector)(f: Length => Codec[L])(
-      implicit
+  def unknownByteOrder[L <: HList, LB <: HList](
+      hexConstant: ByteVector
+  )(f: Length => Codec[L])(implicit
       prepend: Prepend.Aux[L, Unit :: HNil, LB],
       init: Init.Aux[LB, L],
       last: Last.Aux[LB, Unit]

protocols/shared/src/main/scala-2/fs2/protocols/pcapng/BlockCodec.scala

@@ -60,6 +60,7 @@ package object reactivestreams {
     *                   A high number can be useful if the publisher is triggering from IO, like requesting elements from a database.
     *                   The publisher can use this `bufferSize` to query elements in batch.
     *                   A high number will also lead to more elements in memory.
+    *                   The stream will not emit new element until, either the `Chunk` is filled or the publisher finishes.
     */
   def fromPublisher[F[_]: Async, A](p: Publisher[A], bufferSize: Int): Stream[F, A] =
     Stream
@@ -87,6 +88,7 @@ package object reactivestreams {
       *                   A high number can be useful if the publisher is triggering from IO, like requesting elements from a database.
       *                   The publisher can use this `bufferSize` to query elements in batch.
       *                   A high number will also lead to more elements in memory.
+      *                   The stream will not emit new element until, either the `Chunk` is filled or the publisher finishes.
       */
     def toStreamBuffered[F[_]: Async](bufferSize: Int): Stream[F, A] =
       fromPublisher(publisher, bufferSize)