問題タブ [iterate]

ストリーム（遅延リスト）を使用して、多数のソース（簡単にするために2つなど）からオンデマンドでプルすることができます。反復は、単一のソースからのデータを処理するために使用できます。

複数の入力ソースを処理するためのIterateeのような機能概念はありますか？状態がどのソースからプルしたいのかを示すIterateeを想像することができます。

Iteratee を介してファイルを S3 に送信する可能性についていくつか読んだことがあります。これにより、ファイルを受信したときにファイルの S3 チャンクを送信し、たとえば大きなファイルの OutOfMemory を回避できるようです。

おそらく私がする必要があるこの SO 投稿を見つけました: Play 2.x : Reactive file upload with Iteratees Sadek Brodi は、foldM は Play 2.1 で利用可能であると述べているため、例としてのみ)

Iteratees についてのブログを読んだことがあり、まだ Scala/Play2 の専門家ではない人のために、誰かがこれを簡単に説明できますか?

マルチパート ボディ パーサーなどを使用する必要があるかどうかもわかりませんが、このコードが何をしているのか理解できないことが 1 つあります。

ちなみに、従来のJava InputStream / OutputStreamを使用した場合と比較して、メモリ消費量の違いはどうなるでしょうか。私は実際に、Java + AsyncHttpClient + Grizzly を使用して、Iteratees を使用せずに、非常に低いメモリ消費量で非ブロッキングの方法で 500 MB のファイルを S3 に転送できます (ただし、Netty でも動作すると思います)。

では、Iteratee を使用する利点は何でしょうか?

私が見ることができる 1 つの違いは、取得して S3 に転送する InputStream は、私の場合、一時ファイル (これは CXF の動作です) によってサポートされているため、Play Iteratee ほど反応的ではない可能性があることです。

しかし、Iteratee では、Enumerator が接続によって受信されたバイトを生成し、Iteratee を介して S3 に転送する場合、S3 への接続が良好ではなく、バイトを非常に高速に転送できない場合、「保留中」のバイトが格納されます。 ?

重複の可能性:
クライアントへの Play2 フレームワーク プロキシ ストリーミング コンテンツは、ストリーミングが完了した後も接続を開いたままにします

Web サービスからクライアントに 11 MB のファイルをストリーミングしています。これは基本的にパススルー プロキシです。ここに私のコードがあります:

そして私はそれを呼びます：

遅いサービスの場合、これはうまく機能し、ファイル全体を取得します。高速なサービスの場合、ファイルのごく一部 (サイズは異なります) のみを取得し、残りの画像は切り取ります。画像がたとえば 11 MB の場合、.close() メソッドがストリームを強制終了する前に 2 MB しか取得できません。

Iteratee がすべてのデータを取得することは知っていますが (印刷すると表示されます)、Enumerator の .close() 呼び出しの実行が早すぎて、interatee が実行できるようになる前に列挙子を切断するようです。それを埋めます。

いくつかの質問を聞きたいんです：

1. これはプロキシを実行する正しい方法ですか? 私は最初に従来の Java IO ストリームでこれを行いました。これは正常に機能しましたが、慣用的またはノンブロッキングではありませんでした

2. これがこの問題を解決するための合理的な方法である場合、Enumerator.close() メソッドはどこに行くべきですか? .onRedeem には .orTimeout と同じ問題があります

ありがとうございました！

以下のコードは、クライアントにストリーミングを戻します。私が収集するのは、JavaのIOストリームを使用するよりも慣用的な方法です。ただし、問題があります。ストリームが実行された後、接続が開いたままになります。

これは、内部APIからリクエスターに大きな（> 1 mb）ファイルをストリーミングすることを目的としています。

問題は、なぜ接続を開いたままにするのかということです。アップストリームサーバーに期待するものはありますか？curlを使用してアップストリームサーバーをテストしましたが、接続が閉じられます。このプロキシを通過したときに接続が閉じられません。

sbt play プロジェクトで、次のconsole-projectコードを実行して貼り付けます。

ただし、タイムアウトで死亡します。ここで何か間違ったことをしていますか？

このコードは、 「Lazyv.Yield」という論文に基づいています。データストリームのプロデューサーとコンシューマーを切り離す方法についてです。コードのHaskellの部分は理解していますが、O'Caml / F＃ではわかりません。次の理由でこのコードを理解していません。

