diff --git a/java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/adminconsole/ReplicationPanel.java b/java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/adminconsole/ReplicationPanel.java index a6018d13f81..f6984db16e9 100755 --- a/java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/adminconsole/ReplicationPanel.java +++ b/java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/adminconsole/ReplicationPanel.java @@ -85,13 +85,17 @@ public class ReplicationPanel extends FlowPanel { } private void stopReplication() { + stopReplicationButton.setEnabled(false); sailingService.stopReplicatingFromMaster(new AsyncCallback() { @Override public void onFailure(Throwable caught) { errorReporter.reportError(caught.getMessage()); + stopReplicationButton.setEnabled(true); } @Override public void onSuccess(Void result) { + addButton.setEnabled(true); + stopReplicationButton.setEnabled(false); updateReplicaList(); } }); @@ -105,12 +109,15 @@ public class ReplicationPanel extends FlowPanel { registeredMasters.removeRow(0); registeredMasters.insertRow(0); registeredMasters.setWidget(0, 0, new Label(stringMessages.loading())); + addButton.setEnabled(false); + stopReplicationButton.setEnabled(false); sailingService.startReplicatingFromMaster(masterNameAndExchangeNameAndMessagingPortNumberAndServletPortNumber.getA().getA(), masterNameAndExchangeNameAndMessagingPortNumberAndServletPortNumber.getA().getB(), masterNameAndExchangeNameAndMessagingPortNumberAndServletPortNumber.getC(), masterNameAndExchangeNameAndMessagingPortNumberAndServletPortNumber.getB(), new AsyncCallback() { @Override public void onFailure(Throwable e) { + addButton.setEnabled(true); errorReporter.reportError(stringMessages.errorStartingReplication( masterNameAndExchangeNameAndMessagingPortNumberAndServletPortNumber.getA().getA(), masterNameAndExchangeNameAndMessagingPortNumberAndServletPortNumber.getA().getB(), e.getMessage())); @@ -118,8 +125,9 @@ public class ReplicationPanel extends FlowPanel { @Override public void onSuccess(Void arg0) { - updateReplicaList(); addButton.setEnabled(false); + stopReplicationButton.setEnabled(true); + updateReplicaList(); } }); } 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 475f85c182d..1af7a1f486c 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 @@ -70,6 +70,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec private final String exchangeName; private Replicator replicator; + private Thread replicatorThread; public ReplicationServiceImpl(String exchangeName, final ReplicationInstancesManager replicationInstancesManager) throws IOException { this.replicationInstancesManager = replicationInstancesManager; @@ -203,7 +204,8 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec logger.info("Initial load URL is "+initialLoadURL); replicator = new Replicator(master, this, /* startSuspended */ true, consumer); // start receiving messages already now, but start in suspended mode - new Thread(replicator, "Replicator receiving from "+master.getHostname()+"/"+master.getExchangeName()).start(); + replicatorThread = new Thread(replicator, "Replicator receiving from "+master.getHostname()+"/"+master.getExchangeName()); + replicatorThread.start(); logger.info("Started replicator thread"); InputStream is = initialLoadURL.openStream(); final RacingEventService racingEventService = getRacingEventService(); @@ -234,19 +236,25 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec return replicaUUID; } - protected void deregisterReplicaWithMaster(ReplicationMasterDescriptor master) throws IOException { - URL replicationDeRegistrationRequestURL = master.getReplicationDeRegistrationRequestURL(); - 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); + protected void deregisterReplicaWithMaster(ReplicationMasterDescriptor master) { + try { + URL replicationDeRegistrationRequestURL = master.getReplicationDeRegistrationRequestURL(); + 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 } - content.close(); } @Override @@ -275,7 +283,12 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec replicator.stop(); deregisterReplicaWithMaster(descriptor); descriptor.getConsumer().getChannel().close(); + replicatingFromMaster = null; replicaUUIDs.clear(); + + // this is needed because QueuingConsumer.nextDelivery() wont unblock + // if the connection is closed by application. + replicatorThread.interrupt(); } } } 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 76a31452964..6f3377cf59f 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 @@ -90,10 +90,6 @@ public class Replicator implements Runnable { } try { Delivery delivery = consumer.nextDelivery(); - - /* Delivery is blocking, upon unblock we will not check for - * stopping because we want to receive at least the last event. - */ byte[] bytesFromMessage = delivery.getBody(); checksPerformed = 0; // Set this object's class's class loader as context for de-serialization so that all exported classes @@ -103,6 +99,8 @@ public class Replicator implements Runnable { new ByteArrayInputStream(bytesFromMessage)); RacingEventServiceOperation operation = (RacingEventServiceOperation) ois.readObject(); applyOrQueue(operation); + } catch (InterruptedException irr) { + logger.info("Application requested shutdown."); } catch (ShutdownSignalException sse) { /* make sure to respond to a stop event without waiting */ if (isBeingStopped()) { @@ -209,8 +207,13 @@ public class Replicator implements Runnable { /* make sure to apply everything in queue before stopping this thread */ applyQueue(); } - stopped = true; logger.info("Signaled Replicator thread to stop asap."); + stopped = true; + try { + master.getConsumer().getChannel().getConnection().close(1); + } catch (Exception ex) { + // ignore any exception during abort. + } } public synchronized boolean isBeingStopped() {