Java 类javax.jms.BytesMessage 实例源码

@Test
public void validateBytesConvertedToBytesMessageOnSend() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    publisher.publish(destinationName, "hellomq".getBytes());

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    byte[] bytes = new byte[7];
    ((BytesMessage) receivedMessage).readBytes(bytes);
    assertEquals("hellomq", new String(bytes));

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Test
public void client_sends_five_messages_to_queue_and_receives() throws Exception {
    try (JMSClient client = server.createClient()) {
        DummyListener listener = new DummyListener(5);
        client.addQueueListener(DUMMY_QUEUE, listener);
        client.startListening();

        for (int i = 0; i < 5; i++) {
            client.sendToQueue(DUMMY_QUEUE, i + "");
        }

        List<BytesMessage> bytesMessages = listener.awaitMessages();
        assertThat(bytesMessages.size(), is(5));
        assertNotNull(bytesMessages.get(0));

        // Ensure messages are received in the same order they were sent
        for (int i = 0; i < 5; i++) {
            assertThat(JMSHelper.objectFromMsg(bytesMessages.get(i)), is(i + ""));
        }
    }
}

@Test
public void client_sends_to_topic_and_receives() throws Exception {
    DummyListener listener = new DummyListener(1);
    try (JMSClient client = server.createClient()) {

        server.getServer().createTopic(DUMMY_TOPIC);

        client.addTopicListener(DUMMY_TOPIC, listener);
        client.startListening();

        client.sendToTopic(DUMMY_TOPIC, DUMMY_PAYLOAD);

        List<BytesMessage> bytesMessages = listener.awaitMessages();
        assertReceivedMessageIs(bytesMessages, DUMMY_PAYLOAD);
    }
}

@Test
public void client_sends_five_messages_to_topic_and_receives() throws Exception {
    DummyListener listener = new DummyListener(5);
    try (JMSClient client = server.createClient()) {

        server.getServer().createTopic(DUMMY_TOPIC);
        client.addTopicListener(DUMMY_TOPIC, listener);
        client.startListening();

        for (int i = 0; i < 5; i++) {
            client.sendToTopic(DUMMY_TOPIC, i + "");
        }

        List<BytesMessage> bytesMessages = listener.awaitMessages();

        assertNotNull(server);
        assertThat(bytesMessages.size(), is(5));
        assertNotNull(bytesMessages.get(0));

        // Ensure messages are received in the same order they were sent
        for (int i = 0; i < 5; i++) {
            assertThat(JMSHelper.objectFromMsg(bytesMessages.get(i)), is(i + ""));
        }
    }
}

@Test
public void validateBytesConvertedToBytesMessageOnSendOverJNDI() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsJndiTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    publisher.publish(destinationName, "hellomq".getBytes());

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    byte[] bytes = new byte[7];
    ((BytesMessage) receivedMessage).readBytes(bytes);
    assertEquals("hellomq", new String(bytes));

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributesOverJNDI() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsJndiTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributes() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

/**
 * Convert a message into a Kafka Connect SourceRecord.
 * 
 * @param context            the JMS context to use for building messages
 * @param topic              the Kafka topic
 * @param messageBodyJms     whether to interpret MQ messages as JMS messages
 * @param message            the message
 * 
 * @return the Kafka Connect SourceRecord
 * 
 * @throws JMSException      Message could not be converted
 */
@Override public SourceRecord toSourceRecord(JMSContext context, String topic, boolean messageBodyJms, Message message) throws JMSException {
    byte[] payload;
    if (message instanceof BytesMessage) {
        payload = message.getBody(byte[].class);
    }
    else if (message instanceof TextMessage) {
        String s = message.getBody(String.class);
        payload = s.getBytes(UTF_8);
    }
    else {
        log.error("Unsupported JMS message type {}", message.getClass());
        throw new ConnectException("Unsupported JMS message type");
    }

    SchemaAndValue sv = converter.toConnectData(topic, payload);
    return new SourceRecord(null, null, topic, sv.schema(), sv.value());
}

