replica will be considered
- * in the result of {@link #getReplicaInfo()} when this call has succeeded.
+ * Registers a replica with this master instance. The replica will be considered in the result of
+ * {@link #getReplicaInfo()} when this call has succeeded.
*/
- void registerReplica(ReplicaDescriptor replica) throws JMSException;
+ void registerReplica(ReplicaDescriptor replica);
- void unregisterReplica(ReplicaDescriptor replica) throws JMSException;
+ void unregisterReplica(ReplicaDescriptor replica) throws IOException;
/**
* For a replica replicating off this master, provides statistics in the form of number of operations sent to that
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Activator.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Activator.java
index ef23c840cff..ae8231604b6 100644
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Activator.java
+++ b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Activator.java
@@ -1,7 +1,5 @@
package com.sap.sailing.server.replication.impl;
-import java.io.File;
-import java.io.FileNotFoundException;
import java.util.logging.Logger;
import org.osgi.framework.BundleActivator;
@@ -12,72 +10,25 @@ import com.sap.sailing.server.replication.ReplicationService;
public class Activator implements BundleActivator {
private static final Logger logger = Logger.getLogger(Activator.class.getName());
- private static final String REPLICATION_PERSISTENCE_DIR_PROPERTY = "replication.persistenceDir";
-
- private static final String BROKER_URL_PROPERTY = "replication.brokerURL";
-
- private static final String REPLICATION_USE_JMX_PROPERTY = "replication.useJMX";
-
- private MessageBrokerManager messageBrokerManager;
+ private static final String PROPERTY_NAME_EXCHANGE_NAME = "replication.exchangeName";
private ReplicationInstancesManager replicationInstancesManager;
private static BundleContext defaultContext;
-
+
public void start(BundleContext bundleContext) throws Exception {
defaultContext = bundleContext;
- String replicationPersistenceDirectory = bundleContext.getProperty(REPLICATION_PERSISTENCE_DIR_PROPERTY);
- if (replicationPersistenceDirectory == null) {
- replicationPersistenceDirectory = System.getProperty("java.io.tmpdir");
+ String exchangeName = bundleContext.getProperty(PROPERTY_NAME_EXCHANGE_NAME);
+ if (exchangeName == null) {
+ exchangeName = "sapsailinganalytics";
}
- String brokerURL = bundleContext.getProperty(BROKER_URL_PROPERTY);
- if (brokerURL == null) {
- brokerURL = "tcp://localhost:61616";
- }
- final File brokerPersistenceDir = new File(replicationPersistenceDirectory, "kahadb");
- removeTemporaryTestBrokerPersistenceDirectory(brokerPersistenceDir);
- MessageBrokerConfiguration brokerConfig = new MessageBrokerConfiguration("SailingServerReplicationBroker",
- brokerURL, brokerPersistenceDir.getAbsolutePath());
- messageBrokerManager = new MessageBrokerManager(brokerConfig);
- String useJMX = bundleContext.getProperty(REPLICATION_USE_JMX_PROPERTY);
- if (useJMX == null || useJMX.length() == 0) {
- useJMX = "false";
- }
- messageBrokerManager.startMessageBroker(Boolean.valueOf(useJMX));
- messageBrokerManager.createAndStartConnection();
replicationInstancesManager = new ReplicationInstancesManager();
- ReplicationService serverReplicationMasterService = new ReplicationServiceImpl(replicationInstancesManager, messageBrokerManager);
+ ReplicationService serverReplicationMasterService = new ReplicationServiceImpl(exchangeName, replicationInstancesManager);
bundleContext.registerService(ReplicationService.class, serverReplicationMasterService, null);
+ logger.info("Registered replication service "+serverReplicationMasterService);
}
- public static void removeTemporaryTestBrokerPersistenceDirectory(File brokerPersistenceDir) throws FileNotFoundException {
- if (brokerPersistenceDir.exists() && brokerPersistenceDir.isDirectory()) {
- logger.info("Deleting message broker persistence director "+brokerPersistenceDir);
- deleteRecursive(brokerPersistenceDir);
- }
- File failoverStore = new File("activemq-data");
- if (failoverStore.exists() && failoverStore.isDirectory()) {
- logger.info("Deleting message broker failover store "+failoverStore);
- deleteRecursive(failoverStore);
- }
- }
-
- public static boolean deleteRecursive(File path) throws FileNotFoundException{
- if (!path.exists()) throw new FileNotFoundException(path.getAbsolutePath());
- boolean ret = true;
- if (path.isDirectory()){
- for (File f : path.listFiles()){
- ret = ret && deleteRecursive(f);
- }
- }
- return ret && path.delete();
- }
-
-
public void stop(BundleContext bundleContext) throws Exception {
- messageBrokerManager.closeSessions();
- messageBrokerManager.closeConnections();
- messageBrokerManager.stopMessageBroker();
}
public static BundleContext getDefaultContext() {
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/MessageBrokerConfiguration.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/MessageBrokerConfiguration.java
deleted file mode 100644
index 73b9a03a5ec..00000000000
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/MessageBrokerConfiguration.java
+++ /dev/null
@@ -1,32 +0,0 @@
-package com.sap.sailing.server.replication.impl;
-
-/**
- * A simple configuration for the message broker
- */
-public class MessageBrokerConfiguration {
- private final String brokerName;
-
- private final String brokerUrl;
-
- private final String dataStoreDirectory;
-
- public MessageBrokerConfiguration(String brokerName, String brokerUrl, String dataStoreDirectory) {
- super();
- this.brokerName = brokerName;
- this.brokerUrl = brokerUrl;
- this.dataStoreDirectory = dataStoreDirectory;
- }
-
- public String getBrokerName() {
- return brokerName;
- }
-
- public String getDataStoreDirectory() {
- return dataStoreDirectory;
- }
-
- public String getBrokerUrl() {
- return brokerUrl;
- }
-
-}
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/MessageBrokerManager.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/MessageBrokerManager.java
deleted file mode 100644
index 7e472639180..00000000000
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/MessageBrokerManager.java
+++ /dev/null
@@ -1,73 +0,0 @@
-package com.sap.sailing.server.replication.impl;
-
-import java.net.URI;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Session;
-
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.TransportConnector;
-
-public class MessageBrokerManager {
- private final MessageBrokerConfiguration configuration;
-
- private ActiveMQConnectionFactory connectionFactory;
- private Connection connection;
- private Session session;
-
- private BrokerService broker;
-
- public MessageBrokerManager(final MessageBrokerConfiguration configuration) {
- this.configuration = configuration;
- }
-
- public void startMessageBroker(boolean useJmx) throws Exception {
- broker = new BrokerService();
- broker.setBrokerName(configuration.getBrokerName());
- if (configuration.getDataStoreDirectory() != null) {
- broker.setDataDirectory(configuration.getDataStoreDirectory());
- }
- broker.setUseJmx(useJmx);
- TransportConnector transportConnector = new TransportConnector();
- transportConnector.setUri(new URI(configuration.getBrokerUrl()));
- broker.addConnector(transportConnector);
- broker.start();
- }
-
- public void stopMessageBroker() throws Exception {
- if (broker != null) {
- broker.stop();
- }
- }
-
- public void createAndStartConnection() throws JMSException {
- connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
- ActiveMQConnection.DEFAULT_PASSWORD, configuration.getBrokerUrl());
- connection = connectionFactory.createConnection();
- connection.start();
- }
-
- public void closeConnections() throws JMSException {
- if (connection != null) {
- connection.close();
- }
- }
-
- public Session createSession(boolean transacted) throws JMSException {
- session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
- return session;
- }
-
- public void closeSessions() throws JMSException {
- if (session != null) {
- session.close();
- }
- }
-
- public Session getSession() {
- return session;
- }
-}
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationFactoryImpl.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationFactoryImpl.java
index 0fb74745e07..50837ce3b7b 100755
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationFactoryImpl.java
+++ b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationFactoryImpl.java
@@ -4,10 +4,9 @@ import com.sap.sailing.server.replication.ReplicationFactory;
import com.sap.sailing.server.replication.ReplicationMasterDescriptor;
public class ReplicationFactoryImpl implements ReplicationFactory {
-
@Override
- public ReplicationMasterDescriptor createReplicationMasterDescriptor(String hostname, int servletPort, int jmsPort) {
- return new ReplicationMasterDescriptorImpl(hostname, servletPort, jmsPort);
+ public ReplicationMasterDescriptor createReplicationMasterDescriptor(String hostname, String exchangeName, int servletPort, int messagingPort) {
+ return new ReplicationMasterDescriptorImpl(hostname, exchangeName, servletPort, messagingPort);
}
}
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationMasterDescriptorImpl.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationMasterDescriptorImpl.java
index 3086d48ea40..b4ef987e6af 100755
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationMasterDescriptorImpl.java
+++ b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationMasterDescriptorImpl.java
@@ -1,32 +1,27 @@
package com.sap.sailing.server.replication.impl;
-import java.net.InetAddress;
+import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
-import java.net.UnknownHostException;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Session;
-import javax.jms.Topic;
-import javax.jms.TopicSubscriber;
-
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQConnectionFactory;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.QueueingConsumer;
import com.sap.sailing.server.replication.ReplicationMasterDescriptor;
-import com.sap.sailing.server.replication.ReplicationService;
public class ReplicationMasterDescriptorImpl implements ReplicationMasterDescriptor {
private static final String REPLICATION_SERVLET = "/replication/replication";
private final String hostname;
+ private final String exchangeName;
private final int servletPort;
private final int jmsPort;
- public ReplicationMasterDescriptorImpl(String hostname, int servletPort, int jmsPort) {
+ public ReplicationMasterDescriptorImpl(String hostname, String exchangeName, int servletPort, int jmsPort) {
this.hostname = hostname;
this.servletPort = servletPort;
this.jmsPort = jmsPort;
+ this.exchangeName = exchangeName;
}
@Override
@@ -42,15 +37,17 @@ public class ReplicationMasterDescriptorImpl implements ReplicationMasterDescrip
}
@Override
- public TopicSubscriber getTopicSubscriber(String clientID) throws JMSException, UnknownHostException {
- ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
- ActiveMQConnection.DEFAULT_PASSWORD, "tcp://" + hostname + ":" + jmsPort);
- connectionFactory.setClientID(clientID);
- Connection connection = connectionFactory.createConnection();
- connection.start();
- Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Topic topic = session.createTopic(ReplicationService.SAILING_SERVER_REPLICATION_TOPIC);
- return session.createDurableSubscriber(topic, InetAddress.getLocalHost().getHostAddress());
+ public QueueingConsumer getConsumer() throws IOException {
+ ConnectionFactory connectionFactory = new ConnectionFactory();
+ connectionFactory.setHost(getHostname());
+ Connection connection = connectionFactory.newConnection();
+ Channel channel = connection.createChannel();
+ channel.exchangeDeclare(exchangeName, "fanout");
+ QueueingConsumer consumer = new QueueingConsumer(channel);
+ String queueName = channel.queueDeclare().getQueue();
+ channel.queueBind(queueName, exchangeName, "");
+ channel.basicConsume(queueName, consumer);
+ return consumer;
}
@Override
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationServiceImpl.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationServiceImpl.java
index 57825a57263..a4ac435d8e0 100644
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationServiceImpl.java
+++ b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationServiceImpl.java
@@ -11,16 +11,12 @@ import java.net.URLConnection;
import java.util.HashMap;
import java.util.Map;
-import javax.jms.BytesMessage;
-import javax.jms.DeliveryMode;
-import javax.jms.JMSException;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.Topic;
-import javax.jms.TopicSubscriber;
-
import org.osgi.util.tracker.ServiceTracker;
+import com.rabbitmq.client.AMQP.Exchange;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.QueueingConsumer;
import com.sap.sailing.server.OperationExecutionListener;
import com.sap.sailing.server.RacingEventService;
import com.sap.sailing.server.RacingEventServiceOperation;
@@ -31,8 +27,8 @@ import com.sap.sailing.server.replication.ReplicationService;
/**
* Can observe a {@link RacingEventService} for the operations it performs that require replication. Only observes as
* long as there are replicas registered. If the last replica is de-registered, the service stops observing the
- * {@link RacingEventService}. Operations received that require replication are broadcast to the
- * {@link #getReplicationTopic() replication topic}.
+ * {@link RacingEventService}. Operations received that require replication are sent to the {@link Exchange} to which
+ * replica queues can bind. The exchange name is provided to this service during construction.
*
*
* This service object {@link RacingEventService#addOperationExecutionListener(OperationExecutionListener) registers} as
@@ -46,12 +42,6 @@ import com.sap.sailing.server.replication.ReplicationService;
public class ReplicationServiceImpl implements ReplicationService, OperationExecutionListener, HasRacingEventService {
private final ReplicationInstancesManager replicationInstancesManager;
- private final MessageBrokerManager messageBrokerManager;
-
- private MessageProducer messageProducer;
-
- private Topic replicationTopic;
-
private ServiceTrackerGET request, sends a serialized copy of the {@link RacingEventService} to
@@ -60,11 +59,7 @@ public class ReplicationServlet extends SailingServerHttpServlet {
String action = req.getParameter(ACTION);
switch (Action.valueOf(action)) {
case REGISTER:
- try {
- registerClientWithReplicationService(req, resp);
- } catch (JMSException e) {
- resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, e.getMessage());
- }
+ registerClientWithReplicationService(req, resp);
break;
case INITIAL_LOAD:
ObjectOutputStream oos = new ObjectOutputStream(resp.getOutputStream());
@@ -84,7 +79,7 @@ public class ReplicationServlet extends SailingServerHttpServlet {
}
private void registerClientWithReplicationService(HttpServletRequest req, HttpServletResponse resp)
- throws JMSException, IOException {
+ throws IOException {
ReplicaDescriptor replica = getReplicaDescriptor(req);
getReplicationService().registerReplica(replica);
resp.setContentType("text/plain");
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Replicator.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Replicator.java
index 48f89d11783..294d93be3a6 100755
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Replicator.java
+++ b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/Replicator.java
@@ -6,12 +6,12 @@ import java.io.ObjectInputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.logging.Logger;
-import javax.jms.BytesMessage;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-
+import com.rabbitmq.client.ConsumerCancelledException;
+import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.QueueingConsumer.Delivery;
+import com.rabbitmq.client.ShutdownSignalException;
import com.sap.sailing.domain.base.DomainFactory;
import com.sap.sailing.server.RacingEventService;
import com.sap.sailing.server.RacingEventServiceOperation;
@@ -30,10 +30,13 @@ import com.sap.sailing.server.replication.ReplicationMasterDescriptor;
* @author Axel Uhl (d043530)
*
*/
-public class Replicator implements MessageListener {
+public class Replicator implements Runnable {
+ private final static Logger logger = Logger.getLogger(Replicator.class.getName());
+
private final ReplicationMasterDescriptor master;
private final HasRacingEventService racingEventServiceTracker;
private final Listoperation is queued, otherwise immediately applied to
* the receiving replica.
@@ -134,12 +143,6 @@ public class Replicator implements MessageListener {
return suspended;
}
- private byte[] getBytes(BytesMessage m) throws JMSException {
- byte[] buf = new byte[(int) m.getBodyLength()];
- m.readBytes(buf);
- return buf;
- }
-
@Override
public String toString() {
return "Replicator for master "+master+", queue size: "+queue.size();
diff --git a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/SampleMessageConsumer.java b/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/SampleMessageConsumer.java
deleted file mode 100644
index da2c4ad8c72..00000000000
--- a/java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/SampleMessageConsumer.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package com.sap.sailing.server.replication.impl;
-
-import javax.jms.ExceptionListener;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.TextMessage;
-
-public class SampleMessageConsumer implements MessageListener, ExceptionListener {
-
- synchronized public void onException(JMSException ex) {
- System.out.println("JMS Exception occured: " + ex.getMessage());
- }
-
- public void onMessage(Message message) {
- if (message instanceof TextMessage) {
- TextMessage textMessage = (TextMessage) message;
- try {
- System.out.println("Received message: " + textMessage.getText());
- } catch (JMSException ex) {
- System.out.println("Error reading message: " + ex);
- }
- } else {
- System.out.println("Received: " + message);
- }
- }
-}