@Test(timeout = 20000) public void testConnectionReceivesFragmentedData() throws Exception { final int FRAME_SIZE = 5317; ByteBuf sendBuffer = Unpooled.buffer(FRAME_SIZE); for (int i = 0; i < FRAME_SIZE; ++i) { sendBuffer.writeByte('A' + (i % 10)); } try (NettyEchoServer server = createEchoServer(createServerOptions())) { server.setMaxFrameSize(FRAME_SIZE); server.setFragmentWrites(true); server.start(); int port = server.getServerPort(); URI serverLocation = new URI("tcp: List<Transport> transports = new ArrayList<Transport>(); TransportOptions createClientOptions = createClientOptions(); createClientOptions.setTraceBytes(true); NettyTransportListener wsListener = new NettyTransportListener(true); Transport transport = createTransport(serverLocation, wsListener, 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(Wait.waitFor(new Wait.Condition() { @Override public boolean isSatisfied() throws Exception { LOG.debug("Checking completion: read {} expecting {}", bytesRead.get(), FRAME_SIZE); return bytesRead.get() == FRAME_SIZE || !transport.isConnected(); } }, 10000, 50)); assertTrue("Connection failed while receiving.", transport.isConnected()); transport.close(); assertEquals("Expected 2 data packets due to seperate websocket frames", 2, data.size()); ByteBuf receivedBuffer = Unpooled.buffer(FRAME_SIZE); for(ByteBuf buf : data) { buf.readBytes(receivedBuffer, buf.readableBytes()); } assertEquals("Unexpected data length", FRAME_SIZE, receivedBuffer.readableBytes()); assertTrue("Unexpected data", ByteBufUtil.equals(sendBuffer, 0, receivedBuffer, 0, FRAME_SIZE)); } finally { for (ByteBuf buf : data) { buf.release(); } } assertTrue(exceptions.isEmpty()); }
@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);  }