Java 类io.grpc.ServerStreamTracer 实例源码

@Override
@CheckReturnValue
protected NettyServer buildTransportServer(
    List<ServerStreamTracer.Factory> streamTracerFactories) {
  ProtocolNegotiator negotiator = protocolNegotiator;
  if (negotiator == null) {
    negotiator = sslContext != null ? ProtocolNegotiators.serverTls(sslContext) :
            ProtocolNegotiators.serverPlaintext();
  }

  return new NettyServer(
      address, channelType, channelOptions, bossEventLoopGroup, workerEventLoopGroup,
      negotiator, streamTracerFactories, transportTracerFactory,
      maxConcurrentCallsPerConnection, flowControlWindow,
      maxMessageSize, maxHeaderListSize, keepAliveTimeInNanos, keepAliveTimeoutInNanos,
      maxConnectionIdleInNanos,
      maxConnectionAgeInNanos, maxConnectionAgeGraceInNanos,
      permitKeepAliveWithoutCalls, permitKeepAliveTimeInNanos);
}

@Test
public void getPort_notStarted() throws Exception {
  InetSocketAddress addr = new InetSocketAddress(0);
  NettyServer ns = new NettyServer(
      addr,
      NioServerSocketChannel.class,
      new HashMap<ChannelOption<?>, Object>(),
      null, // no boss group
      null, // no event group
      new ProtocolNegotiators.PlaintextNegotiator(),
      Collections.<ServerStreamTracer.Factory>emptyList(),
      TransportTracer.getDefaultFactory(),
      1, // ignore
      1, // ignore
      1, // ignore
      1, // ignore
      1, // ignore
      1, 1, // ignore
      1, 1, // ignore
      true, 0); // ignore

  assertThat(ns.getPort()).isEqualTo(-1);
}

private void startServer(int maxStreamsPerConnection, int maxHeaderListSize) throws IOException {
  server = new NettyServer(
      TestUtils.testServerAddress(0),
      NioServerSocketChannel.class,
      new HashMap<ChannelOption<?>, Object>(),
      group, group, negotiator,
      Collections.<ServerStreamTracer.Factory>emptyList(),
      TransportTracer.getDefaultFactory(),
      maxStreamsPerConnection,
      DEFAULT_WINDOW_SIZE, DEFAULT_MAX_MESSAGE_SIZE, maxHeaderListSize,
      DEFAULT_SERVER_KEEPALIVE_TIME_NANOS, DEFAULT_SERVER_KEEPALIVE_TIMEOUT_NANOS,
      MAX_CONNECTION_IDLE_NANOS_DISABLED,
      MAX_CONNECTION_AGE_NANOS_DISABLED, MAX_CONNECTION_AGE_GRACE_NANOS_INFINITE, true, 0);
  server.start(serverListener);
  address = TestUtils.testServerAddress(server.getPort());
  authority = GrpcUtil.authorityFromHostAndPort(address.getHostString(), address.getPort());
}

@Override
public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
  TagContext parentCtx = headers.get(statsHeader);
  if (parentCtx == null) {
    parentCtx = tagger.empty();
  }
  parentCtx =
      tagger
          .toBuilder(parentCtx)
          .put(RpcMeasureConstants.RPC_METHOD, TagValue.create(fullMethodName))
          .build();
  return new ServerTracer(
      CensusStatsModule.this,
      fullMethodName,
      parentCtx,
      stopwatchSupplier,
      tagger,
      recordStartedRpcs,
      recordFinishedRpcs);
}

@VisibleForTesting
final List<ServerStreamTracer.Factory> getTracerFactories() {
  ArrayList<ServerStreamTracer.Factory> tracerFactories =
      new ArrayList<ServerStreamTracer.Factory>();
  if (statsEnabled) {
    CensusStatsModule censusStats = this.censusStatsOverride;
    if (censusStats == null) {
      censusStats = new CensusStatsModule(GrpcUtil.STOPWATCH_SUPPLIER, true);
    }
    tracerFactories.add(
        censusStats.getServerTracerFactory(recordStartedRpcs, recordFinishedRpcs));
  }
  if (tracingEnabled) {
    CensusTracingModule censusTracing =
        new CensusTracingModule(Tracing.getTracer(),
            Tracing.getPropagationComponent().getBinaryFormat());
    tracerFactories.add(censusTracing.getServerTracerFactory());
  }
  tracerFactories.addAll(streamTracerFactories);
  return tracerFactories;
}

