Używanie Iteratees i enumeratorów w Play Scala do przesyłania danych do S3

Question

Buduję aplikację Play Framework w Scali, w której chciałbym przesłać tablicę bajtów do S3. Używam biblioteki Play-S3, aby to zrobić. W „Multipart upload plików” sekcji Dokumentacja jest co istotne tutaj:

// Retrieve an upload ticket 
val result:Future[BucketFileUploadTicket] = 
    bucket initiateMultipartUpload BucketFile(fileName, mimeType) 

// Upload the parts and save the tickets 
val result:Future[BucketFilePartUploadTicket] = 
    bucket uploadPart (uploadTicket, BucketFilePart(partNumber, content)) 

// Complete the upload using both the upload ticket and the part upload tickets 
val result:Future[Unit] = 
    bucket completeMultipartUpload (uploadTicket, partUploadTickets) 

staram się robić to samo w mojej aplikacji, ale z Iteratee s oraz Enumerator s.

Strumienie i asynchroniczność domiar trochę skomplikowane, ale tutaj jest to, co mam tak daleko (Uwaga uploadTicket jest zdefiniowane wcześniej w kodzie):

val partNumberStream = Stream.iterate(1)(_ + 1).iterator 
val partUploadTicketsIteratee = Iteratee.fold[Array[Byte], Future[Vector[BucketFilePartUploadTicket]]](Future.successful(Vector.empty[BucketFilePartUploadTicket])) { (partUploadTickets, bytes) => 
    bucket.uploadPart(uploadTicket, BucketFilePart(partNumberStream.next(), bytes)).flatMap(partUploadTicket => partUploadTickets.map(_ :+ partUploadTicket)) 
} 
(body |>>> partUploadTicketsIteratee).andThen { 
    case result => 
    result.map(_.map(partUploadTickets => bucket.completeMultipartUpload(uploadTicket, partUploadTickets))) match { 
     case Success(x) => x.map(d => println("Success")) 
     case Failure(t) => throw t 
    } 
} 

Wszystko kompiluje i działa bez incydentu. W rzeczywistości drukowane jest "Success", ale żaden plik nigdy nie pojawia się na S3.

Answer 1

Może być wiele problemów z kodem. Jest to trochę nieczytelne z powodu wywołań metodowych map. Możesz mieć problem z przyszłym składem. Kolejny problem może wynikać z faktu, że wszystkie porcje (z wyjątkiem ostatniego) powinny wynosić co najmniej 5 MB.

Poniższy kod nie został przetestowany, ale pokazuje inne podejście. Podejście iteracyjne to takie, w którym można tworzyć małe bloki i komponować je w potok operacji.

Aby dokonać kompilacji kodu I dodaną cechę i kilka metod

trait BucketFilePartUploadTicket 
val uploadPart: (Int, Array[Byte]) => Future[BucketFilePartUploadTicket] = ??? 
val completeUpload: Seq[BucketFilePartUploadTicket] => Future[Unit] = ??? 
val body: Enumerator[Array[Byte]] = ??? 

Tutaj tworzymy kilka części

// Create 5MB chunks 
val chunked = { 
    val take5MB = Traversable.takeUpTo[Array[Byte]](1024 * 1024 * 5) 
    Enumeratee.grouped(take5MB transform Iteratee.consume()) 
} 

// Add a counter, used as part number later on 
val zipWithIndex = Enumeratee.scanLeft[Array[Byte]](0 -> Array.empty[Byte]) { 
    case ((counter, _), bytes) => (counter + 1) -> bytes 
} 

// Map the (Int, Array[Byte]) tuple to a BucketFilePartUploadTicket 
val uploadPartTickets = Enumeratee.mapM[(Int, Array[Byte])](uploadPart.tupled) 

// Construct the pipe to connect to the enumerator 
// the ><> operator is an alias for compose, it is more intuitive because of 
// it's arrow like structure 
val pipe = chunked ><> zipWithIndex ><> uploadPartTickets 

// Create a consumer that ends by finishing the upload 
val consumeAndComplete = 
    Iteratee.getChunks[BucketFilePartUploadTicket] mapM completeUpload 

przebiegu to się robi po prostu połączenia części

// This is the result, a Future[Unit] 
val result = body through pipe run consumeAndComplete 

Pamiętaj, że nie przetestowałem żadnego kodu i mogłem popełnić błędy w moim podejściu. To jednak pokazuje inny sposób radzenia sobie z problemem i prawdopodobnie powinno pomóc w znalezieniu dobrego rozwiązania.

Należy zauważyć, że to podejście czeka na zakończenie jednej części, zanim przejmie kolejną część. Jeśli połączenie z serwera do Amazon jest wolniejsze niż połączenie z przeglądarki do serwera, ten mechanizm spowolni wprowadzanie danych.

Można zastosować inne podejście, w którym nie trzeba czekać na zakończenie ładowania części do Future. Spowoduje to kolejny krok, w którym użyjesz Future.sequence do konwersji sekwencji przyszłych transakcji do jednej przyszłości zawierającej sekwencję wyników. Rezultatem byłby mechanizm wysyłania części do Amazona, gdy tylko masz wystarczającą ilość danych.