Avoid double evaluation of pattern matchers in Chunk.collect

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

@@ -87,7 +87,8 @@ abstract class Chunk[+O] extends Serializable with ChunkPlatform[O] with ChunkRu
   def collect[O2](pf: PartialFunction[O, O2]): Chunk[O2] = {
     val b = makeArrayBuilder[Any]
     b.sizeHint(size)
-    foreach(o => if (pf.isDefinedAt(o)) b += pf(o))
+    val f = pf.runWith(b += _)
+    foreach(f(_): Unit)
     Chunk.array(b.result()).asInstanceOf[Chunk[O2]]
   }
 

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

@@ -31,6 +31,7 @@ import scodec.bits.ByteVector
 
 import java.nio.ByteBuffer
 import java.nio.CharBuffer
+import java.util.concurrent.atomic.AtomicInteger
 import scala.reflect.ClassTag
 
 class ChunkSuite extends Fs2Suite {
@@ -108,6 +109,36 @@ class ChunkSuite extends Fs2Suite {
         assert(Chunk.javaList(c.asJava) eq c)
       }
     }
+
+    test("Chunk.collect behaves as filter + map") {
+      forAll { (c: Chunk[Int]) =>
+        val extractor = new OddStringExtractor
+        val pf: PartialFunction[Int, String] = { case extractor(s) => s }
+
+        val result = c.collect(pf)
+
+        assertEquals(result, c.filter(pf.isDefinedAt).map(pf))
+      }
+    }
+
+    test("Chunk.collect evaluates pattern matchers once per item") {
+      forAll { (c: Chunk[Int]) =>
+        val extractor = new OddStringExtractor
+
+        val _ = c.collect { case extractor(s) => s }
+
+        assertEquals(extractor.callCounter.get(), c.size)
+      }
+    }
+
+    class OddStringExtractor {
+      val callCounter: AtomicInteger = new AtomicInteger(0)
+
+      def unapply(i: Int): Option[String] = {
+        callCounter.incrementAndGet()
+        if (i % 2 != 0) Some(i.toString) else None
+      }
+    }
   }
 
   def testChunk[A: Arbitrary: ClassTag: CommutativeMonoid: Eq: Cogen](