@Test public void testCleanupResources() throws JMSException, NoSuchFieldException, IllegalAccessException { final PutJMS putJMS = new PutJMS(); final TestRunner runnerPut = TestRunners.newTestRunner(putJMS); runnerPut.setProperty(JmsProperties.JMS_PROVIDER, TEST_PROVIDER); runnerPut.setProperty(JmsProperties.URL, TEST_URL); runnerPut.setProperty(JmsProperties.DESTINATION_TYPE, TEST_DEST_TYPE); runnerPut.setProperty(JmsProperties.DESTINATION_NAME, TEST_DEST_NAME + testQueueSuffix()); final Queue<WrappedMessageProducer> wrappedMessageProducerQueue = (Queue) spy(new LinkedBlockingQueue<>()); injectFieldValue(PutJMS.class, putJMS, "producerQueue", wrappedMessageProducerQueue); final WrappedMessageProducer wrappedProducer = JmsFactory.createMessageProducer(runnerPut.getProcessContext(), true); wrappedMessageProducerQueue.offer(wrappedProducer); assertNotNull(wrappedMessageProducerQueue.peek()); putJMS.cleanupResources(); assertNull(wrappedMessageProducerQueue.peek()); }
@OnStopped public void cleanupResources() { WrappedMessageProducer wrappedProducer = producerQueue.poll(); while (wrappedProducer != null) { wrappedProducer.close(getLogger()); wrappedProducer = producerQueue.poll(); } }
PutJMS extends AbstractProcessor { @OnStopped public void cleanupResources() { WrappedMessageProducer wrappedProducer = producerQueue.poll(); while (wrappedProducer != null) { wrappedProducer.close(getLogger()); wrappedProducer = producerQueue.poll(); } } }
PutJMS extends AbstractProcessor { @OnStopped public void cleanupResources() { WrappedMessageProducer wrappedProducer = producerQueue.poll(); while (wrappedProducer != null) { wrappedProducer.close(getLogger()); wrappedProducer = producerQueue.poll(); } } PutJMS(); }
PutJMS extends AbstractProcessor { @OnStopped public void cleanupResources() { WrappedMessageProducer wrappedProducer = producerQueue.poll(); while (wrappedProducer != null) { wrappedProducer.close(getLogger()); wrappedProducer = producerQueue.poll(); } } PutJMS(); @Override Set<Relationship> getRelationships(); @OnStopped void cleanupResources(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
PutJMS extends AbstractProcessor { @OnStopped public void cleanupResources() { WrappedMessageProducer wrappedProducer = producerQueue.poll(); while (wrappedProducer != null) { wrappedProducer.close(getLogger()); wrappedProducer = producerQueue.poll(); } } PutJMS(); @Override Set<Relationship> getRelationships(); @OnStopped void cleanupResources(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); static final Charset UTF8; static final int DEFAULT_MESSAGE_PRIORITY; static final Relationship REL_SUCCESS; static final Relationship REL_FAILURE; }