@Test public void testClearQueue() throws JMSException, EtiqetException { Queue queue = mock(Queue.class); Session session = mock(Session.class); when(session.createQueue(anyString())).thenReturn(queue); MessageConsumer consumer = mock(MessageConsumer.class); when(session.createConsumer(queue)).thenReturn(consumer); javax.jms.Message msg1 = mock(javax.jms.Message.class); javax.jms.Message msg2 = mock(javax.jms.Message.class); javax.jms.Message msg3 = mock(javax.jms.Message.class); javax.jms.Message msg4 = mock(javax.jms.Message.class); javax.jms.Message msg5 = mock(javax.jms.Message.class); when(consumer.receive(anyLong())).thenReturn(msg1, msg2, msg3, msg4, msg5, null); transport.setSession(session); transport.clearQueue("testQueue"); verify(msg1, times(1)).acknowledge(); verify(msg2, times(1)).acknowledge(); verify(msg3, times(1)).acknowledge(); verify(msg4, times(1)).acknowledge(); verify(msg5, times(1)).acknowledge(); }
@Override public void clearQueue(String queueName) throws EtiqetException { try { Queue queue = session.createQueue(queueName); int i = 0; MessageConsumer consumer = session.createConsumer(queue); Message consumed; while ((consumed = consumer.receive(5000)) != null) { i++; consumed.acknowledge(); logger.debug("Consumed message {} from queue {}", consumed, queueName); } consumer.close(); logger.info("Removed {} messages from queue {}", i, queueName); } catch (JMSException e) { throw new EtiqetException("JMS Exception thrown trying to clear queue " + queueName, e); } }
JmsTransport implements BrokerTransport { @Override public void clearQueue(String queueName) throws EtiqetException { try { Queue queue = session.createQueue(queueName); int i = 0; MessageConsumer consumer = session.createConsumer(queue); Message consumed; while ((consumed = consumer.receive(5000)) != null) { i++; consumed.acknowledge(); logger.debug("Consumed message {} from queue {}", consumed, queueName); } consumer.close(); logger.info("Removed {} messages from queue {}", i, queueName); } catch (JMSException e) { throw new EtiqetException("JMS Exception thrown trying to clear queue " + queueName, e); } } }
JmsTransport implements BrokerTransport { @Override public void clearQueue(String queueName) throws EtiqetException { try { Queue queue = session.createQueue(queueName); int i = 0; MessageConsumer consumer = session.createConsumer(queue); Message consumed; while ((consumed = consumer.receive(5000)) != null) { i++; consumed.acknowledge(); logger.debug("Consumed message {} from queue {}", consumed, queueName); } consumer.close(); logger.info("Removed {} messages from queue {}", i, queueName); } catch (JMSException e) { throw new EtiqetException("JMS Exception thrown trying to clear queue " + queueName, e); } }  }
JmsTransport implements BrokerTransport { @Override public void clearQueue(String queueName) throws EtiqetException { try { Queue queue = session.createQueue(queueName); int i = 0; MessageConsumer consumer = session.createConsumer(queue); Message consumed; while ((consumed = consumer.receive(5000)) != null) { i++; consumed.acknowledge(); logger.debug("Consumed message {} from queue {}", consumed, queueName); } consumer.close(); logger.info("Removed {} messages from queue {}", i, queueName); } catch (JMSException e) { throw new EtiqetException("JMS Exception thrown trying to clear queue " + queueName, e); } }  @Override void init(String configPath); @Override void start(); @Override void stop(); @Override void send(Cdr msg); @Override void send(Cdr msg, String topicName); @Override boolean isLoggedOn(); @Override String getDefaultSessionId(); @Override void setTransportDelegate(TransportDelegate<String, Cdr> transDel); @Override @SuppressWarnings("unchecked") void setCodec(Codec c); @Override Codec getCodec(); @Override void setDelegate(ClientDelegate delegate); @Override ClientDelegate getDelegate(); @Override void subscribeToTopic(Optional<String> topicName, final Consumer<Cdr> cdrListener); @Override void subscribeToQueue(String queueName, Consumer<Cdr> cdrListener); @Override void sendToTopic(Cdr cdr, Optional<String> maybeTopicName); @Override void sendToQueue(Cdr cdr, String queueName); @Override void clearQueue(String queueName); @Override Cdr subscribeAndConsumeFromTopic(final Optional<String> topicName, final Duration timeout); @Override Cdr subscribeAndConsumeFromQueue(final String queueName, final Duration timeout); }
JmsTransport implements BrokerTransport { @Override public void clearQueue(String queueName) throws EtiqetException { try { Queue queue = session.createQueue(queueName); int i = 0; MessageConsumer consumer = session.createConsumer(queue); Message consumed; while ((consumed = consumer.receive(5000)) != null) { i++; consumed.acknowledge(); logger.debug("Consumed message {} from queue {}", consumed, queueName); } consumer.close(); logger.info("Removed {} messages from queue {}", i, queueName); } catch (JMSException e) { throw new EtiqetException("JMS Exception thrown trying to clear queue " + queueName, e); } }  @Override void init(String configPath); @Override void start(); @Override void stop(); @Override void send(Cdr msg); @Override void send(Cdr msg, String topicName); @Override boolean isLoggedOn(); @Override String getDefaultSessionId(); @Override void setTransportDelegate(TransportDelegate<String, Cdr> transDel); @Override @SuppressWarnings("unchecked") void setCodec(Codec c); @Override Codec getCodec(); @Override void setDelegate(ClientDelegate delegate); @Override ClientDelegate getDelegate(); @Override void subscribeToTopic(Optional<String> topicName, final Consumer<Cdr> cdrListener); @Override void subscribeToQueue(String queueName, Consumer<Cdr> cdrListener); @Override void sendToTopic(Cdr cdr, Optional<String> maybeTopicName); @Override void sendToQueue(Cdr cdr, String queueName); @Override void clearQueue(String queueName); @Override Cdr subscribeAndConsumeFromTopic(final Optional<String> topicName, final Duration timeout); @Override Cdr subscribeAndConsumeFromQueue(final String queueName, final Duration timeout);  }