diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index c307cb8ff8..5cf0bac475 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -19,6 +19,7 @@ import java.util.LinkedHashSet; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.jmx.export.annotation.ManagedAttribute; /** * @author Dave Syer @@ -69,6 +70,33 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It public abstract Iterator iterator(); + @ManagedAttribute + public int getMessageCountForAllMessageGroups() { + int count = 0; + for (MessageGroup group : this) { + count += group.size(); + } + return count; + } + + @ManagedAttribute + public int getMarkedMessageCountForAllMessageGroups() { + int count = 0; + for (MessageGroup group : this) { + count += group.getMarked().size(); + } + return count; + } + + @ManagedAttribute + public int getMessageGroupCount() { + int count = 0; + for (@SuppressWarnings("unused") MessageGroup group : this) { + count ++; + } + return count; + } + private void expire(MessageGroup group) { RuntimeException exception = null; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index 2c395a378b..4b4d1a697f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -13,6 +13,7 @@ package org.springframework.integration.store; import org.springframework.integration.Message; +import org.springframework.jmx.export.annotation.ManagedAttribute; /** * Interface for storage operations on groups of messages linked by a group id. @@ -24,6 +25,36 @@ import org.springframework.integration.Message; */ public interface MessageGroupStore { + /** + * Optional attribute giving the number of messages in the store over all groups. Implementations may decline to + * respond by throwing an exception. + * + * @return the number of messages + * @throws UnsupportedOperationException if not implemented + */ + @ManagedAttribute + int getMessageCountForAllMessageGroups(); + + /** + * Optional attribute giving the number of marked messages in the store for all groups. Implementations may decline + * to respond by throwing an exception. + * + * @return the number of marked messages in each group + * @throws UnsupportedOperationException if not implemented + */ + @ManagedAttribute + int getMarkedMessageCountForAllMessageGroups(); + + /** + * Optional attribute giving the number of message groups. Implementations may decline + * to respond by throwing an exception. + * + * @return the number message groups + * @throws UnsupportedOperationException if not implemented + */ + @ManagedAttribute + int getMessageGroupCount(); + /** * Return all Messages currently in the MessageStore that were stored using * {@link #addMessageToGroup(Object, Message)} with this group id. @@ -43,14 +74,14 @@ public interface MessageGroupStore { /** * Persist the mark on all the messages from the group. The group is modified in the process as all its unmarked * messages become marked. - * + * * @param group a MessageGroup with no unmarked messages */ MessageGroup markMessageGroup(MessageGroup group); /** - * Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is no - * longer present in the group. + * Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is + * no longer present in the group. * @param key the groupId for the group containing the message * @param messageToRemove the message to be removed */ @@ -66,14 +97,14 @@ public interface MessageGroupStore { /** * Remove the message group with this id. - * + * * @param groupId the id of the group to remove */ void removeMessageGroup(Object groupId); /** * Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}. - * + * * @param callback a callback to execute when a message group is cleaned up */ void registerMessageGroupExpiryCallback(MessageGroupCallback callback); @@ -83,10 +114,10 @@ public interface MessageGroupStore { * each of the registered callbacks on them in turn. For example: call with a timeout of 100 to expire all groups * that were created more than 100 milliseconds ago, and are not yet complete. Use a timeout of 0 (or negative to be * on the safe side) to expire all message groups. - * + * * @param timeout the timeout threshold to use * @return the number of message groups expired - * + * * @see #registerMessageGroupExpiryCallback(MessageGroupCallback) */ int expireMessageGroups(long timeout); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java index 99b6a44e0f..f5caa96edb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java @@ -19,6 +19,7 @@ package org.springframework.integration.store; import java.util.UUID; import org.springframework.integration.Message; +import org.springframework.jmx.export.annotation.ManagedAttribute; /** * Strategy interface for storing and retrieving messages. @@ -39,8 +40,8 @@ public interface MessageStore { /** * Put the provided Message into the MessageStore. The store may need to mutate the message internally, and if it * does then the return value can be different than the input. The id of the return value will be used as an index - * so that the {@link #getMessage(UUID)} and {@link #removeMessage(UUID)} behave properly. Since messages are immutable, putting - * the same message more than once is a no-op. + * so that the {@link #getMessage(UUID)} and {@link #removeMessage(UUID)} behave properly. Since messages are + * immutable, putting the same message more than once is a no-op. * * @return the message that was stored */ @@ -52,4 +53,14 @@ public interface MessageStore { */ Message removeMessage(UUID id); + /** + * Optional attribute giving the number of messages in the store. Implementations may decline to respond by throwing + * an exception. + * + * @return the number of messages + * @throws UnsupportedOperationException if not implemented + */ + @ManagedAttribute + int getMessageCount(); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index dec3c394a8..019b5f53d2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -13,17 +13,19 @@ package org.springframework.integration.store; -import org.springframework.integration.Message; -import org.springframework.integration.MessagingException; -import org.springframework.integration.util.UpperBound; -import org.springframework.util.Assert; - import java.util.HashSet; import java.util.Iterator; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.util.UpperBound; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.util.Assert; + /** * Map-based implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the * store. @@ -34,6 +36,7 @@ import java.util.concurrent.ConcurrentMap; * * @since 2.0 */ +@ManagedResource public class SimpleMessageStore extends AbstractMessageGroupStore implements MessageStore, MessageGroupStore { private final ConcurrentMap> idToMessage; @@ -71,6 +74,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes public SimpleMessageStore() { this(0); } + + @ManagedAttribute + public int getMessageCount() { + return idToMessage.size(); + } public Message addMessage(Message message) { if (!individualUpperBound.tryAcquire(0)) { @@ -93,7 +101,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes else return null; } - + public MessageGroup getMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); SimpleMessageGroup group = groupIdToMessageGroup.get(groupId); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index 76386f6537..1577a81fc0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -16,29 +16,32 @@ package org.springframework.integration.store; +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Iterator; +import java.util.List; + import org.junit.Test; import org.springframework.integration.Message; import org.springframework.integration.message.GenericMessage; import org.springframework.test.util.ReflectionTestUtils; -import java.util.*; - -import static org.junit.Assert.assertEquals; - /** * @author Dave Syer */ public class MessageStoreTests { - @Test public void shouldRegisterCallbacks() throws Exception { TestMessageStore store = new TestMessageStore(); - store.setExpiryCallbacks(Arrays.asList(new MessageGroupCallback() { + store.setExpiryCallbacks(Arrays. asList(new MessageGroupCallback() { public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } })); - assertEquals(1, ((Collection)ReflectionTestUtils.getField(store, "expiryCallbacks")).size()); + assertEquals(1, ((Collection) ReflectionTestUtils.getField(store, "expiryCallbacks")).size()); } @Test @@ -58,11 +61,30 @@ public class MessageStoreTests { assertEquals(0, store.getMessageGroup("bar").size()); } - + + @Test + public void testGroupCount() throws Exception { + TestMessageStore store = new TestMessageStore(); + assertEquals(1, store.getMessageGroupCount()); + } + + @Test + public void testGroupSizes() throws Exception { + TestMessageStore store = new TestMessageStore(); + assertEquals(1, store.getMessageCountForAllMessageGroups()); + } + + @Test + public void testMarkedGroupSizes() throws Exception { + TestMessageStore store = new TestMessageStore(); + assertEquals(0, store.getMarkedMessageCountForAllMessageGroups()); + } + private static class TestMessageStore extends AbstractMessageGroupStore { @SuppressWarnings("unchecked") MessageGroup testMessages = new SimpleMessageGroup(Arrays.asList(new GenericMessage("foo")), "bar"); + private boolean removed = false; @Override @@ -95,7 +117,7 @@ public class MessageStoreTests { removed = true; } } - + } } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 71b4577817..48becc9884 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -29,7 +29,6 @@ import javax.sql.DataSource; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.core.serializer.support.DeserializingConverter; @@ -49,6 +48,8 @@ import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.core.SingleColumnRowMapper; import org.springframework.jdbc.support.lob.DefaultLobHandler; import org.springframework.jdbc.support.lob.LobHandler; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -60,6 +61,7 @@ import org.springframework.util.StringUtils; * @author Dave Syer * @since 2.0 */ +@ManagedResource public class JdbcMessageStore extends AbstractMessageGroupStore implements MessageStore { private static final Log logger = LogFactory.getLog(JdbcMessageStore.class); @@ -71,6 +73,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private static final String GET_MESSAGE = "SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"; + private static final String GET_MESSAGE_COUNT = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE where REGION=?"; + private static final String DELETE_MESSAGE = "DELETE from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"; private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, REGION, CREATED_DATE, MESSAGE_BYTES)" @@ -78,6 +82,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private static final String LIST_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE, GROUP_KEY, MESSAGE_BYTES, MARKED from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by CREATED_DATE"; + private static final String COUNT_ALL_GROUPS = "SELECT COUNT(GROUP_KEY) from %PREFIX%MESSAGE_GROUP where REGION=?"; + + private static final String COUNT_ALL_MARKED_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where MARKED=1 AND REGION=?"; + + private static final String COUNT_ALL_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where REGION=?"; + private static final String MARK_MESSAGES_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MARKED=0 and GROUP_KEY=? and REGION=?"; private static final String MARK_MESSAGE_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MESSAGE_ID=? and MARKED=0 and GROUP_KEY=? and REGION=?"; @@ -238,6 +248,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa return null; } + @ManagedAttribute + public int getMessageCount() { + return jdbcTemplate.queryForInt(getQuery(GET_MESSAGE_COUNT), region); + } + public Message getMessage(UUID id) { List> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id), region }, mapper); if (list.isEmpty()) { @@ -297,6 +312,21 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } + @ManagedAttribute + public int getMessageGroupCount() { + return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_GROUPS), region); + } + + @ManagedAttribute + public int getMessageCountForAllMessageGroups() { + return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MESSAGES_IN_GROUPS), region); + } + + @ManagedAttribute + public int getMarkedMessageCountForAllMessageGroups() { + return jdbcTemplate.queryForInt(getQuery(COUNT_ALL_MARKED_MESSAGES_IN_GROUPS), region); + } + public MessageGroup getMessageGroup(Object groupId) { String key = getKey(groupId); final List> marked = new ArrayList>(); @@ -305,12 +335,14 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, new RowCallbackHandler() { int count = 0; + public void processRow(ResultSet rs) throws SQLException { int markedFlag = rs.getInt("MARKED"); Message message = mapper.mapRow(rs, count++); if (markedFlag > 0) { marked.add(message); - } else { + } + else { unmarked.add(message); } date.set(rs.getTimestamp("CREATED_DATE")); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index 7bac021445..e31e52887a 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -37,7 +37,6 @@ import javax.sql.DataSource; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -85,6 +84,14 @@ public class JdbcMessageStoreTests { assertNotNull(result.getHeaders().get(JdbcMessageStore.CREATED_DATE_KEY)); } + @Test + @Transactional + public void testSize() throws Exception { + Message message = MessageBuilder.withPayload("foo").build(); + messageStore.addMessage(message); + assertEquals(1, messageStore.getMessageCount()); + } + @Test @Transactional public void testSerializer() throws Exception { @@ -181,6 +188,35 @@ public class JdbcMessageStoreTests { assertEquals(0, group.size()); } + @Test + @Transactional + public void testMessageGroupCount() throws Exception { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").build(); + messageStore.addMessageToGroup(groupId, message); + assertEquals(1, messageStore.getMessageGroupCount()); + } + + @Test + @Transactional + public void testMessageGroupSizes() throws Exception { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").build(); + messageStore.addMessageToGroup(groupId, message); + assertEquals(1, messageStore.getMessageCountForAllMessageGroups()); + } + + @Test + @Transactional + public void testMarkedMessageGroupSizes() throws Exception { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").build(); + messageStore.addMessageToGroup(groupId, message); + assertEquals(0, messageStore.getMarkedMessageCountForAllMessageGroups()); + messageStore.markMessageGroup(messageStore.getMessageGroup(groupId)); + assertEquals(1, messageStore.getMarkedMessageCountForAllMessageGroups()); + } + @Test @Transactional public void testOrderInMessageGroup() throws Exception { @@ -213,7 +249,7 @@ public class JdbcMessageStoreTests { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); messageStore.addMessageToGroup(groupId, message); - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); MessageGroup group = messageStore.markMessageFromGroup(groupId, message); assertEquals(1, group.getMarked().size()); } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests-context.xml new file mode 100644 index 0000000000..ed6d339c21 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests-context.xml @@ -0,0 +1,21 @@ + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests.java new file mode 100644 index 0000000000..2f47d37e92 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests.java @@ -0,0 +1,51 @@ +/* + * Copyright 2002-2010 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. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.jmx.config; + +import static org.junit.Assert.assertEquals; + +import java.util.Set; + +import javax.management.MBeanServer; +import javax.management.ObjectName; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * @since 2.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class MessageStoreTests { + + @Autowired + private MBeanServer server; + + @Test + public void testHandlerMBeanRegistration() throws Exception { + Set names = server.queryNames(new ObjectName("test.MessageStore:type=SimpleMessageStore,*"), null); + assertEquals(1, names.size()); + ObjectName name = names.iterator().next(); + assertEquals(0, server.getAttribute(name, "MessageCount")); + assertEquals(0, server.getAttribute(name, "MessageGroupCount")); + assertEquals(0, server.getAttribute(name, "MessageCountForAllMessageGroups")); + assertEquals(0, server.getAttribute(name, "MarkedMessageCountForAllMessageGroups")); + } + +}