replication panel shows replicas and their statistical data

2026-09-30 09:26:44 +00:00 · 2012-05-09 18:03:39 +02:00
parent 96ffdcbb98
commit b654340264
17 changed files with 262 additions and 34 deletions
@@ -20,4 +20,10 @@ public interface ReplicationMasterDescriptor {
    URL getInitialLoadURL() throws MalformedURLException;

    TopicSubscriber getTopicSubscriber(String clientID) throws JMSException, UnknownHostException;
+    
+    int getJMSPort();
+    
+    int getServletPort();
+
+    String getHostname();
 }
@@ -1,16 +1,26 @@
 package com.sap.sailing.server.replication;

 import java.io.IOException;
-import java.util.List;
+import java.util.Map;

 import javax.jms.JMSException;

+import com.sap.sailing.server.RacingEventServiceOperation;
 import com.sap.sailing.server.replication.impl.ReplicationServlet;

 public interface ReplicationService {
    static String SAILING_SERVER_REPLICATION_TOPIC = "SailingServerReplicationTopic";

-    List<String> getHostnamesOfReplica();
+    /**
+     * Tells about replicas registered with this master.
+     */
+    Iterable<ReplicaDescriptor> getReplicaInfo();
+    
+    /**
+     * If this server instance is a replica of some other master server instance, this method returns that master's
+     * descriptor. Otherwise (if this instance is not a replica), returns <code>null</code>.
+     */
+    ReplicationMasterDescriptor isReplicatingFromMaster();
    
    /**
     * Performs a servlet request to the server's {@link ReplicationServlet}, first registering this replica, ensuring
@@ -22,9 +32,15 @@ public interface ReplicationService {
    /**
     * Registers a replica with this master instance. If the replication topic hasn't been created in the
     * JMS message broker yet, it will be when this method returns. The <code>replica</code> will be considered
-     * in the result of {@link #getHostnamesOfReplica()} when this call has succeeded.
+     * in the result of {@link #getReplicaInfo()} when this call has succeeded.
     */
    void registerReplica(ReplicaDescriptor replica) throws JMSException;

    void unregisterReplica(ReplicaDescriptor replica) throws JMSException;
+
+    /**
+     * For a replica replicating off this master, provides statistics in the form of number of operations sent to that
+     * replica by type, where the operation type is the key, represented as the operation's class name
+     */
+    Map<Class<? extends RacingEventServiceOperation<?>>, Integer> getStatistics(ReplicaDescriptor replicaDescriptor);
 }
@@ -1,9 +1,12 @@
 package com.sap.sailing.server.replication.impl;

+import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
-import java.util.Iterator;
+import java.util.Map;
 import java.util.Set;

+import com.sap.sailing.server.RacingEventServiceOperation;
 import com.sap.sailing.server.replication.ReplicaDescriptor;
 import com.sap.sailing.server.replication.ReplicationMasterDescriptor;