@Test
public void traceHeadersPropagateSpanContext() throws Exception {
  CensusTracingModule.ClientCallTracer callTracer =
      censusTracing.newClientCallTracer(fakeClientParentSpan, method);
  Metadata headers = new Metadata();
  callTracer.newClientStreamTracer(CallOptions.DEFAULT, headers);

  verify(mockTracingPropagationHandler).toByteArray(same(fakeClientSpanContext));
  verifyNoMoreInteractions(mockTracingPropagationHandler);
  verify(tracer).spanBuilderWithExplicitParent(
      eq("Sent.package1.service2.method3"), same(fakeClientParentSpan));
  verify(spyClientSpanBuilder).setRecordEvents(eq(true));
  verifyNoMoreInteractions(tracer);
  assertTrue(headers.containsKey(censusTracing.tracingHeader));

  ServerStreamTracer serverTracer =
      censusTracing.getServerTracerFactory().newServerStreamTracer(
          method.getFullMethodName(), headers);
  verify(mockTracingPropagationHandler).fromByteArray(same(binarySpanContext));
  verify(tracer).spanBuilderWithRemoteParent(
      eq("Recv.package1.service2.method3"), same(spyClientSpan.getContext()));
  verify(spyServerSpanBuilder).setRecordEvents(eq(true));

  Context filteredContext = serverTracer.filterContext(Context.ROOT);
  assertSame(spyServerSpan, ContextUtils.CONTEXT_SPAN_KEY.get(filteredContext));
}

@Test
public void serverTracingSampledToLocalSpanStore() {
  ServerStreamTracer.Factory tracerFactory = censusTracing.getServerTracerFactory();
  ServerStreamTracer serverStreamTracer =
      tracerFactory.newServerStreamTracer(sampledMethod.getFullMethodName(), new Metadata());

  serverStreamTracer.filterContext(Context.ROOT);

  serverStreamTracer.serverCallStarted(
      new ServerCallInfoImpl<String, String>(sampledMethod, Attributes.EMPTY, null));

  serverStreamTracer.streamClosed(Status.CANCELLED);

  verify(spyServerSpan).end(
      EndSpanOptions.builder()
          .setStatus(io.opencensus.trace.Status.CANCELLED)
          .setSampleToLocalSpanStore(true)
          .build());
}

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

public static InternalServer serverBuilderBuildTransportServer(
    AbstractServerImplBuilder<?> builder,
    List<ServerStreamTracer.Factory> streamTracerFactories,
    TransportTracer.Factory transportTracerFactory) {
  builder.transportTracerFactory = transportTracerFactory;
  return builder.buildTransportServer(streamTracerFactories);
}

@Override
public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
  ServerStreamTracerInfo.InteropServerStreamTracer tracer
      = new ServerStreamTracerInfo.InteropServerStreamTracer();
  serverStreamTracers.add(new ServerStreamTracerInfo(fullMethodName, tracer));
  return tracer;
}

NettyServerTransport(
    Channel channel,
    ChannelPromise channelUnused,
    ProtocolNegotiator protocolNegotiator,
    List<ServerStreamTracer.Factory> streamTracerFactories,
    TransportTracer transportTracer,
    int maxStreams,
    int flowControlWindow,
    int maxMessageSize,
    int maxHeaderListSize,
    long keepAliveTimeInNanos,
    long keepAliveTimeoutInNanos,
    long maxConnectionIdleInNanos,
    long maxConnectionAgeInNanos,
    long maxConnectionAgeGraceInNanos,
    boolean permitKeepAliveWithoutCalls,
    long permitKeepAliveTimeInNanos) {
  this.channel = Preconditions.checkNotNull(channel, "channel");
  this.channelUnused = channelUnused;
  this.protocolNegotiator = Preconditions.checkNotNull(protocolNegotiator, "protocolNegotiator");
  this.streamTracerFactories =
      Preconditions.checkNotNull(streamTracerFactories, "streamTracerFactories");
  this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
  this.maxStreams = maxStreams;
  this.flowControlWindow = flowControlWindow;
  this.maxMessageSize = maxMessageSize;
  this.maxHeaderListSize = maxHeaderListSize;
  this.keepAliveTimeInNanos = keepAliveTimeInNanos;
  this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos;
  this.maxConnectionIdleInNanos = maxConnectionIdleInNanos;
  this.maxConnectionAgeInNanos = maxConnectionAgeInNanos;
  this.maxConnectionAgeGraceInNanos = maxConnectionAgeGraceInNanos;
  this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;
  this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;
}

