haskell - 在下一步中使用结果流式传输管道

Question

我正在使用 streaming 包。我想通过保留常量内存，将 S.store 定义的一个步骤的结果用作管道中后续步骤的参数。 myStream 从文件中加载和解析。

我有一个类型检查的例子:

import qualified Streaming.Prelude as S
import qualified Data.Map.Strict as M

data A = MkA deriving (Show)

insertA :: MonadIO m => S.Stream (S.Of A) m r -> m (M.Map String Int)
insertA = undefined

insertB :: MonadIO m => M.Map String Int -> S.Stream (S.Of A) m r -> m Int
insertB = undefined

myStream :: S.Stream (S.Of A) IO r
myStream = undefined

run :: IO ()
run =
  myStream
    & S.store insertA
    & insertB M.empty
    & print

但是，& insertB M.empty 行采用的是空 map ，但我想使用上一步中来自 insertA 函数的 map 。insertB 函数然后使用此 Map 进行查找。

我能想到的解决方案如下:

run :: IO ()
run =
  myStream
    & S.store insertA
    & ( \e -> do
          resultMap <- S.effects e
          insertB resultMap e
      )
    & print

问题

这是否保留了像在常量内存中运行一样的流式传输优势？它如何在后台解决这个问题，因为流需要作为一个整体进行处理才能获得 Map？它多次传递相同的流 - 从文件加载它 2 次以保留常量内存？

如果是这种情况(加载文件 2 次)，如果流的源不是来自解析文件而是来自某些只能读取一次的数据流怎么办？

对于这个问题，是否还有其他优雅的解决方案同时具有流式处理的优势，即管道中的下一步需要使用上一步的结果？

Answer 1

这里建议的代码有问题:

  resultMap <- S.effects e
  insertB resultMap e

问题是您正在“运行”相同 流两次，即 usually problematic for IO-based streams .

例如，假设 myStream 从文件句柄中读取。当我们为第二遍调用 insertB 时，effects 已经到达文件末尾!从句柄的任何进一步读取都不会返回任何数据。

当然，我们可以用两个不同的流读取同一个文件两次。这保留了流式传输，但需要两次传递。

---

应该注意的是，对于某些具有内置资源管理的基础 monad，例如 resourcet ，您可以两次运行相同的 Stream 值，因为流代码足够“智能”，可以在每次运行流时分配和释放底层资源。

例如Stream type的版本出现在 linear-base支持函数 readFile :

readFile :: FilePath -> Stream (Of Text) RIO ()

它返回一个在 resource-aware IO 中工作的 Stream .

也就是说，我不喜欢在流式传输管道中隐藏此类重复读取文件，这让我感到困惑。