Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1561: add counters to message stores

Browse Source
This commit is contained in:
Dave Syer
2010-11-09 12:29:35 +00:00
parent 35aeaff97b
commit 941b1127d1
9 changed files with 268 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java
View File

@@ -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<MessageGroup> 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;

45
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java
View File

@@ -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);

15
spring-integration-core/src/main/java/org/springframework/integration/store/MessageStore.java
View File

@@ -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();
}

20
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -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<UUID, Message<?>> idToMessage;
@@ -71,6 +74,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
public SimpleMessageStore() {
this(0);
}
@ManagedAttribute
public int getMessageCount() {
return idToMessage.size();
}
public <T> Message<T> addMessage(Message<T> 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);

40
spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java
View File

@@ -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.<MessageGroupCallback>asList(new MessageGroupCallback() {
store.setExpiryCallbacks(Arrays.<MessageGroupCallback> 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<String>("foo")), "bar");
private boolean removed = false;
@Override
@@ -95,7 +117,7 @@ public class MessageStoreTests {
removed = true;
}
}
}
}

36
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -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<Message<?>> 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<Message<?>> marked = new ArrayList<Message<?>>();
@@ -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"));

40
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -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<String> 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<String> message = MessageBuilder.withPayload("foo").build();
messageStore.addMessageToGroup(groupId, message);
assertEquals(1, messageStore.getMessageGroupCount());
}
@Test
@Transactional
public void testMessageGroupSizes() throws Exception {
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").build();
messageStore.addMessageToGroup(groupId, message);
assertEquals(1, messageStore.getMessageCountForAllMessageGroups());
}
@Test
@Transactional
public void testMarkedMessageGroupSizes() throws Exception {
String groupId = "X";
Message<String> 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<String> 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());
}

21
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests-context.xml Normal file
View File

@@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:jmx="http://www.springframework.org/schema/integration/jmx"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jmx
http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd">
<context:mbean-server id="mbs" />
<context:mbean-export server="mbs" default-domain="test.MessageStore"/>
<bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore"/>
</beans>

51
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests.java Normal file
View File

@@ -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<ObjectName> 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"));
}
}