NettyServer(
    SocketAddress address, Class<? extends ServerChannel> channelType,
    Map<ChannelOption<?>, ?> channelOptions,
    @Nullable EventLoopGroup bossGroup, @Nullable EventLoopGroup workerGroup,
    ProtocolNegotiator protocolNegotiator, List<ServerStreamTracer.Factory> streamTracerFactories,
    TransportTracer.Factory transportTracerFactory,
    int maxStreamsPerConnection, int flowControlWindow, int maxMessageSize, int maxHeaderListSize,
    long keepAliveTimeInNanos, long keepAliveTimeoutInNanos,
    long maxConnectionIdleInNanos,
    long maxConnectionAgeInNanos, long maxConnectionAgeGraceInNanos,
    boolean permitKeepAliveWithoutCalls, long permitKeepAliveTimeInNanos) {
  this.address = address;
  this.channelType = checkNotNull(channelType, "channelType");
  checkNotNull(channelOptions, "channelOptions");
  this.channelOptions = new HashMap<ChannelOption<?>, Object>(channelOptions);
  this.bossGroup = bossGroup;
  this.workerGroup = workerGroup;
  this.protocolNegotiator = checkNotNull(protocolNegotiator, "protocolNegotiator");
  this.streamTracerFactories = checkNotNull(streamTracerFactories, "streamTracerFactories");
  this.usingSharedBossGroup = bossGroup == null;
  this.usingSharedWorkerGroup = workerGroup == null;
  this.transportTracerFactory = transportTracerFactory;
  this.maxStreamsPerConnection = maxStreamsPerConnection;
  this.flowControlWindow = flowControlWindow;
  this.maxMessageSize = maxMessageSize;
  this.maxHeaderListSize = maxHeaderListSize;
  this.keepAliveTimeInNanos = keepAliveTimeInNanos;
  this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos;
  this.maxConnectionIdleInNanos = maxConnectionIdleInNanos;
  this.maxConnectionAgeInNanos = maxConnectionAgeInNanos;
  this.maxConnectionAgeGraceInNanos = maxConnectionAgeGraceInNanos;
  this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;
  this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;
}

@Test
public void getPort() throws Exception {
  InetSocketAddress addr = new InetSocketAddress(0);
  NettyServer ns = new NettyServer(
      addr,
      NioServerSocketChannel.class,
      new HashMap<ChannelOption<?>, Object>(),
      null, // no boss group
      null, // no event group
      new ProtocolNegotiators.PlaintextNegotiator(),
      Collections.<ServerStreamTracer.Factory>emptyList(),
      TransportTracer.getDefaultFactory(),
      1, // ignore
      1, // ignore
      1, // ignore
      1, // ignore
      1, // ignore
      1, 1, // ignore
      1, 1, // ignore
      true, 0); // ignore
  ns.start(new ServerListener() {
    @Override
    public ServerTransportListener transportCreated(ServerTransport transport) {
      return null;
    }

    @Override
    public void serverShutdown() {}
  });

  // Check that we got an actual port.
  assertThat(ns.getPort()).isGreaterThan(0);

  // Cleanup
  ns.shutdown();
}

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

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

@VisibleForTesting
InProcessServer(
    String name, ObjectPool<ScheduledExecutorService> schedulerPool,
    List<ServerStreamTracer.Factory> streamTracerFactories) {
  this.name = name;
  this.schedulerPool = schedulerPool;
  this.streamTracerFactories =
      Collections.unmodifiableList(checkNotNull(streamTracerFactories, "streamTracerFactories"));
}

/**
 * Factory method for the server-side.
 */
public static StatsTraceContext newServerContext(
    List<ServerStreamTracer.Factory> factories, String fullMethodName, Metadata headers) {
  if (factories.isEmpty()) {
    return NOOP;
  }
  StreamTracer[] tracers = new StreamTracer[factories.size()];
  for (int i = 0; i < tracers.length; i++) {
    tracers[i] = factories.get(i).newServerStreamTracer(fullMethodName, headers);
  }
  return new StatsTraceContext(tracers);
}

/**
 * See {@link ServerStreamTracer#filterContext}.  For server-side only.
 *
 * <p>Called from {@link io.grpc.internal.ServerImpl}.
 */
