@Test(timeout = 60000) public void testTransportDetectsConnectionDropWhenServerEnforcesMaxFrameSize() throws Exception { final int FRAME_SIZE = 1024; ByteBuf sendBuffer = Unpooled.buffer(FRAME_SIZE); for (int i = 0; i < FRAME_SIZE; ++i) { sendBuffer.writeByte('A'); } try (NettyEchoServer server = createEchoServer(createServerOptions())) { server.setMaxFrameSize(FRAME_SIZE / 2); server.start(); int port = server.getServerPort(); URI serverLocation = new URI("tcp: List<Transport> transports = new ArrayList<Transport>(); final Transport transport = createTransport(serverLocation, testListener, createClientOptions()); try { transport.setMaxFrameSize(FRAME_SIZE); transport.connect(null, null); transports.add(transport); transport.writeAndFlush(sendBuffer.copy()); } catch (Exception e) { fail("Should have connected to the server at " + serverLocation + " but got exception: " + e); } assertTrue("Transport should have lost connection", Wait.waitFor(new Wait.Condition() { @Override public boolean isSatisfied() throws Exception { try { transport.writeAndFlush(sendBuffer.copy()); } catch (IOException e) { LOG.info("Transport send caught error:", e); return true; } return false; } }, 10000, 10)); transport.close(); } }
@Override public void writeAndFlush(ByteBuf output) throws IOException { checkConnected(); int length = output.readableBytes(); if (length == 0) { return; } LOG.trace("Attempted write and flush of: {} bytes", length); channel.writeAndFlush(new BinaryWebSocketFrame(output), channel.voidPromise()); }
NettyWsTransport extends NettyTcpTransport { @Override public void writeAndFlush(ByteBuf output) throws IOException { checkConnected(); int length = output.readableBytes(); if (length == 0) { return; } LOG.trace("Attempted write and flush of: {} bytes", length); channel.writeAndFlush(new BinaryWebSocketFrame(output), channel.voidPromise()); } }
NettyWsTransport extends NettyTcpTransport { @Override public void writeAndFlush(ByteBuf output) throws IOException { checkConnected(); int length = output.readableBytes(); if (length == 0) { return; } LOG.trace("Attempted write and flush of: {} bytes", length); channel.writeAndFlush(new BinaryWebSocketFrame(output), channel.voidPromise()); } NettyWsTransport(URI remoteLocation, TransportOptions options, boolean secure);  NettyWsTransport(TransportListener listener, URI remoteLocation, TransportOptions options, boolean secure); }
NettyWsTransport extends NettyTcpTransport { @Override public void writeAndFlush(ByteBuf output) throws IOException { checkConnected(); int length = output.readableBytes(); if (length == 0) { return; } LOG.trace("Attempted write and flush of: {} bytes", length); channel.writeAndFlush(new BinaryWebSocketFrame(output), channel.voidPromise()); } NettyWsTransport(URI remoteLocation, TransportOptions options, boolean secure);  NettyWsTransport(TransportListener listener, URI remoteLocation, TransportOptions options, boolean secure); @Override void write(ByteBuf output); @Override void writeAndFlush(ByteBuf output); }
NettyWsTransport extends NettyTcpTransport { @Override public void writeAndFlush(ByteBuf output) throws IOException { checkConnected(); int length = output.readableBytes(); if (length == 0) { return; } LOG.trace("Attempted write and flush of: {} bytes", length); channel.writeAndFlush(new BinaryWebSocketFrame(output), channel.voidPromise()); } NettyWsTransport(URI remoteLocation, TransportOptions options, boolean secure);  NettyWsTransport(TransportListener listener, URI remoteLocation, TransportOptions options, boolean secure); @Override void write(ByteBuf output); @Override void writeAndFlush(ByteBuf output);  }