@Test public void shouldConsumeByAttachingConsumerToQueue() throws IOException { RabbitMQMessageQueue<TestMsg> queue = new RabbitMQMessageQueue<>(channel, QUEUE_NAME, TestMsg.class, metrics); queue.consume(msg -> null); verify(channel).basicConsume(Matchers.eq(QUEUE_NAME), Matchers.eq(false), Matchers.any(FunctionConsumer.class)); }
@Override public Closeable consume(Function<T, Void> processor) { Consumer consumer = new FunctionConsumer<>(channel, processor, type, name, metrics); try { String tag = channel.basicConsume(name, false, consumer); log.info("Set up consumer '{}' for queue '{}'.", tag, name); return () -> channel.basicCancel(tag); } catch (IOException e) { throw new MessageQueueException("Unable to set up consumer.", e); } }
RabbitMQMessageQueue implements MessageQueue<T> { @Override public Closeable consume(Function<T, Void> processor) { Consumer consumer = new FunctionConsumer<>(channel, processor, type, name, metrics); try { String tag = channel.basicConsume(name, false, consumer); log.info("Set up consumer '{}' for queue '{}'.", tag, name); return () -> channel.basicCancel(tag); } catch (IOException e) { throw new MessageQueueException("Unable to set up consumer.", e); } } }
RabbitMQMessageQueue implements MessageQueue<T> { @Override public Closeable consume(Function<T, Void> processor) { Consumer consumer = new FunctionConsumer<>(channel, processor, type, name, metrics); try { String tag = channel.basicConsume(name, false, consumer); log.info("Set up consumer '{}' for queue '{}'.", tag, name); return () -> channel.basicCancel(tag); } catch (IOException e) { throw new MessageQueueException("Unable to set up consumer.", e); } } RabbitMQMessageQueue(Channel channel, String name, Class<T> type, MetricRegistry metrics); }
RabbitMQMessageQueue implements MessageQueue<T> { @Override public Closeable consume(Function<T, Void> processor) { Consumer consumer = new FunctionConsumer<>(channel, processor, type, name, metrics); try { String tag = channel.basicConsume(name, false, consumer); log.info("Set up consumer '{}' for queue '{}'.", tag, name); return () -> channel.basicCancel(tag); } catch (IOException e) { throw new MessageQueueException("Unable to set up consumer.", e); } } RabbitMQMessageQueue(Channel channel, String name, Class<T> type, MetricRegistry metrics); @Override void publish(T message); @Override Closeable consume(Function<T, Void> processor); @Override T consumeNext(); }
RabbitMQMessageQueue implements MessageQueue<T> { @Override public Closeable consume(Function<T, Void> processor) { Consumer consumer = new FunctionConsumer<>(channel, processor, type, name, metrics); try { String tag = channel.basicConsume(name, false, consumer); log.info("Set up consumer '{}' for queue '{}'.", tag, name); return () -> channel.basicCancel(tag); } catch (IOException e) { throw new MessageQueueException("Unable to set up consumer.", e); } } RabbitMQMessageQueue(Channel channel, String name, Class<T> type, MetricRegistry metrics); @Override void publish(T message); @Override Closeable consume(Function<T, Void> processor); @Override T consumeNext();  }