WMQConsumer extends ScheduledPollConsumer implements SuspendableService { @Override protected int poll() throws Exception { Exchange exchange = getEndpoint().createExchange(); Message in = exchange.getIn(); WMQComponent component = (WMQComponent) getEndpoint().getComponent(); if (mqQueueManager == null) { mqQueueManager = component.getQueueManager(getEndpoint().getQueueManagerName(), getEndpoint().getQueueManagerHostname(), getEndpoint().getQueueManagerPort(), getEndpoint().getQueueManagerChannel(), getEndpoint().getQueueManagerUserID(), getEndpoint().getQueueManagerPassword(), getEndpoint().getQueueManagerCCSID()); } MQDestination destination = null; try { LOGGER.debug("Consuming from {}", getEndpoint().getDestinationName()); if (getEndpoint().getDestinationName().startsWith("topic:")) { String destinationName = getEndpoint().getDestinationName().substring("topic:".length()); int options = CMQC.MQSO_CREATE | CMQC.MQSO_RESUME | CMQC.MQSO_DURABLE | CMQC.MQSO_FAIL_IF_QUIESCING; destination = mqQueueManager.accessTopic(destinationName, null, options, null, null); } else { String destinationName = getEndpoint().getDestinationName(); if (destinationName.startsWith("queue:")) { destinationName = destinationName.substring("queue:".length()); } destination = mqQueueManager.accessQueue(destinationName, MQConstants.MQOO_INPUT_AS_Q_DEF, null, null, null); } MQMessage message = new MQMessage(); MQGetMessageOptions options = new MQGetMessageOptions(); options.options = MQConstants.MQGMO_WAIT + MQConstants.MQGMO_PROPERTIES_COMPATIBILITY + MQConstants.MQGMO_ALL_SEGMENTS_AVAILABLE + MQConstants.MQGMO_COMPLETE_MSG + MQConstants.MQGMO_ALL_MSGS_AVAILABLE; options.waitInterval = MQConstants.MQWI_UNLIMITED; LOGGER.info("Waiting for message ..."); destination.get(message, options); LOGGER.info("Message consumed"); LOGGER.info("Dealing with MQMD headers"); LOGGER.info("\tmq.mqmd.format: {}", message.format); in.setHeader("mq.mqmd.format", message.format); LOGGER.info("\tmq.mqmd.charset: {}", message.characterSet); in.setHeader("mq.mqmd.charset", message.characterSet); LOGGER.info("\tmq.mqmd.expiry: {}", message.expiry); in.setHeader("mq.mqmd.expiry", message.expiry); LOGGER.info("\tmq.mqmd.put.appl.name: {}", message.putApplicationName); in.setHeader("mq.mqmd.put.appl.name", message.putApplicationName); LOGGER.info("\tmq.mqmd.group.id: {}", message.groupId); in.setHeader("mq.mqmd.group.id", message.groupId); LOGGER.info("\tmq.mqmd.msg.seq.number: {}", message.messageSequenceNumber); in.setHeader("mq.mqmd.msg.seq.number", message.messageSequenceNumber); LOGGER.info("\tmq.mqmd.msg.accounting.token: {}", message.accountingToken); in.setHeader("mq.mqmd.msg.accounting.token", message.accountingToken); LOGGER.info("\tmq.mqmd.correl.id: {}", message.correlationId); in.setHeader("mq.mqmd.correl.id", message.correlationId); LOGGER.info("\tmq.mqmd.replyto.q: {}", message.replyToQueueName); in.setHeader("mq.mqmd.replyto.q", message.replyToQueueName); LOGGER.info("\tmq.mqmd.replyto.q.mgr: {}", message.replyToQueueManagerName); in.setHeader("mq.mqmd.replyto.q.mgr", message.replyToQueueManagerName); LOGGER.info("\tmq.mqmd.putdatetime: {}", message.putDateTime); in.setHeader("mq.mqmd.putdatetime", message.putDateTime); LOGGER.info("\tmq.mqmd.user.id: {}", message.userId); in.setHeader("mq.mqmd.user.id", message.userId); LOGGER.info("\tmq.mqmd.type", message.messageType); in.setHeader("mq.mqmd.type", message.messageType); LOGGER.info("\tmq.mqmd.priority", message.priority); in.setHeader("mq.mqmd.priority", message.priority); LOGGER.info("\tmq.mqmd.persistence", message.persistence); in.setHeader("mq.mqmd.persistence", message.persistence); LOGGER.info("\tmq.mqmd.backout.count", message.backoutCount); in.setHeader("mq.mqmd.backout.count", message.backoutCount); LOGGER.info("\tmq.mqmd.report", message.report); in.setHeader("mq.mqmd.report", message.report); LOGGER.info("\tmq.mqmd.feedback", message.feedback); in.setHeader("mq.mqmd.feedback", message.feedback); LOGGER.info("\tmq.mqmd.original.length", message.originalLength); in.setHeader("mq.mqmd.original.length", message.originalLength); LOGGER.info("\tmq.mqmd.appl.type", message.putApplicationType); in.setHeader("mq.mqmd.appl.type", message.putApplicationType); LOGGER.info("\tmq.mqmd.appl.id.data", message.applicationIdData); in.setHeader("mq.mqmd.appl.id.data", message.applicationIdData); LOGGER.info("\tmq.mqmd.appl.origin.data", message.applicationOriginData); in.setHeader("mq.mqmd.appl.origin.data", message.applicationOriginData); LOGGER.info("\tmq.mqmd.id", message.messageId); in.setHeader("mq.mqmd.id", message.messageId); LOGGER.info("\tmq.mqmd.offset", message.offset); in.setHeader("mq.mqmd.offset", message.offset); LOGGER.info("\tmq.mqmd.flags", message.messageFlags); in.setHeader("mq.mqmd.flags", message.messageFlags); LOGGER.info("\tmq.mqmd.length.total", message.getTotalMessageLength()); in.setHeader("mq.mqmd.length.total", message.getTotalMessageLength()); LOGGER.info("\tmq.mqmd.length.data", message.getDataLength()); in.setHeader("mq.mqmd.length.data", message.getDataLength()); LOGGER.info("\tmq.mqmd.encoding", message.encoding); in.setHeader("mq.mqmd.encoding", message.encoding); MQHeaderList headerList = this.mqHeaderListFactory.apply(message); int index = headerList.indexOf("MQRFH2"); if (index >= 0) { LOGGER.info("MQRFH2 header detected (index " + index + ")"); MQRFH2 rfh = (MQRFH2) headerList.get(index); LOGGER.info("\tmq.rfh2.format: " + rfh.getFormat()); in.setHeader("mq.rfh2.format", rfh.getFormat()); LOGGER.info("\tmq.rfh2.struct.id: " + rfh.getStrucId()); in.setHeader("mq.rfh2.struct.id", rfh.getStrucId()); LOGGER.info("\tmq.rfh2.encoding: " + rfh.getEncoding()); in.setHeader("mq.rfh2.encoding", rfh.getEncoding()); LOGGER.info("\tmq.rfh2.coded.charset.id: " + rfh.getCodedCharSetId()); in.setHeader("mq.rfh2.coded.charset.id", rfh.getCodedCharSetId()); LOGGER.info("\tmq.rfh2.flags: " + rfh.getFlags()); in.setHeader("mq.rfh2.flags", rfh.getFlags()); LOGGER.info("\tmq.rfh2.version: " + rfh.getVersion()); in.setHeader("mq.rfh2.version", rfh.getVersion()); MQRFH2.Element[] folders = rfh.getFolders(); for (MQRFH2.Element folder : folders) { LOGGER.info("mq.rfh2.folder " + folder.getName() + ": " + folder.toXML()); in.setHeader("mq.rfh2.folder." + folder.getName(), folder.toXML()); } } LOGGER.info("Reading body"); byte[] buffer = new byte[message.getDataLength()]; message.readFully(buffer); saveBody(in, buffer); getProcessor().process(exchange); } catch (Exception e) { LOGGER.error("Error occurred while consuming message, waiting for {}", getEndpoint().getDelayOnException(), e); Thread.sleep(getEndpoint().getDelayOnException()); exchange.setException(e); } finally { if (destination != null) destination.close(); } if (exchange.getException() != null) { getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException()); } return 1; } WMQConsumer(WMQEndpoint endpoint, Processor processor);  WMQConsumer(WMQEndpoint endpoint, Processor processor,
                       Function<MQMessage, MQHeaderList> mqHeaderListFactory); @Override WMQEndpoint getEndpoint();  }
@Test public void pool_noOutputConfiguration_setBodyAsUTF8String() throws Exception { this.consumer.poll(); final String value = (String) bodyCaptor.getValue(); CharsetDetector charsetDetector = new CharsetDetector(); charsetDetector.setText(value.getBytes()); assertThat(charsetDetector.detect().getName()).isEqualTo("UTF-8"); }
@Test public void pool_bodyTypeAsBytes_setBodyAsByteBuffer() throws Exception { when(wmqEndpoint.getBodyType()).thenReturn("bytes"); this.consumer.poll(); assertThat(bodyCaptor.getValue()).isInstanceOf(ByteBuffer.class); assertThat(bodyClassCaptor.getValue()).isEqualTo(ByteBuffer.class); }