private void doSendBinaryMessage( final Session session, final Destination destination,
                                  final byte[] bytes,
                                  final Map<String, ?> properties ) throws JMSException {

    try {
        BytesMessage message = session.createBytesMessage();
        message.writeBytes(bytes);
        if (properties != null) {
            // Note: Setting any properties (including JMS fields) using
            // setObjectProperty might not be supported by all providers
            // Tested with: ActiveMQ
            for (final Entry<String, ?> property : properties.entrySet()) {
                message.setObjectProperty(property.getKey(), property.getValue());
            }
        }
        final MessageProducer producer = session.createProducer(destination);
        producer.send(message);
    } finally {
        releaseSession(false);
    }
}

@Test
public void validateBytesConvertedToBytesMessageOnSend() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    publisher.publish(destinationName, "hellomq".getBytes());

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    byte[] bytes = new byte[7];
    ((BytesMessage) receivedMessage).readBytes(bytes);
    assertEquals("hellomq", new String(bytes));

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Test
public void validateBytesConvertedToBytesMessageOnSendOverJNDI() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsJndiTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    publisher.publish(destinationName, "hellomq".getBytes());

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    byte[] bytes = new byte[7];
    ((BytesMessage) receivedMessage).readBytes(bytes);
    assertEquals("hellomq", new String(bytes));

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributesOverJNDI() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsJndiTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributes() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

public void send(byte[] message) {
    if (session == null) {
        logger.warning("Session is null - sending monitoring message aborted");
        return;
    }
    if (queue == null) {
        logger.warning("Queue is null - sending monitoring message aborted");
        return;
    }
    if (producer == null) {
        logger.warning("Producer is null - sending monitoring message aborted");
        return;
    }
    try {
        BytesMessage bytesMessage = session.createBytesMessage();
        bytesMessage.writeBytes(message);
        producer.send(bytesMessage);
    } catch (JMSException ex) {
        logger.log(Level.SEVERE, "Message sending error", ex);
    }
    //logger.info("Sent monitoring message");
}

@Override
public void sendMessage(final Message message) {
    this.jmsTemplate.send(createDestination(message.getDestination()), new MessageCreator() {
        @Override
        public javax.jms.Message createMessage(Session session) throws JMSException {
            BytesMessage bytesMessage = session.createBytesMessage();
            bytesMessage.writeBytes(message.getBody());
            if(!CollectionUtils.isEmpty(message.getProperties())){
                if(message.getProperties().get("JMSXGroupID") != null){
                    bytesMessage.setStringProperty("JMSXGroupID", message.getProperties().get("JMSXGroupID").toString());
                }
                if(message.getProperties().get("JMSXGroupSeq") != null){
                    String JMSXGroupSeq = message.getProperties().get("JMSXGroupSeq").toString();
                    if(StringUtil.isNumeric(JMSXGroupSeq)){
                        bytesMessage.setIntProperty("JMSXGroupSeq", Integer.valueOf(JMSXGroupSeq));
                    }
                }
            }
            return bytesMessage;
        }
    });
}

void createBytesMessage() throws Exception {
  BytesMessage message = mock(BytesMessage.class);
  when(message.getBodyLength()).thenReturn((long)BYTES.length);
  when(message.readBytes(any(byte[].class))).then(new Answer<Integer>() {
    @Override
    public Integer answer(InvocationOnMock invocation) throws Throwable {
      byte[] buffer = (byte[])invocation.getArguments()[0];
      if (buffer != null) {
        assertEquals(buffer.length, BYTES.length);
        System.arraycopy(BYTES, 0, buffer, 0, BYTES.length);
      }
      return BYTES.length;
    }
  });
  this.message = message;
}

