Java 类io.grpc.Server 实例源码

@Override
public Lifecycle newInstance(KernelContext kernelContext, final Dependencies dependencies) throws Throwable {
    return new LifecycleAdapter() {

        private Server server;

        @Override
        public void start() throws Throwable {
            server  = ServerBuilder.forPort(9999).addService(new Neo4jGRPCService(dependencies.getGraphDatabaseService())).build();
            server.start();
            System.out.println("Started gRPC Server.");
        }

        @Override
        public void shutdown() throws Throwable {
            server.shutdown();
        }
    };
}

/**
 * Start Netty Grpc Server.
 *
 * @return Server gRPC Server
 * @throws IOException - when something went wrong starting the grpc server
 */
final Server start() throws IOException {
  final int port = 8080;

  log.info("Starting grpc server on port '{}'...", port);
  final Server server =
      NettyServerBuilder
          .forPort(port)
          .addService(productReadService)
          .addService(productUpdateService)
          .addService(ServerInterceptors.intercept(echoService, serviceInterceptor))
          .build();

  server.start();
  log.info("grpc (port={}) server started successfully.", port);

  return server;
}

@Test
public void stopSendingWhenClusterIsDown() throws Exception {
  servers.values().forEach(Server::shutdownNow);
  messageSender.onConnected();

  Thread thread = new Thread(() -> messageSender.send(event));
  thread.start();

  // we don't want to keep sending on cluster down
  await().atMost(2, SECONDS).until(() -> thread.isAlive() && thread.getState().equals(WAITING));

  assertThat(eventsMap.get(8080).isEmpty(), is(true));
  assertThat(eventsMap.get(8090).isEmpty(), is(true));

  startServerOnPort(8080);
  startServerOnPort(8090);

  await().atMost(2, SECONDS).until(() -> connected.get(8080).size() == 2 || connected.get(8090).size() == 2);
  await().atMost(2, SECONDS).until(() -> eventsMap.get(8080).size() == 1 || eventsMap.get(8090).size() == 1);
}

/**
 * Shutdown the gRPC {@link Server} when this object is closed.
 */
@Override
public void close() throws Exception {
    final Server server = server();

    if (server != null) {
        server.shutdown();

        try {
            // TODO: Maybe we should catch the InterruptedException from this?
            server.awaitTermination(shutdownWaitTimeInMillis, TimeUnit.MILLISECONDS);
        } finally {
            server.shutdownNow();

            this.server = null;
        }
    }
}

@Test
public void startDoesNotStartServerWithoutServices() throws Exception {
    final int port = ThreadLocalRandom.current().nextInt(1000, 10000);
    final long shutdownWaitTimeInMillis = ThreadLocalRandom.current().nextLong(1000, 10000);

    final ApplicationContext applicationContext = mock(ApplicationContext.class);
    final Server server = mock(Server.class, new TriesToReturnSelf());
    final GrpcServerFactory factory = mock(GrpcServerFactory.class);

    when(server.getPort()).thenReturn(port);

    // Configure application context to contain no gRPC services.
    when(applicationContext.getBeansWithAnnotation(eq(GrpcService.class))).thenReturn(ImmutableMap.of());

    GrpcServerHost runner = new GrpcServerHost(port, shutdownWaitTimeInMillis, factory);
    runner.setApplicationContext(applicationContext);

    assertThatThrownBy(runner::start).isInstanceOf(IOException.class);

    // Make sure the server builder was not used.
    verify(factory, never()).buildServerForServices(anyInt(), any());

    assertThat(runner.server()).isNull();
}

@Test
public void getPortReturnsServerPortForRunningServer() throws Exception {
    final int configPort = ThreadLocalRandom.current().nextInt(1000, 2000);
    final int serverPort = ThreadLocalRandom.current().nextInt(2000, 3000);
    final int serviceCount = ThreadLocalRandom.current().nextInt(5, 10);
    final long shutdownWaitTimeInMillis = ThreadLocalRandom.current().nextLong(1000, 10000);

    final ApplicationContext applicationContext = mock(ApplicationContext.class);
    final Server server = mock(Server.class, new TriesToReturnSelf());
    final GrpcServerFactory factory = (p, s) -> server;

    final Map<String, Object> services = IntStream.range(0, serviceCount)
            .mapToObj(i -> mock(BindableService.class))
            .collect(Collectors.toMap(s -> UUID.randomUUID().toString(), s -> s));

    when(applicationContext.getBeansWithAnnotation(eq(GrpcService.class))).thenReturn(services);

    when(server.getPort()).thenReturn(serverPort);

    GrpcServerHost runner = new GrpcServerHost(configPort, shutdownWaitTimeInMillis, factory);
    runner.setApplicationContext(applicationContext);

    runner.start();

    assertThat(runner.getPort()).isEqualTo(serverPort);
}

