You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2018/07/05 13:14:18 UTC
[1/2] qpid-broker-j git commit: QPID-8214: [Broker-J][JDBC] Reduce
the sizes of table names in the JDBC configuration store
Repository: qpid-broker-j
Updated Branches:
refs/heads/master c018e1ac9 -> 38d5844e5
QPID-8214: [Broker-J][JDBC] Reduce the sizes of table names in the JDBC configuration store
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/fbeb2084
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/fbeb2084
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/fbeb2084
Branch: refs/heads/master
Commit: fbeb2084635c5b15bfeed25674513e2783457ccf
Parents: c018e1a
Author: Alex Rudyy <or...@apache.org>
Authored: Sat Jun 30 00:01:50 2018 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Sat Jun 30 00:01:50 2018 +0100
----------------------------------------------------------------------
...stractDurableConfigurationStoreTestCase.java | 9 +-
.../jdbc/AbstractJDBCConfigurationStore.java | 198 ++++++++++++++++++-
.../jdbc/GenericJDBCConfigurationStoreTest.java | 149 +++++++++++++-
3 files changed, 343 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fbeb2084/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java b/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
index 00d75c8..4951b71 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
@@ -173,7 +173,7 @@ public abstract class AbstractDurableConfigurationStoreTestCase extends UnitTest
org.apache.qpid.server.model.Exchange.LIFETIME_POLICY, LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS.name())));
}
- private Map<String,Object> map(Object... vals)
+ protected Map<String,Object> map(Object... vals)
{
Map<String,Object> map = new HashMap<>();
boolean isValue = false;
@@ -205,11 +205,16 @@ public abstract class AbstractDurableConfigurationStoreTestCase extends UnitTest
verify(_handler, never()).handle(any(ConfiguredObjectRecord.class));
}
- private ConfiguredObjectRecord matchesRecord(UUID id, String type, Map<String, Object> attributes)
+ protected ConfiguredObjectRecord matchesRecord(UUID id, String type, Map<String, Object> attributes)
{
return argThat(new ConfiguredObjectMatcher(id, type, attributes, ANY_MAP));
}
+ protected ConfiguredObjectRecord matchesRecord(UUID id, String type, Map<String, Object> attributes, Map<String,UUID> parents)
+ {
+ return argThat(new ConfiguredObjectMatcher(id, type, attributes, parents));
+ }
+
private static class ConfiguredObjectMatcher extends ArgumentMatcher<ConfiguredObjectRecord>
{
private final Map<String,Object> _matchingMap;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fbeb2084/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCConfigurationStore.java
----------------------------------------------------------------------
diff --git a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCConfigurationStore.java b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCConfigurationStore.java
index c87d719..1b57755 100644
--- a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCConfigurationStore.java
+++ b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCConfigurationStore.java
@@ -25,6 +25,7 @@ import static org.apache.qpid.server.store.jdbc.AbstractJDBCConfigurationStore.S
import java.io.ByteArrayInputStream;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
@@ -58,8 +59,12 @@ import org.apache.qpid.server.util.Action;
public abstract class AbstractJDBCConfigurationStore implements MessageStoreProvider, DurableConfigurationStore
{
private final static String CONFIGURATION_VERSION_TABLE_NAME_SUFFIX = "QPID_CONFIG_VERSION";
- private final static String CONFIGURED_OBJECTS_TABLE_NAME_SUFFIX = "QPID_CONFIGURED_OBJECTS";
- private final static String CONFIGURED_OBJECT_HIERARCHY_TABLE_NAME_SUFFIX = "QPID_CONFIGURED_OBJECT_HIERARCHY";
+ private final static String VERSION_1_CONFIGURED_OBJECTS_TABLE_NAME_SUFFIX = "QPID_CONFIGURED_OBJECTS";
+ private final static String VERSION_1_CONFIGURED_OBJECT_HIERARCHY_TABLE_NAME_SUFFIX = "QPID_CONFIGURED_OBJECT_HIERARCHY";
+ private final static String CONFIGURATION_STRUCTURE_VERSION_TABLE_NAME_SUFFIX = "QPID_CFG_VERSION";
+ private final static String CONFIGURED_OBJECTS_TABLE_NAME_SUFFIX = "QPID_CFG_OBJECTS";
+ private final static String CONFIGURED_OBJECT_HIERARCHY_TABLE_NAME_SUFFIX = "QPID_CFG_HIERARCHY";
+ private static final int CONFIG_DB_VERSION = 2;
private static final int DEFAULT_CONFIG_VERSION = 0;
private final Set<Action<Connection>> _deleteActions = Collections.newSetFromMap(new ConcurrentHashMap<>());;
@@ -153,6 +158,21 @@ public abstract class AbstractJDBCConfigurationStore implements MessageStoreProv
return _tableNamePrefix + CONFIGURED_OBJECT_HIERARCHY_TABLE_NAME_SUFFIX;
}
+ private String getConfigStructureVersionTableName()
+ {
+ return _tableNamePrefix + CONFIGURATION_STRUCTURE_VERSION_TABLE_NAME_SUFFIX;
+ }
+
+ private String getVersion1ConfiguredObjectsTableName()
+ {
+ return _tableNamePrefix + VERSION_1_CONFIGURED_OBJECTS_TABLE_NAME_SUFFIX;
+ }
+
+ private String getVersion1ConfiguredObjectHierarchyTableName()
+ {
+ return _tableNamePrefix + VERSION_1_CONFIGURED_OBJECT_HIERARCHY_TABLE_NAME_SUFFIX;
+ }
+
private Collection<ConfiguredObjectRecordImpl> doVisitAllConfiguredObjectRecords(ConfiguredObjectRecordHandler handler) throws SQLException
{
Map<UUID, ConfiguredObjectRecordImpl> configuredObjects = new HashMap<UUID, ConfiguredObjectRecordImpl>();
@@ -246,6 +266,31 @@ public abstract class AbstractJDBCConfigurationStore implements MessageStoreProv
+ configVersion);
}
}
+ else
+ {
+ try (PreparedStatement statement = connection.prepareStatement(String.format("SELECT version FROM %s",
+ getConfigStructureVersionTableName())))
+ {
+ try (ResultSet rs = statement.executeQuery())
+ {
+ if (!rs.next())
+ {
+ throw new StoreException(getConfigStructureVersionTableName()
+ + " does not contain the configuration database version");
+ }
+ int version = rs.getInt(1);
+ switch (version)
+ {
+ case 1:
+ upgradeFromConfigVersion1();
+ case CONFIG_DB_VERSION:
+ return;
+ default:
+ throw new StoreException("Unknown configuration database version: " + version);
+ }
+ }
+ }
+ }
}
catch (SQLException se)
{
@@ -258,6 +303,117 @@ public abstract class AbstractJDBCConfigurationStore implements MessageStoreProv
}
+ private void upgradeFromConfigVersion1() throws SQLException
+ {
+ Connection connection = newConnection();
+ try
+ {
+ try (PreparedStatement stmt = connection.prepareStatement(String.format(
+ "SELECT id, object_type, attributes FROM %s",
+ getVersion1ConfiguredObjectsTableName())))
+ {
+ try (ResultSet rs = stmt.executeQuery())
+ {
+ while (rs.next())
+ {
+ String id = rs.getString(1);
+ String objectType = rs.getString(2);
+ String attributes = getBlobAsString(rs, 3);
+
+ try (PreparedStatement insertStmt = connection.prepareStatement(String.format(
+ "INSERT INTO %s ( id, object_type, attributes) VALUES (?,?,?)",
+ getConfiguredObjectsTableName())))
+ {
+ insertStmt.setString(1, id);
+ insertStmt.setString(2, objectType);
+ if (attributes == null)
+ {
+ insertStmt.setNull(3, Types.BLOB);
+ }
+ else
+ {
+ byte[] attributesAsBytes = attributes.getBytes(StandardCharsets.UTF_8);
+
+ try(ByteArrayInputStream bis = new ByteArrayInputStream(attributesAsBytes))
+ {
+ insertStmt.setBinaryStream(3, bis, attributesAsBytes.length);
+ }
+ catch (IOException e)
+ {
+ throw new StoreException("Unexpected exception: " + e.getMessage(), e);
+ }
+ }
+ insertStmt.execute();
+ }
+ }
+ }
+ }
+
+ try (PreparedStatement stmt = connection.prepareStatement(String.format(
+ "SELECT child_id, parent_type, parent_id FROM %s",
+ getVersion1ConfiguredObjectHierarchyTableName())))
+ {
+ try (ResultSet rs = stmt.executeQuery())
+ {
+ while (rs.next())
+ {
+ String childId = rs.getString(1);
+ String parentType = rs.getString(2);
+ String parentId = rs.getString(3);
+
+ try (PreparedStatement insertStmt = connection.prepareStatement(String.format(
+ "INSERT INTO %s ( child_id, parent_type, parent_id) VALUES (?,?,?)",
+ getConfiguredObjectHierarchyTableName())))
+ {
+ insertStmt.setString(1, childId);
+ insertStmt.setString(2, parentType);
+ insertStmt.setString(3, parentId);
+
+ insertStmt.execute();
+ }
+ }
+ }
+ }
+
+ updateConfigStructureVersionTableName(connection, 2);
+ connection.commit();
+
+ }
+ catch(SQLException | RuntimeException e)
+ {
+ try
+ {
+ connection.rollback();
+ }
+ catch(SQLException re)
+ {
+ }
+ throw e;
+ }
+ finally
+ {
+ connection.close();
+ }
+
+ try(Connection c = newAutoCommitConnection())
+ {
+ JdbcUtils.dropTables(c,
+ getLogger(),
+ Arrays.asList(getVersion1ConfiguredObjectHierarchyTableName(),
+ getVersion1ConfiguredObjectsTableName()));
+ }
+ }
+
+ private void updateConfigStructureVersionTableName(Connection conn, int newVersion) throws SQLException
+ {
+ try (PreparedStatement statement = conn.prepareStatement("UPDATE " + getConfigStructureVersionTableName()
+ + " SET version = ?"))
+ {
+ statement.setInt(1, newVersion);
+ statement.execute();
+ }
+ }
+
private void upgradeFromV7(ConfiguredObject<?> parent) throws SQLException
{
@SuppressWarnings("serial")
@@ -446,6 +602,7 @@ public abstract class AbstractJDBCConfigurationStore implements MessageStoreProv
createConfiguredObjectsTable(conn);
createConfiguredObjectHierarchyTable(conn);
+ createConfigurationStructureVersionTable(conn);
}
catch (SQLException e)
{
@@ -457,18 +614,38 @@ public abstract class AbstractJDBCConfigurationStore implements MessageStoreProv
}
}
- private void dropConfigVersionTable(final Connection conn) throws SQLException
+ private void createConfigurationStructureVersionTable(final Connection conn) throws SQLException
{
- if(!tableExists(getConfigurationVersionTableName(), conn))
+ if (!tableExists(getConfigStructureVersionTableName(), conn))
{
- Statement stmt = conn.createStatement();
- try
+ try (Statement stmt = conn.createStatement())
{
- stmt.execute("DROP TABLE " + getConfigurationVersionTableName());
+ stmt.execute(String.format("CREATE TABLE %s ( version int not null )",
+ getConfigStructureVersionTableName()));
}
- finally
+
+ int version = tableExists(getVersion1ConfiguredObjectsTableName(), conn) ? 1 : CONFIG_DB_VERSION;
+ try (PreparedStatement pstmt = conn.prepareStatement(String.format("INSERT INTO %s ( version ) VALUES ( ? )",
+ getConfigStructureVersionTableName())))
{
- stmt.close();
+ pstmt.setInt(1, version);
+ pstmt.execute();
+ }
+ }
+ }
+
+ private void dropConfigVersionTable(final Connection conn) throws SQLException
+ {
+ dropTable(conn, getConfigurationVersionTableName());
+ }
+
+ private void dropTable(final Connection conn, final String tableName) throws SQLException
+ {
+ if(!tableExists(tableName, conn))
+ {
+ try (Statement stmt = conn.createStatement())
+ {
+ stmt.execute(String.format("DROP TABLE %s", tableName));
}
}
}
@@ -848,7 +1025,8 @@ public abstract class AbstractJDBCConfigurationStore implements MessageStoreProv
JdbcUtils.dropTables(conn,
getLogger(),
Arrays.asList(getConfiguredObjectsTableName(),
- getConfiguredObjectHierarchyTableName()));
+ getConfiguredObjectHierarchyTableName(),
+ getConfigStructureVersionTableName()));
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fbeb2084/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStoreTest.java
----------------------------------------------------------------------
diff --git a/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStoreTest.java b/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStoreTest.java
index b55acc3..2e7caf5 100644
--- a/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStoreTest.java
+++ b/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStoreTest.java
@@ -24,13 +24,21 @@ package org.apache.qpid.server.store.jdbc;
import static org.apache.qpid.server.store.jdbc.TestJdbcUtils.assertTablesExistence;
import static org.apache.qpid.server.store.jdbc.TestJdbcUtils.getTableNames;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
import java.sql.Connection;
+import java.sql.PreparedStatement;
import java.sql.SQLException;
+import java.sql.Statement;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
@@ -40,6 +48,8 @@ import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.model.VirtualHostNode;
import org.apache.qpid.server.store.AbstractDurableConfigurationStoreTestCase;
import org.apache.qpid.server.store.DurableConfigurationStore;
+import org.apache.qpid.server.store.StoreException;
+import org.apache.qpid.server.store.handler.ConfiguredObjectRecordHandler;
import org.apache.qpid.server.virtualhostnode.jdbc.JDBCVirtualHostNode;
public class GenericJDBCConfigurationStoreTest extends AbstractDurableConfigurationStoreTestCase
@@ -66,7 +76,7 @@ public class GenericJDBCConfigurationStoreTest extends AbstractDurableConfigurat
@Test
public void testOnDelete() throws Exception
{
- try(Connection connection = openConnection())
+ try (Connection connection = openConnection())
{
GenericJDBCConfigurationStore store = (GenericJDBCConfigurationStore) getConfigurationStore();
Collection<String> expectedTables = Arrays.asList(store.getConfiguredObjectHierarchyTableName(),
@@ -91,6 +101,143 @@ public class GenericJDBCConfigurationStoreTest extends AbstractDurableConfigurat
assertEquals("Delete action was not invoked", true, deleted.get());
}
+ @Test
+ public void testUpgradeStoreStructure() throws Exception
+ {
+ GenericJDBCConfigurationStore store = (GenericJDBCConfigurationStore) getConfigurationStore();
+ store.closeConfigurationStore();
+ store.onDelete(getVirtualHostNode());
+
+ GenericJDBCConfigurationStore store2 = (GenericJDBCConfigurationStore) createConfigStore();
+
+ UUID hostId = UUID.randomUUID();
+ UUID queueId = UUID.randomUUID();
+ try (Connection connection = openConnection())
+ {
+ assertTablesExistence(Arrays.asList("QPID_CONFIGURED_OBJECTS", "QPID_CONFIGURED_OBJECT_HIERARCHY"),
+ getTableNames(connection), false);
+ try (Statement stmt = connection.createStatement())
+ {
+ stmt.execute("CREATE TABLE QPID_CONFIGURED_OBJECTS ( id VARCHAR(36) not null,"
+ + " object_type varchar(255), attributes blob, PRIMARY KEY (id))");
+ stmt.execute("CREATE TABLE QPID_CONFIGURED_OBJECT_HIERARCHY ( child_id VARCHAR(36) not null,"
+ + " parent_type varchar(255), parent_id VARCHAR(36), PRIMARY KEY (child_id, parent_type))");
+ }
+
+ try (PreparedStatement insertStmt = connection.prepareStatement(
+ "INSERT INTO QPID_CONFIGURED_OBJECTS ( id, object_type, attributes) VALUES (?,?,?)"))
+ {
+ insertStmt.setString(1, hostId.toString());
+ insertStmt.setString(2, "VirtualHost");
+ final byte[] attributesAsBytes = "{\"name\":\"testHost\"}".getBytes(StandardCharsets.UTF_8);
+ try (ByteArrayInputStream bis = new ByteArrayInputStream(attributesAsBytes))
+ {
+ insertStmt.setBinaryStream(3, bis, attributesAsBytes.length);
+ }
+ insertStmt.execute();
+ }
+ try (PreparedStatement insertStmt = connection.prepareStatement(
+ "INSERT INTO QPID_CONFIGURED_OBJECTS ( id, object_type, attributes) VALUES (?,?,?)"))
+ {
+ insertStmt.setString(1, queueId.toString());
+ insertStmt.setString(2, "Queue");
+ final byte[] attributesAsBytes = "{\"name\":\"testQueue\"}".getBytes(StandardCharsets.UTF_8);
+ try (ByteArrayInputStream bis = new ByteArrayInputStream(attributesAsBytes))
+ {
+ insertStmt.setBinaryStream(3, bis, attributesAsBytes.length);
+ }
+ insertStmt.execute();
+ }
+
+ try (PreparedStatement insertStmt = connection.prepareStatement(
+ "INSERT INTO QPID_CONFIGURED_OBJECT_HIERARCHY "
+ + " ( child_id, parent_type, parent_id) VALUES (?,?,?)"))
+ {
+ insertStmt.setString(1, queueId.toString());
+ insertStmt.setString(2, "VirtualHost");
+ insertStmt.setString(3, hostId.toString());
+
+ insertStmt.execute();
+ }
+ }
+
+ store2.init(getVirtualHostNode());
+ store2.upgradeStoreStructure();
+
+ try (Connection connection = openConnection())
+ {
+ try
+ {
+ assertTablesExistence(Arrays.asList("QPID_CONFIGURED_OBJECTS", "QPID_CONFIGURED_OBJECT_HIERARCHY"),
+ getTableNames(connection), false);
+
+ assertTablesExistence(Collections.singletonList("QPID_CFG_VERSION"), getTableNames(connection), true);
+ }
+ finally
+ {
+ JdbcUtils.dropTables(connection,
+ store2.getLogger(),
+ Arrays.asList("QPID_CONFIGURED_OBJECTS", "QPID_CONFIGURED_OBJECT_HIERARCHY"));
+ }
+ }
+
+ ConfiguredObjectRecordHandler handler = mock(ConfiguredObjectRecordHandler.class);
+ store2.openConfigurationStore(handler);
+
+ verify(handler).handle(matchesRecord(hostId, "VirtualHost", map("name", "testHost")));
+ verify(handler).handle(matchesRecord(queueId,
+ "Queue",
+ map("name", "testQueue"),
+ Collections.singletonMap("VirtualHost", hostId)));
+ }
+
+ @Test
+ public void testUpgradeStoreStructureFromUnknownVersion() throws Exception
+ {
+ GenericJDBCConfigurationStore store = (GenericJDBCConfigurationStore) getConfigurationStore();
+ store.closeConfigurationStore();
+ store.onDelete(getVirtualHostNode());
+
+ GenericJDBCConfigurationStore store2 = (GenericJDBCConfigurationStore) createConfigStore();
+
+ try (Connection connection = openConnection())
+ {
+ assertTablesExistence(Collections.singletonList("QPID_CFG_VERSION"),
+ getTableNames(connection), false);
+ try (Statement stmt = connection.createStatement())
+ {
+ stmt.execute("CREATE TABLE QPID_CFG_VERSION ( version int not null )");
+ }
+ try (PreparedStatement insertStmt = connection.prepareStatement(
+ "INSERT INTO QPID_CFG_VERSION ( version) VALUES (?)"))
+ {
+ insertStmt.setInt(1, 0);
+ insertStmt.execute();
+ }
+ }
+
+ store2.init(getVirtualHostNode());
+
+ try
+ {
+ store2.upgradeStoreStructure();
+ fail("Exception is expected");
+ }
+ catch (StoreException e)
+ {
+ // pass
+ }
+ finally
+ {
+ try (Connection connection = openConnection())
+ {
+ JdbcUtils.dropTables(connection,
+ store2.getLogger(),
+ Arrays.asList("QPID_CFG_VERSION"));
+ }
+ }
+ }
+
@Override
protected VirtualHostNode createVirtualHostNode(final String storeLocation, final ConfiguredObjectFactory factory)
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/2] qpid-broker-j git commit: QPID-8216: [Broker-J] Restore
reporting of pperational log message CHN-1011 about moving message to dead
letter queue
Posted by or...@apache.org.
QPID-8216: [Broker-J] Restore reporting of pperational log message CHN-1011 about moving message to dead letter queue
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/38d5844e
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/38d5844e
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/38d5844e
Branch: refs/heads/master
Commit: 38d5844e5df591e8a2850f6e302763a764cc50c8
Parents: fbeb208
Author: Alex Rudyy <or...@apache.org>
Authored: Thu Jul 5 14:13:58 2018 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Thu Jul 5 14:13:58 2018 +0100
----------------------------------------------------------------------
.../qpid/server/queue/QueueEntryImpl.java | 2 +-
.../server/queue/QueueEntryImplTestBase.java | 27 ++++++++++++++++++++
2 files changed, 28 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/38d5844e/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
index aed7cfe..51cee72 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
@@ -615,7 +615,7 @@ public abstract class QueueEntryImpl implements QueueEntry
}
});
- int enqueues = result.send(txn, null);
+ int enqueues = result.send(txn, action);
if(autocommit)
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/38d5844e/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java b/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
index 5f41c45..1bfafea 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
@@ -32,6 +32,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Field;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@@ -45,9 +46,11 @@ import org.apache.qpid.server.message.MessageInstance.StealableConsumerAcquiredS
import org.apache.qpid.server.message.MessageInstance.UnstealableConsumerAcquiredState;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.AlternateBinding;
import org.apache.qpid.server.model.BrokerTestHelper;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.store.TransactionLogResource;
+import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.StateChangeListener;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
import org.apache.qpid.test.utils.UnitTestBase;
@@ -358,4 +361,28 @@ public abstract class QueueEntryImplTestBase extends UnitTestBase
next = next.getNextValidEntry();
assertNull("The next entry after the last should be null", next);
}
+
+ @Test
+ public void testRouteToAlternateInvokesAction()
+ {
+ String dlqName = "dlq";
+ Map<String, Object> dlqAttributes = new HashMap<>();
+ dlqAttributes.put(Queue.ID, UUID.randomUUID());
+ dlqAttributes.put(Queue.NAME, dlqName);
+ Queue<?> dlq = (Queue<?>) _queueEntry.getQueue().getVirtualHost().createChild(Queue.class, dlqAttributes);
+
+ final AlternateBinding alternateBinding = mock(AlternateBinding.class);
+ when(alternateBinding.getDestination()).thenReturn(dlqName);
+ _queueEntry.getQueue().setAttributes(Collections.singletonMap(Queue.ALTERNATE_BINDING, alternateBinding));
+
+ final Action<? super MessageInstance> action = mock(Action.class);
+ when(_queueEntry.getMessage().isResourceAcceptable(dlq)).thenReturn(true);
+ _queueEntry.acquire();
+ int enqueues = _queueEntry.routeToAlternate(action, null);
+
+ assertEquals("Unexpected number of enqueues", 1, enqueues);
+ verify(action).performAction(any());
+
+ assertEquals("Unexpected number of messages on DLQ", 1, dlq.getQueueDepthMessages());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org