@Test public void testWrite() throws Exception { final NoSecurityImpl wireSecurity = new NoSecurityImpl(); MsgHandler msgHandler = new MsgHandler(wireSecurity); SocketChannelHandler handler = mock(SocketChannelHandler.class); SocketChannel channel = mock(SocketChannel.class); final List<SerializedMsg> msgs = new ArrayList<SerializedMsg>(); Answer<Long> bulkWrite = new Answer<Long>() { @Override public Long answer(InvocationOnMock invocation) throws Throwable { ByteBuffer[] buffers = (ByteBuffer[]) invocation.getArguments()[0]; long bytes = 0; for (ByteBuffer b : buffers) { bytes += b.remaining(); } assertEquals("Wrong number of buffers", 8, buffers.length); assertEquals("Wrong header buffer size", AbstractMessageHandler.HEADER_BYTE_SIZE, buffers[0].remaining()); assertEquals("Invalid magic", WireSizes.MAGIC_NUMBER, buffers[0].getInt()); assertEquals("Invalid message buffer length", buffers[0].getInt(), buffers[1].remaining()); assertEquals("Expected order 0", 0, buffers[0].getLong()); msgs.add((SerializedMsg) wireSecurity.fromWireForm(buffers[1])); assertEquals("Wrong header buffer size", AbstractMessageHandler.HEADER_BYTE_SIZE, buffers[2].remaining()); assertEquals("Invalid magic", WireSizes.MAGIC_NUMBER, buffers[2].getInt()); assertEquals("Invalid message buffer length", buffers[2].getInt(), buffers[3].remaining()); assertEquals("Expected order 1", 1, buffers[2].getLong()); msgs.add((SerializedMsg) wireSecurity.fromWireForm(buffers[3])); assertEquals("Wrong header buffer size", AbstractMessageHandler.HEADER_BYTE_SIZE, buffers[4].remaining()); assertEquals("Invalid magic", WireSizes.MAGIC_NUMBER, buffers[4].getInt()); assertEquals("Invalid message buffer length", buffers[4].getInt(), buffers[5].remaining()); assertEquals("Expected order 2", 2, buffers[4].getLong()); msgs.add((SerializedMsg) wireSecurity.fromWireForm(buffers[5])); assertEquals("Wrong header buffer size", AbstractMessageHandler.HEADER_BYTE_SIZE, buffers[6].remaining()); assertEquals("Invalid magic", WireSizes.MAGIC_NUMBER, buffers[6].getInt()); assertEquals("Invalid message buffer length", buffers[6].getInt(), buffers[7].remaining()); assertEquals("Expected order 3", 3, buffers[6].getLong()); msgs.add((SerializedMsg) wireSecurity.fromWireForm(buffers[7])); return bytes; } }; ByteBuffer[] template = new ByteBuffer[0]; when(handler.getChannel()).thenReturn(channel); when(channel.write(any(template.getClass()), anyInt(), anyInt())).thenReturn(0L).thenAnswer(bulkWrite); msgHandler.connect(handler); SerializedMsg msg1 = new SerializedMsg("Give me Slack"); SerializedMsg msg2 = new SerializedMsg("Or give me Food"); SerializedMsg msg3 = new SerializedMsg("Or kill me"); SerializedMsg msg4 = new SerializedMsg("Hello World"); msgHandler.send(msg1); msgHandler.send(msg2); msgHandler.send(msg3); msgHandler.send(msg4); msgHandler.writeReady(); assertEquals("Messages were written", 4, msgs.size()); assertEquals("Wrong message", msg1.getObject(), msgs.get(0).getObject()); assertEquals("Wrong message", msg2.getObject(), msgs.get(1).getObject()); assertEquals("Wrong message", msg3.getObject(), msgs.get(2).getObject()); assertEquals("Wrong message", msg4.getObject(), msgs.get(3).getObject()); }
private boolean write(ByteBuffer[] buffers) { try { long written = handler.getChannel().write(buffers); if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes written", written)); } if (written < 0) { close(); return false; } else if (buffers[buffers.length - 1].hasRemaining()) { long plusWritten = handler.getChannel().write(buffers); if (plusWritten < 0) { close(); return false; } if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes +written", plusWritten)); } } } catch (ClosedChannelException e) { if (getLog().isTraceEnabled()) { getLog().trace(format("shutting down handler due to other side closing [%s]", this), e); } error(); return false; } catch (IOException ioe) { if (getLog().isWarnEnabled() && !isClose(ioe)) { getLog().warn("shutting down handler", ioe); } error(); return false; } return true; }
AbstractMessageHandler implements CommunicationsHandler { private boolean write(ByteBuffer[] buffers) { try { long written = handler.getChannel().write(buffers); if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes written", written)); } if (written < 0) { close(); return false; } else if (buffers[buffers.length - 1].hasRemaining()) { long plusWritten = handler.getChannel().write(buffers); if (plusWritten < 0) { close(); return false; } if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes +written", plusWritten)); } } } catch (ClosedChannelException e) { if (getLog().isTraceEnabled()) { getLog().trace(format("shutting down handler due to other side closing [%s]", this), e); } error(); return false; } catch (IOException ioe) { if (getLog().isWarnEnabled() && !isClose(ioe)) { getLog().warn("shutting down handler", ioe); } error(); return false; } return true; } }
AbstractMessageHandler implements CommunicationsHandler { private boolean write(ByteBuffer[] buffers) { try { long written = handler.getChannel().write(buffers); if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes written", written)); } if (written < 0) { close(); return false; } else if (buffers[buffers.length - 1].hasRemaining()) { long plusWritten = handler.getChannel().write(buffers); if (plusWritten < 0) { close(); return false; } if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes +written", plusWritten)); } } } catch (ClosedChannelException e) { if (getLog().isTraceEnabled()) { getLog().trace(format("shutting down handler due to other side closing [%s]", this), e); } error(); return false; } catch (IOException ioe) { if (getLog().isWarnEnabled() && !isClose(ioe)) { getLog().warn("shutting down handler", ioe); } error(); return false; } return true; } AbstractMessageHandler(WireSecurity wireSecurity); }
AbstractMessageHandler implements CommunicationsHandler { private boolean write(ByteBuffer[] buffers) { try { long written = handler.getChannel().write(buffers); if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes written", written)); } if (written < 0) { close(); return false; } else if (buffers[buffers.length - 1].hasRemaining()) { long plusWritten = handler.getChannel().write(buffers); if (plusWritten < 0) { close(); return false; } if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes +written", plusWritten)); } } } catch (ClosedChannelException e) { if (getLog().isTraceEnabled()) { getLog().trace(format("shutting down handler due to other side closing [%s]", this), e); } error(); return false; } catch (IOException ioe) { if (getLog().isWarnEnabled() && !isClose(ioe)) { getLog().warn("shutting down handler", ioe); } error(); return false; } return true; } AbstractMessageHandler(WireSecurity wireSecurity); @Override void readReady(); void shutdown(); @Override void writeReady(); }
AbstractMessageHandler implements CommunicationsHandler { private boolean write(ByteBuffer[] buffers) { try { long written = handler.getChannel().write(buffers); if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes written", written)); } if (written < 0) { close(); return false; } else if (buffers[buffers.length - 1].hasRemaining()) { long plusWritten = handler.getChannel().write(buffers); if (plusWritten < 0) { close(); return false; } if (getLog().isTraceEnabled()) { getLog().trace(format("%s bytes +written", plusWritten)); } } } catch (ClosedChannelException e) { if (getLog().isTraceEnabled()) { getLog().trace(format("shutting down handler due to other side closing [%s]", this), e); } error(); return false; } catch (IOException ioe) { if (getLog().isWarnEnabled() && !isClose(ioe)) { getLog().warn("shutting down handler", ioe); } error(); return false; } return true; } AbstractMessageHandler(WireSecurity wireSecurity); @Override void readReady(); void shutdown(); @Override void writeReady();  }