public <ReqT, RespT> Context serverFilterContext(Context context) {
  Context ctx = checkNotNull(context, "context");
  for (StreamTracer tracer : tracers) {
    ctx = ((ServerStreamTracer) tracer).filterContext(ctx);
    checkNotNull(ctx, "%s returns null context", tracer);
  }
  return ctx;
}

@SuppressWarnings("ReferenceEquality")
@Override
public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
  SpanContext remoteSpan = headers.get(tracingHeader);
  if (remoteSpan == SpanContext.INVALID) {
    remoteSpan = null;
  }
  return new ServerTracer(fullMethodName, remoteSpan);
}

@Test
public void getPort_notStarted() throws Exception {
  InProcessServer s =
      new InProcessServer("name", GrpcUtil.TIMER_SERVICE,
          Collections.<ServerStreamTracer.Factory>emptyList());

  Truth.assertThat(s.getPort()).isEqualTo(-1);
}

@Test
public void serverHoldsRefToScheduler() throws Exception {
  final ScheduledExecutorService ses = new FakeClock().getScheduledExecutorService();
  class RefCountingObjectPool implements ObjectPool<ScheduledExecutorService> {
    private int count;

    @Override
    public ScheduledExecutorService getObject() {
      count++;
      return ses;
    }

    @Override
    public ScheduledExecutorService returnObject(Object returned) {
      count--;
      return null;
    }
  }

  RefCountingObjectPool pool = new RefCountingObjectPool();
  InProcessServer s =
      new InProcessServer("name", pool, Collections.<ServerStreamTracer.Factory>emptyList());
  Truth.assertThat(pool.count).isEqualTo(0);
  s.start(new ServerListener() {
    @Override public ServerTransportListener transportCreated(ServerTransport transport) {
      throw new UnsupportedOperationException();
    }

    @Override public void serverShutdown() {}
  });
  Truth.assertThat(pool.count).isEqualTo(1);
  s.shutdown();
  Truth.assertThat(pool.count).isEqualTo(0);
}

@Test
public void getTracerFactories_default() {
  builder.addStreamTracerFactory(DUMMY_USER_TRACER);
  List<ServerStreamTracer.Factory> factories = builder.getTracerFactories();
  assertEquals(3, factories.size());
  assertThat(factories.get(0)).isInstanceOf(CensusStatsModule.ServerTracerFactory.class);
  assertThat(factories.get(1)).isInstanceOf(CensusTracingModule.ServerTracerFactory.class);
  assertThat(factories.get(2)).isSameAs(DUMMY_USER_TRACER);
}

@Test
public void getTracerFactories_disableStats() {
  builder.addStreamTracerFactory(DUMMY_USER_TRACER);
  builder.setStatsEnabled(false);
  List<ServerStreamTracer.Factory> factories = builder.getTracerFactories();
  assertEquals(2, factories.size());
  assertThat(factories.get(0)).isInstanceOf(CensusTracingModule.ServerTracerFactory.class);
  assertThat(factories.get(1)).isSameAs(DUMMY_USER_TRACER);
}

@Test
public void getTracerFactories_disableTracing() {
  builder.addStreamTracerFactory(DUMMY_USER_TRACER);
  builder.setTracingEnabled(false);
  List<ServerStreamTracer.Factory> factories = builder.getTracerFactories();
  assertEquals(2, factories.size());
  assertThat(factories.get(0)).isInstanceOf(CensusStatsModule.ServerTracerFactory.class);
  assertThat(factories.get(1)).isSameAs(DUMMY_USER_TRACER);
}

@Test
public void getTracerFactories_disableBoth() {
  builder.addStreamTracerFactory(DUMMY_USER_TRACER);
  builder.setTracingEnabled(false);
  builder.setStatsEnabled(false);
  List<ServerStreamTracer.Factory> factories = builder.getTracerFactories();
  assertThat(factories).containsExactly(DUMMY_USER_TRACER);
}

@Test
public void serverTracingNotSampledToLocalSpanStore_whenServerCallNotCreated() {
  ServerStreamTracer.Factory tracerFactory = censusTracing.getServerTracerFactory();
  ServerStreamTracer serverStreamTracer =
      tracerFactory.newServerStreamTracer(sampledMethod.getFullMethodName(), new Metadata());

  serverStreamTracer.streamClosed(Status.CANCELLED);

  verify(spyServerSpan).end(
      EndSpanOptions.builder()
          .setStatus(io.opencensus.trace.Status.CANCELLED)
          .setSampleToLocalSpanStore(false)
          .build());
}

