@Test public void testHandleMessage() { Message message = mock(Message.class); managedMessageListener.onMessage(message, context); verify(messageListener).onMessage(eq(message), eq(context)); }
private void handleMessage(Message message, Context context, boolean isError) { ContextThreadLocal.setContext(context); Thread currentThread = Thread.currentThread(); String origThreadName = currentThread.getName(); MessageLoggingContext messageLoggingContext = buildMessageLoggingContext(message); Map<String, String> messageHeaderValuesForLogging = messageLoggingContext.getMessageHeaderValuesForLogging(); if (logger.isDebugEnabled()) { logger.debug("currentThread {}", currentThread); logger.debug("origThreadName {}", origThreadName); messageHeaderValuesForLogging.values().forEach(logger::debug); } try (CounterMetric autoCounter = activeJobs.inc(); StopWatch stopWatch = messageProcessingTimer.getStopWatch( new MapMetricState(messageHeaderValuesForLogging))) { queueExecutionState.setExecutionState(messageHeaderValuesForLogging); currentThread.setName(messageLoggingContext.getThreadName()); if (logger.isDebugEnabled()) { logger.debug("Executing message listener"); } if (isError) { handleErrorMessage(message, context); } else { messageListener.onMessage(message, context); } if (logger.isDebugEnabled()) { logger.debug("Done executing message listener"); } } catch (Exception ex) { logger.warn("Error while processing message. {}", ex.getMessage()); errorCounter.inc(); throw ex; } finally { try { queueExecutionState.clearExecutionState(); } finally { currentThread.setName(origThreadName); } } }
ManagedMessageListenerImpl implements ManagedMessageListener { private void handleMessage(Message message, Context context, boolean isError) { ContextThreadLocal.setContext(context); Thread currentThread = Thread.currentThread(); String origThreadName = currentThread.getName(); MessageLoggingContext messageLoggingContext = buildMessageLoggingContext(message); Map<String, String> messageHeaderValuesForLogging = messageLoggingContext.getMessageHeaderValuesForLogging(); if (logger.isDebugEnabled()) { logger.debug("currentThread {}", currentThread); logger.debug("origThreadName {}", origThreadName); messageHeaderValuesForLogging.values().forEach(logger::debug); } try (CounterMetric autoCounter = activeJobs.inc(); StopWatch stopWatch = messageProcessingTimer.getStopWatch( new MapMetricState(messageHeaderValuesForLogging))) { queueExecutionState.setExecutionState(messageHeaderValuesForLogging); currentThread.setName(messageLoggingContext.getThreadName()); if (logger.isDebugEnabled()) { logger.debug("Executing message listener"); } if (isError) { handleErrorMessage(message, context); } else { messageListener.onMessage(message, context); } if (logger.isDebugEnabled()) { logger.debug("Done executing message listener"); } } catch (Exception ex) { logger.warn("Error while processing message. {}", ex.getMessage()); errorCounter.inc(); throw ex; } finally { try { queueExecutionState.clearExecutionState(); } finally { currentThread.setName(origThreadName); } } } }
ManagedMessageListenerImpl implements ManagedMessageListener { private void handleMessage(Message message, Context context, boolean isError) { ContextThreadLocal.setContext(context); Thread currentThread = Thread.currentThread(); String origThreadName = currentThread.getName(); MessageLoggingContext messageLoggingContext = buildMessageLoggingContext(message); Map<String, String> messageHeaderValuesForLogging = messageLoggingContext.getMessageHeaderValuesForLogging(); if (logger.isDebugEnabled()) { logger.debug("currentThread {}", currentThread); logger.debug("origThreadName {}", origThreadName); messageHeaderValuesForLogging.values().forEach(logger::debug); } try (CounterMetric autoCounter = activeJobs.inc(); StopWatch stopWatch = messageProcessingTimer.getStopWatch( new MapMetricState(messageHeaderValuesForLogging))) { queueExecutionState.setExecutionState(messageHeaderValuesForLogging); currentThread.setName(messageLoggingContext.getThreadName()); if (logger.isDebugEnabled()) { logger.debug("Executing message listener"); } if (isError) { handleErrorMessage(message, context); } else { messageListener.onMessage(message, context); } if (logger.isDebugEnabled()) { logger.debug("Done executing message listener"); } } catch (Exception ex) { logger.warn("Error while processing message. {}", ex.getMessage()); errorCounter.inc(); throw ex; } finally { try { queueExecutionState.clearExecutionState(); } finally { currentThread.setName(origThreadName); } } } ManagedMessageListenerImpl(
            Context context,
            MessageListener messageListener,
            ManagedInputQueue managedQueue,
            int threadOrdinal); }
