Java 类io.grpc.netty.NettyServerBuilder 实例源码

/**
 * 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;
}

public PluginGrpcServer(int port) {
    this.pluginConnections = new HashMap<>();

    PlayerEvents playerEvents = new PlayerEvents(this);

    this.server = NettyServerBuilder.forPort(port)
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)

            .addService(playerEvents)

            .directExecutor()
            .channelType(EpollServerSocketChannel.class)
            .bossEventLoopGroup(new EpollEventLoopGroup())
            .workerEventLoopGroup(new EpollEventLoopGroup())
            .build();

    // demoPluginConnections();
}

/**
 * Before the test has started, create the server and channel.
 */
@Override
protected void before() throws Throwable {
    serviceRegistry = new MutableHandlerRegistry();

    NettyServerBuilder serverBuilder = NettyServerBuilder
            .forPort(0)
            .fallbackHandlerRegistry(serviceRegistry);

    if (useDirectExecutor) {
        serverBuilder.directExecutor();
    }

    configureServerBuilder.accept(serverBuilder);
    server = serverBuilder.build().start();
    port = server.getPort();

    NettyChannelBuilder channelBuilder = NettyChannelBuilder.forAddress("localhost", port).usePlaintext(true);
    configureChannelBuilder.accept(channelBuilder);
    channel = channelBuilder.build();
}

@Activate
protected void activate(ComponentContext context) throws IOException {
    modified(context);

    log.debug("Server starting on {}", listenPort);
    try {
        server  = NettyServerBuilder.forPort(listenPort)
                .addService(DeviceProviderRegistryRpcGrpc.bindService(new DeviceProviderRegistryServerProxy()))
                .addService(LinkProviderServiceRpcGrpc.bindService(new LinkProviderServiceServerProxy(this)))
                .build().start();
    } catch (IOException e) {
        log.error("Failed to start gRPC server", e);
        throw e;
    }

    log.info("Started on {}", listenPort);
}

public static void main(String[] args) throws Exception {
    NettyServerBuilder.forAddress(LocalAddress.ANY).forPort(19876)
        .maxConcurrentCallsPerConnection(12).maxMessageSize(16777216)
        .addService(new MockApplicationRegisterService())
        .addService(new MockInstanceDiscoveryService())
        .addService(new MockJVMMetricsService())
        .addService(new MockServiceNameDiscoveryService())
        .addService(new MockTraceSegmentService()).build().start();

    Server jettyServer = new Server(new InetSocketAddress("0.0.0.0",
        Integer.valueOf(12800)));
    String contextPath = "/";
    ServletContextHandler servletContextHandler = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
    servletContextHandler.setContextPath(contextPath);
    servletContextHandler.addServlet(GrpcAddressHttpService.class, GrpcAddressHttpService.SERVLET_PATH);
    servletContextHandler.addServlet(ReceiveDataService.class, ReceiveDataService.SERVLET_PATH);
    servletContextHandler.addServlet(ClearReceiveDataService.class, ClearReceiveDataService.SERVLET_PATH);
    jettyServer.setHandler(servletContextHandler);
    jettyServer.start();
}

@Before
public void setUp() throws IOException {
  serviceRegistry.addService(new MaintenanceImplBase() {
     @Override
     public void snapshot(SnapshotRequest request, StreamObserver<SnapshotResponse> observer) {
       try {
         observerQueue.put(observer);
       } catch (InterruptedException e) {
         throw new RuntimeException(e);
       }
     }
   }
  );

  fakeServer = NettyServerBuilder.forPort(TestUtil.findNextAvailablePort())
      .fallbackHandlerRegistry(serviceRegistry)
      .directExecutor()
      .build()
      .start();

  client = Client.builder().endpoints("http://127.0.0.1:" + fakeServer.getPort()).build();
  maintenance = client.getMaintenanceClient();
}

@Override
public void start() throws Exception {
    LOG.info("Configuring server on port " + port);
    server = NettyServerBuilder.forPort(port)
        .addService(chatHandler)
        .build();
    LOG.info("Starting server");
    server.start();
    Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
            ChatServerImpl.this.stop();
        }
    });
    blockUntilShutdown();
}

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();
}