static NettyServerHandler newHandler(
    ServerTransportListener transportListener,
    ChannelPromise channelUnused,
    List<ServerStreamTracer.Factory> streamTracerFactories,
    TransportTracer transportTracer,
    int maxStreams,
    int flowControlWindow,
    int maxHeaderListSize,
    int maxMessageSize,
    long keepAliveTimeInNanos,
    long keepAliveTimeoutInNanos,
    long maxConnectionIdleInNanos,
    long maxConnectionAgeInNanos,
    long maxConnectionAgeGraceInNanos,
    boolean permitKeepAliveWithoutCalls,
    long permitKeepAliveTimeInNanos) {
  Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive");
  Http2FrameLogger frameLogger = new Http2FrameLogger(LogLevel.DEBUG, NettyServerHandler.class);
  Http2HeadersDecoder headersDecoder = new GrpcHttp2ServerHeadersDecoder(maxHeaderListSize);
  Http2FrameReader frameReader = new Http2InboundFrameLogger(
      new DefaultHttp2FrameReader(headersDecoder), frameLogger);
  Http2FrameWriter frameWriter =
      new Http2OutboundFrameLogger(new DefaultHttp2FrameWriter(), frameLogger);
  return newHandler(
      channelUnused,
      frameReader,
      frameWriter,
      transportListener,
      streamTracerFactories,
      transportTracer,
      maxStreams,
      flowControlWindow,
      maxHeaderListSize,
      maxMessageSize,
      keepAliveTimeInNanos,
      keepAliveTimeoutInNanos,
      maxConnectionIdleInNanos,
      maxConnectionAgeInNanos,
      maxConnectionAgeGraceInNanos,
      permitKeepAliveWithoutCalls,
      permitKeepAliveTimeInNanos);
}

@VisibleForTesting
static NettyServerHandler newHandler(
    ChannelPromise channelUnused,
    Http2FrameReader frameReader,
    Http2FrameWriter frameWriter,
    ServerTransportListener transportListener,
    List<ServerStreamTracer.Factory> streamTracerFactories,
    TransportTracer transportTracer,
    int maxStreams,
    int flowControlWindow,
    int maxHeaderListSize,
    int maxMessageSize,
    long keepAliveTimeInNanos,
    long keepAliveTimeoutInNanos,
    long maxConnectionIdleInNanos,
    long maxConnectionAgeInNanos,
    long maxConnectionAgeGraceInNanos,
    boolean permitKeepAliveWithoutCalls,
    long permitKeepAliveTimeInNanos) {
  Preconditions.checkArgument(maxStreams > 0, "maxStreams must be positive");
  Preconditions.checkArgument(flowControlWindow > 0, "flowControlWindow must be positive");
  Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive");
  Preconditions.checkArgument(maxMessageSize > 0, "maxMessageSize must be positive");

  final Http2Connection connection = new DefaultHttp2Connection(true);
  WeightedFairQueueByteDistributor dist = new WeightedFairQueueByteDistributor(connection);
  dist.allocationQuantum(16 * 1024); // Make benchmarks fast again.
  DefaultHttp2RemoteFlowController controller =
      new DefaultHttp2RemoteFlowController(connection, dist);
  connection.remote().flowController(controller);
  final KeepAliveEnforcer keepAliveEnforcer = new KeepAliveEnforcer(
      permitKeepAliveWithoutCalls, permitKeepAliveTimeInNanos, TimeUnit.NANOSECONDS);

  // Create the local flow controller configured to auto-refill the connection window.
  connection.local().flowController(
      new DefaultHttp2LocalFlowController(connection, DEFAULT_WINDOW_UPDATE_RATIO, true));
  frameWriter = new WriteMonitoringFrameWriter(frameWriter, keepAliveEnforcer);
  Http2ConnectionEncoder encoder = new DefaultHttp2ConnectionEncoder(connection, frameWriter);
  Http2ConnectionDecoder decoder = new DefaultHttp2ConnectionDecoder(connection, encoder,
      frameReader);

  Http2Settings settings = new Http2Settings();
  settings.initialWindowSize(flowControlWindow);
  settings.maxConcurrentStreams(maxStreams);
  settings.maxHeaderListSize(maxHeaderListSize);

  return new NettyServerHandler(
      channelUnused,
      connection,
      transportListener,
      streamTracerFactories,
      transportTracer,
      decoder, encoder, settings,
      maxMessageSize,
      keepAliveTimeInNanos, keepAliveTimeoutInNanos,
      maxConnectionIdleInNanos,
      maxConnectionAgeInNanos, maxConnectionAgeGraceInNanos,
      keepAliveEnforcer);
}

