アッカストリーム：何mapMaterializedValue私は<a href="http://doc.akka.io/docs/akka/current/java/stream/stream-flows-and-basics.html#stream-materialization" rel="nofollow noreferrer">Akka streams materialization concept</a>を読んだ

Question

を意味し、ストリームのマテリアを理解しないことである。

実行するために、ストリーム記述（グラフ）を取り、それを必要とするすべての必要なリソースを割り当てる処理。

mapMaterializedValueを使用してキューにメッセージを送信すると、akkaストリームを作成する1つの例が続きました。コードの目的は、ストリームブループリント後にキューにメッセージをプッシュすることで構築しているとコードが機能していますが、私は本当にmapMaterrializaedValueがコードで何をするのか理解していない：

Promise<SourceQueueWithComplete<String>> promise = new Promise.DefaultPromise<>(); 

Source<String, SourceQueueWithComplete<String>> s = Source 
    .queue(100, OverflowStrategy.fail()) 
    .mapMaterializaedValue(queue -> { 
     promise.trySuccess(queue); 
    }); 

source.toMat(Sink.foreach(x -> System.out.println(x)), Keep.left()).run(materIalizer); 

promise.<SourceQueueWithComplete<String>>future().map(mapMapperFunction(), actorSystem.dispatcher()); 

Answer 1

mapMaterializedValueの目的はマテリアライズされた値は、マテリアライズされた直後に変換されます。たとえば、あなたがこのようなコールバックを受け入れ、サードパーティのライブラリを持っているとします

interface Callback<T> { 
    void onNext(T next); 
    void onError(Throwable t); 
    void onComplete(); 
} 

その後マテリアライズド値あなたはすぐにそのサードパーティのライブラリに渡すことができるSource<T, Callback<T>>を返すメソッドを作成することができますストリームが実際に実行されます。

<T> Source<T, Callback<T>> callbackSource() { 
    return Source.queue(1024, OverflowStrategy.fail()) 
     .mapMaterializedValue(queue -> new Callback<T> { 
      // an implementation of Callback which pushes the data 
      // to the queue 
     }); 
} 

Source<Integer, Callback<Integer>> source = callbackSource(); 

Callback<Integer> callback = source 
    .toMat(Sink.foreach(System.out::println), Keep.left()) 
    .run(materializer); 

thirdPartyApiObject.runSomethingWithCallback(callback); 

あなたはこのキュー行うので、これは、サードパーティのAPIのような種類を使用する必要があり、コードを簡略化することができることをここで見ることができます - 一度だけ>コールバック変換をして、それをカプセル化方法。

ただし、実際には必要ありません。あなたは、あなただけの直接そのマテリア後マテリアライズド値を使用することができますので、完全に不要であるマテリアライズド値、外部の約束を完了するために、mapMaterializedValueを使用している：

Source<String, SourceQueueWithComplete<String>> s = Source 
    .queue(100, OverflowStrategy.fail()); 

SourceQueueWithComplete<String> queue = source 
    .toMat(Sink.foreach(x -> System.out.println(x)), Keep.left()) 
    .run(materIalizer); 

mapMapperFunction().apply(queue); 

は