引数として例外を取り、単位を返す関数にはどのような動作が期待できますか？

消費者はどのように特定の例外に投影しますか？（どういう意味ですか？）

消費者の例は何でしょうか？

Enumerator または EnumeratorM を使用して任意の関数を構成し、iteratee にプッシュされる個々のデータ項目が関数を適用することによって最初に前処理されるようにすることは可能ですか?

iteratee は怠け者だと聞いたことがありますが、具体的にどのくらい怠けているのでしょうか? あるいは、中間データ構造を構築する必要がないように、反復対象を後処理関数と融合できますか?

たとえば、 iteratee で aStream[Option[String]] からjava.io.BufferedReader100 万個のアイテムを作成し、その後、NoneStream 全体をメモリに保持する必要なく、構成的な方法で s を除外できますか? 同時に、スタックを爆破しないことを保証しますか? またはそのようなもの - を使用する必要はありませんStream。

私は現在 Scalaz 6 を使用していますが、他の iteratee 実装がより良い方法でこれを行うことができれば、知りたいです。

該当する場合は、 を閉じてBufferedReaderを呼び出すなど、完全な解決策を提供してください。unsafePerformIO

Scalaz 7のiterateeライブラリjava.io.BufferedReaderを使用して、ファイルを1行ずつ読み取るための列挙子を作成しようとしています。このライブラリは、現在、の（非常に遅い）列挙子のみを提供します。java.io.Reader

私が遭遇している問題は、私が使用した他のすべてのiterateeライブラリ（たとえば、Play2.0やHaskellのJohnMillikin）が、その型のコンストラクターの1つとしてエラー状態になっているという事実とScalaz7に関連していenumeratorStepます。そうではありません。

私の現在の実装

これが私が現在持っているものです。まず、いくつかのインポートとIOラッパーについて：

そして、物事を少しクリーンアップするためのタイプエイリアス：

そして今、tryIOヘルパーは、次のものをモデルにしています（大まかに、そしておそらく間違っています）enumerator：

BufferedReaderそれ自体の列挙子：

そして最後に、リーダーの開閉を担当する列挙子：

たとえば、少なくとも25'0'文字を含むファイル内のすべての行をリストにまとめたいとします。私は書くことができます：

多くの点で、これは見事に機能しているように見えます。アクションを開始するunsafePerformIOと、数千万行とギガバイトのデータを数分で、一定のメモリ内で、スタックを爆破することなくチャンクし、リーダーを閉じます。終わったら。存在しないファイルの名前を付けると、。でラップされた例外が忠実に返され、読み取り中に例外が発生した場合でもLeft、enumBuffered少なくとも適切に動作するように見えます。

潜在的な問題

ただし、特に。の実装について懸念がありtryIOます。たとえば、いくつかの反復を作成しようとしたとします。

これを実行すると、次のようになります。

GHCiで同じことを試してみるとenumerator、結果は私が期待するものに似ています。

iterateeライブラリ自体にエラー状態がない状態でこの動作を取得する方法がわかりません。

私の質問

私は反復の専門家であるとは言いませんが、いくつかのプロジェクトでさまざまなHaskellの実装を使用し、基本的な概念を多かれ少なかれ理解しているように感じ、Olegと一度コーヒーを飲みました。しかし、私はここで途方に暮れています。これは、エラー状態がない場合に例外を処理するための合理的な方法ですか？バージョンtryIOのように動作する実装方法はありますか？enumerator私の実装の動作が異なるという事実で、私を待っているある種の時限爆弾はありますか？

Iteratee.scalaのソースコードを読んでいます：https：//github.com/playframework/Play20/blob/master/framework/src/iteratees/src/main/scala/play/api/libs/iteratee/Iteratee。スカラ

具体的には、「フォールド」イテラティーを構築するための便利な方法です。

それぞれのcaseステートメントで、DoneまたはContコンストラクターを呼び出しています。しかし、これらのコンストラクターはどこで定義されていますか？これらはIterateeトレイトの実装者である必要があると推測しますが、「extends Iteratee」に対して、ctrl+Fを実行してもそれらを見つけることができませんでした。