Comment combiner 2 puits d'un type différent?

Question

Je veux télécharger le fichier dans S3 en utilisant Alpakka et en même temps l'analyser avec Tika pour obtenir son MimeType.

J'ai 3 parties du graphique au moment:

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 

Je voudrais obtenir les résultats à un endroit, quelque chose comme:

Future[(MultipartUploadResult, MediaType.Binary)] 

je ne conteste pas partie de la radiodiffusion:

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

source ~> broadcast ~> fileUpload 
      broadcast ~> mimeTypeDetection 

Cependant, j'ai de la difficulté à composer des puits. Les méthodes que j'ai trouvées dans l'API et dans la documentation supposent que les deux puits combinés sont du même type ou que je suis Zipping Flows, pas les puits.

Quelle est l'approche suggérée dans un tel cas?

Answer 1

Deux façons:

1) à l'aide alsoToMat (plus facile, pas GraphDSL, assez pour votre exemple)

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

2) à l'aide GraphDSL avec des valeurs matérialisées sur mesure (plus bavard, plus souple). Plus d'informations à ce sujet dans le 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()