Переслать поток загрузки файла на S3 через Iteratee с помощью Play2/Scala

Я прочитал кое-что о возможности отправки файла на S3 через Iteratee, что, по-видимому, позволяет отправлять так S3-фрагменты файла по мере их получения и избегать OutOfMemory для больших файлов, например.

Я нашел этот пост SO, который, вероятно, почти то, что мне нужно сделать: Play 2.x: Загрузка активного файла с помощью Iteratees Я действительно не понимаю, как это сделать, или если это действительно доступно в Play 2.0.2 (потому что Садек Броди говорит, что foldM доступен в Play 2.1 только для примера)

Может кто-нибудь объяснить это простым способом, для тех, кто прочитал какой-то блог об Iteratees и еще не является экспертом Scala/Play2?

Я даже не знаю, должен ли я использовать парсер партирования или что-то в этом роде, но одна вещь, которую я знаю, это то, что я не понимаю, что делает этот код:

val consumeAMB = 
  Traversable.takeUpTo[Array[Byte]](1028*1028) &>> Iteratee.consume()

val rechunkAdapter:Enumeratee[Array[Byte],Array[Byte]] =
  Enumeratee.grouped(consumeAMB)

val writeToStore: Iteratee[Array[Byte],_] =
  Iteratee.foldM[Array[Byte],_](connectionHandle){ (c,bytes) => 
    // write bytes and return next handle, probable in a Future
  }

BodyParser( rh => (rechunkAdapter &>> writeToStore).map(Right(_)))

Кстати, какова будет разница в потреблении памяти по сравнению с использованием классического Java InputStream/OutputStream. Я действительно могу переслать 500-мегабайтный файл на S3 неблокирующим способом с очень низким потреблением памяти без использования Iteratees, используя Java + AsyncHttpClient + Grizzly (но, я думаю, он также будет работать с Netty).

Итак, какое преимущество использования Iteratee?

Единственное отличие, которое я вижу, это то, что InputStream, который я получаю и пересылаю на S3, в моем случае поддерживается временным файлом (это поведение CXF), поэтому он может быть не так реактивным, как Play Iteratee

Но с помощью Iteratees, если Enumerator создает байты, полученные соединением, и перенаправляет их на S3 через Iteratee, тогда, если соединение с S3 не является хорошим, и байты не могут быть перенаправлены очень быстро, где хранятся "ожидающие" байты?

Ответ 1

Простое объяснение? Я попробую.:)

Вы строите трубопровод из компонентов. После того, как вы построили конвейер, его можно отправить. Это Iteratee, поэтому он знает, как выполнять итерацию данных.

Файл, который вы хотите загрузить, содержится в теле запроса, а BodyParser - это то, что обрабатывает тела запросов в Play. Таким образом, вы помещаете свой итерационный конвейер в BodyParser. Когда запрос будет сделан, ваш конвейер будет отправлен (он будет перебирать его).

Ваш конвейер (rechunkAdapter &>> writeToStore) разбивает данные на бит 1 МБ, а затем отправляет их на S3.

Первая часть конвейера (rechunkAdapter) выполняет разделение. На самом деле у него есть собственный мини-трубопровод, который выполняет разбиение (consumeAMB). Как только мини-труба получила достаточное количество данных, чтобы сделать кусок, он отправляет его по основному конвейеру.

Вторая часть конвейера (writeToStore) похожа на цикл, который вызывается на каждом фрагменте, что дает вам возможность отправлять каждый фрагмент на S3.

Преимущества итераций?

Как только вы знаете, что происходит, вы можете создавать итерационные конвейеры, подключая компоненты. И проверка типа будет чаще всего рассказывать вам, когда вы неправильно подключаете что-то вместе.

Например, мы можем изменить конвейер выше, чтобы исправить тот факт, что он медленный. Это, вероятно, медленное, потому что загрузка запроса приостанавливается всякий раз, когда кусок готов к загрузке на S3. Важно замедлить загрузку запроса, чтобы у нас не хватило памяти, но мы могли бы быть немного более прощающими, добавляя буфер фиксированного размера. Поэтому просто добавьте Concurrent.buffer(2) в середину конвейера для буферизации до 2-х кусков.

Iteratees обеспечивают функциональный подход к потокам. Это преимущество или недостаток, в зависимости от того, как вы относитесь к функциональному программированию.:) По сравнению с ленивыми потоками (другой функциональный подход) итераторы предлагают точный контроль над ресурсом.

Наконец, итерации позволяют нам сделать очень сложное асинхронное потоковое программирование относительно (!) просто. Мы можем обрабатывать IO без хранения потоков, что является огромной победой для масштабируемости. Для классического примера Java InputStream/OutputStream требуется 2 потока.