假設我們有 2 個 fs2 流：
val stream1 = fs2.Stream.bracket(IO { println("Acquire 1"); 2})(_ => IO { println("Release 1") })
.flatMap(p => fs2.Stream.range(1,p))
val stream2 = fs2.Stream.bracket(IO { println("Acquire 2"); 4})(_ => IO { println("Release 2") })
.flatMap(p => fs2.Stream.range(1,p))
我想相互聯系：
def connect[F[_]]: (fs2.Stream[F, Int], fs2.Stream[F, Int]) => fs2.Stream[F, Int] = {
def go(stream1: fs2.Stream[F, Int], stream2: fs2.Stream[F, Int]): Pull[F, Int, Unit] =
stream1.pull.uncons1.flatMap { stream1Element =>
stream2.pull.uncons1.flatMap { stream2Element =>
(stream1Element, stream2Element) match {
case (Some((stream1Head, stream1Tail)), Some((stream2Head, stream2Tail))) =>
println("Some, Some")
Pull.output1(stream1Head stream2Head) >> go(stream1Tail, stream2Tail)
case (Some((stream1Head, stream1Tail)), None) =>
println("1 Stream still available")
Pull.output1(stream1Head) >> go(fs2.Stream.empty, stream1Tail)
case (None, Some((stream2Head, stream2Tail))) =>
println("2 Stream still available")
Pull.output1(stream2Head) >> go(fs2.Stream.empty, stream2Tail)
case _ => Pull.output1(-1)
}
}
}
(one, two) => go(one, two).stream
}
現在檢查日志我看到：
Acquire 1
Acquire 2
Some, Some
Release 2
Release 1
2 Stream still available
2 Stream still available
這對我來說有點令人驚訝，因為似乎一旦第一個 Stream 完成，第二個 Stream 的資源也將關閉。假設現在資源是到資料庫的連接，那么第二個流中的元素無法再獲取。
這是正確的行為嗎？有沒有辦法避免關閉第二個流的資源？令人驚訝的是，如果第一個流的元素比第二個多，一切都按預期作業（當第二個流完成時，流 1 的資源不會關閉）
uj5u.com熱心網友回復：
通過檢查 zipAllWith 函式的實作，我發現在這種情況下確實應該避免 uncons1 。最終的解決方案是使用stepLeg函式而不是 uncons1。所以上面的函式應該是這樣的：
def connect[F[_]]: (fs2.Stream[F, Int], fs2.Stream[F, Int]) => fs2.Stream[F, Int] = {
def go(stream1: fs2.Stream[F, Int], stream2: fs2.Stream[F, Int]): Pull[F, Int, Unit] =
stream1.pull.stepLeg.flatMap { stream1Element =>
stream2.pull.stepLeg.flatMap { stream2Element =>
(stream1Element, stream2Element) match {
case (Some(sl1), Some(sl2)) =>
println("Some, Some")
val one = sl1.head(0)
val two = sl2.head(0)
Pull.output1(one two) >> go(sl1.stream, sl2.stream)
case (Some(sl1), None) =>
val one = sl1.head(0)
println("1 Stream still available")
Pull.output1(one) >> go(sl1.stream, fs2.Stream.empty)
case (None, Some(sl2)) =>
val two = sl2.head(0)
println("2 Stream still available")
Pull.output1(two) >> go(fs2.Stream.empty, sl2.stream)
case _ => Pull.output1(-1)
}
}
}
(one, two) => {
go(one.flatMap(fs2.Stream.emit), two.flatMap(fs2.Stream.emit)).stream
}
}
和日志：
Acquire 1
Acquire 2
Some, Some
Release 1
2 Stream still available
2 Stream still available
Release 2
可以在此處找到此問題的另一個示例： uncons vs stepLeg
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/394694.html
上一篇：最大請求大小的ZIOHTTP配置