@Test public void testMap2FlowFileBytesMessage() throws Exception { TestRunner runner = TestRunners.newTestRunner(GetJMSQueue.class); BytesMessage bytesMessage = new ActiveMQBytesMessage(); String sourceString = "Apache NiFi is an easy to use, powerful, and reliable system to process and distribute data.!"; byte[] payload = sourceString.getBytes("UTF-8"); bytesMessage.writeBytes(payload); bytesMessage.reset(); ProcessContext context = runner.getProcessContext(); ProcessSession session = runner.getProcessSessionFactory().createSession(); ProcessorInitializationContext pic = new MockProcessorInitializationContext(runner.getProcessor(), (MockProcessContext) runner.getProcessContext()); JmsProcessingSummary summary = JmsConsumer.map2FlowFile(context, session, bytesMessage, true, pic.getLogger()); assertEquals("BytesMessage content length should equal to FlowFile content size", payload.length, summary.getLastFlowFile().getSize()); final byte[] buffer = new byte[payload.length]; runner.clearTransferState(); session.read(summary.getLastFlowFile(), new InputStreamCallback() { @Override public void process(InputStream in) throws IOException { StreamUtils.fillBuffer(in, buffer, false); } }); String contentString = new String(buffer, "UTF-8"); assertEquals("", sourceString, contentString); }
public static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger) throws Exception { final AtomicInteger msgsThisFlowFile = new AtomicInteger(1); FlowFile flowFile = session.create(); try { if (message instanceof MapMessage) { MapMessage mapMessage = (MapMessage) message; flowFile = session.putAllAttributes(flowFile, createMapMessageValues(mapMessage)); } else { flowFile = session.write(flowFile, new OutputStreamCallback() { @Override public void process(final OutputStream rawOut) throws IOException { try (final OutputStream out = new BufferedOutputStream(rawOut, 65536)) { final byte[] messageBody = JmsFactory.createByteArray(message); out.write(messageBody); } catch (final JMSException e) { throw new ProcessException("Failed to receive JMS Message due to " + e.getMessage(), e); } } }); } if (addAttributes) { flowFile = session.putAllAttributes(flowFile, JmsFactory.createAttributeMap(message)); } session.getProvenanceReporter().receive(flowFile, context.getProperty(URL).getValue()); session.transfer(flowFile, REL_SUCCESS); logger.info("Created {} from {} messages received from JMS Server and transferred to 'success'", new Object[]{flowFile, msgsThisFlowFile.get()}); return new JmsProcessingSummary(flowFile.getSize(), message, flowFile); } catch (Exception e) { session.remove(flowFile); throw e; } }
JmsConsumer extends AbstractProcessor { public static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger) throws Exception { final AtomicInteger msgsThisFlowFile = new AtomicInteger(1); FlowFile flowFile = session.create(); try { if (message instanceof MapMessage) { MapMessage mapMessage = (MapMessage) message; flowFile = session.putAllAttributes(flowFile, createMapMessageValues(mapMessage)); } else { flowFile = session.write(flowFile, new OutputStreamCallback() { @Override public void process(final OutputStream rawOut) throws IOException { try (final OutputStream out = new BufferedOutputStream(rawOut, 65536)) { final byte[] messageBody = JmsFactory.createByteArray(message); out.write(messageBody); } catch (final JMSException e) { throw new ProcessException("Failed to receive JMS Message due to " + e.getMessage(), e); } } }); } if (addAttributes) { flowFile = session.putAllAttributes(flowFile, JmsFactory.createAttributeMap(message)); } session.getProvenanceReporter().receive(flowFile, context.getProperty(URL).getValue()); session.transfer(flowFile, REL_SUCCESS); logger.info("Created {} from {} messages received from JMS Server and transferred to 'success'", new Object[]{flowFile, msgsThisFlowFile.get()}); return new JmsProcessingSummary(flowFile.getSize(), message, flowFile); } catch (Exception e) { session.remove(flowFile); throw e; } } }
JmsConsumer extends AbstractProcessor { public static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger) throws Exception { final AtomicInteger msgsThisFlowFile = new AtomicInteger(1); FlowFile flowFile = session.create(); try { if (message instanceof MapMessage) { MapMessage mapMessage = (MapMessage) message; flowFile = session.putAllAttributes(flowFile, createMapMessageValues(mapMessage)); } else { flowFile = session.write(flowFile, new OutputStreamCallback() { @Override public void process(final OutputStream rawOut) throws IOException { try (final OutputStream out = new BufferedOutputStream(rawOut, 65536)) { final byte[] messageBody = JmsFactory.createByteArray(message); out.write(messageBody); } catch (final JMSException e) { throw new ProcessException("Failed to receive JMS Message due to " + e.getMessage(), e); } } }); } if (addAttributes) { flowFile = session.putAllAttributes(flowFile, JmsFactory.createAttributeMap(message)); } session.getProvenanceReporter().receive(flowFile, context.getProperty(URL).getValue()); session.transfer(flowFile, REL_SUCCESS); logger.info("Created {} from {} messages received from JMS Server and transferred to 'success'", new Object[]{flowFile, msgsThisFlowFile.get()}); return new JmsProcessingSummary(flowFile.getSize(), message, flowFile); } catch (Exception e) { session.remove(flowFile); throw e; } } JmsConsumer(); }
JmsConsumer extends AbstractProcessor { public static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger) throws Exception { final AtomicInteger msgsThisFlowFile = new AtomicInteger(1); FlowFile flowFile = session.create(); try { if (message instanceof MapMessage) { MapMessage mapMessage = (MapMessage) message; flowFile = session.putAllAttributes(flowFile, createMapMessageValues(mapMessage)); } else { flowFile = session.write(flowFile, new OutputStreamCallback() { @Override public void process(final OutputStream rawOut) throws IOException { try (final OutputStream out = new BufferedOutputStream(rawOut, 65536)) { final byte[] messageBody = JmsFactory.createByteArray(message); out.write(messageBody); } catch (final JMSException e) { throw new ProcessException("Failed to receive JMS Message due to " + e.getMessage(), e); } } }); } if (addAttributes) { flowFile = session.putAllAttributes(flowFile, JmsFactory.createAttributeMap(message)); } session.getProvenanceReporter().receive(flowFile, context.getProperty(URL).getValue()); session.transfer(flowFile, REL_SUCCESS); logger.info("Created {} from {} messages received from JMS Server and transferred to 'success'", new Object[]{flowFile, msgsThisFlowFile.get()}); return new JmsProcessingSummary(flowFile.getSize(), message, flowFile); } catch (Exception e) { session.remove(flowFile); throw e; } } JmsConsumer(); @Override Set<Relationship> getRelationships(); void consume(final ProcessContext context, final ProcessSession session, final WrappedMessageConsumer wrappedConsumer); static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger); static Map<String, String> createMapMessageValues(final MapMessage mapMessage); }
JmsConsumer extends AbstractProcessor { public static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger) throws Exception { final AtomicInteger msgsThisFlowFile = new AtomicInteger(1); FlowFile flowFile = session.create(); try { if (message instanceof MapMessage) { MapMessage mapMessage = (MapMessage) message; flowFile = session.putAllAttributes(flowFile, createMapMessageValues(mapMessage)); } else { flowFile = session.write(flowFile, new OutputStreamCallback() { @Override public void process(final OutputStream rawOut) throws IOException { try (final OutputStream out = new BufferedOutputStream(rawOut, 65536)) { final byte[] messageBody = JmsFactory.createByteArray(message); out.write(messageBody); } catch (final JMSException e) { throw new ProcessException("Failed to receive JMS Message due to " + e.getMessage(), e); } } }); } if (addAttributes) { flowFile = session.putAllAttributes(flowFile, JmsFactory.createAttributeMap(message)); } session.getProvenanceReporter().receive(flowFile, context.getProperty(URL).getValue()); session.transfer(flowFile, REL_SUCCESS); logger.info("Created {} from {} messages received from JMS Server and transferred to 'success'", new Object[]{flowFile, msgsThisFlowFile.get()}); return new JmsProcessingSummary(flowFile.getSize(), message, flowFile); } catch (Exception e) { session.remove(flowFile); throw e; } } JmsConsumer(); @Override Set<Relationship> getRelationships(); void consume(final ProcessContext context, final ProcessSession session, final WrappedMessageConsumer wrappedConsumer); static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger); static Map<String, String> createMapMessageValues(final MapMessage mapMessage); static final String MAP_MESSAGE_PREFIX; static final Relationship REL_SUCCESS; }