import akka.stream.javadsl.Keep; //导入依赖的package包/类
public AkkaHubProxy(ActorMaterializer mat) {
    this.mat = mat;
    //  Obtain a Sink and Source which will publish and receive from the "bus" respectively.
    Pair<Sink<Object, NotUsed>, Source<Object, NotUsed>> sinkAndSource =
            MergeHub.of(Object.class, 16)
                    .toMat(BroadcastHub.of(Object.class, 256), Keep.both())
                    .run(mat);

    Sink<Object, NotUsed> sink = sinkAndSource.first();
    source = sinkAndSource.second().takeWhile((Predicate<Object>) o -> o != Done.getInstance());
    //source.runWith(Sink.ignore(), mat);
    busFlow = Flow.fromSinkAndSource(sink, source)
            .joinMat(KillSwitches.singleBidi(), Keep.right());
}
 

import akka.stream.javadsl.Keep; //导入依赖的package包/类
@Override
public Removable addUpstream(Source<Object, NotUsed> publisher) {
    UniqueKillSwitch killSwitch =
            publisher.viaMat(busFlow, Keep.right())
                    .to(Sink.ignore())
                    .run(mat);
    subscriptions.put(publisher, killSwitch);
    return () -> AkkaHubProxy.this.removeUpstream(publisher);
}
 

import akka.stream.javadsl.Keep; //导入依赖的package包/类
@Override
public Receive createReceive() {
    return receiveBuilder()
            .match(RequestReply.class, requestReply -> {
                // TODO(barp): Fix the ugly HttpRequest cast here due to java vs scala dsl
                akka.stream.javadsl.Source.single(requestReply.getRequest())
                        .via(_processGraph)
                        .toMat(_sink, Keep.right())
                        .run(_materializer)
                        .whenComplete((done, err) -> {
                            final CompletableFuture<HttpResponse> responseFuture = requestReply.getResponse();
                            if (err == null) {
                                responseFuture.complete(HttpResponse.create().withStatus(200));
                            } else {
                                BAD_REQUEST_LOGGER.warn()
                                        .setMessage("Error handling http post")
                                        .setThrowable(err)
                                        .log();
                                if (err instanceof ParsingException) {
                                    responseFuture.complete(HttpResponse.create().withStatus(400));
                                } else {
                                    responseFuture.complete(HttpResponse.create().withStatus(500));
                                }
                            }
                        });
            })
            .build();
}
 

import akka.stream.javadsl.Keep; //导入依赖的package包/类
private Flow<ByteString, ByteString, CompletionStage<HttpResponse>> createFlow(HttpMethod method, Uri uri, Option<ContentType> contentType, Predicate<HttpResponse> isSuccess, HttpHeader... headers) {
    Sink<ByteString, Publisher<ByteString>> in = Sink.asPublisher(AsPublisher.WITH_FANOUT); // akka internally recreates this twice, on some errors...
    Source<ByteString, Subscriber<ByteString>> out = Source.asSubscriber();
    
    return Flow.fromSinkAndSourceMat(in, out, Keep.both()).mapMaterializedValue(pair -> {
        RequestEntity entity;
        if (contentType.isDefined()) {
            Source<ByteString, NotUsed> inReader = Source.fromPublisher(pair.first());
            entity = HttpEntities.createChunked(contentType.get(), inReader);
        } else {
            entity = HttpEntities.EMPTY;
        }
        HttpRequest rq = HttpRequest.create().withMethod(method).withUri(uri).addHeaders(Arrays.asList(headers)).withEntity(entity);
        
        return http.singleRequest(rq, materializer).thenApply(resp -> {
            if (isSuccess.test(resp)) {
                resp.entity().getDataBytes()
                    .runWith(Sink.fromSubscriber(pair.second()), materializer);
            } else {
                log.info("Http responded error: {} for request {}", resp, rq);
                resp.discardEntityBytes(materializer);
                pair.second().onError(new IllegalStateException("Unsuccessful HTTP response: " + resp + " for " + rq));
            }
            return resp;
        }).exceptionally(x -> {
            Throwable cause = (x instanceof CompletionException) ? x.getCause() : x;
            if (!(cause instanceof IllegalStateException)) {
                log.info("Could not make http request " + rq, cause);
            }
            pair.second().onError(cause);
            throw (cause instanceof RuntimeException) ? (RuntimeException) x : new RuntimeException(cause);
        });
    });
}
 