@Test(timeout = 60000)
public void childChannelOptions() throws Exception {
  final int originalLowWaterMark = 2097169;
  final int originalHighWaterMark = 2097211;

  Map<ChannelOption<?>, Object> channelOptions = new HashMap<ChannelOption<?>, Object>();

  channelOptions.put(ChannelOption.WRITE_BUFFER_WATER_MARK,
      new WriteBufferWaterMark(originalLowWaterMark, originalHighWaterMark));

  final AtomicInteger lowWaterMark = new AtomicInteger(0);
  final AtomicInteger highWaterMark = new AtomicInteger(0);

  final CountDownLatch countDownLatch = new CountDownLatch(1);

  InetSocketAddress addr = new InetSocketAddress(0);
  NettyServer ns = new NettyServer(
      addr,
      NioServerSocketChannel.class,
      channelOptions,
      null, // no boss group
      null, // no event group
      new ProtocolNegotiators.PlaintextNegotiator(),
      Collections.<ServerStreamTracer.Factory>emptyList(),
      TransportTracer.getDefaultFactory(),
      1, // ignore
      1, // ignore
      1, // ignore
      1, // ignore
      1, // ignore
      1, 1, // ignore
      1, 1, // ignore
      true, 0); // ignore
  ns.start(new ServerListener() {
    @Override
    public ServerTransportListener transportCreated(ServerTransport transport) {
      Channel channel = ((NettyServerTransport)transport).channel();
      WriteBufferWaterMark writeBufferWaterMark = channel.config()
          .getOption(ChannelOption.WRITE_BUFFER_WATER_MARK);
      lowWaterMark.set(writeBufferWaterMark.low());
      highWaterMark.set(writeBufferWaterMark.high());

      countDownLatch.countDown();

      return null;
    }

    @Override
    public void serverShutdown() {}
  });

  Socket socket = new Socket();
  socket.connect(new InetSocketAddress("localhost", ns.getPort()), /* timeout= */ 8000);
  countDownLatch.await();
  socket.close();

  assertThat(lowWaterMark.get()).isEqualTo(originalLowWaterMark);
  assertThat(highWaterMark.get()).isEqualTo(originalHighWaterMark);

  ns.shutdown();
}

InProcessServer(
    String name, Resource<ScheduledExecutorService> schedulerResource,
    List<ServerStreamTracer.Factory> streamTracerFactories) {
  this(name, SharedResourcePool.forResource(schedulerResource), streamTracerFactories);
}

List<ServerStreamTracer.Factory> getStreamTracerFactories() {
  return streamTracerFactories;
}

@Override
protected InProcessServer buildTransportServer(
    List<ServerStreamTracer.Factory> streamTracerFactories) {
  return new InProcessServer(name, GrpcUtil.TIMER_SERVICE, streamTracerFactories);
}

/**
 * Returns the server tracer factory.
 */
ServerStreamTracer.Factory getServerTracerFactory(
    boolean recordStartedRpcs, boolean recordFinishedRpcs) {
  return new ServerTracerFactory(recordStartedRpcs, recordFinishedRpcs);
}

/**
 * See {@link ServerStreamTracer#serverCallStarted}.  For server-side only.
 *
 * <p>Called from {@link io.grpc.internal.ServerImpl}.
 */
public void serverCallStarted(ServerCallInfo<?, ?> callInfo) {
  for (StreamTracer tracer : tracers) {
    ((ServerStreamTracer) tracer).serverCallStarted(callInfo);
  }
}

/**
 * Returns the server tracer factory.
 */
ServerStreamTracer.Factory getServerTracerFactory() {
  return serverTracerFactory;
}

@Override
public final T addStreamTracerFactory(ServerStreamTracer.Factory factory) {
  streamTracerFactories.add(checkNotNull(factory, "factory"));
  return thisT();
}

@Override
protected InternalServer newServer(List<ServerStreamTracer.Factory> streamTracerFactories) {
  return new InProcessServer(TRANSPORT_NAME, GrpcUtil.TIMER_SERVICE, streamTracerFactories);
}

@Override
protected InternalServer newServer(
    InternalServer server, List<ServerStreamTracer.Factory> streamTracerFactories) {
  return newServer(streamTracerFactories);
}

@Override
public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
  throw new UnsupportedOperationException();
}