GrpcServerWrapper(TraceCollector collector, int port) throws IOException {
    bossEventLoopGroup = EventLoopGroups.create("Glowroot-IT-Harness-GRPC-Boss-ELG");
    workerEventLoopGroup = EventLoopGroups.create("Glowroot-IT-Harness-GRPC-Worker-ELG");
    executor = Executors.newCachedThreadPool(
            new ThreadFactoryBuilder()
                    .setDaemon(true)
                    .setNameFormat("Glowroot-IT-Harness-GRPC-Executor-%d")
                    .build());
    downstreamService = new DownstreamServiceImpl();
    server = NettyServerBuilder.forPort(port)
            .bossEventLoopGroup(bossEventLoopGroup)
            .workerEventLoopGroup(workerEventLoopGroup)
            .executor(executor)
            .addService(new CollectorServiceImpl(collector).bindService())
            .addService(downstreamService.bindService())
            .maxMessageSize(1024 * 1024 * 100)
            .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;
}

@Activate
protected void activate(ComponentContext context) throws IOException {
    executor = newScheduledThreadPool(1, Tools.groupedThreads("grpc", "%d", log));
    modified(context);

    log.debug("Server starting on {}", listenPort);
    try {
        server  = NettyServerBuilder.forPort(listenPort)
                .addService(new DeviceProviderRegistryServerProxy())
                .addService(new LinkProviderServiceServerProxy(this))
                .build().start();
    } catch (IOException e) {
        log.error("Failed to start gRPC server", e);
        throw e;
    }

    log.info("Started on {}", listenPort);
}

