Merge branch 'master' into bug2026

java/com.sap.sailing.domain.swisstimingreplayadapter/src/com/sap/sailing/domain/swisstimingreplayadapter/impl/SwissTimingReplayServiceImpl.java

@@ -1,7 +1,6 @@
 package com.sap.sailing.domain.swisstimingreplayadapter.impl;
 
 import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
@@ -25,6 +24,7 @@ import com.sap.sailing.domain.swisstimingreplayadapter.SwissTimingReplayListener
 import com.sap.sailing.domain.swisstimingreplayadapter.SwissTimingReplayRace;
 import com.sap.sailing.domain.swisstimingreplayadapter.SwissTimingReplayService;
 import com.sap.sailing.domain.tracking.TrackedRegattaRegistry;
+import com.sap.sailing.util.ByteArrayOutputStreamWithVisibleBuffer;
 
 public class SwissTimingReplayServiceImpl implements SwissTimingReplayService {
 
@@ -119,12 +119,6 @@ public class SwissTimingReplayServiceImpl implements SwissTimingReplayService {
         return startTimeFormat;
     }
 
-    private static class ByteArrayOutputStreamWithVisibleBuffer extends ByteArrayOutputStream {
-        public byte[] getBuffer() {
-            return buf;
-        }
-    }   
-    
     @Override
     public void loadRaceData(String link, SwissTimingReplayListener replayListener) {
         final URL raceDataUrl;

java/com.sap.sailing.domain.test/src/com/sap/sailing/domain/test/TimerTest.java

@@ -0,0 +1,49 @@
+package com.sap.sailing.domain.test;
+
+import static org.junit.Assert.fail;
+
+import java.util.Timer;
+import java.util.TimerTask;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+/**
+ * Tests around the Timer class; e.g., does it kill a timer if a task throws an exception?
+ * 
+ * @author Axel Uhl (d043530)
+ * 
+ */
+public class TimerTest {
+    @Rule
+    public Timeout AbstractTracTracLiveTestTimeout = new Timeout(10 * 1000);
+
+    @Test
+    public void exceptionInTask() throws InterruptedException {
+        Timer t = new Timer("Test");
+        final TimerTask taskThrowingException = new TimerTask() {
+            @Override
+            public void run() {
+                throw new RuntimeException("An exception that may or may not kill the timer");
+            }
+        };
+        t.schedule(taskThrowingException, 1l);
+        Thread.sleep(10);
+        final boolean[] result = new boolean[1];
+        try {
+            t.schedule(new TimerTask() {
+                @Override
+                public void run() {
+                    synchronized (result) {
+                        result[0] = true;
+                        result.notifyAll();
+                    }
+                }
+            }, 1l);
+            fail("Expected IllegalStateException because task terminating abnormally cancels the timer");
+        } catch (IllegalStateException e) {
+            // expected
+        }
+    }
+}

java/com.sap.sailing.domain/src/com/sap/sailing/util/ByteArrayOutputStreamWithVisibleBuffer.java

@@ -0,0 +1,9 @@
+package com.sap.sailing.util;
+
+import java.io.ByteArrayOutputStream;
+
+public class ByteArrayOutputStreamWithVisibleBuffer extends ByteArrayOutputStream {
+    public byte[] getBuffer() {
+        return buf;
+    }
+}

java/com.sap.sailing.server.replication/src/com/sap/sailing/server/replication/impl/ReplicationServiceImpl.java

@@ -144,6 +144,11 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
      * 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
@@ -283,36 +288,26 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
             }
             outboundObjectBuffer.writeObject(bytes);
             outboundBufferClasses.add(operation.getClass());
-            if (sendingTask == null) {
-                sendingTask = new TimerTask() {
-                    @Override
-                    public void run() {
-                        logger.fine("Running timer task to send "+outboundBufferClasses.size()+" operations in one message");
-                        final byte[] bytesToSend;
-                        final List<Class<?>> classesOfOperationsToSend;
-                        synchronized (outboundBufferMonitor) {
-                            logger.fine("Preparing "+outboundBufferClasses.size()+" operations for sending to RabbitMQ exchange");
+            if (outboundBuffer.size() > TRIGGER_MESSAGE_SIZE_IN_BYTES) {
+                logger.info("Triggering replication because buffer holds " + outboundBuffer.size()
+                        + "B which exceeds trigger size " + TRIGGER_MESSAGE_SIZE_IN_BYTES);
+                flushBufferToRabbitMQ();
+            } else {
+                if (sendingTask == null) {
+                    sendingTask = new TimerTask() {
+                        @Override
+                        public void run() {
                             try {
-                                outboundObjectBuffer.close();
-                            } catch (IOException e) {
-                                logger.log(Level.SEVERE, "Error trying to replicate "+outboundBufferClasses.size()+" operations", e);
+                                sendingTask = null;
+                                logger.fine("Running timer task, flushing buffer");
+                                flushBufferToRabbitMQ();
+                            } catch (Exception e) {
+                                logger.log(Level.SEVERE, "Exception while trying to replicate operations", e);
                             }
-                            bytesToSend = outboundBuffer.toByteArray();
-                            classesOfOperationsToSend = outboundBufferClasses;
-                            outboundBuffer = null;
-                            outboundObjectBuffer = null;
-                            outboundBufferClasses = null;
-                            sendingTask = null;
                         }
-                        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);
-                        }
-                    }
-                };
-                timer.schedule(sendingTask, TRANSMISSION_DELAY_MILLIS);
+                    };
+                    timer.schedule(sendingTask, TRANSMISSION_DELAY_MILLIS);
+                }
             }
             if (++messageCount % 10000l == 0) {
                 logger.info("Handled "+messageCount+" messages for replication. Current outbound replication queue size: "+outboundBufferClasses.size());
@@ -506,4 +501,35 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
         return serverUUID;
     }
 
+    /**
+     * 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("Running timer task, trying to acquire monitor");
+        final byte[] bytesToSend;
+        final List<Class<?>> classesOfOperationsToSend;
+        synchronized (outboundBufferMonitor) {
+            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;
+            outboundBuffer = null;
+            outboundObjectBuffer = null;
+            outboundBufferClasses = null;
+        }
+        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);
+        }
+    }
+
 }