import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    ActorSystem system = ActorSystem.create("ServiceB");
    final Http http = Http.get(system);
    final ActorMaterializer materializer = ActorMaterializer.create(system);
    final Main app = new Main();
    final ActorRef serviceBackendActor = system.actorOf(BackendActor.props(), "backendActor");
    final Flow<HttpRequest, HttpResponse, NotUsed> routeFlow =  app.createRoute(serviceBackendActor).flow(system, materializer);

    final CompletionStage<ServerBinding> binding =
            http.bindAndHandle(
                    routeFlow,
                    ConnectHttp.toHost("localhost", 8081),
                    materializer);

    System.out.println("Server online at http://localhost:8081/\nPress RETURN to stop...");
    System.in.read(); // let it run until user presses return
    binding
            .thenCompose(ServerBinding::unbind) // trigger unbinding from the port
            .thenAccept(unbound -> system.terminate()); // and shutdown when done

}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws IOException {
    final ActorSystem system = ActorSystem.create("ServiceA");
    final Http http = Http.get(system);
    final ActorMaterializer materializer = ActorMaterializer.create(system);
    final StreamMain app = new StreamMain();
    final ActorRef streamActor = system.actorOf(StreamActor.props());
    final Flow<HttpRequest, HttpResponse, NotUsed> routeFlow = app.createRoute(streamActor).flow(system, materializer);
    final CompletionStage<ServerBinding> binding =
            http.bindAndHandle(
                    routeFlow,
                    ConnectHttp.toHost("localhost", 8080),
                    materializer);

    System.out.println("Server online at http://localhost:8080/\nPress RETURN to stop...");
    System.in.read(); // let it run until user presses return
    System.in.read(); // let it run until user presses return
    binding
            .thenCompose(ServerBinding::unbind) // trigger unbinding from the port
            .thenAccept(unbound -> system.terminate()); // and shutdown when done

}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public void run() {
  final Config conf = parseString("akka.remote.netty.tcp.hostname=" + config.hostname())
          .withFallback(parseString("akka.remote.netty.tcp.port=" + config.actorPort()))
          .withFallback(ConfigFactory.load("remote"));

  final ActorSystem system = ActorSystem.create("concierge", conf);
  kv = system.actorOf(LinearizableStorage.props(new Cluster(config.cluster().paths(), "kv")), "kv");

  final ActorMaterializer materializer = ActorMaterializer.create(system);

  final Flow<HttpRequest, HttpResponse, NotUsed> theFlow = createRoute().flow(system, materializer);

  final ConnectHttp host = ConnectHttp.toHost(config.hostname(), config.clientPort());
  Http.get(system).bindAndHandle(theFlow, host, materializer);

  LOG.info("Ama up");
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws IOException {
    final ActorSystem system = ActorSystem.create();
    final ActorMaterializer materializer = ActorMaterializer.create(system);

    final Route route = InfillionRoutes.routes();
    final Flow<HttpRequest, HttpResponse, NotUsed> flow = route.flow(system, materializer);

    final Http http = Http.get(system);
    final CompletionStage<ServerBinding> bindings = http.bindAndHandle(flow, ConnectHttp.toHost("127.0.0.1", 8080), materializer);

    System.out.println("Type RETURN to exit");
    System.in.read();

    bindings
            .thenCompose(ServerBinding::unbind)
            .thenAccept(unbound -> system.terminate());

}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
@Override
public Flow<EventEnvelope,Long,?> uploadFlow() {
    ClientConnectionSettings settings = ClientConnectionSettings.create(system.settings().config());
    
    return Flow.<EventEnvelope>create()
        .map(e -> (Message) BinaryMessage.create(serialize(e)))
        .via(Http.get(system).webSocketClientFlow(WebSocketRequest.create(uri), connectionContext, Optional.empty(), settings, system.log()))
        .map(msg -> {
            if (msg.isText()) {
                log.warn("Ignoring unexpected text-type WS message {}", msg);
                return 0l;
            } else {
                EventsPersisted applied = EventsPersisted.parseFrom(
                    msg.asBinaryMessage().getStrictData().iterator().asInputStream());
                return applied.hasOffset() ? applied.getOffset() : 0l;
            }})
        .filter(l -> l > 0);
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws IOException {

        final ActorSystem system = ActorSystem.create("example");
        final ActorMaterializer materializer = ActorMaterializer.create(system);
        final Http http = Http.get(system);

        final MessageDispatcher dispatcher = system.dispatchers().lookup("akka.actor.default-dispatcher");
        final SetSessionJava app = new SetSessionJava(dispatcher);

        final Flow<HttpRequest, HttpResponse, NotUsed> routes = app.createRoutes().flow(system, materializer);
        final CompletionStage<ServerBinding> binding = http.bindAndHandle(routes, ConnectHttp.toHost("localhost", 8080), materializer);

        System.out.println("Server started, press enter to stop");
        System.in.read();

        binding
            .thenCompose(ServerBinding::unbind)
            .thenAccept(unbound -> system.terminate());
    }
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws IOException {

        final ActorSystem system = ActorSystem.create("example");
        final ActorMaterializer materializer = ActorMaterializer.create(system);
        final Http http = Http.get(system);

        final MessageDispatcher dispatcher = system.dispatchers().lookup("akka.actor.default-dispatcher");
        final VariousSessionsJava app = new VariousSessionsJava(dispatcher);

        final Flow<HttpRequest, HttpResponse, NotUsed> routes = app.createRoutes().flow(system, materializer);
        final CompletionStage<ServerBinding> binding = http.bindAndHandle(routes, ConnectHttp.toHost("localhost", 8080), materializer);

        System.out.println("Server started, press enter to stop");
        System.in.read();

        binding
            .thenCompose(ServerBinding::unbind)
            .thenAccept(unbound -> system.terminate());
    }
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws IOException {

        // ** akka-http boiler plate **
        ActorSystem system = ActorSystem.create("example");
        final ActorMaterializer materializer = ActorMaterializer.create(system);
        final Http http = Http.get(system);

        // ** akka-http-session setup **
        MessageDispatcher dispatcher = system.dispatchers().lookup("akka.actor.default-dispatcher");
        final SessionInvalidationJava app = new SessionInvalidationJava(dispatcher);

        // ** akka-http boiler plate continued **
        final Flow<HttpRequest, HttpResponse, NotUsed> routes = app.createRoutes().flow(system, materializer);
        final CompletionStage<ServerBinding> binding = http.bindAndHandle(routes, ConnectHttp.toHost("localhost", 8080), materializer);

        System.out.println("Server started, press enter to stop");
        System.in.read();

        binding
            .thenCompose(ServerBinding::unbind)
            .thenAccept(unbound -> system.terminate());
    }
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws IOException {

        // ** akka-http boiler plate **
        ActorSystem system = ActorSystem.create("example");
        final ActorMaterializer materializer = ActorMaterializer.create(system);
        final Http http = Http.get(system);

        // ** akka-http-session setup **
        MessageDispatcher dispatcher = system.dispatchers().lookup("akka.actor.default-dispatcher");
        final JavaJwtExample app = new JavaJwtExample(dispatcher);

        // ** akka-http boiler plate continued **
        final Flow<HttpRequest, HttpResponse, NotUsed> routes = app.createRoutes().flow(system, materializer);
        final CompletionStage<ServerBinding> binding = http.bindAndHandle(routes, ConnectHttp.toHost("localhost", 8080), materializer);

        System.out.println("Server started, press enter to stop");
        System.in.read();

        binding
            .thenCompose(ServerBinding::unbind)
            .thenAccept(unbound -> system.terminate());
    }
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws IOException {

        // ** akka-http boiler plate **
        ActorSystem system = ActorSystem.create("example");
        final ActorMaterializer materializer = ActorMaterializer.create(system);
        final Http http = Http.get(system);

        // ** akka-http-session setup **
        MessageDispatcher dispatcher = system.dispatchers().lookup("akka.actor.default-dispatcher");
        final JavaExample app = new JavaExample(dispatcher);

        // ** akka-http boiler plate continued **
        final Flow<HttpRequest, HttpResponse, NotUsed> routes = app.createRoutes().flow(system, materializer);
        final CompletionStage<ServerBinding> binding = http.bindAndHandle(routes, ConnectHttp.toHost("localhost", 8080), materializer);

        System.out.println("Server started, press enter to stop");
        System.in.read();

        binding
            .thenCompose(ServerBinding::unbind)
            .thenAccept(unbound -> system.terminate());
    }
 

import akka.http.javadsl.Http; //导入依赖的package包/类
protected CompletableFuture<String> callService(String ids) throws ExecutionException, InterruptedException {
    return Http.get(this.getContext().getSystem())
            .singleRequest(
                    HttpRequest.create()
                            .withUri(URI + ids)
                            .withMethod(HttpMethods.GET),
                    materializer)
            .thenCompose(response ->
                    Unmarshaller.entityToString().unmarshal(
                            response.entity(), this.getContext().dispatcher(), materializer))
            .toCompletableFuture();
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
/**
 * 发送基于ServerChan的微信推送
 *
 * @param message
 */
private void sendWeChatNotification(SendWeChatNotificationMessage message) throws UnsupportedEncodingException {
    String uri = getFinalServerChanUri(message.getScKey(), message.getTitle(), message.getDescription());
    HttpRequest httpRequest = HttpRequest.create()
                                         .withUri(uri);
    Http.get(system)
        .singleRequest(httpRequest, materializer);
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
private void launchActors(final Injector injector) {
    LOGGER.info().setMessage("Launching actors").log();

    // Retrieve the actor system
    final ActorSystem actorSystem = injector.getInstance(ActorSystem.class);

    // Create the status actor
    actorSystem.actorOf(Props.create(Status.class), "status");

    // Create the telemetry connection actor
    actorSystem.actorOf(Props.create(Telemetry.class, injector.getInstance(MetricsFactory.class)), "telemetry");

    // Load supplemental routes
    final ImmutableList.Builder<SupplementalRoutes> supplementalHttpRoutes = ImmutableList.builder();
    _configuration.getSupplementalHttpRoutesClass().ifPresent(clazz -> {
            supplementalHttpRoutes.add(injector.getInstance(clazz));
    });

    // Create and bind Http server
    final Materializer materializer = ActorMaterializer.create(actorSystem);
    final Routes routes = new Routes(
            actorSystem,
            injector.getInstance(PeriodicMetrics.class),
            _configuration.getHttpHealthCheckPath(),
            _configuration.getHttpStatusPath(),
            supplementalHttpRoutes.build());
    final Http http = Http.get(actorSystem);
    final akka.stream.javadsl.Source<IncomingConnection, CompletionStage<ServerBinding>> binding = http.bind(
            ConnectHttp.toHost(
                    _configuration.getHttpHost(),
                    _configuration.getHttpPort()),
            materializer);
    binding.to(
            akka.stream.javadsl.Sink.foreach(
                    connection -> connection.handleWith(routes.flow(), materializer)))
            .run(materializer);
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public AkkaHttpJavaClient(ActorSystem sys, Materializer mat) {
    system = sys;
    materializer = mat;
    settings = Settings.SettingsProvider.get(system);
    connectionFlow = Http.get(system).outgoingConnection(ConnectHttp.toHost(settings.HOST, settings.PORT));
    poolClientFlow = Http.get(system).cachedHostConnectionPool(ConnectHttp.toHost(settings.HOST, settings.PORT), materializer);
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
private HttpResponse send(HttpRequest request) {
    try {
        return Http.get(system)
            .singleRequest(request, materializer)
            .toCompletableFuture()
            .get(timeout, TimeUnit.MILLISECONDS);
    } catch (ExecutionException x) {
        throw (RuntimeException) x.getCause();
    } catch (InterruptedException | TimeoutException e) {
        throw new RuntimeException(e);
    }
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    ActorSystem system = ActorSystem.create("ServiceA");
    final Http http = Http.get(system);
    final ActorMaterializer materializer = ActorMaterializer.create(system);
    final ActorMain app = new ActorMain();
    ActorRef serviceBackendActor = null;

    switch (Integer.parseInt(args[0])) {
        case 0:
            serviceBackendActor = system.actorOf(Step0BackendActor.props(), "step0");
            break;
        case 1:
            serviceBackendActor = system.actorOf(Step1BackendActor.props(), "step1");
            break;
        case 2:
            serviceBackendActor = system.actorOf(Step2BackendActor.props(), "step2");
            break;
        case 3:
            serviceBackendActor = system.actorOf(Step3BackendActor.props(), "step3");
            break;
        default:
            throw new RuntimeException("Unknown example.");
    }

    //final ActorRef serviceBackendActor = system.actorOf(BackendActor.props(), "backendActor");
    final Flow<HttpRequest, HttpResponse, NotUsed> routeFlow = app.createRoute(serviceBackendActor).flow(system, materializer);

    final CompletionStage<ServerBinding> binding =
            http.bindAndHandle(
                    routeFlow,
                    ConnectHttp.toHost("localhost", 8080),
                    materializer);

    System.out.println("Server online at http://localhost:8080/\nPress RETURN to stop...");
    System.in.read(); // let it run until user presses return
    System.in.read(); // let it run until user presses return

    binding
            .thenCompose(ServerBinding::unbind) // trigger unbinding from the port
            .thenAccept(unbound -> system.terminate()); // and shutdown when done

}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public void close() {
    System.out.println("Shutting down client");
    Http.get(system).shutdownAllConnectionPools().whenComplete((s, f) -> system.terminate());
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public <U> CompletionStage<U> requestLevelFutureBased(Optional<String> s,
                                                      Function<HttpResponse, CompletionStage<U>> responseHandler) {
    return Http.get(system)
            .singleRequest(HttpRequest.create().withUri(getUri(s)), materializer)
            .thenComposeAsync(responseHandler);
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public CompletionStage<Void> start(Class<?> eventType, ActorRef shardRegion) {
    Config config = this.config.getConfig("ts-reaktive.replication");
    final String eventTag = getEventTag(eventType);
    
    synchronized(started) {
        if (started.containsKey(eventTag)) return started.get(eventTag);
        
        getEventClassifier(eventType); // will throw exception if the classifier is undefined, so we get an early error
        
        ActorMaterializer materializer = ActorMaterializer.create(system);
        Config tagConfig = config.hasPath(eventTag) ? config.getConfig(eventTag).withFallback(config) : config;
        EventEnvelopeSerializer serializer = new EventEnvelopeSerializer(system);
        
        Seq<DataCenter> remotes = Vector.ofAll(tagConfig.getConfig("remote-datacenters").root().entrySet()).map(e -> {
            String name = e.getKey();
            Config remote = ((ConfigObject) e.getValue()).toConfig();
            // FIXME add config options for ConnectionContext
            return new WebSocketDataCenterClient(system, ConnectionContext.noEncryption(), name, remote.getString("url"), serializer);
        });
        
        DataCenterRepository dataCenterRepository = new DataCenterRepository() {
            @Override
            protected Iterable<DataCenter> listRemotes() {
                return remotes;
            }
            
            @Override
            public String getLocalName() {
                return getLocalDataCenterName();
            }
        };
        
        VisibilityCassandraSession session = new VisibilityCassandraSession(system, "visibilitySession");
        VisibilityRepository visibilityRepo = new VisibilityRepository(session);
        ReadJournal journal = PersistenceQuery.get(system).getReadJournalFor(ReadJournal.class, config.getString("read-journal-plugin-id"));
        
        DataCenterForwarder.startAll(system, materializer, dataCenterRepository, visibilityRepo, eventType,
            (EventsByTagQuery)journal, (CurrentEventsByPersistenceIdQuery) journal);
        
        WebSocketDataCenterServer server = new WebSocketDataCenterServer(config.getConfig("server"), shardRegion);
        final int port = tagConfig.getInt("local-datacenter.port");
        final String host = tagConfig.getString("local-datacenter.host");
        log.debug("Binding to {}:{}", host, port);
        
        CompletionStage<ServerBinding> bind = Http.get(system).bindAndHandle(server.route().flow(system, materializer),
            ConnectHttp.toHost(host, port), materializer);
        
        // The returned future completes when both the HTTP binding is ready, and the cassandra visibility session has initialized.
        started.put(eventTag, bind.thenCompose(binding -> session.getUnderlying().thenApply(s -> binding)).thenApply(b -> {
            log.info("Listening on {}", b.localAddress());
            return null;
        }));
        return started.get(eventTag);
    }
}
 

import akka.http.javadsl.Http; //导入依赖的package包/类
public HttpFlow(Http http, Materializer materializer) {
    this.http = http;
    this.materializer = materializer;
}