@Test public void testInputStreamUsingReceivedMessageWithAmqpValueSectionContainingBinary() throws Exception { byte[] bytes = "myBytes".getBytes(); Message message = Message.Factory.create(); message.setBody(new AmqpValue(new Binary(bytes))); AmqpJmsBytesMessageFacade amqpBytesMessageFacade = createReceivedBytesMessageFacade(createMockAmqpConsumer(), message); InputStream bytesStream = amqpBytesMessageFacade.getInputStream(); byte[] receivedBytes = new byte[bytes.length]; bytesStream.read(receivedBytes); assertTrue(Arrays.equals(bytes, receivedBytes)); assertEquals("Expected input stream to be at end but data was returned", END_OF_STREAM, bytesStream.read(new byte[1])); }
@Override public InputStream getInputStream() throws JMSException { if (bytesOut != null) { throw new IllegalStateException("Body is being written to, cannot perform a read."); } if (bytesIn == null) { Binary body = getBinaryFromBody(); bytesIn = new ByteBufInputStream( Unpooled.wrappedBuffer(body.getArray(), body.getArrayOffset(), body.getLength())); } return bytesIn; }
AmqpJmsBytesMessageFacade extends AmqpJmsMessageFacade implements JmsBytesMessageFacade { @Override public InputStream getInputStream() throws JMSException { if (bytesOut != null) { throw new IllegalStateException("Body is being written to, cannot perform a read."); } if (bytesIn == null) { Binary body = getBinaryFromBody(); bytesIn = new ByteBufInputStream( Unpooled.wrappedBuffer(body.getArray(), body.getArrayOffset(), body.getLength())); } return bytesIn; } }
AmqpJmsBytesMessageFacade extends AmqpJmsMessageFacade implements JmsBytesMessageFacade { @Override public InputStream getInputStream() throws JMSException { if (bytesOut != null) { throw new IllegalStateException("Body is being written to, cannot perform a read."); } if (bytesIn == null) { Binary body = getBinaryFromBody(); bytesIn = new ByteBufInputStream( Unpooled.wrappedBuffer(body.getArray(), body.getArrayOffset(), body.getLength())); } return bytesIn; }  }
AmqpJmsBytesMessageFacade extends AmqpJmsMessageFacade implements JmsBytesMessageFacade { @Override public InputStream getInputStream() throws JMSException { if (bytesOut != null) { throw new IllegalStateException("Body is being written to, cannot perform a read."); } if (bytesIn == null) { Binary body = getBinaryFromBody(); bytesIn = new ByteBufInputStream( Unpooled.wrappedBuffer(body.getArray(), body.getArrayOffset(), body.getLength())); } return bytesIn; }  @Override AmqpJmsBytesMessageFacade copy(); @Override byte getJmsMsgType(); @Override void clearBody(); @Override InputStream getInputStream(); @Override OutputStream getOutputStream(); @Override void reset(); @Override int getBodyLength(); @Override boolean hasBody(); @Override JmsBytesMessage asJmsMessage(); @Override byte[] copyBody(); @Override void onSend(long producerTtl); }
AmqpJmsBytesMessageFacade extends AmqpJmsMessageFacade implements JmsBytesMessageFacade { @Override public InputStream getInputStream() throws JMSException { if (bytesOut != null) { throw new IllegalStateException("Body is being written to, cannot perform a read."); } if (bytesIn == null) { Binary body = getBinaryFromBody(); bytesIn = new ByteBufInputStream( Unpooled.wrappedBuffer(body.getArray(), body.getArrayOffset(), body.getLength())); } return bytesIn; }  @Override AmqpJmsBytesMessageFacade copy(); @Override byte getJmsMsgType(); @Override void clearBody(); @Override InputStream getInputStream(); @Override OutputStream getOutputStream(); @Override void reset(); @Override int getBodyLength(); @Override boolean hasBody(); @Override JmsBytesMessage asJmsMessage(); @Override byte[] copyBody(); @Override void onSend(long producerTtl);  }