class JmsConsumer extends AbstractProcessor {

    public static JmsProcessingSummary map2FlowFile(final ProcessContext context, final ProcessSession session, final Message message, final boolean addAttributes, ComponentLog logger)
            throws Exception {

        // Currently not very useful, because always one Message == one FlowFile
        final AtomicInteger msgsThisFlowFile = new AtomicInteger(1);

        FlowFile flowFile = session.create();
        try {
            // MapMessage is exception, add only name-value pairs to FlowFile attributes
            if (message instanceof MapMessage) {
                MapMessage mapMessage = (MapMessage) message;
                flowFile = session.putAllAttributes(flowFile, createMapMessageValues(mapMessage));
            } else { // all other message types, write Message body to FlowFile content
                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;
        }
    }

    public  JmsConsumer();

}