@Override
public void onMessage(Message message) {

  try {

    if (message instanceof BytesMessage) {

      BytesMessage bytesMessage = (BytesMessage) message;
      byte[] data = new byte[(int) bytesMessage.getBodyLength()];
      bytesMessage.readBytes(data);
      LOG.info("Message received {}", new String(data));

    } else if (message instanceof TextMessage) {

      TextMessage textMessage = (TextMessage) message;
      String text = textMessage.getText();
      LOG.info("Message received {}", text);
    }

  } catch (JMSException jmsEx) {
    jmsEx.printStackTrace();
  }
}

/**
 * 
 * @param session
 * @param frame
 * @return
 * @throws JMSException
 */
public Message toMessage(Session session, Frame frame) throws JMSException {
    // FIXME buffer pool
    final Message msg;
    if (frame.contains(CONTENT_LENGTH)) {
        final ByteBuffer buf = frame.body().get();
        byte[] bytes = new byte[buf.remaining()];
        buf.get(bytes);
        final BytesMessage bm = session.createBytesMessage();
        bm.writeBytes(bytes);
        msg = bm;
    } else {
        msg = session.createTextMessage(UTF_8.decode(frame.body().get()).toString());
    }
    copyHeaders(session, frame, msg);
    return msg;
}

@Test
public void toMessage_bytes() throws JMSException {
    final javax.jms.Session session = mock(javax.jms.Session.class);
    final Frame frame = mock(Frame.class);
    final ByteBuffer buffer = ByteBuffer.wrap(new byte[0]).asReadOnlyBuffer();
    when(frame.body()).thenReturn(Optional.of(buffer));
    when(frame.headers()).thenReturn(new MultivaluedHashMap<>());
    when(frame.contains(Standard.CONTENT_LENGTH)).thenReturn(true);
    final BytesMessage message = mock(BytesMessage.class);
    when(session.createBytesMessage()).thenReturn(message);

    this.factory.toMessage(session, frame);

    verify(frame).body();
    verify(frame, times(2)).headers();
    verify(frame).contains(Standard.CONTENT_LENGTH);
    verify(session).createBytesMessage();
    verify(message).setJMSCorrelationID(null);
    verify(message).writeBytes(new byte[0]);
    verifyNoMoreInteractions(session, frame, message);
}

@Test
public void toFrame_bytesMessage() throws IOException, JMSException {
    final BytesMessage message = mock(BytesMessage.class);
    when(message.getPropertyNames()).thenReturn(Collections.enumeration(Collections.singleton("hello")));

    this.factory.toFrame(message, "subscriptionId");

    verify(message).getPropertyNames();
    verify(message).getJMSMessageID();
    verify(message).getJMSDestination();
    verify(message).getJMSCorrelationID();
    verify(message).getJMSExpiration();
    verify(message).getJMSRedelivered();
    verify(message).getJMSPriority();
    verify(message).getJMSReplyTo();
    verify(message).getJMSTimestamp();
    verify(message).getJMSType();
    verify(message).getStringProperty("hello");
    verify(message).getStringProperty(Standard.CONTENT_TYPE.value());
    verify(message).getBodyLength();
    verify(message).readBytes(new byte[0]);
    verifyNoMoreInteractions(message);
}

/**
 * Read the the payload of the message and return it in a byte array.
 * 
 * @param msg
 * @return
 */
private byte[] getBytes(final Message msg) {
    byte[] data = null;
    try {
        if (msg instanceof BytesMessage) {
            final BytesMessage tmp = (BytesMessage) msg;
            int len;
            len = (int) ((BytesMessage) msg).getBodyLength();
            data = new byte[len];
            tmp.readBytes(data);
        } else if (msg instanceof TextMessage) {
            data = ((TextMessage) msg).getText().getBytes();

        }
    } catch (final JMSException e) {
        logger.error("Error getting bytes from message.", e);
    }
    return data;
}

