StreamJmsP extends AbstractProcessor { @Override public boolean complete() { if (snapshotInProgress) { return false; } while (emitFromTraverser(pendingTraverser)) { try { Message t = consumer.receiveNoWait(); if (t == null) { pendingTraverser = eventTimeMapper.flatMapIdle(); break; } if (guarantee == EXACTLY_ONCE) { if (restoredIdsExpiration == Long.MAX_VALUE) { restoredIdsExpiration = System.nanoTime() + RESTORED_IDS_TTL; } else if (!restoredIds.isEmpty() && restoredIdsExpiration <= System.nanoTime()) { restoredIds = Collections.emptySet(); } Object msgId = messageIdFn.apply(t); if (msgId == null) { throw new JetException("Received a message without an ID. All messages must have an ID, " + "you can specify an extracting function using " + JmsSourceBuilder.class.getSimpleName() + ".messageIdFn()"); } seenIds.add(msgId); if (restoredIds.remove(msgId)) { logFine(getLogger(), "Redelivered message dropped: %s", t); continue; } } T projectedItem = projectionFn.apply(t); pendingTraverser = projectedItem != null ? eventTimeMapper.flatMapEvent(projectedItem, 0, handleJmsTimestamp(t)) : eventTimeMapper.flatMapIdle(); } catch (JMSException e) { throw sneakyThrow(e); } } return false; } StreamJmsP(Connection connection,
               FunctionEx<? super Session, ? extends MessageConsumer> consumerFn,
               FunctionEx<? super Message, ?> messageIdFn,
               FunctionEx<? super Message, ? extends T> projectionFn,
               EventTimePolicy<? super T> eventTimePolicy,
               ProcessingGuarantee guarantee
    ); @Override boolean isCooperative(); @Override boolean complete(); @Override boolean snapshotCommitPrepare(); @Override boolean snapshotCommitFinish(boolean success); @Override void close(); static final int PREFERRED_LOCAL_PARALLELISM; }
@Test public void when_queue() throws Exception { String queueName = randomString(); logger.info("using queue: " + queueName); String message1 = sendMessage(queueName, true); String message2 = sendMessage(queueName, true); initializeProcessor(queueName, true, null); Queue<Object> queue = outbox.queue(0); List<Object> actualOutput = new ArrayList<>(); assertTrueEventually(() -> { outbox.reset(); processor.complete(); Object item = queue.poll(); if (item != null) { actualOutput.add(item); } assertEquals(asList(message1, message2), actualOutput); }); }
@Test public void when_topic() throws Exception { String topicName = randomString(); logger.info("using topic: " + topicName); sendMessage(topicName, false); initializeProcessor(topicName, false, null); processor.complete(); sleepSeconds(1); String message2 = sendMessage(topicName, false); Queue<Object> queue = outbox.queue(0); assertTrueEventually(() -> { processor.complete(); assertEquals(message2, queue.poll()); }); }
@Test public void when_projectionToNull_then_filteredOut() throws Exception { String queueName = randomString(); logger.info("using queue: " + queueName); String message1 = sendMessage(queueName, true); String message2 = sendMessage(queueName, true); initializeProcessor(queueName, true, m -> { String msg = ((TextMessage) m).getText(); return msg.equals(message1) ? null : msg; }); Queue<Object> queue = outbox.queue(0); List<Object> actualOutput = new ArrayList<>(); assertTrueEventually(() -> { outbox.reset(); processor.complete(); Object item = queue.poll(); if (item != null) { actualOutput.add(item); } assertEquals(singletonList(message2), actualOutput); }); }