@VisibleForTesting
void start() throws Exception {
  executor = Executors.newSingleThreadScheduledExecutor();
  SslContext sslContext = null;
  if (useTls) {
    sslContext = GrpcSslContexts.forServer(
            TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key")).build();
  }
  server = NettyServerBuilder.forPort(port)
      .sslContext(sslContext)
      .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .addService(ServerInterceptors.intercept(
          new TestServiceImpl(executor),
          TestServiceImpl.interceptors()))
      .build().start();
}

/**
 * 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();
}

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  // Starts the server with HTTPS.
  try {
    SslProvider sslProvider = SslContext.defaultServerProvider();
    if (sslProvider == SslProvider.OPENSSL && !OpenSsl.isAlpnSupported()) {
      // OkHttp only supports Jetty ALPN on OpenJDK. So if OpenSSL doesn't support ALPN, then we
      // are forced to use Jetty ALPN for Netty instead of OpenSSL.
      sslProvider = SslProvider.JDK;
    }
    SslContextBuilder contextBuilder = SslContextBuilder
        .forServer(TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"));
    GrpcSslContexts.configure(contextBuilder, sslProvider);
    contextBuilder.ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE);
    return NettyServerBuilder.forPort(0)
        .flowControlWindow(65 * 1024)
        .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(contextBuilder.build());
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }
}

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  return NettyServerBuilder.forPort(0)
      .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .compressorRegistry(compressors)
      .decompressorRegistry(decompressors)
      .intercept(new ServerInterceptor() {
          @Override
          public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
              Metadata headers, ServerCallHandler<ReqT, RespT> next) {
            Listener<ReqT> listener = next.startCall(call, headers);
            // TODO(carl-mastrangelo): check that encoding was set.
            call.setMessageCompression(true);
            return listener;
          }
        });
}

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  // Starts the server with HTTPS.
  try {
    return NettyServerBuilder.forPort(0)
        .flowControlWindow(65 * 1024)
        .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
        .sslContext(GrpcSslContexts
            .forServer(TestUtils.loadCert("server1.pem"), TestUtils.loadCert("server1.key"))
            .clientAuth(ClientAuth.REQUIRE)
            .trustManager(TestUtils.loadCert("ca.pem"))
            .ciphers(TestUtils.preferredTestCiphers(), SupportedCipherSuiteFilter.INSTANCE)
            .sslProvider(SslProvider.OPENSSL)
            .build());
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }
}

private void start() throws IOException {
    server = NettyServerBuilder.forPort(port)
            .addService(ServerInterceptors.intercept(
                    new SimpleServiceImpl()))
            .addService(ServerInterceptors.intercept(
                    new LessSimpleServiceImpl("localhost", port))).build().start();
}

@EventListener(ContextStartedEvent.class)
public void start() throws Exception {
    int port = properties.getPort();
    NettyServerBuilder builder = NettyServerBuilder.forPort(port);
    bindableServices.forEach(builder::addService);
    if (properties.isSecure()) {
        configureSsl(builder);
    }
    grpcServer = builder.build();

    try {
        grpcServer.start();
    } catch (IOException e) {
        if (e.getCause() instanceof BindException) {
            throw new BindException("Address already in use: " + port);
        }
        throw new RuntimeException(e);
    }

    if (properties.getHealthCheckPort() > 0) {
        log.info("Starting health check at port {}", properties.getHealthCheckPort());
        healthCheckSocket = new ServerSocket(properties.getHealthCheckPort());
        healthChecker = new Thread(this::healthCheck, "Health Checker");
        healthChecker.setDaemon(true);
        healthChecker.start();
    }

    log.info("Server started, GRPC API listening on {}", grpcServer.getPort());

    String endpointUrl = properties.getEndpointUrl();
    if (endpointUrl == null) {
        log.warn("No endpoint url provided");
    } else {
        for (Address address : ethereumConfig.getAddresses()) {
            ContractsManager contractManager = factory.getContractManager(address);
            EndpointRegistry registry = new EndpointRegistry(contractManager.endpointRegistry());
            registry.registerEndpoint(address, endpointUrl);
        }
    }
}

private void configureSsl(NettyServerBuilder builder) throws NoSuchAlgorithmException, CertificateEncodingException, NoSuchProviderException, InvalidKeyException, SignatureException, SSLException {
    NativeLibraryLoader.loadFirstAvailable(ClassLoader.getSystemClassLoader(),
        "netty_tcnative_osx_x86_64",
        "netty_tcnative_linux_x86_64",
        "netty_tcnative_windows_x86_64"
    );
    ECKeyPair ecKeyPair = ethereumConfig.getMainCredentials().getEcKeyPair();
    KeyPair keyPair = CryptoUtil.decodeKeyPair(ecKeyPair);
    SslContextBuilder contextBuilder = SslContextBuilder.forServer(
        keyPair.getPrivate(),
        CryptoUtil.genCert(keyPair)
    );

    builder.sslContext(GrpcSslContexts.configure(contextBuilder).build());
}

private HelloWorldServer(int port, TlsContext tls) throws SSLException {
  this.stats = new StatsTracerFactory();
  this.bossEventLoopGroup = Netty.newBossEventLoopGroup();
  this.workerEventLoopGroup = Netty.newWorkerEventLoopGroup();
  this.server =
      NettyServerBuilder.forPort(port)
          .bossEventLoopGroup(bossEventLoopGroup)
          .workerEventLoopGroup(workerEventLoopGroup)
          .channelType(Netty.serverChannelType())
          .addStreamTracerFactory(stats)
          .sslContext(tls.toServerContext())
          .addService(new GreeterService())
          .build();
}

private ProxyRpcServer(int port, TlsContext tls, HttpUrl backend) throws SSLException {
  this.stats = new StatsTracerFactory();
  this.bossEventLoopGroup = Netty.newBossEventLoopGroup();
  this.workerEventLoopGroup = Netty.newWorkerEventLoopGroup();
  this.server =
      NettyServerBuilder.forPort(port)
          .bossEventLoopGroup(bossEventLoopGroup)
          .workerEventLoopGroup(workerEventLoopGroup)
          .channelType(Netty.serverChannelType())
          .addStreamTracerFactory(stats)
          .sslContext(tls.toServerContext())
          .fallbackHandlerRegistry(new ProxyHandlerRegistry(backend))
          .build();
}

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

    setServer(server);
}

public static void main(String... args) throws Exception {
  System.out.println("starting server");

  String host = System.getProperty("host", "0.0.0.0");
  int port = Integer.getInteger("port", 8001);
  boolean useEpoll = Boolean.getBoolean("usePoll");

  Class channel;

  if (useEpoll) {
    channel = EpollServerSocketChannel.class;
  } else  {
    channel = NioServerSocketChannel.class;
  }

  ThreadFactory tf = new DefaultThreadFactory("server-elg-", true /*daemon */);
  NioEventLoopGroup boss = new NioEventLoopGroup(1, tf);
  NioEventLoopGroup worker = new NioEventLoopGroup(0, tf);
  NettyServerBuilder builder =
      NettyServerBuilder.forPort(port)
          .bossEventLoopGroup(boss)
          .workerEventLoopGroup(worker)
          .channelType(channel)
          .addService(new DefaultService())
          .directExecutor()
          .maxConcurrentCallsPerConnection(Runtime.getRuntime().availableProcessors() * 256)
          .flowControlWindow(NettyChannelBuilder.DEFAULT_FLOW_CONTROL_WINDOW * 10);

  io.grpc.Server start = builder.build();
  start.start();

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