import akka.stream.javadsl.Keep; //导入依赖的package包/类
@Test
public void testCreateWebSocketFlow() {
    // Injected dependencies to create the controller...
    final TestProbe stocksActorProbe = new TestProbe(system, "stocksActor");
    final TestProbe userParentActorProbe = new TestProbe(system, "userParentActor");

    // Create the controller without having to create a play app...
    final HomeController controller = new HomeController(system,
            materializer,
            webJarAssets,
            stocksActorProbe.ref(),
            userParentActorProbe.ref());

    // input for creating a flow...
    final TestProbe userActorProbe = new TestProbe(system, "userActor");
    TestPublisher.Probe<JsonNode> publisher = TestPublisher.probe(0, system);

    // method under test
    final Flow<JsonNode, JsonNode, NotUsed> flowUnderTest = controller.createWebSocketFlow(publisher, userActorProbe.ref());

    Source<JsonNode, TestPublisher.Probe<JsonNode>> sourceProbe = TestSource.probe(system);

    // create a test source and sink around the flow
    final Pair<TestPublisher.Probe<JsonNode>, TestSubscriber.Probe<JsonNode>> pair = sourceProbe
            .via(flowUnderTest)
            .toMat(TestSink.probe(system), Keep.both())
            .run(materializer);

    final TestPublisher.Probe<JsonNode> pub = pair.first();
    final TestSubscriber.Probe<JsonNode> sub = pair.second();

    // check that a message sent in will come out the other end
    final ObjectNode jsvalue = Json.newObject().put("herp", "derp");
    sub.request(1);
    publisher.sendNext(jsvalue);
    sub.expectNext(jsvalue);
}
 

import akka.stream.javadsl.Keep; //导入依赖的package包/类
@Test
public void testCreateWebSocketFlow() {
    // Injected dependencies to create the controller...
    final TestProbe stocksActorProbe = new TestProbe(system, "stocksActor");
    final TestProbe userParentActorProbe = new TestProbe(system, "userParentActor");

    // Create the controller without having to create a play app...
    final HomeController controller = new HomeController(system,
            materializer,
            stocksActorProbe.ref(),
            userParentActorProbe.ref());

    // input for creating a flow...
    final TestProbe userActorProbe = new TestProbe(system, "userActor");
    TestPublisher.Probe<JsonNode> publisher = TestPublisher.probe(0, system);

    // method under test
    final Flow<JsonNode, JsonNode, NotUsed> flowUnderTest = controller.createWebSocketFlow(publisher, userActorProbe.ref());

    Source<JsonNode, TestPublisher.Probe<JsonNode>> sourceProbe = TestSource.probe(system);

    // create a test source and sink around the flow
    final Pair<TestPublisher.Probe<JsonNode>, TestSubscriber.Probe<JsonNode>> pair = sourceProbe
            .via(flowUnderTest)
            .toMat(TestSink.probe(system), Keep.both())
            .run(materializer);

    final TestPublisher.Probe<JsonNode> pub = pair.first();
    final TestSubscriber.Probe<JsonNode> sub = pair.second();

    // check that a message sent in will come out the other end
    final ObjectNode jsvalue = Json.newObject().put("herp", "derp");
    sub.request(1);
    publisher.sendNext(jsvalue);
    sub.expectNext(jsvalue);
}
 

import akka.stream.javadsl.Keep; //导入依赖的package包/类
@Override
public void emit(Object event) {
    Source.single(event).viaMat(busFlow, Keep.right())
            .to(Sink.ignore())
            .run(mat);
}