/**
 * Attempt to {@link Server#shutdown()} the {@link Server} gracefully. If the max wait time is exceeded, give up and
 * perform a hard {@link Server#shutdownNow()}.
 *
 * @param server the server to be shutdown
 * @param timeout the max amount of time to wait for graceful shutdown to occur
 * @param unit the time unit denominating the shutdown timeout
 * @return the given server
 * @throws InterruptedException if waiting for termination is interrupted
 */
public static Server shutdownGracefully(Server server, long timeout, TimeUnit unit) throws InterruptedException {
    Preconditions.checkNotNull(server, "server");
    Preconditions.checkArgument(timeout > 0, "timeout must be greater than 0");
    Preconditions.checkNotNull(unit, "unit");

    server.shutdown();

    try {
        server.awaitTermination(timeout, unit);
    } finally {
        server.shutdownNow();
    }

    return server;
}

@Test
public void shutdownGracefullyPropagatesAwaitInterrupt() throws Exception {
    final Server server = mock(Server.class);
    final long maxWaitTimeInMillis = ThreadLocalRandom.current().nextLong(1, 10000);

    final InterruptedException interruptedException = new InterruptedException();

    when(server.awaitTermination(anyLong(), any())).thenThrow(interruptedException);

    assertThatThrownBy(() -> Servers.shutdownGracefully(server, maxWaitTimeInMillis))
            .isSameAs(interruptedException);

    InOrder inOrder = Mockito.inOrder(server);

    inOrder.verify(server).shutdown();
    inOrder.verify(server).awaitTermination(eq(maxWaitTimeInMillis), eq(TimeUnit.MILLISECONDS));
    inOrder.verify(server).shutdownNow();
}

public static void main(String[] args) throws InterruptedException {

    Server server=ServerBuilder.forPort(Config.getLocalPortProvider())
                    .addService(new AddService())
                    .build();
    try {
        server.start();

        ProviderBootstrap.init();

        server.awaitTermination();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

}

/**
 * Tries a few times to start a server. If this returns, the server has been started. Throws
 * {@link RuntimeException} on failure.
 */
public static TestServer createAndStart(Optional<SslContext> sslContext) {
  RecordingTestService recordingTestService = new RecordingTestService(
      UNARY_SERVER_RESPONSE,
      STREAMING_SERVER_RESPONSE,
      CLIENT_STREAMING_SERVER_RESPONSE,
      BIDI_SERVER_RESPONSE);

  Random random = new Random();
  for (int i = 0; i < NUM_SERVER_START_TRIES; ++i) {
    int port = random.nextInt(MAX_SERVER_PORT - MIN_SERVER_PORT + 1) + MIN_SERVER_PORT;
    try {
      Server server = tryStartServer(port, recordingTestService, sslContext);

      // If we got this far, we have successfully started the server.
      return new TestServer(server, port, recordingTestService);
    } catch (IOException e) {
      // The random port might have been in use, try again...
      continue;
    }
  }

  // If we got to here, we didn't manage to start a server.
  throw new RuntimeException("Unable to start server after " + NUM_SERVER_START_TRIES + " tries");
}

public static void main(String[] args) throws IOException, InterruptedException {
  RxMetricsServiceGrpc.MetricsServiceImplBase service = new RxMetricsServiceGrpc.MetricsServiceImplBase() {

    @Override
    public Single<Streaming.Average> collect(Flowable<Streaming.Metric> request) {
      return request.map(m -> m.getMetric())
          .map(m -> new State(m, 1))
          .reduce((a, b) -> new State(a.sum + b.sum, a.count + b.count))
          .map(s -> Streaming.Average.newBuilder().setVal(s.sum / s.count).build())
          .toSingle();
    }
  };

  Server server = ServerBuilder.forPort(8080)
      .addService(service)
      .build();

  server.start();
  server.awaitTermination();
}

protected void createAndStartGrpcServer() throws IOException {
    Server localServer = this.server;
    if (localServer == null) {
        this.server = factory.createServer();
        this.server.start();
        logger.info("gRPC Server started, listening on address: "
                + this.factory.getAddress() + ", port: " + this.factory.getPort());

        Thread awaitThread = new Thread(
                "container-" + (serverCounter.incrementAndGet())) {

            @Override
            public void run() {
                try {
                    GrpcServerLifecycle.this.server.awaitTermination();
                }
                catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }

        };
        awaitThread.setDaemon(false);
        awaitThread.start();
    }
}

/** Equivalent of "main", but non-static. */
public void run(String[] args) throws Exception {

    final Server server = newServer();

    server.start();

    Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
            try {
                System.out.println("Server shutting down");
                server.shutdown();
                server.awaitTermination(5, TimeUnit.SECONDS);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    });
}

