問題タブ [scalaz-stream]

scalaz-stream Process1 を使用してモデル化したいステート マシンがあります。

ステート マシンは、クライアントとサーバー間のメッセージ フローをモデル化します。

基本的なデータ型のセットは次のとおりです。

Process1[I, O]私の頭では、これは a 、 where type I = Message、およびによってモデル化されtype O = scalaz.State[Message, ServerState]ます。

それは霧のようなものです - 解決策の輪郭は見えますが、その明確な定義は私から離れます。

現在、私はこのようなものを持っています

これが間違っていることはわかっていますが、状態遷移がどこに存在する必要があるかわかりません。

ドライバーが内部状態を処理している間、プロセスは物理を受け入れてMessage返す必要がありますか?ServerStateProcess1

現時点では実行できないメッセージを「運ぶ」方法を見つけるのに苦労しています。

例：

私はこれをやってみましたが、うまくいきませんでした:

2 つの部分からなる質問があるので、まず背景を説明させてください。私は、次のように私が望むものと同様のことを行うことが可能であることを知っています:

p1ハングせずに以下の出力が得られるように使用できる他のものはありますか(のようになりますprocess1.awaitOption)?

はいの場合、次の質問に答えるのは簡単だと思います。p1ハングせずに以下の出力が得られるように使用できる他のものはありますか(のようになりますprocess1.chunkAll)?

編集：

質問を補足して、より理解しやすくします。次のようなループがある場合:

結果は次のようになります。

私がやろうとしていることは今、はっきりしていると思います。問題は、ループを制御できないことです。キューに値がない場合はループをブロックしてはなりません。

私は scalaz-stream と特に scalaz.stream.tcp の両方に非常に慣れていません。私は自分の教育目的で非常に単純なサーバーを作ろうとしています。リクエストをコマンドに解析し、それらを実行してレスポンスを生成し、レスポンスをクライアントに書き戻します。私が問題を抱えている部分は、受信した各コマンドを標準出力に記録したいということです。

tcp.server に渡す内部プロセスは次のとおりです。

(私はどこにでも型を指定する習慣はありません。物事を把握するためにそうしました。それらを削除する予定です。)

上記のコードは、実際には正しくコンパイルおよび実行されますが、あまりクリーンでも慣用的でもないように感じます。具体的には、loggedCmds の部分に不満があります。.observer または writer.logged/mapW/drainW を使用して、io.stdOutLines を使用したかったのですが、何を試しても、型を正しく並べることができなかったようです。タスクと接続の間で常に型の競合が発生していました。tcp.lift は入力ストリームには役立つようですが、シンクには機能しないようです。LoggedCmds 部分を実行するためのよりクリーンな/より良い方法はありますか (FWIW: 上記のコードの修正または改善を受け入れます)。

結果を io.stdOutLines 経由で stdout に送るだけであれば問題はありません (例で見たように、その場合は「スルー」が機能するようです)。ストリームを io.stdOutLines に送信し、引き続きストリームを使用してクライアントに応答します。

入力ファイルmyInput.txtが次のようになっているとします。

つまり、 で区切られたドキュメントがあります~~~。目的の出力は次のとおりです。

どうすればいいですか？以下はかなり不自然に思えます。さらに、タイトルを失います。

私のアプリケーションでは、並行して動作する最大 N 個のコンシューマーとプロデューサーがいます。コンシューマーはプロデューサーからリソースを取得し、作業を行い、結果を に追加して、追加のupdateQueueリソースを要求します。Producer は最初に利用可能なリソースをいくつか持っており、updateQueue. 新しいリソースがコンシューマに発行される前に、利用可能なすべての更新を適用することが重要です。次のジェネレーターを使用して、消費者が要求を行うたびに「一括」更新を要求し、新しいリソース (消費者には必要ありませんが、後で他の消費者から要求される可能性があります) を確保してみましたticketQueue:

機能しません - 最初に利用可能なリソースが から放出され、ticketQueue.dequeue次に でブロックされているように見えますwye。ログ:

で最初に使用可能なリソースが 2 つあったときticketQueue。ただし、それを単に

期待どおりに機能します（ただし、「新しいリソースを発行する前に更新を適用する」という保証はありません）。適切なタイミングで更新が確実に適用されるようにするにはどうすればよいですか?

編集：次のコードを使用して解決しました：

しかし、なぜオリジナルdefが機能しなかったのかという疑問は残ります...

などの名前の複数のファイルを含むフォルダーがありますfiletype1_ddMMyyyy_hhmm。 filetype2_ddMMyyyy_hhmm

毎日、時間の異なる複数のファイルが存在する可能性があり、時間の最も長いファイルのみを解析する必要があります。非反応的なストリームの世界では、アルゴリズムは groupBy の日付として実装できますが、scalaz-stream でこれに相当するものは何ですか?

scalaz-stream次のコードを高速化するにはどうすればよいですか? 現在、70MB のテキストを処理するのに約 5 分かかります。単純な scala と同等の処理には数秒かかるため、おそらくかなり間違ったことをしているのでしょう。

（別の質問へのフォローアップ）