RepublishMessageRecoverer implements MessageRecoverer { @Override public void recover(Message message, Throwable cause) { MessageProperties messageProperties = message.getMessageProperties(); Map<String, Object> headers = messageProperties.getHeaders(); String exceptionMessage = cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage(); String[] processed = processStackTrace(cause, exceptionMessage); String stackTraceAsString = processed[0]; String truncatedExceptionMessage = processed[1]; if (truncatedExceptionMessage != null) { exceptionMessage = truncatedExceptionMessage; } headers.put(X_EXCEPTION_STACKTRACE, stackTraceAsString); headers.put(X_EXCEPTION_MESSAGE, exceptionMessage); headers.put(X_ORIGINAL_EXCHANGE, messageProperties.getReceivedExchange()); headers.put(X_ORIGINAL_ROUTING_KEY, messageProperties.getReceivedRoutingKey()); Map<? extends String, ?> additionalHeaders = additionalHeaders(message, cause); if (additionalHeaders != null) { headers.putAll(additionalHeaders); } if (messageProperties.getDeliveryMode() == null) { messageProperties.setDeliveryMode(this.deliveryMode); } if (null != this.errorExchangeName) { String routingKey = this.errorRoutingKey != null ? this.errorRoutingKey : this.prefixedOriginalRoutingKey(message); this.errorTemplate.send(this.errorExchangeName, routingKey, message); if (this.logger.isWarnEnabled()) { this.logger.warn("Republishing failed message to exchange '" + this.errorExchangeName + "' with routing key " + routingKey); } } else { final String routingKey = this.prefixedOriginalRoutingKey(message); this.errorTemplate.send(routingKey, message); if (this.logger.isWarnEnabled()) { this.logger.warn("Republishing failed message to the template's default exchange with routing key " + routingKey); } } } RepublishMessageRecoverer(AmqpTemplate errorTemplate);  RepublishMessageRecoverer(AmqpTemplate errorTemplate, String errorExchange);  RepublishMessageRecoverer(AmqpTemplate errorTemplate, String errorExchange, String errorRoutingKey); RepublishMessageRecoverer errorRoutingKeyPrefix(String errorRoutingKeyPrefix); RepublishMessageRecoverer frameMaxHeadroom(int headroom); void setErrorRoutingKeyPrefix(String errorRoutingKeyPrefix); void setDeliveryMode(MessageDeliveryMode deliveryMode); @Override void recover(Message message, Throwable cause); static final String X_EXCEPTION_STACKTRACE; static final String X_EXCEPTION_MESSAGE; static final String X_ORIGINAL_EXCHANGE; static final String X_ORIGINAL_ROUTING_KEY; static final int DEFAULT_FRAME_MAX_HEADROOM; }
@Test void shouldPublishWithRoutingKeyPrefixedWithErrorWhenExchangeIsNotSet() { recoverer = new RepublishMessageRecoverer(amqpTemplate); recoverer.recover(message, cause); verify(amqpTemplate).send("error.some.key", message); }
@Test void nullCauseMessage() { recoverer = new RepublishMessageRecoverer(amqpTemplate); recoverer.recover(message, new RuntimeException(new RuntimeException())); verify(amqpTemplate).send("error.some.key", message); }
@Test void shouldPublishWithSetErrorRoutingKeyWhenExchangeAndErrorRoutingKeyProvided() { recoverer = new RepublishMessageRecoverer(amqpTemplate, "errorExchange", "errorRoutingKey"); recoverer.recover(message, cause); verify(amqpTemplate).send("errorExchange", "errorRoutingKey", message); }
@Test void shouldPublishToProvidedExchange() { recoverer = new RepublishMessageRecoverer(amqpTemplate, "error"); recoverer.recover(message, cause); verify(amqpTemplate).send("error", "error.some.key", message); }
@Test void shouldIncludeTheStacktraceInTheHeaderOfThePublishedMessage() { recoverer = new RepublishMessageRecoverer(amqpTemplate); ByteArrayOutputStream baos = new ByteArrayOutputStream(); cause.printStackTrace(new PrintStream(baos)); final String expectedHeaderValue = baos.toString(); recoverer.recover(message, cause); assertThat(message.getMessageProperties().getHeaders().get("x-exception-stacktrace")).isEqualTo(expectedHeaderValue); }
@Test void shouldIncludeTheCauseMessageInTheHeaderOfThePublishedMessage() { recoverer = new RepublishMessageRecoverer(amqpTemplate); recoverer.recover(message, cause); assertThat(message.getMessageProperties().getHeaders().get("x-exception-message")).isEqualTo(cause.getCause().getMessage()); }
@Test void shouldSetTheOriginalMessageExchangeOnInTheHeaders() { message.getMessageProperties().setReceivedExchange("the.original.exchange"); recoverer = new RepublishMessageRecoverer(amqpTemplate, "error"); recoverer.recover(message, cause); assertThat(message.getMessageProperties().getHeaders().get("x-original-exchange")).isEqualTo("the.original.exchange"); }