public DataRecorderMessage(final Message message, final long timeStamp) {
    try {
        // this.setDelayMillis(delayMillis);
        this.timeStamp = new Date(timeStamp);

        // Get all the properties from the incoming message
        this.properties = getAllProperties(message);

        // Get the payload from the incoming message.
        if (message instanceof BytesMessage) {
            byte[] byteArray = null;
            final BytesMessage tmp = (BytesMessage) message;
            int len;
            len = (int) ((BytesMessage) message).getBodyLength();
            byteArray = new byte[len];
            tmp.readBytes(byteArray);
            this.body = byteArray;
        } else if (message instanceof TextMessage) {
            this.body = ((TextMessage) message).getText();
        }

    } catch (final JMSException e) {
        logger.error("Error reading from the incoming JMS message");
        logger.error("Stacktrace: ", e);
    }
}

@Override
protected <T> boolean publish(Channel<T> channel, T data) {
  MessageProducer producer = getProducerForChannel(channel);
  try {
    byte[] serializedObject = getSerializer().serialize(data);

    BytesMessage bm = messageCreationSession.createBytesMessage();
    bm.writeBytes(serializedObject);

    producer.send(bm);
  } catch (JMSException e) {
    logger.error("JMS message publishing failed", e);
    return false;
  }

  return true;
}

@Override
protected Object extractPayload(Message message) throws JMSException {
    Object payload = extractMessage(message);
    if (message instanceof BytesMessage) {
        try {
            // In case the BytesMessage is going to be received as a user argument:
            // reset it, otherwise it would appear empty to such processing code...
            ((BytesMessage) message).reset();
        }
        catch (JMSException ex) {
            // Continue since the BytesMessage typically won't be used any further.
            logger.debug("Failed to reset BytesMessage after payload extraction", ex);
        }
    }
    return payload;
}

/**
 * Convert a BytesMessage to a Java Object with the specified type.
 * @param message the input message
 * @param targetJavaType the target type
 * @return the message converted to an object
 * @throws JMSException if thrown by JMS
 * @throws IOException in case of I/O errors
 */
protected Object convertFromBytesMessage(BytesMessage message, JavaType targetJavaType)
        throws JMSException, IOException {

    String encoding = this.encoding;
    if (this.encodingPropertyName != null && message.propertyExists(this.encodingPropertyName)) {
        encoding = message.getStringProperty(this.encodingPropertyName);
    }
    byte[] bytes = new byte[(int) message.getBodyLength()];
    message.readBytes(bytes);
    try {
        String body = new String(bytes, encoding);
        return this.objectMapper.readValue(body, targetJavaType);
    }
    catch (UnsupportedEncodingException ex) {
        throw new MessageConversionException("Cannot convert bytes to String", ex);
    }
}

@Test
public void testWithMessageContentsDelegateForBytesMessage() throws Exception {
    BytesMessage bytesMessage = mock(BytesMessage.class);
    // BytesMessage contents must be unwrapped...
    given(bytesMessage.getBodyLength()).willReturn(new Long(TEXT.getBytes().length));
    given(bytesMessage.readBytes(any(byte[].class))).willAnswer(new Answer<Integer>() {
        @Override
        public Integer answer(InvocationOnMock invocation) throws Throwable {
            byte[] bytes = (byte[]) invocation.getArguments()[0];
            ByteArrayInputStream inputStream = new ByteArrayInputStream(TEXT.getBytes());
            return inputStream.read(bytes);
        }
    });

    MessageContentsDelegate delegate = mock(MessageContentsDelegate.class);

    MessageListenerAdapter adapter = new MessageListenerAdapter(delegate);
    adapter.onMessage(bytesMessage);

    verify(delegate).handleMessage(TEXT.getBytes());
}

