haskell - 将导管合二为一

Question

我正在寻找一个可以执行类似于以下操作的函数:

merge :: MonadIO m => [Producer m a] -> Producer m a

我快速浏览了 stm-conduit ，它看起来很相似，但我不确定它是否符合我的要求:

messagesSource :: MonadIO m => AmqpConn -> Ack -> Text -> Producer m (Message, Envelope)
messagesSource conn ack q = loop
  where
    loop = do
      mmsg <- liftIO $ getMsg chan ack q
      case mmsg of
        Just (m, e) -> do
          yield (m, e)
          liftIO $ ackMsg chan (envDeliveryTag e) False
          loop
        Nothing     -> loop
    chan = fst $ amqpChan conn

如您所见，此管道生产者在产生消息后对其进行确认。在一个简单的“单线程”管道中，它运行良好，消息进入接收器，然后被确认。

然而与 stm-conduit这可能会改变，因为据我所知，生产者不会等待接收器消耗消息，而是并行工作，并且消息可能会过早地被确认。

是我对 stm-conduit的理解正确的？
将不同的源合并为一个以获得良好的单流语义的方法是什么？

更新 :根据要求将代码更新为实际工作的 AMQP 示例(但是它可能有点嘈杂)。

更新 2 :我认为我所追求的可能是管道来源的替代实例，所以我可以做类似 let src = src1 <|> src2 的事情.有可能吗？

Answer 1

看看 ZipSource ，这是一个新型包装器，其 Applicative让你结合 Source以您想要的方式。

一旦您拥有 ZipSource ，您可以使用 zipSources 结合 Source s 在 Traversable (例如一个列表)到一个 Source的 Traversable s。

与您想要的结果类型的唯一区别是它是 Source在 Traversable值，而不仅仅是单个值，但这应该不是什么大问题。