private AsyncFuture<Void> start() throws IOException {
    final Server server = NettyServerBuilder
        .forAddress(address)
        .addService(bindService())
        .maxMessageSize(maxFrameSize)
        .bossEventLoopGroup(bossGroup)
        .workerEventLoopGroup(workerGroup)
        .build();

    return async.call(() -> {
        server.start();
        this.server.set(server);
        return null;
    }).directTransform(v -> {
        final InetSocketAddress localAddress = extractInetSocketAddress(server);
        bindFuture.resolve(localAddress);
        return null;
    });
}

private static Server startServer(String bindAddress, int port, boolean https, File centralDir,
        DownstreamServiceImpl downstreamService, CollectorServiceImpl collectorService)
        throws IOException {
    NettyServerBuilder builder =
            NettyServerBuilder.forAddress(new InetSocketAddress(bindAddress, port));
    if (https) {
        builder.useTransportSecurity(
                getHttpsConfFile(centralDir, "grpc-cert.pem", "cert.pem", "certificate"),
                getHttpsConfFile(centralDir, "grpc-key.pem", "key.pem", "private key"));
    }
    return builder.addService(collectorService.bindService())
            .addService(downstreamService.bindService())
            // need to override default max message size of 4mb until streaming is implemented
            // for DownstreamService.EntriesResponse and FullTraceResponse
            .maxMessageSize(64 * 1024 * 1024)
            .build()
            .start();
}

public Server startServer() throws IOException {
  ServerInterceptor headersInterceptor = new TracingMetadataUtils.ServerHeadersInterceptor();
  NettyServerBuilder b =
      NettyServerBuilder.forPort(workerOptions.listenPort)
          .addService(ServerInterceptors.intercept(actionCacheServer, headersInterceptor))
          .addService(ServerInterceptors.intercept(bsServer, headersInterceptor))
          .addService(ServerInterceptors.intercept(casServer, headersInterceptor));

  if (execServer != null) {
    b.addService(ServerInterceptors.intercept(execServer, headersInterceptor));
    b.addService(ServerInterceptors.intercept(watchServer, headersInterceptor));
  } else {
    logger.info("Execution disabled, only serving cache requests.");
  }

  Server server = b.build();
  logger.log(INFO, "Starting gRPC server on port {0,number,#}.", workerOptions.listenPort);
  server.start();

  return server;
}

void run() throws Exception {
  // Port 0 means that the operating system will pick an available port to use.
  Server server = ServerBuilder.forPort(0).addService(new GreeterGrpc.GreeterImplBase() {
    @Override
    public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
      responseObserver.onError(Status.INTERNAL
          .withDescription("Eggplant Xerxes Crybaby Overbite Narwhal").asRuntimeException());
    }
  }).build().start();
  channel =
      ManagedChannelBuilder.forAddress("localhost", server.getPort()).usePlaintext(true).build();

  blockingCall();
  futureCallDirect();
  futureCallCallback();
  asyncCall();
  advancedAsyncCall();

  channel.shutdown();
  server.shutdown();
  channel.awaitTermination(1, TimeUnit.SECONDS);
  server.awaitTermination();
}

void run() throws Exception {
  Server server = ServerBuilder.forPort(0).addService(new GreeterGrpc.GreeterImplBase() {
    @Override
    public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
      Metadata trailers = new Metadata();
      trailers.put(DEBUG_INFO_TRAILER_KEY, DEBUG_INFO);
      responseObserver.onError(Status.INTERNAL.withDescription(DEBUG_DESC)
          .asRuntimeException(trailers));
    }
  }).build().start();
  channel =
      ManagedChannelBuilder.forAddress("localhost", server.getPort()).usePlaintext(true).build();

  blockingCall();
  futureCallDirect();
  futureCallCallback();
  asyncCall();
  advancedAsyncCall();

  channel.shutdown();
  server.shutdown();
  channel.awaitTermination(1, TimeUnit.SECONDS);
  server.awaitTermination();
}

