NettyClientTransport implements ManagedClientTransport { @Override public void start(Listener transportListener) { lifecycleManager = new ClientTransportLifecycleManager( Preconditions.checkNotNull(transportListener, "listener")); handler = newHandler(); negotiationHandler = negotiator.newHandler(handler); Bootstrap b = new Bootstrap(); b.group(group); b.channel(channelType); if (NioSocketChannel.class.isAssignableFrom(channelType)) { b.option(SO_KEEPALIVE, true); } b.handler(negotiationHandler); channel = b.connect(address).addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { ChannelHandlerContext ctx = future.channel().pipeline().context(handler); if (ctx != null) { ctx.fireExceptionCaught(future.cause()); } future.channel().pipeline().fireExceptionCaught(future.cause()); } } }).channel(); handler.startWriteQueue(channel); channel.write(NettyClientHandler.NOOP_MESSAGE).addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { lifecycleManager.notifyTerminated(Utils.statusFromThrowable(future.cause())); } } }); channel.closeFuture().addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { lifecycleManager.notifyTerminated( Status.INTERNAL.withDescription("Connection closed with unknown cause")); } }); } NettyClientTransport(SocketAddress address, Class<? extends Channel> channelType,
                       EventLoopGroup group, ProtocolNegotiator negotiator,
                       int flowControlWindow, int maxMessageSize, int maxHeaderListSize,
                       String authority); @Override void ping(final PingCallback callback, final Executor executor); @Override ClientStream newStream(MethodDescriptor<?, ?> method, Metadata headers); @Override void start(Listener transportListener); @Override void shutdown(); @Override String toString(); @Override String getLogId();  }
@Test public void headersShouldAddDefaultUserAgent() throws Exception { startServer(); NettyClientTransport transport = newTransport(newNegotiator()); transport.start(clientTransportListener); new Rpc(transport).halfClose().waitForResponse(); assertEquals(1, serverListener.streamListeners.size()); Metadata headers = serverListener.streamListeners.get(0).headers; assertEquals(GrpcUtil.getGrpcUserAgent("netty", null), headers.get(USER_AGENT_KEY)); }
@Test public void headersShouldOverrideDefaultUserAgent() throws Exception { startServer(); NettyClientTransport transport = newTransport(newNegotiator()); transport.start(clientTransportListener); String userAgent = "testUserAgent"; Metadata sentHeaders = new Metadata(); sentHeaders.put(USER_AGENT_KEY, userAgent); new Rpc(transport, sentHeaders).halfClose().waitForResponse(); assertEquals(1, serverListener.streamListeners.size()); Metadata receivedHeaders = serverListener.streamListeners.get(0).headers; assertEquals(GrpcUtil.getGrpcUserAgent("netty", userAgent), receivedHeaders.get(USER_AGENT_KEY)); }
@Test public void maxMessageSizeShouldBeEnforced() throws Throwable { startServer(); NettyClientTransport transport = newTransport(newNegotiator(), 1, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE); transport.start(clientTransportListener); try { new Rpc(transport).halfClose().waitForResponse(); fail("Expected the stream to fail."); } catch (ExecutionException e) { Status status = Status.fromThrowable(e); assertEquals(INTERNAL, status.getCode()); System.err.println(status.getDescription()); assertTrue(status.getDescription().contains("deframing")); } }
@Test public void creatingMultipleTlsTransportsShouldSucceed() throws Exception { startServer(); ProtocolNegotiator negotiator = newNegotiator(); for (int index = 0; index < 2; ++index) { NettyClientTransport transport = newTransport(negotiator); transport.start(clientTransportListener); } final List<Rpc> rpcs = new ArrayList<Rpc>(transports.size()); for (NettyClientTransport transport : transports) { rpcs.add(new Rpc(transport).halfClose()); } for (Rpc rpc : rpcs) { rpc.waitForResponse(); } }
@Test public void bufferedStreamsShouldBeClosedWhenConnectionTerminates() throws Exception { startServer(1, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE); NettyClientTransport transport = newTransport(newNegotiator()); transport.start(clientTransportListener); new Rpc(transport).halfClose().waitForResponse(); Rpc[] rpcs = new Rpc[] { new Rpc(transport), new Rpc(transport), new Rpc(transport) }; rpcs[0].waitForResponse(); serverListener.transports.get(0).channel().pipeline().firstContext().close(); for (Rpc rpc : rpcs) { try { rpc.waitForClose(); fail("Expected the RPC to fail"); } catch (ExecutionException e) { } } }
@Test public void maxHeaderListSizeShouldBeEnforcedOnClient() throws Exception { startServer(); NettyClientTransport transport = newTransport(newNegotiator(), DEFAULT_MAX_MESSAGE_SIZE, 1); transport.start(clientTransportListener); try { new Rpc(transport, new Metadata()).halfClose().waitForResponse(); fail("The stream should have been failed due to client received header exceeds header list" + " size limit!"); } catch (Exception e) { Throwable rootCause = getRootCause(e); assertTrue(rootCause instanceof Http2Exception); assertEquals("Header size exceeded max allowed size (1)", rootCause.getMessage()); } }
@Test public void maxHeaderListSizeShouldBeEnforcedOnServer() throws Exception { startServer(100, 1); NettyClientTransport transport = newTransport(newNegotiator()); transport.start(clientTransportListener); try { new Rpc(transport, new Metadata()).halfClose().waitForResponse(); fail("The stream should have been failed due to server received header exceeds header list" + " size limit!"); } catch (Exception e) { Throwable rootCause = getRootCause(e); assertTrue(rootCause.getMessage(), rootCause.getMessage().contains("Header size exceeded max allowed size (1)")); } }