@@ -15,6 +18,8 @@ public class ReplicationInstancesManager {
     */
    private Set<ReplicaDescriptor> replicaDescriptors;
    
+    private Map<ReplicaDescriptor, Map<Class<? extends RacingEventServiceOperation<?>>, Integer>> replicationCounts;
+    
    /**
     * The descriptor of the replication master
     */
@@ -22,6 +27,7 @@ public class ReplicationInstancesManager {

    public ReplicationInstancesManager() {
        replicaDescriptors = new HashSet<ReplicaDescriptor>();
+        replicationCounts = new HashMap<ReplicaDescriptor, Map<Class<? extends RacingEventServiceOperation<?>>,Integer>>();
    }
    
    /**
@@ -34,8 +40,8 @@ public class ReplicationInstancesManager {
        return !replicaDescriptors.isEmpty();
    }
    
-    public Iterator<ReplicaDescriptor> getReplicaDescriptors() {
-        return replicaDescriptors.iterator();
+    public Iterable<ReplicaDescriptor> getReplicaDescriptors() {
+        return Collections.unmodifiableCollection(replicaDescriptors);
    }

    public ReplicationMasterDescriptor getReplicationMasterDescriptor() {
@@ -44,9 +50,40 @@ public class ReplicationInstancesManager {
    
    public void registerReplica(ReplicaDescriptor replica) {
        replicaDescriptors.add(replica);
+        replicationCounts.put(replica, new HashMap<Class<? extends RacingEventServiceOperation<?>>, Integer>());
    }

    public void unregisterReplica(ReplicaDescriptor replica) {
        replicaDescriptors.remove(replica);
+        replicationCounts.remove(replica);
+    }
+    
+    /**
+     * For {@link #replicaDescriptors each replica currently registered}, increases the replication count for the
+     * type of <code>replicatedOperation</code> by one.
+     * 
+     * @see #getStatistics
+     */
+    public <T> void log(RacingEventServiceOperation<T> replicatedOperation) {
+        for (ReplicaDescriptor replica : getReplicaDescriptors()) {
+            Map<Class<? extends RacingEventServiceOperation<?>>, Integer> counts = replicationCounts.get(replica);
+            if (counts == null) {
+                counts = new HashMap<Class<? extends RacingEventServiceOperation<?>>, Integer>();
+                replicationCounts.put(replica, counts);
+            }
+            @SuppressWarnings("unchecked") // safe because replicatedOperation is declared of type RacingEventserviceOperation<T>
+            Class<? extends RacingEventServiceOperation<T>> operationClass = (Class<? extends RacingEventServiceOperation<T>>) replicatedOperation
+                    .getClass();
+            Integer count = counts.get(operationClass);
+            if (count == null) {
+                count = 0;
+            }
+            count++;
+            counts.put(operationClass, count);
+        }
+    }
+    
+    public Map<Class<? extends RacingEventServiceOperation<?>>, Integer> getStatistics(ReplicaDescriptor replica) {
+        return replicationCounts.get(replica);
    }
 }
@@ -53,4 +53,18 @@ public class ReplicationMasterDescriptorImpl implements ReplicationMasterDescrip
        return session.createDurableSubscriber(topic, InetAddress.getLocalHost().getHostAddress());
    }

+    @Override
+    public int getJMSPort() {
+        return jmsPort;
+    }
+
+    @Override
+    public int getServletPort() {
+        return servletPort;
+    }
+
+    @Override
+    public String getHostname() {
+        return hostname;
+    }
 }
@@ -7,10 +7,7 @@ import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.net.URL;
 import java.net.URLConnection;
-import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
 import java.util.Map;

 import javax.jms.BytesMessage;
@@ -58,6 +55,11 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
    
    private final RacingEventService localService;
    
+    /**
+     * <code>null</code>, if this instance is not currently replicating from some master; the master's descriptor otherwise
+     */
+    private ReplicationMasterDescriptor replicatingFromMaster;
+    
    /**
     * The UUIDs with which this replica is registered by the master identified by the corresponding key
     */
@@ -151,6 +153,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
        oos.close();
        operationAsMessage.writeBytes(bos.toByteArray());
        messageProducer.send(operationAsMessage);
+        replicationInstancesManager.log(operation);
    }

    private MessageProducer getMessageProducer(Topic topic) throws JMSException {
@@ -161,17 +164,18 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
    }

    @Override
-    public List<String> getHostnamesOfReplica() {
-        List<String> result = new ArrayList<String>();
-        for (Iterator<ReplicaDescriptor> i=replicationInstancesManager.getReplicaDescriptors(); i.hasNext(); ) {
-            ReplicaDescriptor d = i.next();
-            result.add(d.getIpAddress().getHostAddress());
-        }
-        return result;
+    public Iterable<ReplicaDescriptor> getReplicaInfo() {
+        return replicationInstancesManager.getReplicaDescriptors();
+    }
+
+    @Override
+    public ReplicationMasterDescriptor isReplicatingFromMaster() {
+        return replicatingFromMaster;
    }

    @Override
    public void startToReplicateFrom(ReplicationMasterDescriptor master) throws IOException, ClassNotFoundException, JMSException {
+        replicatingFromMaster = master;
        String uuid = registerReplicaWithMaster(master);
        TopicSubscriber replicationSubscription = master.getTopicSubscriber(uuid);
        URL initialLoadURL = master.getInitialLoadURL();
@@ -214,4 +218,9 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
        replicaUUIDs.put(master, uuid);
    }

+    @Override
+    public Map<Class<? extends RacingEventServiceOperation<?>>, Integer> getStatistics(ReplicaDescriptor replicaDescriptor) {
+        return replicationInstancesManager.getStatistics(replicaDescriptor);
+    }
+
 }