BytesMessageTranslator extends MessageTypeTranslatorImp { public Message translate(AdaptrisMessage msg) throws JMSException { BytesMessage jmsMsg = session.createBytesMessage(); if (msg.getSize() > streamThreshold()) { try (InputStream in = msg.getInputStream(); BytesMessageOutputStream out = new BytesMessageOutputStream(jmsMsg)) { StreamUtil.copyAndClose(in, out); } catch (IOException e) { throw JmsUtils.wrapJMSException(e); } } else { jmsMsg.writeBytes(msg.getPayload()); } return helper.moveMetadata(msg, jmsMsg); }  Message translate(AdaptrisMessage msg); AdaptrisMessage translate(Message msg);  }
@Override @Test public void testMoveMetadataJmsMessageToAdaptrisMessage() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); BytesMessage jmsMsg = createMessage(session); addProperties(jmsMsg); start(trans, session); jmsMsg.reset(); AdaptrisMessage msg = trans.translate(jmsMsg); assertMetadata(msg); } finally { stop(trans); broker.destroy(); } }
@Override @Test public void testMoveJmsHeadersJmsMessageToAdaptrisMessage() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); BytesMessage jmsMsg = createMessage(session); jmsMsg.setJMSCorrelationID("ABC"); jmsMsg.setJMSDeliveryMode(1); jmsMsg.setJMSPriority(4); addProperties(jmsMsg); long timestamp = System.currentTimeMillis(); jmsMsg.setJMSTimestamp(timestamp); trans.setMoveJmsHeaders(true); start(trans, session); jmsMsg.reset(); AdaptrisMessage msg = trans.translate(jmsMsg); assertMetadata(msg); assertEquals("ABC", msg.getMetadataValue(JmsConstants.JMS_CORRELATION_ID)); assertEquals("1", msg.getMetadataValue(JmsConstants.JMS_DELIVERY_MODE)); assertEquals("4", msg.getMetadataValue(JmsConstants.JMS_PRIORITY)); assertEquals(String.valueOf(timestamp), msg.getMetadataValue(JmsConstants.JMS_TIMESTAMP)); } finally { stop(trans); broker.destroy(); } }
@Override @Test public void testMoveMetadataJmsMessageToAdaptrisMessage_RemoveAllFilter() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); trans.setMetadataFilter(new RemoveAllMetadataFilter()); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); BytesMessage jmsMsg = createMessage(session); addProperties(jmsMsg); start(trans, session); jmsMsg.reset(); AdaptrisMessage msg = trans.translate(jmsMsg); assertFalse(msg.containsKey(INTEGER_METADATA)); assertFalse(msg.containsKey(STRING_METADATA)); assertFalse(msg.containsKey(BOOLEAN_METADATA)); } finally { stop(trans); broker.destroy(); } }
@Override @Test public void testMoveMetadata_JmsMessageToAdaptrisMessage_WithFilter() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); RegexMetadataFilter regexp = new RegexMetadataFilter(); regexp.addExcludePattern("IntegerMetadataKey"); trans.setMetadataFilter(regexp); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); BytesMessage jmsMsg = createMessage(session); addProperties(jmsMsg); start(trans, session); jmsMsg.reset(); AdaptrisMessage msg = trans.translate(jmsMsg); assertMetadata(msg, new MetadataElement(STRING_METADATA, STRING_VALUE)); assertMetadata(msg, new MetadataElement(BOOLEAN_METADATA, BOOLEAN_VALUE)); assertFalse(msg.containsKey(INTEGER_METADATA)); } finally { stop(trans); broker.destroy(); } }
@Test public void testBytesMessageToAdaptrisMessage() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); BytesMessage jmsMsg = session.createBytesMessage(); jmsMsg.writeBytes(BYTES); addProperties(jmsMsg); start(trans, session); jmsMsg.reset(); AdaptrisMessage msg = trans.translate(jmsMsg); assertMetadata(msg); assertTrue("Payload is not equal", Arrays.equals(BYTES, msg.getPayload())); } finally { stop(trans); broker.destroy(); } }
@Test public void testBytesMessageToAdaptrisMessage_Alt() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); BytesMessage jmsMsg = session.createBytesMessage(); jmsMsg.writeBytes(BYTES_ALT); addProperties(jmsMsg); start(trans, session); jmsMsg.reset(); AdaptrisMessage msg = trans.translate(jmsMsg); assertMetadata(msg); assertTrue(Arrays.equals(BYTES_ALT, msg.getPayload())); } finally { stop(trans); broker.destroy(); } }
@Test public void testBytesMessageToAdaptrisMessage_StreamFailure() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); BytesMessage jmsMsg = createMessage(session); jmsMsg.writeBytes(TEXT.getBytes()); start(trans, session); trans.registerMessageFactory(new DefectiveMessageFactory()); jmsMsg.reset(); try { AdaptrisMessage msg = trans.translate(jmsMsg); fail(); } catch (JMSException expected) { } } finally { stop(trans); broker.destroy(); } }
@Test public void testBytesMessageToAdaptrisMessage_StreamFailure_CheckedJMSException() throws Exception { BytesMessage jmsMsg = Mockito.mock(BytesMessage.class); Session session = Mockito.mock(Session.class); doThrow(new JMSException(testName.getMethodName())).when(jmsMsg).readByte(); when(session.createBytesMessage()).thenReturn(jmsMsg); BytesMessageTranslator trans = new BytesMessageTranslator() { @Override long streamThreshold() { return TEXT.length() - 1; } }; try { start(trans, session); try { AdaptrisMessage msg = trans.translate(jmsMsg); fail(); } catch (JMSException expected) { } } finally { stop(trans); } }
@Test public void testAdaptrisMessageToBytesMessage() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); MessageTypeTranslatorImp trans = createTranslator(); try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); start(trans, session); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(TEXT); addMetadata(msg); Message jmsMsg = trans.translate(msg); assertTrue("jmsMsg instanceof BytesMessage", jmsMsg instanceof BytesMessage); ((BytesMessage) jmsMsg).reset(); assertEquals(TEXT, new String(getBytes((BytesMessage) jmsMsg))); assertJmsProperties(jmsMsg); } finally { stop(trans); broker.destroy(); } }
@Test public void testAdaptrisMessageToBytesMessage_StreamFailure() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); BytesMessageTranslator trans = new BytesMessageTranslator() { @Override long streamThreshold() { return TEXT.length() - 1; } }; try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); start(trans, session); AdaptrisMessage msg = new DefectiveMessageFactory().newMessage(TEXT); addMetadata(msg); try { Message jmsMsg = trans.translate(msg); fail(); } catch (JMSException expected) { } } finally { stop(trans); broker.destroy(); } }
@Test public void testAdaptrisMessageToBytesMessage_StreamFailure_CheckedJMSException() throws Exception { BytesMessage jmsMsg = Mockito.mock(BytesMessage.class); Session session = Mockito.mock(Session.class); doThrow(new JMSException(testName.getMethodName())).when(jmsMsg).writeByte(anyByte()); when(session.createBytesMessage()).thenReturn(jmsMsg); BytesMessageTranslator trans = new BytesMessageTranslator() { @Override long streamThreshold() { return TEXT.length() - 1; } }; try { start(trans, session); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(TEXT); try { Message m = trans.translate(msg); fail(); } catch (JMSException expected) { } } finally { stop(trans); } }
@Test public void testAdaptrisMessageToBytesMessage_ExceedsThreshold() throws Exception { EmbeddedActiveMq broker = new EmbeddedActiveMq(); BytesMessageTranslator trans = new BytesMessageTranslator() { @Override long streamThreshold() { return TEXT.length() -1; } }; try { broker.start(); Session session = broker.createConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE); start(trans, session); AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage(TEXT); addMetadata(msg); Message jmsMsg = trans.translate(msg); assertTrue("jmsMsg instanceof BytesMessage", jmsMsg instanceof BytesMessage); ((BytesMessage) jmsMsg).reset(); assertEquals(TEXT, new String(getBytes((BytesMessage) jmsMsg))); assertJmsProperties(jmsMsg); } finally { stop(trans); broker.destroy(); } }