From 6ca1aab8d1124de10ed2826f85fcb0b85528638d Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 28 Sep 2015 08:41:34 -0400 Subject: [PATCH] INT-3831: MessageGatewaySupport Statistics JIRA: https://jira.spring.io/browse/INT-3831 Implement `MessageSourceMetrics` in MGS. --- .../aggregator/BarrierMessageHandler.java | 5 ++ .../gateway/MessagingGatewaySupport.java | 77 ++++++++++++++++++- .../gateway/MessagingGatewayTests.java | 4 +- 3 files changed, 84 insertions(+), 2 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java index c21d5a0845..a9bc9221fe 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java @@ -106,6 +106,11 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler this.timeout = timeout; } + @Override + public String getComponentType() { + return "barrier"; + } + @Override protected Object handleRequestMessage(Message requestMessage) { Object key = this.correlationStrategy.getCorrelationKey(requestMessage); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index ba5c96a8af..4fc2f91db0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -16,6 +16,8 @@ package org.springframework.integration.gateway; +import java.util.concurrent.atomic.AtomicLong; + import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; @@ -28,6 +30,7 @@ import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.converter.SimpleMessageConverter; +import org.springframework.integration.support.management.MessageSourceMetrics; import org.springframework.integration.support.management.TrackableComponent; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -47,7 +50,8 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan */ -public abstract class MessagingGatewaySupport extends AbstractEndpoint implements TrackableComponent { +public abstract class MessagingGatewaySupport extends AbstractEndpoint + implements TrackableComponent, MessageSourceMetrics { private static final long DEFAULT_TIMEOUT = 1000L; @@ -62,6 +66,8 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement private final boolean errorOnTimeout; + private final AtomicLong messageCount = new AtomicLong(); + private volatile MessageChannel requestChannel; private volatile String requestChannelName; @@ -83,6 +89,14 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement private volatile AbstractEndpoint replyMessageCorrelator; + private volatile String managedType; + + private volatile String managedName; + + private volatile boolean countsEnabled; + + private volatile boolean loggingEnabled = true; + /** * Construct an instance that will return null if no reply is received. @@ -218,11 +232,61 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement this.historyWritingPostProcessor.setShouldTrack(shouldTrack); } + @Override + public int getMessageCount() { + return (int) this.messageCount.get(); + } + + @Override + public long getMessageCountLong() { + return this.messageCount.get(); + } + + @Override + public void setManagedName(String name) { + this.managedName = name; + } + + @Override + public String getManagedName() { + return this.managedName; + } + + @Override + public void setManagedType(String type) { + this.managedType = type; + } + + @Override + public String getManagedType() { + return this.managedType; + } + @Override public String getComponentType() { return "gateway"; } + @Override + public void setLoggingEnabled(boolean enabled) { + this.loggingEnabled = enabled; + } + + @Override + public boolean isLoggingEnabled() { + return this.loggingEnabled; + } + + @Override + public void setCountsEnabled(boolean countsEnabled) { + this.countsEnabled = countsEnabled; + } + + @Override + public boolean isCountsEnabled() { + return this.countsEnabled; + } + @Override protected void onInit() throws Exception { Assert.state(!(this.requestChannelName != null && this.requestChannel != null), @@ -292,6 +356,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement Assert.state(requestChannel != null, "send is not supported, because no request channel has been configured"); try { + if (this.countsEnabled) { + this.messageCount.incrementAndGet(); + } this.messagingTemplate.convertAndSend(requestChannel, object, this.historyWritingPostProcessor); } catch (Exception e) { @@ -336,6 +403,9 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement Object reply = null; Throwable error = null; try { + if (this.countsEnabled) { + this.messageCount.incrementAndGet(); + } if (shouldConvert) { reply = this.messagingTemplate.convertSendAndReceive(requestChannel, object, null, this.historyWritingPostProcessor); @@ -462,6 +532,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement } } + @Override + public void reset() { + this.messageCount.set(0); + } + private static class DefaultRequestMapper implements InboundMessageMapper { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java index c607f452da..5dd412336f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -70,6 +70,7 @@ public class MessagingGatewayTests { this.messagingGateway.setReplyChannel(replyChannel); TestApplicationContext applicationContext = TestUtils.createTestApplicationContext(); this.messagingGateway.setBeanFactory(applicationContext); + this.messagingGateway.setCountsEnabled(true); this.messagingGateway.afterPropertiesSet(); this.messagingGateway.start(); applicationContext.refresh(); @@ -83,6 +84,7 @@ public class MessagingGatewayTests { Mockito.when(requestChannel.send(messageMock, 1000L)).thenReturn(true); this.messagingGateway.send(messageMock); Mockito.verify(requestChannel).send(messageMock, 1000L); + assertEquals(1, this.messagingGateway.getMessageCount()); } @Test(expected=MessageDeliveryException.class)