Wie 2 Waschbecken eines anderen Typs zu kombinieren?

Question

Ich möchte Datei in S3 mit Alpakka hochladen und gleichzeitig mit Tika parsen, um seinen MimeType zu erhalten.

Ich habe 3 Teile Graph zur Zeit:

val fileSource: Source[ByteString, Any] // comes from Akka-HTTP 
val fileUpload: Sink[ByteString, Future[MultipartUploadResult]] // created by S3Client from Alpakka 
val mimeTypeDetection: Sink[ByteString, Future[MediaType.Binary]] // my implementation using Apache Tika 

Ich mag beiden Ergebnisse an einem Ort erhalten, so etwas wie:

Future[(MultipartUploadResult, MediaType.Binary)] 

Ich habe kein Problem mit Rundfunk-Teil:

val broadcast = builder.add(Broadcast[ByteString](2)) 

source ~> broadcast ~> fileUpload 
      broadcast ~> mimeTypeDetection 

Allerdings habe ich ein Problem, Sinks zu komponieren. Methoden, die ich in der API und in der Dokumentation gefunden habe, gehen davon aus, dass entweder kombinierte Senken vom gleichen Typ sind oder dass ich Flüsse und nicht Senken ziehe.

Was ist der vorgeschlagene Ansatz in einem solchen Fall?

Answer 1

Zwei Möglichkeiten:

1) mit alsoToMat (einfacher, keine GraphDSL ausreichend für Beispiel)

val mat1: (Future[MultipartUploadResult], Future[Binary]) = 
    fileSource 
    .alsoToMat(fileUpload)(Keep.right) 
    .toMat(mimeTypeDetection)(Keep.both) 
    .run() 

2) GraphDSL mit benutzerdefinierten mit materialisierten Werte (ausführlicher, flexibler). Weitere Informationen hierzu in der docs)

val mat2: (Future[MultipartUploadResult], Future[Binary]) = 
    RunnableGraph.fromGraph(GraphDSL.create(fileUpload, mimeTypeDetection)((_, _)) { implicit builder => 
     (fileUpload, mimeTypeDetection) => 
     import GraphDSL.Implicits._ 
     val broadcast = builder.add(Broadcast[ByteString](2)) 

     fileSource ~> broadcast ~> fileUpload 
         broadcast ~> mimeTypeDetection 
     ClosedShape 
    }).run()