@Test public void shouldRejectMessageOnException() throws IOException { FunctionConsumer<TestMsg> consumer = new FunctionConsumer<>(channel, msg -> { throw new RuntimeException("Message processing failed on purpose."); }, TestMsg.class, "name", metrics); consumer.handleDelivery(null, envelope, null, message); verify(channel).basicNack(1L, false, true); }
@Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); Stopwatch stopwatch = Stopwatch.createStarted(); try { T message = MAPPER.readValue(body, type); if (log.isTraceEnabled()) { log.trace("Received message '{}' with data '{}'.", deliveryTag, new String(body)); } processor.apply(message); getChannel().basicAck(deliveryTag, false); stopwatch.stop(); successCount.mark(); successDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.info("Processed {} message '{}' succesfully in {} ms.", type.getSimpleName(), deliveryTag, stopwatch.elapsed(TimeUnit.MILLISECONDS)); } catch (Exception e) { getChannel().basicNack(deliveryTag, false, true); stopwatch.stop(); failureCount.mark(); failureDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.error("Processing {} message '{}' failed with exception:", type.getSimpleName(), deliveryTag, e); } }
FunctionConsumer extends DefaultConsumer { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); Stopwatch stopwatch = Stopwatch.createStarted(); try { T message = MAPPER.readValue(body, type); if (log.isTraceEnabled()) { log.trace("Received message '{}' with data '{}'.", deliveryTag, new String(body)); } processor.apply(message); getChannel().basicAck(deliveryTag, false); stopwatch.stop(); successCount.mark(); successDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.info("Processed {} message '{}' succesfully in {} ms.", type.getSimpleName(), deliveryTag, stopwatch.elapsed(TimeUnit.MILLISECONDS)); } catch (Exception e) { getChannel().basicNack(deliveryTag, false, true); stopwatch.stop(); failureCount.mark(); failureDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.error("Processing {} message '{}' failed with exception:", type.getSimpleName(), deliveryTag, e); } } }
FunctionConsumer extends DefaultConsumer { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); Stopwatch stopwatch = Stopwatch.createStarted(); try { T message = MAPPER.readValue(body, type); if (log.isTraceEnabled()) { log.trace("Received message '{}' with data '{}'.", deliveryTag, new String(body)); } processor.apply(message); getChannel().basicAck(deliveryTag, false); stopwatch.stop(); successCount.mark(); successDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.info("Processed {} message '{}' succesfully in {} ms.", type.getSimpleName(), deliveryTag, stopwatch.elapsed(TimeUnit.MILLISECONDS)); } catch (Exception e) { getChannel().basicNack(deliveryTag, false, true); stopwatch.stop(); failureCount.mark(); failureDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.error("Processing {} message '{}' failed with exception:", type.getSimpleName(), deliveryTag, e); } } FunctionConsumer(Channel channel, Function<T, Void> processor, Class<T> type, String name, MetricRegistry metrics); }
FunctionConsumer extends DefaultConsumer { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); Stopwatch stopwatch = Stopwatch.createStarted(); try { T message = MAPPER.readValue(body, type); if (log.isTraceEnabled()) { log.trace("Received message '{}' with data '{}'.", deliveryTag, new String(body)); } processor.apply(message); getChannel().basicAck(deliveryTag, false); stopwatch.stop(); successCount.mark(); successDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.info("Processed {} message '{}' succesfully in {} ms.", type.getSimpleName(), deliveryTag, stopwatch.elapsed(TimeUnit.MILLISECONDS)); } catch (Exception e) { getChannel().basicNack(deliveryTag, false, true); stopwatch.stop(); failureCount.mark(); failureDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.error("Processing {} message '{}' failed with exception:", type.getSimpleName(), deliveryTag, e); } } FunctionConsumer(Channel channel, Function<T, Void> processor, Class<T> type, String name, MetricRegistry metrics); @Override void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body); }
FunctionConsumer extends DefaultConsumer { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); Stopwatch stopwatch = Stopwatch.createStarted(); try { T message = MAPPER.readValue(body, type); if (log.isTraceEnabled()) { log.trace("Received message '{}' with data '{}'.", deliveryTag, new String(body)); } processor.apply(message); getChannel().basicAck(deliveryTag, false); stopwatch.stop(); successCount.mark(); successDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.info("Processed {} message '{}' succesfully in {} ms.", type.getSimpleName(), deliveryTag, stopwatch.elapsed(TimeUnit.MILLISECONDS)); } catch (Exception e) { getChannel().basicNack(deliveryTag, false, true); stopwatch.stop(); failureCount.mark(); failureDuration.update(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS); log.error("Processing {} message '{}' failed with exception:", type.getSimpleName(), deliveryTag, e); } } FunctionConsumer(Channel channel, Function<T, Void> processor, Class<T> type, String name, MetricRegistry metrics); @Override void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body);  }