/** 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();
}

public HillviewServer(final HostAndPort listenAddress, final IDataSet dataSet) throws IOException {
    this.listenAddress = listenAddress;
    this.server = NettyServerBuilder.forAddress(new InetSocketAddress(listenAddress.getHost(),
                                                                 listenAddress.getPort()))
                                    .executor(executorService)
                                    .workerEventLoopGroup(workerElg)
                                    .bossEventLoopGroup(bossElg)
                                    .addService(this)
                                    .maxMessageSize(MAX_MESSAGE_SIZE)
                                    .build()
                                    .start();
    this.dataSets = new ConcurrentHashMap<Integer, IDataSet>();
    this.put(this.dsIndex.getAndIncrement(), dataSet);
}

/**
 * Build NettyServerBuilder.
 * 1. Enable tls or not
 * 2. Load gRPC specified service definition and then add to gRPC registry.
 */
public Server buildServer() throws IOException {

  SslContext sslContext;
  NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(port).executor(Executors.newFixedThreadPool(8));
  if (useTls) {
    sslContext = GrpcSslContexts
        .forServer(SslUtils.loadCert("server.pem"), SslUtils.loadCert("server.key")).build();
    serverBuilder.sslContext(sslContext);
  }
  serverBuilder = bindService(serverBuilder);

  return serverBuilder.flowControlWindow(1024 * 1024).build();// 1024*1024 = 1MiB
}

@Override
public Server createServer() {
    NettyServerBuilder builder = NettyServerBuilder.forAddress(
            new InetSocketAddress(InetAddresses.forString(getAddress()), getPort()));
    Collection<GrpcServiceDefinition> definitions = discoverer.findGrpcServices();
    for (GrpcServiceDefinition definition : definitions) {
        ServiceDescriptor descriptor = definition.getService().bindService().getServiceDescriptor();
        logger.info("Registered gRPC service: " + descriptor.getName()
                + ", bean: " + definition.getBeanName() + ", class: "
                + definition.getService().getClass().getName());
        builder.addService(definition.getService());
    }

    return builder.build();
}

@SuppressWarnings("PMD.DoNotUseThreads")
private void start() throws IOException {
  server = NettyServerBuilder.forPort(PORT)
      .addService(intercept(new BenchImpl(), new ConnectionInterceptor()))
      .build()
      .start();

  Runtime.getRuntime().addShutdownHook(new Thread() {
    @Override
    public void run() {
      BenchServer.this.stop();
    }
  });
}

@Test
public void testKeepAlive() throws IOException, ExecutionException, InterruptedException {

    // Verify that keep-alive timeout less than permissible by the server results in a failure.
    final ControllerImpl controller = new ControllerImpl(NettyChannelBuilder.forAddress("localhost", serverPort)
            .keepAliveTime(10, TimeUnit.SECONDS).usePlaintext(true),
            ControllerImplConfig.builder().retryAttempts(1).build(), this.executor);
    CompletableFuture<Boolean> createStreamStatus = controller.createStream(StreamConfiguration.builder()
            .streamName("streamdelayed")
            .scope("scope1")
            .scalingPolicy(ScalingPolicy.fixed(1))
            .build());
    AssertExtensions.assertThrows("Should throw RetriesExhaustedException", createStreamStatus,
            throwable -> throwable instanceof RetriesExhaustedException);

    // Verify that the same RPC with permissible keepalive time succeeds.
    int serverPort2 = TestUtils.getAvailableListenPort();
    Server testServer = NettyServerBuilder.forPort(serverPort2)
            .addService(testServerImpl)
            .permitKeepAliveTime(5, TimeUnit.SECONDS)
            .build()
            .start();
    final ControllerImpl controller1 = new ControllerImpl(NettyChannelBuilder.forAddress("localhost", serverPort2)
            .keepAliveTime(10, TimeUnit.SECONDS).usePlaintext(true),
            ControllerImplConfig.builder().retryAttempts(1).build(), this.executor);
    createStreamStatus = controller1.createStream(StreamConfiguration.builder()
            .streamName("streamdelayed")
            .scope("scope1")
            .scalingPolicy(ScalingPolicy.fixed(1))
            .build());
    assertTrue(createStreamStatus.get());
    testServer.shutdownNow();
}