ManagedMessageListenerImpl implements ManagedMessageListener { private void handleMessage(Message message, Context context, boolean isError) { ContextThreadLocal.setContext(context); Thread currentThread = Thread.currentThread(); String origThreadName = currentThread.getName(); MessageLoggingContext messageLoggingContext = buildMessageLoggingContext(message); Map<String, String> messageHeaderValuesForLogging = messageLoggingContext.getMessageHeaderValuesForLogging(); if (logger.isDebugEnabled()) { logger.debug("currentThread {}", currentThread); logger.debug("origThreadName {}", origThreadName); messageHeaderValuesForLogging.values().forEach(logger::debug); } try (CounterMetric autoCounter = activeJobs.inc(); StopWatch stopWatch = messageProcessingTimer.getStopWatch( new MapMetricState(messageHeaderValuesForLogging))) { queueExecutionState.setExecutionState(messageHeaderValuesForLogging); currentThread.setName(messageLoggingContext.getThreadName()); if (logger.isDebugEnabled()) { logger.debug("Executing message listener"); } if (isError) { handleErrorMessage(message, context); } else { messageListener.onMessage(message, context); } if (logger.isDebugEnabled()) { logger.debug("Done executing message listener"); } } catch (Exception ex) { logger.warn("Error while processing message. {}", ex.getMessage()); errorCounter.inc(); throw ex; } finally { try { queueExecutionState.clearExecutionState(); } finally { currentThread.setName(origThreadName); } } } ManagedMessageListenerImpl(
            Context context,
            MessageListener messageListener,
            ManagedInputQueue managedQueue,
            int threadOrdinal); @Override void onMessage(Message message, Context context); @Override void onErrorMessage(Message message, Context context); QueueExecutionState getQueueExecutionState(); }
ManagedMessageListenerImpl implements ManagedMessageListener { private void handleMessage(Message message, Context context, boolean isError) { ContextThreadLocal.setContext(context); Thread currentThread = Thread.currentThread(); String origThreadName = currentThread.getName(); MessageLoggingContext messageLoggingContext = buildMessageLoggingContext(message); Map<String, String> messageHeaderValuesForLogging = messageLoggingContext.getMessageHeaderValuesForLogging(); if (logger.isDebugEnabled()) { logger.debug("currentThread {}", currentThread); logger.debug("origThreadName {}", origThreadName); messageHeaderValuesForLogging.values().forEach(logger::debug); } try (CounterMetric autoCounter = activeJobs.inc(); StopWatch stopWatch = messageProcessingTimer.getStopWatch( new MapMetricState(messageHeaderValuesForLogging))) { queueExecutionState.setExecutionState(messageHeaderValuesForLogging); currentThread.setName(messageLoggingContext.getThreadName()); if (logger.isDebugEnabled()) { logger.debug("Executing message listener"); } if (isError) { handleErrorMessage(message, context); } else { messageListener.onMessage(message, context); } if (logger.isDebugEnabled()) { logger.debug("Done executing message listener"); } } catch (Exception ex) { logger.warn("Error while processing message. {}", ex.getMessage()); errorCounter.inc(); throw ex; } finally { try { queueExecutionState.clearExecutionState(); } finally { currentThread.setName(origThreadName); } } } ManagedMessageListenerImpl(
            Context context,
            MessageListener messageListener,
            ManagedInputQueue managedQueue,
            int threadOrdinal); @Override void onMessage(Message message, Context context); @Override void onErrorMessage(Message message, Context context); QueueExecutionState getQueueExecutionState();  }