問題タブ [scalaz-stream]

Process[F, A]タイマーで a のスループットを制御しようとしていますProcess:

しかし、コンパイラはそれp2がProcess[Object, Int]. zipWith()署名によると、それはProcess[Task, Int].

Process[F, A]if Fis not aの出力を調整するにはどうすればよいTaskですか? 試してみましsleepUntil()たが、同様の問題がありました。

最近の scalaz-stream では、awakeEvery内部がなくなっていることがわかりました。scalaz.stream.Processピリオドで何かを実行する方法は？

次のようなストリームがあります。

アバババババ…

A と B は 1 つ以上のプロセスで構成されています。A の内容に基づいて、B をシンクまたはドレインに渡すことで、何かを実行したいと考えています。

B については評価せずに決定できるようにする必要があります。かなり大容量のメモリは使いにくいかもしれません - GB

エリックのブログ投稿で概説されているものと同様のものが必要かどうか疑問に思っています:
http://etorreborre.blogspot.com.au/2014/03/streaming-with-previous-and-next.html

を持っているとしIterator[A]ます。scalaz streamに変換したいProcess[Nothing, A]と思います。

どのように実装しfooますか?

このような問題を解決するために、scalaz-streams を使用できる (すべきか?) という予感があります。

開始項目 A があります。A を受け取り、A のリストを返す関数があります。

1 つのアイテム (開始アイテム) で始まるワーク キューがあります。各アイテムを処理 ( doSomething) すると、同じワーク キューの最後に多くのアイテムが追加される場合があります。ただし、ある時点 (何百万ものアイテムの後) に、後続の各アイテムdoSomethingが作業キューに追加するアイテムの数が減り始め、最終的には新しいアイテムが追加されなくなります (doSomething はこれらのアイテムに対して Nil を返します)。これにより、計算が最終的に終了することがわかります。

scalaz-streams がこれに適していると仮定すると、これを実装するためにどの全体的な構造または型を調べる必要があるかについてのヒントを教えてください。

単一の「ワーカー」を使用した単純な実装が完了したら、複数のワーカーを使用してキュー アイテムを並行して処理したいと考えていますdoSomething。そのため、このアルゴリズムでも効果 (ワーカーの失敗など) を処理する必要があります。

受信したイベントに関する統計を蓄積し、完了すると最終的な統計を発行する scalaz-stream チャネルを実装しようとしています。

具体的で簡単な例を挙げるとProcess[Task, String]、各文字列が単語である があるとします。Channel[Task, String, (String, Int)]その初期プロセスに適用すると、それを排出し、各単語の出現回数をカウントし、それを出力したいと思います。

私はこれが折りたたみを通して簡単であることを理解しています:

私が書こうとしているのは、明示的にフォールドするのではなく、プロセスをパイプするだけでよい標準アキュムレータのコレクションです。たとえば、次のように記述します。

私は少し途方に暮れています-状態を共有せずにそうする方法を理解することはできません. Sinkaに蓄積する a を書くのMap[String, Int]はかなり簡単ですが、マップの最終状態を取得して、シンクが終了した後にそれを出力する方法はありません。

私は scalaz-stream プロセスを持っています:

オプションを削除するにはどうすればよいですか?

これまでのところ、collect を使用してきましたが、エレガントではありません。

より良い方法はありますか？