You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2013/12/13 19:50:15 UTC
svn commit: r1550805 - in /qpid/branches/java-broker-bdb-ha/qpid/java:
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/
bdbstore/src/test/java/org/apache/qpid/se...
Author: kwall
Date: Fri Dec 13 18:50:15 2013
New Revision: 1550805
URL: http://svn.apache.org/r1550805
Log:
QPID-5411: Initial code introducing replication node into broker model
Added:
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java
qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/ReplicationNode.java
qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/replication/
qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/replication/ReplicationGroupListener.java
Modified:
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java
qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java
qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/UUIDGenerator.java
qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHost.java Fri Dec 13 18:50:15 2013
@@ -32,6 +32,7 @@ import org.apache.qpid.server.logging.ac
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.subjects.MessageStoreLogSubject;
import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.replication.ReplicationGroupListener;
import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.store.DurableConfigurationRecoverer;
import org.apache.qpid.server.store.DurableConfigurationStore;
@@ -80,8 +81,6 @@ public class BDBHAVirtualHost extends Ab
_messageStore.addEventListener(new AfterActivationListener(), Event.AFTER_ACTIVATE);
_messageStore.addEventListener(new BeforeCloseListener(), Event.BEFORE_CLOSE);
-
-
_messageStore.addEventListener(new AfterInitialisationListener(), Event.AFTER_INIT);
_messageStore.addEventListener(new BeforePassivationListener(), Event.BEFORE_PASSIVATE);
@@ -99,7 +98,10 @@ public class BDBHAVirtualHost extends Ab
recoveryHandler
);
- ((ReplicatedEnvironmentFacade)_messageStore.getEnvironmentFacade()).setStateChangeListener(new BDBHAMessageStoreStateChangeListener());
+ // Make the virtualhost model object a replication group listener
+ ReplicatedEnvironmentFacade environmentFacade = (ReplicatedEnvironmentFacade) _messageStore.getEnvironmentFacade();
+ environmentFacade.setReplicationGroupListener((ReplicationGroupListener) virtualHost);
+ environmentFacade.setStateChangeListener(new BDBHAMessageStoreStateChangeListener());
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacade.java Fri Dec 13 18:50:15 2013
@@ -37,7 +37,10 @@ import java.util.concurrent.atomic.Atomi
import org.apache.log4j.Logger;
import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.replication.ReplicationGroupListener;
import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNode;
+import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNodeFactory;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
@@ -56,6 +59,7 @@ import com.sleepycat.je.rep.NetworkResto
import com.sleepycat.je.rep.NetworkRestoreConfig;
import com.sleepycat.je.rep.ReplicatedEnvironment;
import com.sleepycat.je.rep.ReplicationConfig;
+import com.sleepycat.je.rep.ReplicationGroup;
import com.sleepycat.je.rep.ReplicationMutableConfig;
import com.sleepycat.je.rep.ReplicationNode;
import com.sleepycat.je.rep.StateChangeEvent;
@@ -64,6 +68,8 @@ import com.sleepycat.je.rep.util.Replica
public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChangeListener
{
+ private static final Logger LOGGER = Logger.getLogger(ReplicatedEnvironmentFacade.class);
+
@SuppressWarnings("serial")
private static final Map<String, String> REPCONFIG_DEFAULTS = Collections.unmodifiableMap(new HashMap<String, String>()
{{
@@ -101,7 +107,6 @@ public class ReplicatedEnvironmentFacade
put(ReplicationConfig.ENV_UNKNOWN_STATE_TIMEOUT, "5 s");
}});
- private static final Logger LOGGER = Logger.getLogger(ReplicatedEnvironmentFacade.class);
public static final String TYPE = "BDB-HA";
// TODO: get rid of these names
@@ -111,28 +116,34 @@ public class ReplicatedEnvironmentFacade
private volatile ReplicatedEnvironment _environment;
private CommitThreadWrapper _commitThreadWrapper;
- private String _groupName;
- private String _nodeName;
- private String _nodeHostPort;
- private String _helperHostPort;
- private Durability _durability;
- private boolean _designatedPrimary;
- private boolean _coalescingSync;
+ private final String _groupName;
+ private final String _nodeName;
+ private final String _nodeHostPort;
+ private final String _helperHostPort;
+ private final Durability _durability;
+ private final boolean _designatedPrimary;
+ private final boolean _coalescingSync;
private volatile StateChangeListener _stateChangeListener;
- private String _environmentPath;
- private Map<String, String> _environmentParameters;
- private Map<String, String> _replicationEnvironmentParameters;
- private String _name;
- private ExecutorService _executor = Executors.newFixedThreadPool(1);
- private AtomicReference<State> _state = new AtomicReference<State>(State.INITIAL);
+ private final String _environmentPath;
+ private final Map<String, String> _environmentParameters;
+ private final Map<String, String> _replicationEnvironmentParameters;
+ private final String _name;
+ private final ExecutorService _executor = Executors.newFixedThreadPool(1);
+ private final AtomicReference<State> _state = new AtomicReference<State>(State.INITIAL);
private final ConcurrentMap<String, Database> _databases = new ConcurrentHashMap<String, Database>();
- public ReplicatedEnvironmentFacade(String name, String environmentPath, String groupName, String nodeName, String nodeHostPort,
- String helperHostPort, Durability durability, boolean designatedPrimary, boolean coalescingSync,
- Map<String, String> environmentParameters, Map<String, String> replicationEnvironmentParameters)
- {
+ private ReplicationGroupListener _replicationGroupListener;
+ private final RemoteReplicationNodeFactory _remoteReplicationNodeFactory;
+
+ public ReplicatedEnvironmentFacade(String name, String environmentPath,
+ String groupName, String nodeName, String nodeHostPort,
+ String helperHostPort, Durability durability,
+ Boolean designatedPrimary, Boolean coalescingSync,
+ Map<String, String> envConfigMap,
+ Map<String, String> replicationConfig, RemoteReplicationNodeFactory remoteReplicationNodeFactory)
+ {
_name = name;
_environmentPath = environmentPath;
_groupName = groupName;
@@ -142,15 +153,17 @@ public class ReplicatedEnvironmentFacade
_durability = durability;
_designatedPrimary = designatedPrimary;
_coalescingSync = coalescingSync;
- _environmentParameters = environmentParameters;
- _replicationEnvironmentParameters = replicationEnvironmentParameters;
+ _environmentParameters = envConfigMap;
+ _replicationEnvironmentParameters = replicationConfig;
+ _remoteReplicationNodeFactory = remoteReplicationNodeFactory;
_state.set(State.OPENING);
_environment = createEnvironment(environmentPath, groupName, nodeName, nodeHostPort, helperHostPort, durability,
- designatedPrimary, environmentParameters, replicationEnvironmentParameters);
- startCommitThread(name, _environment);
+ designatedPrimary, _environmentParameters, _replicationEnvironmentParameters);
+ startCommitThread(_name, _environment);
}
+
@Override
public StoreFuture commit(final Transaction tx, final boolean syncCommit) throws AMQStoreException
{
@@ -430,6 +443,42 @@ public class ReplicatedEnvironmentFacade
return _state.get();
}
+ /**
+ * Sets the replication group listener. Whenever a new listener is set, the listener
+ * will hear {@link ReplicationGroupListener#onReplicationNodeRecovered(org.apache.qpid.server.model.ReplicationNode)
+ * for every existing remote node.
+ *
+ * @param replicationGroupListener listener
+ */
+ public void setReplicationGroupListener(ReplicationGroupListener replicationGroupListener)
+ {
+ _replicationGroupListener = replicationGroupListener;
+ if (_replicationGroupListener != null)
+ {
+ notifyExistingRemoteReplicationNodes(_replicationGroupListener);
+ }
+ }
+
+ private void notifyExistingRemoteReplicationNodes(ReplicationGroupListener listener)
+ {
+ ReplicationGroup group = _environment.getGroup();
+ Set<ReplicationNode> nodes = new HashSet<ReplicationNode>(group.getElectableNodes());
+ String localNodeName = getNodeName();
+
+ for (ReplicationNode replicationNode : nodes)
+ {
+ String discoveredNodeName = replicationNode.getName();
+ if (!discoveredNodeName.equals(localNodeName))
+ {
+ // TODO remote replication nodes should be cached
+ RemoteReplicationNode remoteNode = _remoteReplicationNodeFactory.create(group.getName(),
+ replicationNode.getName(),
+ replicationNode.getHostName(), replicationNode.getPort());
+ listener.onReplicationNodeRecovered(remoteNode);
+ }
+ }
+ }
+
private ReplicationGroupAdmin createReplicationGroupAdmin()
{
final Set<InetSocketAddress> helpers = new HashSet<InetSocketAddress>();
@@ -652,5 +701,4 @@ public class ReplicatedEnvironmentFacade
CLOSING,
CLOSED
}
-
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeFactory.java Fri Dec 13 18:50:15 2013
@@ -25,6 +25,8 @@ import java.util.Map;
import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNode;
+import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNodeFactory;
import com.sleepycat.je.Durability;
import com.sleepycat.je.Durability.ReplicaAckPolicy;
@@ -81,7 +83,7 @@ public class ReplicatedEnvironmentFacade
}
return new ReplicatedEnvironmentFacade(name, storeLocation, groupName, nodeName, nodeHostPort, helperHostPort, durability,
- designatedPrimary, coalescingSync, envConfigMap, replicationConfig);
+ designatedPrimary, coalescingSync, envConfigMap, replicationConfig, new RemoteReplicationNodeFactoryImpl(virtualHost));
}
private String getValidatedStringAttribute(org.apache.qpid.server.model.VirtualHost virtualHost, String attributeName)
@@ -126,4 +128,19 @@ public class ReplicatedEnvironmentFacade
return defaultVal;
}
+ private static class RemoteReplicationNodeFactoryImpl implements RemoteReplicationNodeFactory
+ {
+ private VirtualHost _virtualHost;
+
+ public RemoteReplicationNodeFactoryImpl(VirtualHost virtualHost)
+ {
+ _virtualHost = virtualHost;
+ }
+
+ @Override
+ public RemoteReplicationNode create(String groupName, String nodeName, String host, int port)
+ {
+ return new RemoteReplicationNode(groupName, nodeName, host, port, _virtualHost);
+ }
+ }
}
Added: qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java?rev=1550805&view=auto
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java (added)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNode.java Fri Dec 13 18:50:15 2013
@@ -0,0 +1,220 @@
+package org.apache.qpid.server.store.berkeleydb.replication;
+
+import java.security.AccessControlException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.qpid.server.model.ConfigurationChangeListener;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.IllegalStateTransitionException;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.ReplicationNode;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.model.UUIDGenerator;
+import org.apache.qpid.server.model.VirtualHost;
+
+/**
+ * Represents a remote replication node in a BDB group.
+ */
+public class RemoteReplicationNode implements ReplicationNode
+{
+
+ private final UUID _id;
+ private final String _groupName;
+ private final String _nodeName;
+ private final String _host;
+ private final int _port;
+ private final VirtualHost _virtualHost;
+
+ public RemoteReplicationNode(String groupName, String nodeName, String host, int port, VirtualHost virtualHost)
+ {
+ super();
+ _id = UUIDGenerator.generateReplicationNodeId(groupName, nodeName);
+ _groupName = groupName;
+ _nodeName = nodeName;
+ _host = host;
+ _port = port;
+ _virtualHost = virtualHost;
+ }
+
+ @Override
+ public UUID getId()
+ {
+ return _id;
+ }
+
+ @Override
+ public String getName()
+ {
+ return _nodeName;
+ }
+
+ @Override
+ public String setName(String currentName, String desiredName)
+ throws IllegalStateException, AccessControlException
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public State getDesiredState()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public State setDesiredState(State currentState, State desiredState)
+ throws IllegalStateTransitionException, AccessControlException
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public State getActualState()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void addChangeListener(ConfigurationChangeListener listener)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean removeChangeListener(ConfigurationChangeListener listener)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T extends ConfiguredObject> T getParent(Class<T> clazz)
+ {
+ if (clazz == VirtualHost.class)
+ {
+ return (T) _virtualHost;
+ }
+ throw new IllegalArgumentException();
+ }
+
+ @Override
+ public boolean isDurable()
+ {
+ return true;
+ }
+
+ @Override
+ public void setDurable(boolean durable) throws IllegalStateException,
+ AccessControlException, IllegalArgumentException
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public LifetimePolicy getLifetimePolicy()
+ {
+ return LifetimePolicy.PERMANENT;
+ }
+
+ @Override
+ public LifetimePolicy setLifetimePolicy(LifetimePolicy expected,
+ LifetimePolicy desired) throws IllegalStateException,
+ AccessControlException, IllegalArgumentException
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getTimeToLive()
+ {
+ return 0;
+ }
+
+ @Override
+ public long setTimeToLive(long expected, long desired)
+ throws IllegalStateException, AccessControlException,
+ IllegalArgumentException
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Collection<String> getAttributeNames()
+ {
+ return ReplicationNode.AVAILABLE_ATTRIBUTES;
+ }
+
+ @Override
+ public Object getAttribute(String name)
+ {
+ if (ReplicationNode.ID.equals(name))
+ {
+ return getId();
+ }
+ else if (ReplicationNode.NAME.equals(name))
+ {
+ return getName();
+ }
+ else if (ReplicationNode.LIFETIME_POLICY.equals(name))
+ {
+ return getLifetimePolicy();
+ }
+ else if (ReplicationNode.DURABLE.equals(name))
+ {
+ return isDurable();
+ }
+ else if (ReplicationNode.HOST_PORT.equals(name))
+ {
+ return _host + ":" + _port;
+ }
+ else if (ReplicationNode.GROUP_NAME.equals(name))
+ {
+ return _groupName;
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, Object> getActualAttributes()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object setAttribute(String name, Object expected, Object desired)
+ throws IllegalStateException, AccessControlException,
+ IllegalArgumentException
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Statistics getStatistics()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <C extends ConfiguredObject> C createChild(Class<C> childClass,
+ Map<String, Object> attributes, ConfiguredObject... otherParents)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setAttributes(Map<String, Object> attributes)
+ throws IllegalStateException, AccessControlException,
+ IllegalArgumentException
+ {
+ throw new UnsupportedOperationException();
+ }
+}
Added: qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java?rev=1550805&view=auto
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java (added)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/RemoteReplicationNodeFactory.java Fri Dec 13 18:50:15 2013
@@ -0,0 +1,26 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.qpid.server.store.berkeleydb.replication;
+
+public interface RemoteReplicationNodeFactory
+{
+ RemoteReplicationNode create(String groupName, String nodeName, String host, int port);
+}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java Fri Dec 13 18:50:15 2013
@@ -22,11 +22,12 @@ package org.apache.qpid.server.store.ber
import java.io.File;
import java.net.InetAddress;
-
import java.util.HashMap;
import java.util.Map;
+
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
+import org.apache.qpid.server.replication.ReplicationGroupListener;
import org.apache.qpid.server.util.BrokerTestHelper;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -65,7 +66,7 @@ public class BDBHAMessageStoreTest exten
FileUtils.delete(new File(_workDir), true);
_configXml = new XMLConfiguration();
- _modelVhost = mock(org.apache.qpid.server.model.VirtualHost.class);
+ _modelVhost = mock(TestVirtualHost.class);
BrokerTestHelper.setUp();
@@ -95,7 +96,7 @@ public class BDBHAMessageStoreTest exten
String vhostName = "test" + _masterPort;
VirtualHostConfiguration configuration = new VirtualHostConfiguration(vhostName, _configXml.subset("virtualhosts.virtualhost." + vhostName), BrokerTestHelper.createBrokerMock());
- _virtualHost = BrokerTestHelper.createVirtualHost(configuration,null,_modelVhost);
+ _virtualHost = BrokerTestHelper.createVirtualHost(configuration, null, _modelVhost);
BDBMessageStore store = (BDBMessageStore) _virtualHost.getMessageStore();
// test whether JVM system settings were applied
@@ -125,7 +126,7 @@ public class BDBHAMessageStoreTest exten
_configXml.addProperty("virtualhosts.virtualhost.name", vhostName);
_configXml.addProperty(vhostPrefix + ".type", BDBHAVirtualHostFactory.TYPE);
- when(_modelVhost.getAttribute(eq(_modelVhost.STORE_PATH))).thenReturn(_workDir + File.separator
+ when(_modelVhost.getAttribute(eq(org.apache.qpid.server.model.VirtualHost.STORE_PATH))).thenReturn(_workDir + File.separator
+ port);
when(_modelVhost.getAttribute(eq("haGroupName"))).thenReturn(_groupName);
when(_modelVhost.getAttribute(eq("haNodeName"))).thenReturn(nodeName);
@@ -163,4 +164,9 @@ public class BDBHAMessageStoreTest exten
}
return _host + ":" + _masterPort;
}
+
+ // TODO: a temporary work around against issue with casting VirtualHost model object to ReplicationGroupListener in the BDBHAVH
+ public static interface TestVirtualHost extends org.apache.qpid.server.model.VirtualHost, ReplicationGroupListener
+ {
+ }
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/ReplicatedEnvironmentFacadeTest.java Fri Dec 13 18:50:15 2013
@@ -20,6 +20,10 @@
*/
package org.apache.qpid.server.store.berkeleydb;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
import java.io.File;
import java.util.ArrayList;
import java.util.Collections;
@@ -36,6 +40,9 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.replication.ReplicationGroupListener;
+import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNode;
+import org.apache.qpid.server.store.berkeleydb.replication.RemoteReplicationNodeFactory;
import org.apache.qpid.test.utils.QpidTestCase;
import com.sleepycat.bind.tuple.IntegerBinding;
@@ -65,6 +72,7 @@ public class ReplicatedEnvironmentFacade
private static final boolean TEST_DESIGNATED_PRIMARY = true;
private static final boolean TEST_COALESCING_SYNC = true;
private final Map<String, ReplicatedEnvironmentFacade> _nodes = new HashMap<String, ReplicatedEnvironmentFacade>();
+ private RemoteReplicationNodeFactory _remoteReplicationNodeFactory = mock(RemoteReplicationNodeFactory.class);;
public void tearDown() throws Exception
{
@@ -136,6 +144,24 @@ public class ReplicatedEnvironmentFacade
assertEquals("Unexpected group members", expectedGroupMembers, new HashSet<Map<String, String>>(groupMembers));
}
+ public void testReplicationGroupListenerHearsAboutExistingRemoteReplicationNodes() throws Exception
+ {
+ ReplicatedEnvironmentFacade replicatedEnvironmentFacade = getEnvironmentFacade();
+ String nodeName2 = TEST_NODE_NAME + "_2";
+ String host = "localhost";
+ int port = getNextAvailable(TEST_NODE_PORT + 1);
+ String node2NodeHostPort = host + ":" + port;
+ joinReplica(nodeName2, node2NodeHostPort);
+
+ List<Map<String, String>> groupMembers = replicatedEnvironmentFacade.getGroupMembers();
+ assertEquals("Unexpected number of nodes at start of test", 2, groupMembers.size());
+
+ ReplicationGroupListener listener = mock(ReplicationGroupListener.class);
+ replicatedEnvironmentFacade.setReplicationGroupListener(listener);
+ verify(listener).onReplicationNodeRecovered(any(RemoteReplicationNode.class));
+ verify(_remoteReplicationNodeFactory).create(TEST_GROUP_NAME, nodeName2, host, port);
+ }
+
public void testRemoveNodeFromGroup() throws Exception
{
ReplicatedEnvironmentFacade environmentFacade = getEnvironmentFacade();
@@ -418,7 +444,7 @@ public class ReplicatedEnvironmentFacade
ReplicatedEnvironmentFacade ref = new ReplicatedEnvironmentFacade(getName(), nodePath, TEST_GROUP_NAME, nodeName,
nodeHostPort, TEST_NODE_HELPER_HOST_PORT, TEST_DURABILITY, designatedPrimary, TEST_COALESCING_SYNC,
- Collections.<String, String> emptyMap(), repConfig);
+ Collections.<String, String> emptyMap(), repConfig, _remoteReplicationNodeFactory);
return ref;
}
Added: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/ReplicationNode.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/ReplicationNode.java?rev=1550805&view=auto
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/ReplicationNode.java (added)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/ReplicationNode.java Fri Dec 13 18:50:15 2013
@@ -0,0 +1,114 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.qpid.server.model;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+public interface ReplicationNode extends ConfiguredObject
+{
+ String ID = "id";
+ String NAME = "name";
+ String STATE = "state";
+ String CREATED = "created";
+ String DURABLE = "durable";
+ String LIFETIME_POLICY = "lifetimePolicy";
+ String TIME_TO_LIVE = "timeToLive";
+ String TYPE = "type";
+ String UPDATED = "updated";
+
+ /** Name of the group to which this replication node belongs */
+ String GROUP_NAME = "groupName";
+
+ /** Node host name/IP and port separated by semicolon*/
+ String HOST_PORT = "hostPort";
+
+ /** Node helper host name/IP and port separated by semicolon*/
+ String HELPER_HOST_PORT = "helperHostPort";
+
+ /** Durability settings*/
+ String DURABILITY = "durability";
+
+ /** Sync multiple transactions on disc at the same time*/
+ String COALESCING_SYNC = "coalescingSync";
+
+ /** A designated primary setting for 2-nodes group*/
+ String DESIGNATED_PRIMARY = "designatedPrimary";
+
+ /** Node priority*/
+ String PRIORITY = "priority";
+
+ /** The overridden minimum number of group nodes required to commit transaction on this node instead of simple majority*/
+ String QUORUM_OVERRIDE = "quorumOverride";
+
+ /** Node role: MASTER,REPLICA,UNKNOWN,DETACHED*/
+ String ROLE = "role";
+
+ /** Time when node joined the group */
+ String JOIN_TIME = "joinTime";
+
+ /** Last known replication transaction id */
+ String LAST_KNOWN_REPLICATION_TRANSACTION_ID= "lastKnownReplicationTransactionId";
+
+ /** Map with additional implementation specific node settings */
+ String PARAMETERS = "parameters";
+
+ /** Map with additional implementation specific replication parameters*/
+ String REPLICATION_PARAMETERS = "replicationParameters";
+
+ /** Name of the node which is a master at the moment*/
+ String MASTER_NODE = "masterNode";
+
+ /** Store path */
+ String STORE_PATH = "storePath";
+
+ // Attributes
+ public static final Collection<String> AVAILABLE_ATTRIBUTES =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ ID,
+ NAME,
+ TYPE,
+ STATE,
+ DURABLE,
+ LIFETIME_POLICY,
+ TIME_TO_LIVE,
+ CREATED,
+ UPDATED,
+ GROUP_NAME,
+ HOST_PORT,
+ HELPER_HOST_PORT,
+ DURABILITY,
+ COALESCING_SYNC,
+ DESIGNATED_PRIMARY,
+ PRIORITY,
+ QUORUM_OVERRIDE,
+ ROLE,
+ JOIN_TIME,
+ LAST_KNOWN_REPLICATION_TRANSACTION_ID,
+ PARAMETERS,
+ REPLICATION_PARAMETERS,
+ MASTER_NODE,
+ STORE_PATH
+ ));
+
+}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/UUIDGenerator.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/UUIDGenerator.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/UUIDGenerator.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/UUIDGenerator.java Fri Dec 13 18:50:15 2013
@@ -97,4 +97,9 @@ public class UUIDGenerator
{
return createUUID(PreferencesProvider.class.getName(), authenticationProviderName, preferencesProviderName);
}
+
+ public static UUID generateReplicationNodeId(String groupName, String nodeName)
+ {
+ return createUUID(ReplicationNode.class.getName(), groupName, nodeName);
+ }
}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java Fri Dec 13 18:50:15 2013
@@ -131,6 +131,7 @@ public interface VirtualHost extends Con
Collection<Connection> getConnections();
Collection<Queue> getQueues();
Collection<Exchange> getExchanges();
+ Collection<ReplicationNode> getReplicationNodes();
Exchange createExchange(String name, State initialState, boolean durable,
LifetimePolicy lifetime, long ttl, String type, Map<String, Object> attributes)
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java Fri Dec 13 18:50:15 2013
@@ -55,6 +55,7 @@ import org.apache.qpid.server.model.Port
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.QueueType;
+import org.apache.qpid.server.model.ReplicationNode;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.Statistics;
import org.apache.qpid.server.model.UUIDGenerator;
@@ -66,6 +67,7 @@ import org.apache.qpid.server.protocol.A
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.replication.ReplicationGroupListener;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.access.Operation;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
@@ -82,7 +84,7 @@ import org.apache.qpid.server.virtualhos
import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.qpid.server.virtualhost.plugins.QueueExistsException;
-public final class VirtualHostAdapter extends AbstractAdapter implements VirtualHost, VirtualHostListener
+public final class VirtualHostAdapter extends AbstractAdapter implements VirtualHost, VirtualHostListener, ReplicationGroupListener
{
private static final Logger LOGGER = Logger.getLogger(VirtualHostAdapter.class);
@@ -111,6 +113,8 @@ public final class VirtualHostAdapter ex
private final List<VirtualHostAlias> _aliases = new ArrayList<VirtualHostAlias>();
private StatisticsGatherer _brokerStatisticsGatherer;
+ private final List<ReplicationNode> _replicationNodes = new ArrayList<ReplicationNode>();
+
public VirtualHostAdapter(UUID id, Map<String, Object> attributes, Broker broker, StatisticsGatherer brokerStatisticsGatherer, TaskExecutor taskExecutor)
{
super(id, null, MapValueConverter.convert(attributes, ATTRIBUTE_TYPES, false), taskExecutor, false);
@@ -257,6 +261,15 @@ public final class VirtualHostAdapter ex
}
}
+ @Override
+ public Collection<ReplicationNode> getReplicationNodes()
+ {
+ synchronized (_replicationNodes)
+ {
+ return Collections.unmodifiableList(_replicationNodes);
+ }
+ }
+
public Exchange createExchange(Map<String, Object> attributes)
throws AccessControlException, IllegalArgumentException
@@ -1252,4 +1265,11 @@ public final class VirtualHostAdapter ex
throw new AccessControlException("Setting of virtual host attributes is denied");
}
}
+
+ @Override
+ public void onReplicationNodeRecovered(ReplicationNode node)
+ {
+ _replicationNodes.add(node);
+ }
+
}
Added: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/replication/ReplicationGroupListener.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/replication/ReplicationGroupListener.java?rev=1550805&view=auto
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/replication/ReplicationGroupListener.java (added)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/replication/ReplicationGroupListener.java Fri Dec 13 18:50:15 2013
@@ -0,0 +1,59 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.qpid.server.replication;
+
+import org.apache.qpid.server.model.ReplicationNode;
+
+public interface ReplicationGroupListener
+{
+ /**
+ * Fired when a remote replication node is added to a group. This event happens
+ * exactly once just after a new replication node is created.
+ */
+ //void onReplicationNodeAddedToGroup(ReplicationNode node);
+
+ /**
+ * Fired exactly once for each existing remote node. Used to inform the application
+ * on any existing nodes as it starts up for the first time.
+ */
+ void onReplicationNodeRecovered(ReplicationNode node);
+
+ /**
+ * Fired when a remote replication node is (permanently) removed from group. This event
+ * happens exactly once just after the existing replication node is deleted.
+ */
+ //void onReplicationNodeRemovedFromGroup(ReplicationNode node);
+
+ /**
+ * Fired when a remote replication node (that is already a member of the group) joins
+ * the group. This will typically occur when another replication node is started perhaps
+ * because the broker has been started.
+ */
+ //void onReplicationNodeUp();
+
+ /**
+ * Fired when a remote replication node (that is already a member of the group) leaves
+ * the group. This will typically occur when another replication node is stopped perhaps
+ * because its broker has been stopped.
+ */
+ //void onReplicationNodeDown();
+
+}
Modified: qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java?rev=1550805&r1=1550804&r2=1550805&view=diff
==============================================================================
--- qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java (original)
+++ qpid/branches/java-broker-bdb-ha/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java Fri Dec 13 18:50:15 2013
@@ -25,7 +25,6 @@ import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledFuture;
import org.apache.qpid.AMQException;
-import org.apache.qpid.AMQSecurityException;
import org.apache.qpid.common.Closeable;
import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.server.connection.IConnectionRegistry;
@@ -33,7 +32,6 @@ import org.apache.qpid.server.exchange.E
import org.apache.qpid.server.plugin.ExchangeType;
import org.apache.qpid.server.protocol.LinkRegistry;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.store.DurableConfigurationStore;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org