問題タブ [scalaz-stream]

TCP から読み取り、標準出力に書き込む単純な scalaz-stream の例を実行できません。

何も印刷せずに、ただそこに座っています。

また、 を使用してさまざまなアレンジを試みましたがto、常にtcp.liftを取得するための呪文がProcess[Connection, A]含まれています。

これはコンパイルさえしません。

wyeソース ストリームと印刷ストリームを一緒にする必要がありますか? 元のプルリクエストで見つけたtcp置換の例nioはこれを示しているように見えましたがwye、Process.

編集すると、Paul によって説明された型の問題に加えて、内部プロセスを「手動で」実行する必要があることがわかりましたp.map(_.run.run).run.run。これが慣用的な方法だとは思いませんが、うまくいきます。

scalaz-stream を使用してストリームを分割/フォークしてから再結合することは可能ですか?

例として、次の関数があるとしましょう

scalaz-stream を使用すると、この例では期待どおりの結果が得られます - 1 から 10 までの数値のタプルがシンクに渡されます。

streamOfNumbersただし、 IO を必要とするものに置き換えると、実際には IO アクションが 2 回実行されます。

を使用するTopicと、ストリーム内の要素を正しく複製する pub/sub プロセスを作成できますが、バッファリングは行われません。シンクが消費するペースに関係なく、ソース全体をできるだけ速く消費するだけです。

これを制限された Queue でラップすることはできますが、最終結果は必要以上に複雑に感じられます。

ソースから IO アクションを複製せずに scalaz-stream でストリームを分割する簡単な方法はありますか?

いくつかの scalaz-stream ドキュメントから例を挙げてみましょう。ただし、理論的なひねりがあります。

この場合、ファイルには double 以外の文字列が含まれている可能性が非常に高く、 はfahrenheitToCelsiusいくつかのNumberFormatException. この場合、このエラーをログに記録し、その後のストリーム処理のために無視したいとしましょう。それを行う慣用的な方法は何ですか？いくつかの例を見てきましたが、通常collectFromはストリームです。

最初にリクエストをサーバーに送信し、その返信を待つクライアントアプリを実装したい(httpと同様)

私のクライアントプロセスは

ここにアピがあります

次に、残りの部分を実装する方法はask ？

ログ ファイルの解析に役立つプロセスを作成しました。ログ ファイルは、文字列タグでタグ付けする必要があります。レコードがこの文字列タグを提供できるわけではないので、各ログ イベントで何らかの状態を維持して、各イベントにタグが付くようにする必要があります。イベントにタグがない場合は、最後に見つかったタグを使用したいと考えています。これにより、以下のようなログを解析し、メッセージ行が eventindex04 レコード ヘッダーの一部であっても、「タグ」をメッセージ行に関連付けることができます。イベントをエンベロープでラップし、ロジックをエンベロープに含めることもできますが、どこにでもエンベロープがあり、重いように見えます。

私はうまく機能するこのプロセスをまとめました：

これは、スキャン/フォールドで表現する必要があるように感じますが、機能させることができませんでした。これをスキャン/フォールドとして書き換えることはできますか?

slick 3.0.0 クエリから返されたデータを db.stream(yourquery) 経由で scalaz-stream にストリーミングしたかったのです。

react-streams.org は、さまざまなライブラリが実装する API とデータフロー モデルを使用しているようです。

scalaz-stream プロセスから洗練されたパブリッシャーに逆流するバック プレッシャーをどのように処理しますか?

チャンクして連結する慣用的な方法はありますか?

私が見つけた方法 (バイトの例):

1.

ただし、この場合、中間Vector(からchunk) は実際には必要ありません。

2. process1.chunk からのコピー/貼り付けに基づく:

Process既存の 'esを組み合わせて同じことを行う方法はありますか?

副次的な質問: ?repeatの代わりに使用できます。++ goこれは前のものと同じですか：