@Test
public void fromBytesMessage() throws Exception {
    BytesMessage bytesMessageMock = mock(BytesMessage.class);
    Map<String, String> unmarshalled = Collections.singletonMap("foo", "bar");

    byte[] bytes = "{\"foo\":\"bar\"}".getBytes();
    final ByteArrayInputStream byteStream = new ByteArrayInputStream(bytes);

    given(bytesMessageMock.getStringProperty("__typeid__")).willReturn(Object.class.getName());
    given(bytesMessageMock.propertyExists("__encoding__")).willReturn(false);
    given(bytesMessageMock.getBodyLength()).willReturn(new Long(bytes.length));
    given(bytesMessageMock.readBytes(any(byte[].class))).willAnswer(
            new Answer<Integer>() {
                @Override
                public Integer answer(InvocationOnMock invocation) throws Throwable {
                    return byteStream.read((byte[]) invocation.getArguments()[0]);
                }
            });

    Object result = converter.fromMessage(bytesMessageMock);
    assertEquals("Invalid result", result, unmarshalled);
}

@Test
public void testByteArrayConversion() throws JMSException {
    Session session = mock(Session.class);
    BytesMessage message = mock(BytesMessage.class);

    byte[] content = "test".getBytes();
    final ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(content);

    given(session.createBytesMessage()).willReturn(message);
    given(message.getBodyLength()).willReturn((long) content.length);
    given(message.readBytes(any(byte[].class))).willAnswer(new Answer<Integer>() {
        @Override
        public Integer answer(InvocationOnMock invocation) throws Throwable {
            return byteArrayInputStream.read((byte[]) invocation.getArguments()[0]);
        }
    });

    SimpleMessageConverter converter = new SimpleMessageConverter();
    Message msg = converter.toMessage(content, session);
    assertEquals(content.length, ((byte[]) converter.fromMessage(msg)).length);

    verify(message).writeBytes(content);
}

private String getPayload(Message message) throws Exception {
    String payload = null;

    if (message instanceof TextMessage) {
        payload = ((TextMessage) message).getText();
    } 
    else if(message instanceof BytesMessage) {
        BytesMessage bMessage = (BytesMessage) message;
        int payloadLength = (int)bMessage.getBodyLength();
        byte payloadBytes[] = new byte[payloadLength];
        bMessage.readBytes(payloadBytes);
        payload = new String(payloadBytes);
    }
    else {
        log.warn("Message not recognized as a TextMessage or BytesMessage.  It is of type: "+message.getClass().toString());
        payload = message.toString();
    }
    return payload;
}

/**
 * Called when new data arrives to the sink, and forwards it to RMQ.
 *
 * @param value
 *            The incoming data
 */
@Override
public void invoke(IN value) {
    try {
        byte[] bytes = serializationSchema.serialize(value);
        BytesMessage message = session.createBytesMessage();
        message.writeBytes(bytes);
        producer.send(message);
    } catch (JMSException e) {
        if (logFailuresOnly) {
            LOG.error("Failed to send message to ActiveMQ", e);
        } else {
            throw new RuntimeException("Failed to send message to ActiveMQ", e);
        }
    }
}

@Override
public void run(SourceContext<OUT> ctx) throws Exception {
    while (runningChecker.isRunning()) {
        exceptionListener.checkErroneous();

        Message message = consumer.receive(1000);
        if (! (message instanceof BytesMessage)) {
            LOG.warn("Active MQ source received non bytes message: {}", message);
            return;
        }
        BytesMessage bytesMessage = (BytesMessage) message;
        byte[] bytes = new byte[(int) bytesMessage.getBodyLength()];
        bytesMessage.readBytes(bytes);
        OUT value = deserializationSchema.deserialize(bytes);
        synchronized (ctx.getCheckpointLock()) {
            ctx.collect(value);
            if (!autoAck) {
                addId(bytesMessage.getJMSMessageID());
                unacknowledgedMessages.put(bytesMessage.getJMSMessageID(), bytesMessage);
            }
        }
    }
}