/**
 * Creates and starts a new {@link TestServiceImpl} server.
 */
private Server newServer() throws CertificateException, IOException {
  File serverCertChainFile = TestUtils.loadCert("server1.pem");
  File serverPrivateKeyFile = TestUtils.loadCert("server1.key");
  X509Certificate[] serverTrustedCaCerts = {
    TestUtils.loadX509Cert("ca.pem")
  };

  SslContext sslContext =
      GrpcSslContexts.forServer(serverCertChainFile, serverPrivateKeyFile)
                     .trustManager(serverTrustedCaCerts)
                     .clientAuth(ClientAuth.REQUIRE)
                     .build();

  return NettyServerBuilder.forPort(0)
      .sslContext(sslContext)
      .addService(new TestServiceImpl(serverExecutor))
      .build()
      .start();
}

public static void main(String[] args) throws InterruptedException, IOException {

        // Build server
        Server server = ServerBuilder.forPort(SERVER_PORT)
                .addService(new ChatServiceImpl())
                .build();

        // Start server
        System.out.println("Starting server on port " + SERVER_PORT);
        server.start();

        // Keep it running
        System.out.println("Server started!");
        server.awaitTermination();
    }

public static void main(String[] args) throws IOException, InterruptedException {

        // Build server
        Server server = ServerBuilder.forPort(SERVER_PORT)
                .addService(new GreetingServiceImpl())
                .build();

        // Start server
        System.out.println("Starting GreeterServer on port " + SERVER_PORT);
        server.start();

        // Keep it running
        System.out.println("GreeterServer started!");
        server.awaitTermination();
    }

private static void startServerOnPort(int port) {
  ServerBuilder<?> serverBuilder = ServerBuilder.forPort(port);
  serverBuilder.addService(new MyTxEventService(connected.get(port), eventsMap.get(port), delays.get(port)));
  Server server = serverBuilder.build();

  try {
    server.start();
    servers.put(port, server);
  } catch (IOException e) {
    fail(e.getMessage());
  }
}

private int killServerReceivedMessage() {
  for (int port : eventsMap.keySet()) {
    if (!eventsMap.get(port).isEmpty()) {
      Server serverToKill = servers.get(port);
      serverToKill.shutdownNow();
      return port;
    }
  }
  throw new IllegalStateException("None of the servers received any message");
}

@Override
public Server buildServerForServices(int port, Collection<BindableService> services) {
    ServerBuilder builder = ServerBuilder.forPort(port);
    setupServer(builder);
    services.forEach(service -> registerService(builder, service));
    return builder.build();
}

/**
 * Returns the actual port of the running gRPC server. The port is only available once the server is up and running.
 *
 * @throws IllegalStateException if called either before the server has started or after it has stopped
 */
public final int getPort() {
    final Server server = server();

    if (server == null) {
        throw new IllegalStateException("Cannot fetch port until server has started.");
    }

    return server.getPort();
}

@Test
public void startStartsServerWithServices() throws Exception {
    final int port = ThreadLocalRandom.current().nextInt(1000, 10000);
    final int serviceCount = ThreadLocalRandom.current().nextInt(5, 10);
    final long shutdownWaitTimeInMillis = ThreadLocalRandom.current().nextLong(1000, 10000);

    final ApplicationContext applicationContext = mock(ApplicationContext.class);
    final Server server = mock(Server.class, new TriesToReturnSelf());
    when(server.getPort()).thenReturn(port);

    final Map<String, Object> services = IntStream.range(0, serviceCount)
            .mapToObj(i -> mock(BindableService.class))
            .collect(Collectors.toMap(s -> UUID.randomUUID().toString(), s -> s));

    AtomicBoolean built = new AtomicBoolean(false);

    GrpcServerFactory fakeFactory = (p, s) -> {
        built.set(true);
        assertThat(p).isEqualTo(port);
        s.forEach(ss -> assertThat(services.values().contains(ss)).isTrue());
        return server;
    };

    when(applicationContext.getBeansWithAnnotation(eq(GrpcService.class))).thenReturn(services);

    GrpcServerHost runner = new GrpcServerHost(port, shutdownWaitTimeInMillis, fakeFactory);
    runner.setApplicationContext(applicationContext);

    runner.start();
    assertThat(built.get()).isTrue();

    verify(server).start();

    assertThat(runner.server()).isEqualTo(server);
}