private static Server createServer(BindableService service, InetSocketAddress socket)
    throws IOException {
  Server server =
      NettyServerBuilder.forPort(socket.getPort())
          .addService(service)
          // Set the message size to max value here. The actual size is governed by the
          // buffer size in the layers above.
          .maxMessageSize(Integer.MAX_VALUE)
          .build();
  server.start();
  return server;
}

@Override
protected Server newServer() {
  return NettyServerBuilder
      .forPort(1234)
      .flowControlWindow(flowWindow)
      .addService(newService())
      .build();
}

private void start(int port) throws IOException {
  server =
      NettyServerBuilder.forPort(port)
          .sslContext(
              SslContextFactory.buildServerContext(
                  TlsConfig.fromConfig("xio.testServer.settings.tls")))
          .addService(new GreeterImpl())
          .build()
          .start();
}

private void start() {
    try {
        server = NettyServerBuilder.forPort(port)
                .addService(new MulticastRouteGrpcService(multicastRouteService))
                .build()
                .start();
        log.info("gRPC server started listening on " + port);
    } catch (IOException e) {
        log.error("Failed to start gRPC server", e);
    }
}

@Override
protected InternalServer newServer(List<ServerStreamTracer.Factory> streamTracerFactories) {
  return AccessProtectedHack.serverBuilderBuildTransportServer(
      NettyServerBuilder
        .forPort(0)
        .flowControlWindow(65 * 1024),
      streamTracerFactories,
      fakeClockTransportTracer);
}

@Override
protected InternalServer newServer(
    InternalServer server, List<ServerStreamTracer.Factory> streamTracerFactories) {
  int port = server.getPort();
  return AccessProtectedHack.serverBuilderBuildTransportServer(
      NettyServerBuilder
          .forPort(port)
          .flowControlWindow(65 * 1024),
      streamTracerFactories,
      fakeClockTransportTracer);
}

private ServerBuilder<?> serverBuilder(int port, File serverCertChainFile,
    File serverPrivateKeyFile, X509Certificate[] serverTrustedCaCerts) throws IOException {
  SslContextBuilder sslContextBuilder
      = SslContextBuilder.forServer(serverCertChainFile, serverPrivateKeyFile);
  GrpcSslContexts.configure(sslContextBuilder, sslProvider);
  sslContextBuilder.trustManager(serverTrustedCaCerts)
      .clientAuth(ClientAuth.REQUIRE);

  return NettyServerBuilder.forPort(port)
      .sslContext(sslContextBuilder.build());
}

private void startServer(int serverFlowControlWindow) {
  ServerBuilder<?> builder =
      NettyServerBuilder.forAddress(new InetSocketAddress("localhost", 0))
      .flowControlWindow(serverFlowControlWindow);
  builder.addService(ServerInterceptors.intercept(
      new TestServiceImpl(Executors.newScheduledThreadPool(2)),
      ImmutableList.<ServerInterceptor>of()));
  try {
    server = builder.build().start();
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}

@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  return NettyServerBuilder
      .forAddress(new LocalAddress("in-process-1"))
      .flowControlWindow(65 * 1024)
      .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .channelType(LocalServerChannel.class);
}

LoadWorker(int driverPort, int serverPort) throws Exception {
  this.serverPort = serverPort;
  NioEventLoopGroup singleThreadGroup = new NioEventLoopGroup(1,
      new ThreadFactoryBuilder()
          .setDaemon(true)
          .setNameFormat("load-worker-%d")
          .build());
  this.driverServer = NettyServerBuilder.forPort(driverPort)
      .directExecutor()
      .workerEventLoopGroup(singleThreadGroup)
      .bossEventLoopGroup(singleThreadGroup)
      .addService(new WorkerServiceImpl())
      .build();
}