@Before
public void before() throws Exception {
    connectionFactory = mock(ActiveMQConnectionFactory.class);
    producer = mock(MessageProducer.class);
    session = mock(Session.class);
    connection = mock(Connection.class);
    destination = mock(Destination.class);
    message = mock(BytesMessage.class);

    when(connectionFactory.createConnection()).thenReturn(connection);
    when(connection.createSession(anyBoolean(), anyInt())).thenReturn(session);
    when(session.createProducer(null)).thenReturn(producer);
    when(session.createBytesMessage()).thenReturn(message);
    serializationSchema = new SimpleStringSchema();

    AMQSinkConfig<String> config = new AMQSinkConfig.AMQSinkConfigBuilder<String>()
        .setConnectionFactory(connectionFactory)
        .setDestinationName(DESTINATION_NAME)
        .setSerializationSchema(serializationSchema)
        .build();
    amqSink = new AMQSink<>(config);
    amqSink.open(new Configuration());
}

@Override
public void onMessage(Message message) {
    try {
        if (message instanceof BytesMessage) {
            Destination source = message.getJMSDestination();
            BytesMessage bytesMsg = (BytesMessage) message;

            byte[] payload = new byte[(int) bytesMsg.getBodyLength()];
            bytesMsg.readBytes(payload);

            listeners.forEach(listener -> {
                listener.onMessage(source.toString(), payload);
            });
        } else {
            LOG.debug("Received message type we don't yet handle: {}", message);
        }

        // TODO - Handle other message types.

    } catch (Exception ex) {
        LOG.error("Error delivering incoming message to listeners: {}", ex.getMessage());
        LOG.trace("Error detail", ex);
    }
}

@Test
public void testConsumeBytesMessage() throws Exception {
    endpoint.expectedMessageCount(1);

    jmsTemplate.setPubSubDomain(false);
    jmsTemplate.send("test.bytes", new MessageCreator() {
        public Message createMessage(Session session) throws JMSException {
            BytesMessage bytesMessage = session.createBytesMessage();
            bytesMessage.writeByte((byte) 1);
            bytesMessage.writeByte((byte) 2);
            bytesMessage.writeByte((byte) 3);
            return bytesMessage;
        }
    });

    endpoint.assertIsSatisfied();
    assertCorrectBytesReceived();
}

@Override
public void onMessage(final Message message, final Session session) throws JMSException {
  final Destination replyTo = message.getJMSReplyTo();
  if (replyTo == null) {
    throw new IllegalArgumentException("No JMSReplyTo destination set.");
  }
  final byte[] requestBytes = JmsByteArrayHelper.extractBytes(message);
  s_logger.debug("Dispatching request {} of size {} to underlying", message.getJMSMessageID(), requestBytes.length);
  final byte[] responseBytes = getUnderlying().requestReceived(requestBytes);
  s_logger.debug("Returning response of size {} to {}", responseBytes.length, replyTo);
  final MessageProducer mp = session.createProducer(replyTo);
  try {
    final BytesMessage bytesMessage = session.createBytesMessage();
    bytesMessage.writeBytes(responseBytes);
    bytesMessage.setJMSCorrelationID(message.getJMSMessageID());
    mp.send(bytesMessage);
  } finally {
    mp.close();
  }
}

private void send() {
  DistributionSpecification distributionSpec = getDistributor().getDistributionSpec();

  LiveDataValueUpdateBean liveDataValueUpdateBean = new LiveDataValueUpdateBean(
      _lastSequenceNumber, 
      distributionSpec.getFullyQualifiedLiveDataSpecification(), 
      _cumulativeDelta.getLastKnownValues());
  s_logger.debug("{}: Sending Live Data update {}", this, liveDataValueUpdateBean);

  FudgeMsg fudgeMsg = LiveDataValueUpdateBeanFudgeBuilder.toFudgeMsg(new FudgeSerializer(_fudgeContext), liveDataValueUpdateBean);
  String destinationName = distributionSpec.getJmsTopic();
  final byte[] bytes = _fudgeContext.toByteArray(fudgeMsg);

  _jmsConnector.getJmsTemplateTopic().send(destinationName, new MessageCreator() {
    @Override
    public Message createMessage(Session session) throws JMSException {
      // TODO kirk 2009-10-30 -- We want to put stuff in the properties as well I think.
      BytesMessage bytesMessage = session.createBytesMessage();
      bytesMessage.writeBytes(bytes);
      return bytesMessage;
    }
  });

  _cumulativeDelta.clear();
}

