Haskell流-复制后如何将1流分成2流？

Question

在haskell 流式传输中，有一个复制的例子。

>>> (S.toList . mapped S.toList . chunksOf 5) $  (S.toList . mapped S.toList . chunksOf 3) $ S.copy $ each [1..10]
[[1,2,3,4,5],[6,7,8,9,10]] :> ([[1,2,3],[4,5,6],[7,8,9],[10]] :> ())

是否有可能把它分成两个“干净”的流，这样就可以打印下面的结果？

>>>S.print stream1
[[1,2,3,4,5],[6,7,8,9,10]]
>>>S.print stream2
[[1,2,3],[4,5,6],[7,8,9],[10]]

注意，在上面的结果中不再有':>‘了。更广泛地说，我不确定是否有一些函数可以“简化”m或(Of a)部件在Stream (Of a) m r中嵌套的流(或流)。

f1::Stream (Of a) (Stream (Of b) m) r -> Stream (Of b) m r
f2::Stream (Of a) (Stream (Of b) m) r -> Stream (Of a) m r
f3::Stream (Stream (Of a) m) r -> Stream (Of a) m r

更新

这个问题的背景是，我正在寻找惯用的方式重复使用一个潜在的流多次。流是从数据库中提取的，IO可能很昂贵。我还希望获得对中间流的引用，以便更好地构造代码。一些模拟代码：

my-stream-fn = do
  original_stream <- pull_from_database 
  let  (o1, s1) = calc_moving_average $ S.copy original_stream
       (o2, s2) = calc_max $ S.copy o1
       (o3, s3) = calc_min $ S.copy o2
  S.print $ S.zipWith3 (\x y z-> (x, y, z)) s1 s2 s3

我希望o1、o2和o3与original_stream和pull_from_database IO操作完全相同，当original_stream被提取时只执行一次。

Answer 1

f1 = S.effects @(Stream (Of _) _) 
  :: Monad m 
  => Stream (Of a) (Stream (Of b) m) r 
  -> Stream (Of b) r
f2 = hoist @(Stream (Of _)) S.effects
  :: Monad m
  => Stream (Of a) (Stream (Of b) m) r
  -> Stream (Of a) m r

(类型重命名是为了清晰，请参阅effects文档)，而f3不进行类型检查。

这感觉就像你试图击败流媒体的观点。构建管道、源到接收器并运行它-关键是没有(隐式的)中间值积累。您的问题稍微松散，因此要精确地回答不同的问题，但是如果您希望运行第一个流的所有效果，然后运行第二个流的所有效果，那么您必须愿意存储(代表)第二个流的计算，直到第一个流完成对=>的影响--您已经积累了第二个流(因此不是真正的流)。因此，为什么S.copy被设计为交错的效果。请参阅这个github问题。

对最新情况的答复

我认为让你困惑的部分原因是，你使用的是纯流，而在没有效果的情况下，限制的动机就不那么明显了。使用管道组件的标识符，而不是部分结果。同样，在你的例子中，你应该组合折叠，例如。

import qualified Control.Foldl as L
import qualified Streaming.Prelude as S

myStreamFn =
  let movingAvg n = {-# ... #-}
      combinedAcc = (,,) <$> L.minimum <*> L.maximum <*> movingAvg 10
  in  S.print 
   $  L.purely S.fold combinedAcc 
   $  pullFromDatabase

您可能想考虑的另一个函数是S.store，例如。

myStreamFn 
  = pullFromDatabase
  & S.store S.maximum
  & S.store (L.purely S.fold L.minimum)
  & S.store movingAvg
  & S.print