haskell - 根据时间限制管道？

Question

是否可以创建管道来获取在特定时间段内向下游发送的所有值？我正在实现一个服务器，该协议(protocol)允许我连接传出数据包并将它们压缩在一起，所以我想有效地“清空”下游队列ByteString每 100ms 和 mappend它们在一起，然后屈服于下一个进行压缩的管道。

Answer 1

这是使用 pipes-concurrency 的解决方案.你给它任何Input它会定期排出所有值的输入:

import Control.Applicative ((<|>))
import Control.Concurrent (threadDelay)
import Data.Foldable (forM_)
import Pipes
import Pipes.Concurrent

drainAll :: Input a -> STM (Maybe [a])
drainAll i = do
    ma <- recv i
    case ma of
        Nothing -> return Nothing
        Just a  -> loop (a:)
  where
    loop diffAs = do
        ma <- recv i <|> return Nothing
        case ma of
            Nothing -> return (Just (diffAs []))
            Just a  -> loop (diffAs . (a:))

bucketsEvery :: Int -> Input a -> Producer [a] IO ()
bucketsEvery microseconds i = loop
  where
    loop = do
        lift $ threadDelay microseconds
        ma <- lift $ atomically $ drainAll i
        forM_ ma $ \a -> do
            yield a
            loop

通过选择 Buffer 的类型，您可以更好地控制如何使用上游元素。您用来构建 Input .

如果您是 pipes-concurrency 的新手, 你可以阅读 the tutorial其中解释了如何使用 spawn , Buffer和 Input .