/**
 * Attempt to {@link Server#shutdown()} the {@link Server} gracefully when the JVM terminates. If the max wait time
 * is exceeded, give up and perform a hard {@link Server#shutdownNow()}.
 *
 * @param server the server to be shutdown
 * @param maxWaitTimeInMillis the max amount of time to wait for graceful shutdown to occur
 * @return the given server
 */
public static Server shutdownWithJvm(Server server, long maxWaitTimeInMillis) {
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        try {
            shutdownGracefully(server, maxWaitTimeInMillis);
        } catch (InterruptedException ex) {
            // do nothing
        }
    }));

    return server;
}

@Test
public void uniqueSessionIdPerChannel() throws Exception {
    GreeterGrpc.GreeterImplBase svc = new GreeterGrpc.GreeterImplBase() {
        @Override
        public void sayHello(HelloRequest request, StreamObserver<HelloResponse> responseObserver) {
            responseObserver.onNext(HelloResponse.newBuilder().setMessage(SessionIdServerInterceptor.SESSION_ID.get().toString()).build());
            responseObserver.onCompleted();
        }
    };

    Server server = InProcessServerBuilder.forName("uniqueSessionIdPerChannel")
            .addTransportFilter(new ClientSessionTransportFilter())
            .intercept(new SessionIdServerInterceptor())
            .addService(svc)
            .build()
            .start();

    ManagedChannel channel1 = InProcessChannelBuilder.forName("uniqueSessionIdPerChannel")
            .usePlaintext(true)
            .build();
    GreeterGrpc.GreeterBlockingStub stub1 = GreeterGrpc.newBlockingStub(channel1);

    ManagedChannel channel2 = InProcessChannelBuilder.forName("uniqueSessionIdPerChannel")
            .usePlaintext(true)
            .build();
    GreeterGrpc.GreeterBlockingStub stub2 = GreeterGrpc.newBlockingStub(channel2);

    try {
        String sessionId1 = stub1.sayHello(HelloRequest.getDefaultInstance()).getMessage();
        String sessionId2 = stub2.sayHello(HelloRequest.getDefaultInstance()).getMessage();

        assertThat(sessionId1).isNotEqualTo(sessionId2);
    } finally {
        channel1.shutdown();
        channel2.shutdown();
        server.shutdown();
    }
}

@Test
public void interceptorThrowsIfMissingTransportFilter() throws Exception {
    GreeterGrpc.GreeterImplBase svc = new GreeterGrpc.GreeterImplBase() {
        @Override
        public void sayHello(HelloRequest request, StreamObserver<HelloResponse> responseObserver) {
            responseObserver.onNext(HelloResponse.newBuilder().setMessage(SessionIdServerInterceptor.SESSION_ID.get().toString()).build());
            responseObserver.onCompleted();
        }
    };

    Server server = InProcessServerBuilder.forName("interceptorThrowsIfMissingTransportFilter")
            .intercept(new SessionIdServerInterceptor())
            .addService(svc)
            .build()
            .start();

    ManagedChannel channel = InProcessChannelBuilder.forName("interceptorThrowsIfMissingTransportFilter")
            .usePlaintext(true)
            .build();
    GreeterGrpc.GreeterBlockingStub stub = GreeterGrpc.newBlockingStub(channel);

    try {
        assertThatThrownBy(() -> stub.sayHello(HelloRequest.getDefaultInstance()).getMessage()).isInstanceOf(StatusRuntimeException.class);

    } finally {
        channel.shutdown();
        server.shutdown();
    }
}

@Test
public void shutdownGracefullyThrowsIfMaxWaitTimeInMillisIsZero() {
    final Server server = mock(Server.class);

    assertThatThrownBy(() -> Servers.shutdownGracefully(server, 0))
            .isInstanceOf(IllegalArgumentException.class)
            .hasMessageContaining("timeout must be greater than 0");
}

@Test
public void shutdownGracefullyThrowsIfMaxWaitTimeInMillisIsLessThanZero() {
    final long maxWaitTimeInMillis = ThreadLocalRandom.current().nextLong(Long.MIN_VALUE, 0);
    final Server server = mock(Server.class);

    assertThatThrownBy(() -> Servers.shutdownGracefully(server, maxWaitTimeInMillis))
            .isInstanceOf(IllegalArgumentException.class)
            .hasMessageContaining("timeout must be greater than 0");
}

