非決定的インタリーブ導管のソース

Question

私は

interleave :: WhateverIOMonadClassItWouldWant m => [(k, Source m a)] -> Source m (k, a) 

ユースケースのようなタイプの署名で、ソースに対する非決定性インターリーブ動作を確認するために期待していたが、私は上の多くのノードへのオープン接続を維持してP2Pアプリケーションを持っているということですほとんどの場合、それらのいずれかからのメッセージを待っているだけです。メッセージが到着すると、メッセージはどこから来たかに関わらず、できるだけ早くメッセージを処理する必要があります。理論的には、この種のアプリケーションは少なくともGHCのIOマネージャをバイパスしてselect/epoll/etcを実行することができます。直接呼びますが、動作している限り、その実装方法は特に気にしません。

これはコンジットでも可能ですか？それほど一般的ではありませんが、おそらくもっと実現可能な方法は、すべてのソケットでの受信を処理する[(k, Socket)] -> Source m (k, ByteString)関数を書くことです。

私はコンジットでResumableSourceの操作を気付いたが、少なくともこの操作では少しの抽象リークのような感じの特定のSinkを知りたいと思うようだ。

Answer 1

stm-conduitパッケージは、あなたが探しているものと同じではありませんが、類似のものを実行するmergeSourcesを提供します。それはおそらく始めるのに良い場所です。

Answer 2

はい、可能です。

あなたは、各スレッドにあなたには、いくつかの並行処理チャネルに出力を送信SinkとまでSourceをペア場所をポーリングするスレッドをフォークからブロックされることなく、Source Sの束をポーリングすることができます

concur :: (WhateverIOMonadClassItWouldWant m) => TChan a -> Sink a m r 

...その後、あなたはそのチャンネルから読み取っSourceを定義します。

これは単なるソケット自体をポーリングするスレッドをフォークの違いはなく、それだろうと

synchronize :: (WhateverIOMonadClassItWouldWant m) => TChan a -> Source a m r 

お知らせもっと一般的であるため、conduitのSourceを使ってソケット以外のものをポーリングしたいかもしれない他のユーザーにとっては役に立ちます。

使用すると、1つの関数にこれらの機能を組み合わせる場合は、コールの全体的なAPIは次のようになります：

poll :: (WhateverIOMonadClassItWouldWant m) => [Source a m r] -> m (Source a m r) 

...しかし、あなたが望むなら、あなたはまだ、これらのk秒で投げることができます。