@Test public void AbstractAmqpExceptionを継承した例外の場合は例外IDがヘッダに設定される() { class AmqpException extends AmqpApplicationRecoverableException { private static final long serialVersionUID = 1L; public AmqpException(String code, String routingKey) { super(code, routingKey); } } AmqpException e = new AmqpException("code", "dummy"); exchanger.recover(message, e); ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class); ArgumentCaptor<String> routingKeyCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> exchangeCaptor = ArgumentCaptor.forClass(String.class); verify(mock).send(exchangeCaptor.capture(), routingKeyCaptor.capture(), messageCaptor.capture()); MessageProperties prop = messageCaptor.getValue().getMessageProperties(); assertThat(prop.getHeaders().get("x-exception-stacktrace").toString(), is(ExceptionUtils.getStackTrace(e))); assertThat(prop.getHeaders().get("x-exception-message").toString(), is(e.getMessage())); assertThat(prop.getHeaders().get("x-original-exchange").toString(), is("the.original.exchange")); assertThat(prop.getHeaders().get("x-original-routing-key").toString(), is("the.original.routing-key")); assertThat(prop.getHeaders().get("x-exception-id").toString(), is(e.getId())); assertThat(exchangeCaptor.getValue(), is("error.exchange")); assertThat(routingKeyCaptor.getValue(), is(e.getRoutingKey())); }
@Override public void recover(Message message, Throwable throwable) { Args.checkNotNull(exchange, R.getObject("E-AMQP-RETRY#0004")); Map<String, Object> headers = message.getMessageProperties().getHeaders(); headers.put(HEADER_KEY_EXCEPTION_STACKTRACE, ExceptionUtils.getStackTrace(throwable)); headers.put(HEADER_KEY_EXCEPTION_MESSAGE, throwable.getMessage()); headers.put(HEADER_KEY_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange()); headers.put(HEADER_KEY_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey()); String rk = defaultRoutingKey; Throwable cause = throwable; if (throwable instanceof ListenerExecutionFailedException) { cause = throwable.getCause(); } if (cause instanceof AbstractAmqpException) { headers.put(HEADER_KEY_EXCEPTION_ID, ((AbstractAmqpException) cause).getId()); rk = ((AbstractAmqpException) cause).getRoutingKey(); } amqpTemplate.send(exchange, rk, message); logging(exchange, rk, message); }
ExceptionMessageExchanger implements MessageRecoverer, InitializingBean { @Override public void recover(Message message, Throwable throwable) { Args.checkNotNull(exchange, R.getObject("E-AMQP-RETRY#0004")); Map<String, Object> headers = message.getMessageProperties().getHeaders(); headers.put(HEADER_KEY_EXCEPTION_STACKTRACE, ExceptionUtils.getStackTrace(throwable)); headers.put(HEADER_KEY_EXCEPTION_MESSAGE, throwable.getMessage()); headers.put(HEADER_KEY_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange()); headers.put(HEADER_KEY_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey()); String rk = defaultRoutingKey; Throwable cause = throwable; if (throwable instanceof ListenerExecutionFailedException) { cause = throwable.getCause(); } if (cause instanceof AbstractAmqpException) { headers.put(HEADER_KEY_EXCEPTION_ID, ((AbstractAmqpException) cause).getId()); rk = ((AbstractAmqpException) cause).getRoutingKey(); } amqpTemplate.send(exchange, rk, message); logging(exchange, rk, message); } }
ExceptionMessageExchanger implements MessageRecoverer, InitializingBean { @Override public void recover(Message message, Throwable throwable) { Args.checkNotNull(exchange, R.getObject("E-AMQP-RETRY#0004")); Map<String, Object> headers = message.getMessageProperties().getHeaders(); headers.put(HEADER_KEY_EXCEPTION_STACKTRACE, ExceptionUtils.getStackTrace(throwable)); headers.put(HEADER_KEY_EXCEPTION_MESSAGE, throwable.getMessage()); headers.put(HEADER_KEY_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange()); headers.put(HEADER_KEY_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey()); String rk = defaultRoutingKey; Throwable cause = throwable; if (throwable instanceof ListenerExecutionFailedException) { cause = throwable.getCause(); } if (cause instanceof AbstractAmqpException) { headers.put(HEADER_KEY_EXCEPTION_ID, ((AbstractAmqpException) cause).getId()); rk = ((AbstractAmqpException) cause).getRoutingKey(); } amqpTemplate.send(exchange, rk, message); logging(exchange, rk, message); } ExceptionMessageExchanger(); }
ExceptionMessageExchanger implements MessageRecoverer, InitializingBean { @Override public void recover(Message message, Throwable throwable) { Args.checkNotNull(exchange, R.getObject("E-AMQP-RETRY#0004")); Map<String, Object> headers = message.getMessageProperties().getHeaders(); headers.put(HEADER_KEY_EXCEPTION_STACKTRACE, ExceptionUtils.getStackTrace(throwable)); headers.put(HEADER_KEY_EXCEPTION_MESSAGE, throwable.getMessage()); headers.put(HEADER_KEY_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange()); headers.put(HEADER_KEY_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey()); String rk = defaultRoutingKey; Throwable cause = throwable; if (throwable instanceof ListenerExecutionFailedException) { cause = throwable.getCause(); } if (cause instanceof AbstractAmqpException) { headers.put(HEADER_KEY_EXCEPTION_ID, ((AbstractAmqpException) cause).getId()); rk = ((AbstractAmqpException) cause).getRoutingKey(); } amqpTemplate.send(exchange, rk, message); logging(exchange, rk, message); } ExceptionMessageExchanger(); @Override void recover(Message message, Throwable throwable); @Override void afterPropertiesSet(); void setAmqpTemplate(AmqpTemplate amqpTemplate); void setExchange(String exchange); void setDefaultRoutingKey(String defaultRoutingKey); }
ExceptionMessageExchanger implements MessageRecoverer, InitializingBean { @Override public void recover(Message message, Throwable throwable) { Args.checkNotNull(exchange, R.getObject("E-AMQP-RETRY#0004")); Map<String, Object> headers = message.getMessageProperties().getHeaders(); headers.put(HEADER_KEY_EXCEPTION_STACKTRACE, ExceptionUtils.getStackTrace(throwable)); headers.put(HEADER_KEY_EXCEPTION_MESSAGE, throwable.getMessage()); headers.put(HEADER_KEY_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange()); headers.put(HEADER_KEY_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey()); String rk = defaultRoutingKey; Throwable cause = throwable; if (throwable instanceof ListenerExecutionFailedException) { cause = throwable.getCause(); } if (cause instanceof AbstractAmqpException) { headers.put(HEADER_KEY_EXCEPTION_ID, ((AbstractAmqpException) cause).getId()); rk = ((AbstractAmqpException) cause).getRoutingKey(); } amqpTemplate.send(exchange, rk, message); logging(exchange, rk, message); } ExceptionMessageExchanger(); @Override void recover(Message message, Throwable throwable); @Override void afterPropertiesSet(); void setAmqpTemplate(AmqpTemplate amqpTemplate); void setExchange(String exchange); void setDefaultRoutingKey(String defaultRoutingKey);  }