@Test public void shouldPublishPersistentMessageToQueue() throws IOException { RabbitMQMessageQueue<TestMsg> queue = new RabbitMQMessageQueue<>(channel, QUEUE_NAME, TestMsg.class, metrics); TestMsg message = new TestMsg("blah", 5); byte[] messageBytes = Jackson.newObjectMapper().writeValueAsBytes(message); queue.publish(message); verify(channel).basicPublish(Matchers.eq(""), Matchers.eq(QUEUE_NAME), Matchers.eq(MessageProperties.PERSISTENT_TEXT_PLAIN), Matchers.eq(messageBytes)); }
@Override public void publish(T message) { try { channel.basicPublish("", name, MessageProperties.PERSISTENT_TEXT_PLAIN, MAPPER.writeValueAsBytes(message)); publish.mark(); if (log.isTraceEnabled()) { log.trace("Published to '{}' with data '{}'.", name, MAPPER.writeValueAsString(message)); } } catch (IOException e) { throw new MessageQueueException("Unable to publish to queue.", e); } }
RabbitMQMessageQueue implements MessageQueue<T> { @Override public void publish(T message) { try { channel.basicPublish("", name, MessageProperties.PERSISTENT_TEXT_PLAIN, MAPPER.writeValueAsBytes(message)); publish.mark(); if (log.isTraceEnabled()) { log.trace("Published to '{}' with data '{}'.", name, MAPPER.writeValueAsString(message)); } } catch (IOException e) { throw new MessageQueueException("Unable to publish to queue.", e); } } }
RabbitMQMessageQueue implements MessageQueue<T> { @Override public void publish(T message) { try { channel.basicPublish("", name, MessageProperties.PERSISTENT_TEXT_PLAIN, MAPPER.writeValueAsBytes(message)); publish.mark(); if (log.isTraceEnabled()) { log.trace("Published to '{}' with data '{}'.", name, MAPPER.writeValueAsString(message)); } } catch (IOException e) { throw new MessageQueueException("Unable to publish to queue.", e); } } RabbitMQMessageQueue(Channel channel, String name, Class<T> type, MetricRegistry metrics); }
RabbitMQMessageQueue implements MessageQueue<T> { @Override public void publish(T message) { try { channel.basicPublish("", name, MessageProperties.PERSISTENT_TEXT_PLAIN, MAPPER.writeValueAsBytes(message)); publish.mark(); if (log.isTraceEnabled()) { log.trace("Published to '{}' with data '{}'.", name, MAPPER.writeValueAsString(message)); } } catch (IOException e) { throw new MessageQueueException("Unable to publish to queue.", 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 void publish(T message) { try { channel.basicPublish("", name, MessageProperties.PERSISTENT_TEXT_PLAIN, MAPPER.writeValueAsBytes(message)); publish.mark(); if (log.isTraceEnabled()) { log.trace("Published to '{}' with data '{}'.", name, MAPPER.writeValueAsString(message)); } } catch (IOException e) { throw new MessageQueueException("Unable to publish to queue.", 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();  }