mirror of
https://github.com/eclipse-sailing-analytics/sailing-analytics.git
synced 2026-10-20 11:10:00 +00:00
579 lines
26 KiB
Java
Executable File
579 lines
26 KiB
Java
Executable File
package com.sap.sse.replication.impl;
|
|
|
|
import java.io.BufferedReader;
|
|
import java.io.ByteArrayOutputStream;
|
|
import java.io.IOException;
|
|
import java.io.InputStream;
|
|
import java.io.InputStreamReader;
|
|
import java.io.ObjectInputStream;
|
|
import java.io.ObjectOutputStream;
|
|
import java.net.ConnectException;
|
|
import java.net.URL;
|
|
import java.net.URLConnection;
|
|
import java.util.ArrayList;
|
|
import java.util.HashMap;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.Timer;
|
|
import java.util.TimerTask;
|
|
import java.util.UUID;
|
|
import java.util.logging.Level;
|
|
import java.util.logging.Logger;
|
|
import java.util.zip.GZIPInputStream;
|
|
import java.util.zip.GZIPOutputStream;
|
|
|
|
import org.osgi.util.tracker.ServiceTracker;
|
|
|
|
import com.rabbitmq.client.AMQP.Exchange;
|
|
import com.rabbitmq.client.AMQP.Queue.DeleteOk;
|
|
import com.rabbitmq.client.Channel;
|
|
import com.rabbitmq.client.ConnectionFactory;
|
|
import com.rabbitmq.client.QueueingConsumer;
|
|
import com.sap.sse.BuildVersion;
|
|
import com.sap.sse.replication.OperationExecutionListener;
|
|
import com.sap.sse.replication.OperationWithResult;
|
|
import com.sap.sse.replication.Replicable;
|
|
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
|
import com.sap.sse.replication.ReplicationService;
|
|
|
|
/**
|
|
* Can observe a {@link Replicable} 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 Replicable}. Operations received that require replication are sent to the {@link Exchange} to which replica
|
|
* queues can bind, using a {@link Replicator}. The exchange name and connectivity information for the message queuing
|
|
* system is provided to this service during construction.
|
|
* <p>
|
|
*
|
|
* This service object {@link Replicable#addOperationExecutionListener(OperationExecutionListener) registers} as
|
|
* listener at the {@link Replicable} so it {@link #executed(OperationWithResult) receives} notifications about
|
|
* operations executed by the {@link Replicable} that require replication if and only if there is at least one
|
|
* replica registered.
|
|
*
|
|
* @author Frank Mittag, Axel Uhl (d043530)
|
|
*
|
|
*/
|
|
public class ReplicationServiceImpl<S, O extends OperationWithResult<S, ?>> implements ReplicationService<S>,
|
|
OperationExecutionListener<S>, HasReplicable<S, O> {
|
|
private static final Logger logger = Logger.getLogger(ReplicationServiceImpl.class.getName());
|
|
|
|
private final ReplicationInstancesManager<S> replicationInstancesManager;
|
|
|
|
private final ServiceTracker<Replicable<S, O>, Replicable<S, O>> replicableTracker;
|
|
|
|
private final Replicable<S, O> 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
|
|
*/
|
|
private final Map<ReplicationMasterDescriptor, String> replicaUUIDs;
|
|
|
|
/**
|
|
* Channel used by a master server to publish replication operations; <code>null</code> in servers that don't have replicas registered
|
|
*/
|
|
private Channel masterChannel;
|
|
|
|
/**
|
|
* The name of the RabbitMQ exchange to which this replication service sends its replication operations in
|
|
* serialized form. Clients need to know this name to be able to bind their queues to the exchange.
|
|
*/
|
|
private final String exchangeName;
|
|
|
|
/**
|
|
* The host on which the RabbitMQ server runs
|
|
*/
|
|
private final String exchangeHost;
|
|
|
|
/**
|
|
* The port on which to reach the RabbitMQ server, or 0 for the RabbitMQ default port
|
|
*/
|
|
private final int exchangePort;
|
|
|
|
/**
|
|
* UUID that identifies this server
|
|
*/
|
|
private final UUID serverUUID;
|
|
|
|
private Replicator<S, O> replicator;
|
|
private Thread replicatorThread;
|
|
|
|
/**
|
|
* Used to synchronize write access and replacements of {@link #outboundBuffer}, {@link #outboundObjectBuffer} and
|
|
* {@link #outboundBufferClasses} when the timer scoops up the messages to send.
|
|
*/
|
|
private final Object outboundBufferMonitor = "";
|
|
|
|
/**
|
|
* Sending operations as serialized Java objects using binary RabbitMQ messages comes at an overhead. To reduce the overhead,
|
|
* several operations can be serialized into a single message. The actual serialization of the buffer happens after a short duration
|
|
* has passed since the last sending, managed by a {@link Timer}. Writers need to synchronize on {@link #outboundBufferMonitor}
|
|
* which protects all of {@link #outboundBuffer}, {@link #outboundObjectBuffer} and {@link #outboundBufferClasses} which
|
|
* are replaced or cleared when the timer scoops up the currently buffered operations to send them out.
|
|
*/
|
|
private ByteArrayOutputStream outboundBuffer;
|
|
|
|
/**
|
|
* An object output stream that writes to {@link #outboundBuffer}. Operations are serialized into this stream until the timer
|
|
* acquires the {@link #outboundBufferMonitor}, closes the stream and transmits the contents of {@link #outboundBuffer} as a
|
|
* RabbitMQ message. While still holding the monitor, the timer task creates a new {@link #outboundBuffer} and a new
|
|
* {@link #outboundObjectBuffer} wrapping the {@link #outboundBuffer}.
|
|
*/
|
|
private ObjectOutputStream outboundObjectBuffer;
|
|
|
|
/**
|
|
* Remembers the classes of the operations serialized into {@link #outboundObjectBuffer}. The list of classes in this list
|
|
* matches with the sequence of objects written to {@link #outboundObjectBuffer} as long as the {@link #outboundBufferMonitor}
|
|
* is being held.
|
|
*/
|
|
private List<Class<?>> outboundBufferClasses;
|
|
|
|
/**
|
|
* Used to schedule the sending of all operations in {@link #outboundBuffer} using the {@link #sendingTask}.
|
|
*/
|
|
private final Timer timer;
|
|
|
|
/**
|
|
* Sends all operations in {@link #outboundBuffer}. When holding the monitor of {@link #outboundBuffer},
|
|
* the following rules hold:
|
|
*
|
|
* <ul>
|
|
* <li>if <code>null</code>, adding an operation to {@link #outboundBuffer} needs to create and assign a new timer that
|
|
* schedules a sending task.
|
|
* </li>
|
|
* <li>if not <code>null</code>, an operation added to {@link #outboundBuffer} is guaranteed to be sent by the timer
|
|
* </li>
|
|
* </ul>
|
|
*
|
|
*/
|
|
private TimerTask sendingTask;
|
|
|
|
/**
|
|
* Defines for how many milliseconds the {@link #timer} will wait since the first operation has been added to an empty
|
|
* {@link #outboundBuffer} until it carries out the actual transmission task. The longer this duration, the more operations
|
|
* are likely to be sent per message transmitted, reducing overhead but correspondingly increasing latency.
|
|
*/
|
|
private final long TRANSMISSION_DELAY_MILLIS = 100;
|
|
|
|
/**
|
|
* Defines at which message size in bytes the message will be sent regardless the {@link #TRANSMISSION_DELAY_MILLIS}.
|
|
*/
|
|
private final int TRIGGER_MESSAGE_SIZE_IN_BYTES = 1024*1024;
|
|
|
|
/**
|
|
* Counts the messages sent out by this replicator
|
|
*/
|
|
private long messageCount;
|
|
|
|
/**
|
|
* @param exchangeName name of the fan-out exchange to which replication operations will be sent
|
|
* @param exchangeHost name of the host under which the RabbitMQ server can be reached
|
|
* @param exchangePort port of the RabbitMQ server, or 0 for default port
|
|
*/
|
|
public ReplicationServiceImpl(String exchangeName, String exchangeHost, int exchangePort, final ReplicationInstancesManager<S> replicationInstancesManager) throws IOException {
|
|
this(exchangeName, exchangeHost, exchangePort, replicationInstancesManager, /* localService */ null, /* create RacingEventServiceTracker */ true);
|
|
}
|
|
|
|
private ReplicationServiceImpl(String exchangeName, String exchangeHost,
|
|
int exchangePort, final ReplicationInstancesManager<S> replicationInstancesManager,
|
|
Replicable<S, O> localService, boolean createRacingEventServiceTracker) throws IOException {
|
|
timer = new Timer("ReplicationServiceImpl timer for delayed task sending");
|
|
this.replicationInstancesManager = replicationInstancesManager;
|
|
replicaUUIDs = new HashMap<ReplicationMasterDescriptor, String>();
|
|
if (createRacingEventServiceTracker) {
|
|
replicableTracker = getReplicableTracker();
|
|
replicableTracker.open();
|
|
} else {
|
|
replicableTracker = null;
|
|
}
|
|
this.localService = localService;
|
|
this.exchangeName = exchangeName;
|
|
this.exchangeHost = exchangeHost;
|
|
this.exchangePort = exchangePort;
|
|
replicator = null;
|
|
serverUUID = UUID.randomUUID();
|
|
logger.info("Setting " + serverUUID.toString() + " as unique replication identifier.");
|
|
|
|
}
|
|
|
|
/**
|
|
* Like {@link #ReplicationServiceImpl(String, ReplicationInstancesManager)}, only that instead of using an OSGi
|
|
* service tracker to discover the {@link RacingEventService}, the service to replicate is "injected" here.
|
|
*
|
|
* @param exchangeName
|
|
* the name of the exchange to which replicas can bind
|
|
*/
|
|
public ReplicationServiceImpl(String exchangeName, String exchangeHost,
|
|
final ReplicationInstancesManager<S> replicationInstancesManager, Replicable<S, O> localService) throws IOException {
|
|
this(exchangeName, exchangeHost, 0, replicationInstancesManager, localService, /* create RacingEventServiceTracker */ false);
|
|
}
|
|
|
|
protected ServiceTracker<Replicable<S, O>, Replicable<S, O>> getReplicableTracker() {
|
|
return new ServiceTracker<Replicable<S, O>, Replicable<S, O>>(
|
|
Activator.getDefaultContext(), Replicable.class.getName(), null);
|
|
}
|
|
|
|
private Channel createMasterChannelAndDeclareFanoutExchange() throws IOException {
|
|
Channel result = createMasterChannel();
|
|
result.exchangeDeclare(exchangeName, "fanout");
|
|
logger.info("Created fanout exchange "+exchangeName+" successfully.");
|
|
return result;
|
|
}
|
|
|
|
@Override
|
|
public Channel createMasterChannel() throws IOException, ConnectException {
|
|
final ConnectionFactory connectionFactory = new ConnectionFactory();
|
|
connectionFactory.setHost(exchangeHost);
|
|
if (exchangePort != 0) {
|
|
connectionFactory.setPort(exchangePort);
|
|
}
|
|
Channel result = null;
|
|
try {
|
|
result = connectionFactory.newConnection().createChannel();
|
|
} catch (ConnectException ex) {
|
|
// make sure to log something meaningful
|
|
logger.severe("Could not connect to messaging queue on " + connectionFactory.getHost() + ":" + connectionFactory.getPort() + "/" + exchangeName);
|
|
throw ex;
|
|
}
|
|
logger.info("Connected to " + connectionFactory.getHost() + ":" + connectionFactory.getPort());
|
|
return result;
|
|
}
|
|
|
|
@Override
|
|
public Replicable<S, O> getReplicable() {
|
|
Replicable<S, O> result;
|
|
if (localService != null) {
|
|
result = localService;
|
|
} else {
|
|
result = replicableTracker.getService();
|
|
}
|
|
return result;
|
|
}
|
|
|
|
@Override
|
|
public void registerReplica(ReplicaDescriptor replica) throws IOException {
|
|
if (!replicationInstancesManager.hasReplicas()) {
|
|
addAsListenerToRacingEventService();
|
|
synchronized (this) {
|
|
if (masterChannel == null) {
|
|
masterChannel = createMasterChannelAndDeclareFanoutExchange();
|
|
}
|
|
}
|
|
}
|
|
replicationInstancesManager.registerReplica(replica);
|
|
logger.info("Registered replica " + replica);
|
|
}
|
|
|
|
private void addAsListenerToRacingEventService() {
|
|
getReplicable().addOperationExecutionListener(this);
|
|
}
|
|
|
|
@Override
|
|
public void unregisterReplica(ReplicaDescriptor replica) throws IOException {
|
|
logger.info("Unregistering replica " + replica);
|
|
replicationInstancesManager.unregisterReplica(replica);
|
|
if (!replicationInstancesManager.hasReplicas()) {
|
|
removeAsListenerFromRacingEventService();
|
|
synchronized (this) {
|
|
if (masterChannel != null) {
|
|
masterChannel.close();
|
|
masterChannel = null;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private void removeAsListenerFromRacingEventService() {
|
|
getReplicable().removeOperationExecutionListener(this);
|
|
}
|
|
|
|
/**
|
|
* Schedules a single operation for broadcast. The operation is added to {@link #outboundBuffer}, and if not already scheduled,
|
|
* a {@link #timer} is created and scheduled to send in {@link #TRANSMISSION_DELAY_MILLIS} milliseconds.
|
|
*/
|
|
private void broadcastOperation(OperationWithResult<S, ?> operation) throws IOException {
|
|
// need to write the operations one by one, making sure the ObjectOutputStream always writes
|
|
// identical objects again if required because they may have changed state in between
|
|
ByteArrayOutputStream bos = new ByteArrayOutputStream();
|
|
ObjectOutputStream oos = new ObjectOutputStream(bos);
|
|
oos.writeObject(operation);
|
|
oos.close();
|
|
final byte[] bytes = bos.toByteArray();
|
|
synchronized (outboundBufferMonitor) {
|
|
if (outboundBuffer == null) {
|
|
outboundBuffer = new ByteArrayOutputStream();
|
|
GZIPOutputStream zipper = new GZIPOutputStream(outboundBuffer);
|
|
outboundObjectBuffer = new ObjectOutputStream(zipper);
|
|
outboundBufferClasses = new ArrayList<>();
|
|
}
|
|
outboundObjectBuffer.writeObject(bytes);
|
|
outboundBufferClasses.add(operation.getClass());
|
|
if (outboundBuffer.size() > TRIGGER_MESSAGE_SIZE_IN_BYTES) {
|
|
logger.info("Triggering replication because buffer holds " + outboundBuffer.size()
|
|
+ " bytes which exceeds trigger size " + TRIGGER_MESSAGE_SIZE_IN_BYTES+" bytes");
|
|
flushBufferToRabbitMQ();
|
|
} else {
|
|
if (sendingTask == null) {
|
|
sendingTask = new TimerTask() {
|
|
@Override
|
|
public void run() {
|
|
try {
|
|
sendingTask = null;
|
|
logger.fine("Running timer task, flushing buffer");
|
|
flushBufferToRabbitMQ();
|
|
} catch (Exception e) {
|
|
logger.log(Level.SEVERE, "Exception while trying to replicate operations", e);
|
|
}
|
|
}
|
|
};
|
|
timer.schedule(sendingTask, TRANSMISSION_DELAY_MILLIS);
|
|
}
|
|
}
|
|
if (++messageCount % 10000l == 0) {
|
|
logger.info("Handled "+messageCount+" messages for replication. Current outbound replication queue size: "+outboundBufferClasses.size());
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Obtains the monitor on {@link #outboundBufferMonitor}, copies the references to the buffers, nulls out
|
|
* the buffers, then releases the monitor and broadcasts the buffer.
|
|
*/
|
|
private void flushBufferToRabbitMQ() {
|
|
logger.fine("Trying to acquire monitor");
|
|
final byte[] bytesToSend;
|
|
final List<Class<?>> classesOfOperationsToSend;
|
|
final boolean doSend;
|
|
synchronized (outboundBufferMonitor) {
|
|
if (outboundBuffer != null) {
|
|
logger.fine("Preparing " + outboundBufferClasses.size() + " operations for sending to RabbitMQ exchange");
|
|
try {
|
|
outboundObjectBuffer.close();
|
|
logger.fine("Sucessfully closed ObjectOutputStream");
|
|
} catch (IOException e) {
|
|
logger.log(Level.SEVERE, "Error trying to replicate " + outboundBufferClasses.size()
|
|
+ " operations", e);
|
|
}
|
|
bytesToSend = outboundBuffer.toByteArray();
|
|
logger.fine("Successfully produced bytesToSend array of length " + bytesToSend.length);
|
|
classesOfOperationsToSend = outboundBufferClasses;
|
|
doSend = true;
|
|
outboundBuffer = null;
|
|
outboundObjectBuffer = null;
|
|
outboundBufferClasses = null;
|
|
} else {
|
|
logger.fine("No buffer set; probably two timer tasks were scheduled concurrently. No problem, just not sending this time around.");
|
|
doSend = false;
|
|
bytesToSend = null;
|
|
classesOfOperationsToSend = null;
|
|
}
|
|
}
|
|
if (doSend) {
|
|
try {
|
|
broadcastOperations(bytesToSend, classesOfOperationsToSend);
|
|
logger.fine("Successfully handed " + classesOfOperationsToSend.size() + " operations to broadcaster");
|
|
} catch (Exception e) {
|
|
logger.log(Level.SEVERE, "Error trying to replicate " + classesOfOperationsToSend.size() + " operations", e);
|
|
}
|
|
}
|
|
}
|
|
|
|
private void broadcastOperations(byte[] bytesToSend, List<Class<?>> classesOfOperationsToSend) throws IOException {
|
|
logger.fine("broadcasting "+classesOfOperationsToSend.size()+" operations as "+bytesToSend.length+" bytes");
|
|
if (masterChannel != null) {
|
|
logger.fine("buffer to broadcast has "+bytesToSend.length+" bytes ("+(bytesToSend.length/1024/1024)+"MB)");
|
|
long startTime = System.currentTimeMillis();
|
|
masterChannel.basicPublish(exchangeName, /* routingKey */"", /* properties */null, bytesToSend);
|
|
logger.fine("successfully published "+bytesToSend.length+" bytes, taking "+(System.currentTimeMillis()-startTime)+"ms");
|
|
replicationInstancesManager.log(classesOfOperationsToSend, bytesToSend.length);
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public Iterable<ReplicaDescriptor> getReplicaInfo() {
|
|
return replicationInstancesManager.getReplicaDescriptors();
|
|
}
|
|
|
|
@Override
|
|
public ReplicationMasterDescriptor isReplicatingFromMaster() {
|
|
return replicatingFromMaster;
|
|
}
|
|
|
|
@Override
|
|
public void startToReplicateFrom(ReplicationMasterDescriptor master) throws IOException, ClassNotFoundException, InterruptedException {
|
|
logger.info("Starting to replicate from "+master);
|
|
try {
|
|
registerReplicaWithMaster(master);
|
|
} catch (Exception ex) {
|
|
logger.log(Level.SEVERE, "ERROR", ex);
|
|
throw ex;
|
|
}
|
|
replicatingFromMaster = master;
|
|
logger.info("Registered replica with master.");
|
|
QueueingConsumer consumer = null;
|
|
// logging exception here because it will not propagate
|
|
// thru the client with all details
|
|
try {
|
|
logger.info("Connecting to message queue " + master);
|
|
consumer = master.getConsumer();
|
|
} catch (Exception ex) {
|
|
logger.log(Level.SEVERE, "ERROR", ex);
|
|
replicatingFromMaster = null;
|
|
throw ex;
|
|
}
|
|
logger.info("Connection to exchange successful.");
|
|
URL initialLoadURL = master.getInitialLoadURL();
|
|
logger.info("Initial load URL is "+initialLoadURL);
|
|
replicator = new Replicator<S, O>(master, this, /* startSuspended */ true, consumer);
|
|
// start receiving messages already now, but start in suspended mode
|
|
replicatorThread = new Thread(replicator, "Replicator receiving from "+master.getMessagingHostname()+"/"+master.getExchangeName());
|
|
final Replicable<?, ?> replicable = getReplicable();
|
|
// clear racingEventService state here, before starting to receive and de-serialized operations which builds up new state, e.g., in competitor store
|
|
replicable.clearReplicaState(); // see also bug 2437
|
|
replicatorThread.start();
|
|
logger.info("Started replicator thread");
|
|
InputStream is = initialLoadURL.openStream();
|
|
final String queueName = new BufferedReader(new InputStreamReader(is)).readLine();
|
|
try {
|
|
RabbitInputStreamProvider rabbitInputStreamProvider = new RabbitInputStreamProvider(master.createChannel(), queueName);
|
|
ObjectInputStream ois = replicable.createObjectInputStreamResolvingAgainstCache(
|
|
new GZIPInputStream(rabbitInputStreamProvider.getInputStream()));
|
|
logger.info("Starting to receive initial load");
|
|
replicable.initiallyFillFrom(ois);
|
|
logger.info("Done receiving initial load");
|
|
replicator.setSuspended(false); // apply queued operations
|
|
} finally {
|
|
// delete initial load queue
|
|
DeleteOk deleteOk = consumer.getChannel().queueDelete(queueName);
|
|
logger.info("Deleted queue "+queueName+" used for initial load: "+deleteOk.toString());
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @return the UUID that the master generated for this client which is also entered into {@link #replicaUUIDs}
|
|
*/
|
|
private String registerReplicaWithMaster(ReplicationMasterDescriptor master) throws IOException, ClassNotFoundException {
|
|
URL replicationRegistrationRequestURL = master.getReplicationRegistrationRequestURL(getServerIdentifier(), BuildVersion.getBuildVersion());
|
|
final URLConnection registrationRequestConnection = replicationRegistrationRequestURL.openConnection();
|
|
registrationRequestConnection.connect();
|
|
InputStream content = (InputStream) registrationRequestConnection.getContent();
|
|
StringBuilder uuid = new StringBuilder();
|
|
byte[] buf = new byte[256];
|
|
int read = content.read(buf);
|
|
while (read != -1) {
|
|
uuid.append(new String(buf, 0, read));
|
|
read = content.read(buf);
|
|
}
|
|
String replicaUUID = uuid.toString();
|
|
registerReplicaUuidForMaster(replicaUUID, master);
|
|
return replicaUUID;
|
|
}
|
|
|
|
protected void deregisterReplicaWithMaster(ReplicationMasterDescriptor master) {
|
|
try {
|
|
URL replicationDeRegistrationRequestURL = master.getReplicationDeRegistrationRequestURL(getServerIdentifier());
|
|
final URLConnection deregistrationRequestConnection = replicationDeRegistrationRequestURL.openConnection();
|
|
deregistrationRequestConnection.connect();
|
|
StringBuilder uuid = new StringBuilder();
|
|
InputStream content = (InputStream) deregistrationRequestConnection.getContent();
|
|
byte[] buf = new byte[256];
|
|
int read = content.read(buf);
|
|
while (read != -1) {
|
|
uuid.append(new String(buf, 0, read));
|
|
read = content.read(buf);
|
|
}
|
|
content.close();
|
|
} catch (Exception ex) {
|
|
// ignore exceptions here - they will mostly be caused by an incompatible server
|
|
// it is also not problematic if the server does not get this deregistration
|
|
// a new registration will overwrite the current one
|
|
}
|
|
}
|
|
|
|
/**
|
|
* {@link #broadcastOperation(RacingEventServiceOperation) Broadcasts} the <code>operation</code> to all registered
|
|
* replicas by publishing it to the fan-out exchange.
|
|
*/
|
|
@Override
|
|
public <T> void executed(OperationWithResult<S, T> operation) {
|
|
try {
|
|
broadcastOperation(operation);
|
|
} catch (Exception e) {
|
|
throw new RuntimeException(e);
|
|
}
|
|
}
|
|
|
|
protected void registerReplicaUuidForMaster(String uuid, ReplicationMasterDescriptor master) {
|
|
replicaUUIDs.put(master, uuid);
|
|
}
|
|
|
|
@Override
|
|
public Map<Class<? extends OperationWithResult<S, ?>>, Integer> getStatistics(ReplicaDescriptor replicaDescriptor) {
|
|
return replicationInstancesManager.getStatistics(replicaDescriptor);
|
|
}
|
|
|
|
@Override
|
|
public double getAverageNumberOfOperationsPerMessage(ReplicaDescriptor replicaDescriptor) {
|
|
return replicationInstancesManager.getAverageNumberOfOperationsPerMessage(replicaDescriptor);
|
|
}
|
|
|
|
@Override
|
|
public long getNumberOfMessagesSent(ReplicaDescriptor replica) {
|
|
return replicationInstancesManager.getNumberOfMessagesSent(replica);
|
|
}
|
|
|
|
@Override
|
|
public long getNumberOfBytesSent(ReplicaDescriptor replica) {
|
|
return replicationInstancesManager.getNumberOfBytesSent(replica);
|
|
}
|
|
|
|
@Override
|
|
public double getAverageNumberOfBytesPerMessage(ReplicaDescriptor replica) {
|
|
return replicationInstancesManager.getAverageNumberOfBytesPerMessage(replica);
|
|
}
|
|
|
|
@Override
|
|
public void stopToReplicateFromMaster() throws IOException {
|
|
ReplicationMasterDescriptor descriptor = isReplicatingFromMaster();
|
|
if (descriptor != null) {
|
|
synchronized(replicaUUIDs) {
|
|
if (replicator != null) {
|
|
replicator.stop();
|
|
deregisterReplicaWithMaster(descriptor);
|
|
replicatingFromMaster = null;
|
|
replicaUUIDs.clear();
|
|
|
|
// this is needed because QueuingConsumer.nextDelivery() wont unblock
|
|
// if the connection is closed by application.
|
|
replicatorThread.interrupt();
|
|
replicator = null;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public void stopAllReplica() throws IOException {
|
|
if (replicationInstancesManager.hasReplicas()) {
|
|
replicationInstancesManager.removeAll();
|
|
removeAsListenerFromRacingEventService();
|
|
synchronized (this) {
|
|
if (masterChannel != null) {
|
|
masterChannel.close();
|
|
masterChannel = null;
|
|
}
|
|
}
|
|
logger.info("Unregistered all replicas from this server!");
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public UUID getServerIdentifier() {
|
|
return serverUUID;
|
|
}
|
|
|
|
}
|