@Test
public void shutdownGracefullyCausesServerToProperlyShutdown() throws Exception {
    final Server server = mock(Server.class);
    final long maxWaitTimeInMillis = ThreadLocalRandom.current().nextLong(1, 10000);

    // Also assert that chaining works as expected.
    assertThat(Servers.shutdownGracefully(server, maxWaitTimeInMillis)).isSameAs(server);

    InOrder inOrder = Mockito.inOrder(server);

    inOrder.verify(server).shutdown();
    inOrder.verify(server).awaitTermination(eq(maxWaitTimeInMillis), eq(TimeUnit.MILLISECONDS));
    inOrder.verify(server).shutdownNow();
}

@Test
public void perSessionShouldFailMissingTransportFilter() throws Exception {
    class TestService extends GreeterGrpc.GreeterImplBase {
        @Override
        public void sayHello(HelloRequest request, StreamObserver<HelloResponse> responseObserver) {
            responseObserver.onNext(HelloResponse.newBuilder().setMessage(Integer.toString(System.identityHashCode(this))).build());
            responseObserver.onCompleted();
        }
    }

    ClientSessionTransportFilter tf = new ClientSessionTransportFilter();
    Server server = InProcessServerBuilder.forName("perSessionShouldInstantiateOneInstancePerSession")
            .addService(new PerSessionService<>(() -> new TestService(), tf))
            .build()
            .start();

    ManagedChannel channel = InProcessChannelBuilder.forName("perSessionShouldInstantiateOneInstancePerSession")
            .usePlaintext(true)
            .build();
    GreeterGrpc.GreeterBlockingStub stub = GreeterGrpc.newBlockingStub(channel);

    try {
        assertThatThrownBy(() -> stub.sayHello(HelloRequest.getDefaultInstance()).getMessage()).isInstanceOf(StatusRuntimeException.class);
    } finally {
        channel.shutdown();
        server.shutdown();
    }
}

@Test
public void serverRunsAndRespondsCorrectly() throws ExecutionException,
        IOException,
        InterruptedException,
        TimeoutException {
    final String name = UUID.randomUUID().toString();

    Server server = ServerBuilder.forPort(9999)
            .addService(new GreeterImpl())
            .build();

    server.start();

    ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", server.getPort())
            .usePlaintext(true)
            .build();

    GreeterGrpc8.GreeterCompletableFutureStub stub = GreeterGrpc8.newCompletableFutureStub(channel);

    CompletableFuture<HelloResponse> response = stub.sayHello(HelloRequest.newBuilder().setName(name).build());

    await().atMost(3, TimeUnit.SECONDS).until(() -> response.isDone() && response.get().getMessage().contains(name));

    channel.shutdown();
    channel.awaitTermination(1, TimeUnit.MINUTES);
    channel.shutdownNow();

    server.shutdown();
    server.awaitTermination(1, TimeUnit.MINUTES);
    server.shutdownNow();
}

/**
 * Start a grpc service
 * @throws IOException
 */
default void start() throws IOException {
    Server server = NettyServerBuilder
            .forPort(getPort())
            .addService(getServiceDefinition())
            .build()
            .start();

    setServer(server);
}

public GrpcClusterBuilder withGrpcServer(Server grpcServer) {
    if (grpcServerFactory != null) {
        throw new IllegalArgumentException("grpcServerFactory already set");
    }
    this.grpcServer = grpcServer;
    return this;
}

@Override
public Server create(ClusterConfig config) {
    return ServerBuilder
            //TODO
            .forPort(1337)
            .build();
}

/** Starts a grpc server on the given port, throws {@link IOException} on failure. */
private static Server tryStartServer(
    int port,
    TestServiceImplBase testService,
    Optional<SslContext> sslContext) throws IOException {
  NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(port)
      .addService(testService)
      .addService(ProtoReflectionService.newInstance());
  if (sslContext.isPresent()) {
    serverBuilder.sslContext(sslContext.get());
  }
  return serverBuilder.build().start();
}

static public void main(String [] args) throws IOException, InterruptedException {
  Server server = ServerBuilder.forPort(8080)
      .addService(new GreetingServiceImpl()).build();

  System.out.println("Starting server...");
  server.start();
  System.out.println("Server started!");
  server.awaitTermination();
}

static public void main(String[] args) throws IOException, InterruptedException {

    Server server = ServerBuilder.forPort(8080)
        .addService(new EchoServiceImpl()).build();

    System.out.println("Starting server...");
    server.start();
    System.out.println("Server started!");
    server.awaitTermination();
  }