@Override
@SuppressWarnings("rawtypes")
public long getCommittedWindowId(String appId, int operatorId)
{
  logger.debug("Getting committed windowId appId {} operatorId {}", appId, operatorId);

  try {
    beginTransaction();
    BytesMessage message = (BytesMessage)consumer.receive();
    logger.debug("Retrieved committed window messageId: {}, messageAppOperatorIdProp: {}", message.getJMSMessageID(),
        message.getStringProperty(APP_OPERATOR_ID));
    long windowId = message.readLong();

    writeWindowId(appId, operatorId, windowId);
    commitTransaction();
    logger.debug("metaQueueName: " + metaQueueName);
    logger.debug("Retrieved windowId {}", windowId);
    return windowId;
  } catch (JMSException ex) {
    throw new RuntimeException(ex);
  }
}

@Override
protected void prepareMessage(final Message m) throws JMSException {
   super.prepareMessage(m);

   BytesMessage bm = (BytesMessage) m;

   bm.writeBoolean(true);
   bm.writeByte((byte) 3);
   bm.writeBytes(new byte[]{(byte) 4, (byte) 5, (byte) 6});
   bm.writeChar((char) 7);
   bm.writeDouble(8.0);
   bm.writeFloat(9.0f);
   bm.writeInt(10);
   bm.writeLong(11L);
   bm.writeShort((short) 12);
   bm.writeUTF("this is an UTF String");
   bm.reset();
}

public void testNoSlowConsumerAdvisory() throws Exception {
   Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   Queue queue = s.createQueue(getClass().getName());
   MessageConsumer consumer = s.createConsumer(queue);
   consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(Message message) {
      }
   });
   Topic advisoryTopic = AdvisorySupport.getSlowConsumerAdvisoryTopic((ActiveMQDestination) queue);
   s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   MessageConsumer advisoryConsumer = s.createConsumer(advisoryTopic);
   // start throwing messages at the consumer
   MessageProducer producer = s.createProducer(queue);
   for (int i = 0; i < MESSAGE_COUNT; i++) {
      BytesMessage m = s.createBytesMessage();
      m.writeBytes(new byte[1024]);
      producer.send(m);
   }
   Message msg = advisoryConsumer.receive(1000);
   assertNull(msg);
}

@Override
protected void assertEquivalent(final Message m, final int mode, final boolean redelivered) throws JMSException {
   super.assertEquivalent(m, mode, redelivered);

   BytesMessage bm = (BytesMessage) m;

   ProxyAssertSupport.assertEquals(true, bm.readBoolean());
   ProxyAssertSupport.assertEquals((byte) 3, bm.readByte());
   byte[] bytes = new byte[3];
   bm.readBytes(bytes);
   ProxyAssertSupport.assertEquals((byte) 4, bytes[0]);
   ProxyAssertSupport.assertEquals((byte) 5, bytes[1]);
   ProxyAssertSupport.assertEquals((byte) 6, bytes[2]);
   ProxyAssertSupport.assertEquals((char) 7, bm.readChar());
   ProxyAssertSupport.assertEquals(new Double(8.0), new Double(bm.readDouble()));
   ProxyAssertSupport.assertEquals(new Float(9.0), new Float(bm.readFloat()));
   ProxyAssertSupport.assertEquals(10, bm.readInt());
   ProxyAssertSupport.assertEquals(11L, bm.readLong());
   ProxyAssertSupport.assertEquals((short) 12, bm.readShort());
   ProxyAssertSupport.assertEquals("this is an UTF String", bm.readUTF());
}