Added message counting ability to the server (but not to the client)

Added synchronous CompleteBatch read by the client
Started implementing the synchronizer
Added support for null callbacks

bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedBulletinBoardClient.java

@@ -84,6 +84,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -101,6 +102,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -118,6 +120,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -136,6 +139,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -153,6 +157,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -171,6 +176,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -188,6 +194,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -205,6 +212,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
 
             @Override
             public void onFailure(Throwable t) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         });
@@ -233,6 +241,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
         localClient.readBatch(batchSpecificationMessage, new FutureCallback<CompleteBatch>() {
             @Override
             public void onSuccess(CompleteBatch result) {
+                if (callback != null)
                     callback.onSuccess(result);     // Read from local client was successful
             }
 
@@ -255,6 +264,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
                             public void onFailure(Throwable t) {}
                         });
 
+                        if (callback != null)
                             callback.onSuccess(result);
 
                     }
@@ -263,6 +273,7 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
                     public void onFailure(Throwable t) {
 
                         // Read from remote was unsuccessful: report error
+                        if (callback != null)
                             callback.onFailure(t);
 
                     }
@@ -301,11 +312,16 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
     }
 
     @Override
-    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) {
+    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException {
         subscriber.subscribe(filterList, new SubscriptionStoreCallback());
         return localClient.readMessages(filterList);
     }
 
+    @Override
+    public CompleteBatch readBatch(BatchSpecificationMessage batchSpecificationMessage) throws CommunicationException {
+        return localClient.readBatch(batchSpecificationMessage);
+    }
+
     @Override
     public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException {
         return localClient.generateSyncQuery(generateSyncQueryParams);
@@ -327,12 +343,4 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
         subscriber.subscribe(filterList, startEntry, callback);
     }
 
-    public int syncStatus(){
-        return 0;
-    }
-
-    public void reSync(){
-
-    }
-
 }

bulletin-board-client/src/main/java/meerkat/bulletinboard/LocalBulletinBoardClient.java

@@ -291,6 +291,31 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
 
     }
 
+    private class BatchDataReader implements Callable<List<BatchData>> {
+
+        private final BatchSpecificationMessage batchSpecificationMessage;
+
+        public BatchDataReader(BatchSpecificationMessage batchSpecificationMessage) {
+            this.batchSpecificationMessage = batchSpecificationMessage;
+        }
+
+        @Override
+        public List<BatchData> call() throws Exception {
+
+            ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
+            MessageOutputStream<BatchData> outputStream = new MessageOutputStream<>(byteOutputStream);
+            server.readBatch(batchSpecificationMessage, outputStream);
+
+            MessageInputStream<BatchData> inputStream =
+                    MessageInputStreamFactory.createMessageInputStream(
+                            new ByteArrayInputStream(byteOutputStream.toByteArray()),
+                            BatchData.class);
+
+            return inputStream.asList();
+
+        }
+    }
+
     @Override
     public void readMessages(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {
         Futures.addCallback(executorService.submit(new MessageReader(filterList)), callback);
@@ -310,6 +335,7 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
         public void onSuccess(List<BulletinBoardMessage> result) {
 
             // Report new messages to user
+            if (callback != null)
                 callback.onSuccess(result);
 
             MessageFilterList.Builder filterBuilder = filterList.toBuilder();
@@ -339,6 +365,7 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
         public void onFailure(Throwable t) {
 
             // Notify caller about failure and terminate subscription
+            if (callback != null)
                 callback.onFailure(t);
 
         }
@@ -503,7 +530,7 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
     }
 
     @Override
-    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) {
+    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException{
 
         try {
 
@@ -511,7 +538,40 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
             return reader.call();
 
         } catch (Exception e){
-            return null;
+            throw new CommunicationException("Error reading from server");
+        }
+
+    }
+
+    @Override
+    public CompleteBatch readBatch(BatchSpecificationMessage batchSpecificationMessage) throws CommunicationException {
+
+        MessageFilterList filterList = MessageFilterList.newBuilder()
+                .addFilter(MessageFilter.newBuilder()
+                        .setType(FilterType.TAG)
+                        .setTag(BulletinBoardConstants.BATCH_TAG)
+                        .build())
+                .addFilter(MessageFilter.newBuilder()
+                        .setType(FilterType.TAG)
+                        .setTag(BulletinBoardConstants.BATCH_ID_TAG_PREFIX + batchSpecificationMessage.getBatchId())
+                        .build())
+                .addFilter(MessageFilter.newBuilder()
+                        .setType(FilterType.SIGNER_ID)
+                        .setId(batchSpecificationMessage.getSignerId())
+                        .build())
+                .build();
+
+        BulletinBoardMessage batchMessage = readMessages(filterList).get(0);
+
+        BatchDataReader batchDataReader = new BatchDataReader(batchSpecificationMessage);
+
+        try {
+
+            List<BatchData> batchDataList = batchDataReader.call();
+            return new CompleteBatch(batchMessage, batchDataList);
+
+        } catch (Exception e) {
+            throw new CommunicationException("Error reading batch");
         }
 
     }
@@ -525,8 +585,11 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
     public void deleteMessage(MessageID msgID, FutureCallback<Boolean> callback) {
 
         try {
-            callback.onSuccess(server.deleteMessage(msgID).getValue());
+            Boolean deleted = server.deleteMessage(msgID).getValue();
+            if (callback != null)
+                callback.onSuccess(deleted);
         } catch (CommunicationException e) {
+            if (callback != null)
                 callback.onFailure(e);
         }
 
@@ -536,8 +599,11 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
     public void deleteMessage(long entryNum, FutureCallback<Boolean> callback) {
 
         try {
-            callback.onSuccess(server.deleteMessage(entryNum).getValue());
+            Boolean deleted = server.deleteMessage(entryNum).getValue();
+            if (callback != null)
+                callback.onSuccess(deleted);
         } catch (CommunicationException e) {
+            if (callback != null)
                 callback.onFailure(e);
         }
 

bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java

@@ -74,6 +74,7 @@ public abstract class MultiServerWorker<IN, OUT> extends BulletinClientWorker<IN
      */
     protected void succeed(OUT result){
         if (returnedResult.compareAndSet(false, true)) {
+            if (futureCallback != null)
                 futureCallback.onSuccess(result);
         }
     }
@@ -85,6 +86,7 @@ public abstract class MultiServerWorker<IN, OUT> extends BulletinClientWorker<IN
      */
     protected void fail(Throwable t){
         if (returnedResult.compareAndSet(false, true)) {
+            if (futureCallback != null)
                 futureCallback.onFailure(t);
         }
     }

bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java

@@ -2,6 +2,8 @@ package meerkat.bulletinboard;
 
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Timestamp;
+import meerkat.bulletinboard.workers.singleserver.SingleServerReadBatchWorker;
+import meerkat.bulletinboard.workers.singleserver.SingleServerReadMessagesWorker;
 import meerkat.comm.CommunicationException;
 import meerkat.comm.MessageInputStream;
 import meerkat.crypto.Digest;
@@ -138,32 +140,71 @@ public class SimpleBulletinBoardClient implements BulletinBoardClient{
     @Override
     public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) {
 
-        WebTarget webTarget;
-        Response response;
-        BulletinBoardMessageList messageList;
-
         // Replace null filter list with blank one.
         if (filterList == null){
-            filterList = MessageFilterList.newBuilder().build();
+            filterList = MessageFilterList.getDefaultInstance();
         }
 
         for (String db : meerkatDBs) {
 
             try {
-                webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(READ_MESSAGES_PATH);
 
-                response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(filterList, Constants.MEDIATYPE_PROTOBUF));
+                SingleServerReadMessagesWorker worker = new SingleServerReadMessagesWorker(db, filterList, 0);
 
-                messageList = response.readEntity(BulletinBoardMessageList.class);
+                List<BulletinBoardMessage> result = worker.call();
 
-                if (messageList != null){
+                return result;
-                    return messageList.getMessageList();
+
+            } catch (Exception ignored) {}
         }
 
-            } catch (Exception e) {}
+        return null;
 
     }
 
+    @Override
+    public CompleteBatch readBatch(BatchSpecificationMessage batchSpecificationMessage) throws CommunicationException {
+
+        // Create job with no retries for retrieval of the Bulletin Board Message that defines the batch
+
+        MessageFilterList filterList = MessageFilterList.newBuilder()
+                .addFilter(MessageFilter.newBuilder()
+                        .setType(FilterType.TAG)
+                        .setTag(BulletinBoardConstants.BATCH_TAG)
+                        .build())
+                .addFilter(MessageFilter.newBuilder()
+                        .setType(FilterType.TAG)
+                        .setTag(BulletinBoardConstants.BATCH_ID_TAG_PREFIX + batchSpecificationMessage.getBatchId())
+                        .build())
+                .addFilter(MessageFilter.newBuilder()
+                        .setType(FilterType.SIGNER_ID)
+                        .setId(batchSpecificationMessage.getSignerId())
+                        .build())
+                .build();
+
+        for (String db : meerkatDBs) {
+
+            try {
+                SingleServerReadMessagesWorker messagesWorker = new SingleServerReadMessagesWorker(db, filterList, 0);
+
+                List<BulletinBoardMessage> messages = messagesWorker.call();
+
+                if (messages == null || messages.size() < 1)
+                    continue;
+
+                BulletinBoardMessage batchMessage = messages.get(0);
+
+                SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(db, batchSpecificationMessage, 0);
+
+                List<BatchData> batchDataList = batchWorker.call();
+
+                CompleteBatch result = new CompleteBatch(batchMessage, batchDataList);
+
+                return result;
+
+            } catch (Exception ignored) {}
+        }
+
         return null;
 
     }

bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardSynchronizer.java

@@ -0,0 +1,188 @@
+package meerkat.bulletinboard;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.protobuf.ByteString;
+import meerkat.comm.CommunicationException;
+import meerkat.protobuf.BulletinBoardAPI.*;
+import meerkat.util.BulletinBoardUtils;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Created by Arbel on 13/04/2016.
+ * Simple, straightforward implementation of the {@link BulletinBoardSynchronizer} interface
+ */
+public class SimpleBulletinBoardSynchronizer implements BulletinBoardSynchronizer {
+
+    private DeletableSubscriptionBulletinBoardClient localClient;
+    private AsyncBulletinBoardClient remoteClient;
+
+    private volatile SyncStatus syncStatus;
+
+    private List<FutureCallback<Integer>> messageCountCallbacks;
+    private List<FutureCallback<SyncStatus>> syncStatusCallbacks;
+
+    private static final MessageFilterList EMPTY_FILTER = MessageFilterList.getDefaultInstance();
+    private static final int SLEEP_INTERVAL = 10000; // 10 Seconds
+
+    private class SyncCallback implements FutureCallback<List<BulletinBoardMessage>> {
+
+        @Override
+        public void onSuccess(List<BulletinBoardMessage> result) {
+
+            SyncStatus newStatus = SyncStatus.PENDING;
+
+            if (result.size() == 0) {
+                newStatus = SyncStatus.SYNCHRONIZED;
+            }
+
+            else{   // Upload messages
+
+                for (BulletinBoardMessage message : result){
+
+                    if (message.getMsg().getTagList().contains(BulletinBoardConstants.BATCH_TAG)){
+
+                        // This is a batch message: need to upload batch data as well as the message itself
+                        ByteString signerId = message.getSig(0).getSignerId();
+                        long batchID = Long.parseLong(BulletinBoardUtils.findTagWithPrefix(message, BulletinBoardConstants.BATCH_ID_TAG_PREFIX));
+
+                        BatchSpecificationMessage batchSpecificationMessage = BatchSpecificationMessage.newBuilder().build();
+
+                        localClient.readBatch(batchSpecificationMessage, null);
+
+                    }
+                    else{
+
+                        // This is a regular message: post it
+                        try {
+                            remoteClient.postMessage(message);
+                        } catch (CommunicationException e) {
+                            newStatus = SyncStatus.SERVER_ERROR;
+                        }
+
+                    }
+                }
+
+            }
+
+            updateSyncStatus(newStatus);
+
+        }
+
+        @Override
+        public void onFailure(Throwable t) {
+
+            updateSyncStatus(SyncStatus.SERVER_ERROR);
+
+        }
+
+    }
+
+    public SimpleBulletinBoardSynchronizer() {
+        this.syncStatus = SyncStatus.STOPPED;
+    }
+
+    private synchronized void updateSyncStatus(SyncStatus newStatus) {
+
+        if (newStatus != syncStatus){
+
+            syncStatus = newStatus;
+
+            for (FutureCallback<SyncStatus> callback : syncStatusCallbacks){
+                if (callback != null)
+                    callback.onSuccess(syncStatus);
+            }
+
+        }
+
+    }
+
+    @Override
+    public void init(DeletableSubscriptionBulletinBoardClient localClient, AsyncBulletinBoardClient remoteClient) {
+
+        updateSyncStatus(SyncStatus.STOPPED);
+
+        this.localClient = localClient;
+        this.remoteClient = remoteClient;
+
+        messageCountCallbacks = new LinkedList<>();
+        syncStatusCallbacks = new LinkedList<>();
+
+    }
+
+    @Override
+    public SyncStatus getSyncStatus() {
+        return syncStatus;
+    }
+
+    @Override
+    public void subscribeToSyncStatus(FutureCallback<SyncStatus> callback) {
+        syncStatusCallbacks.add(callback);
+    }
+
+    @Override
+    public List<BulletinBoardMessage> getRemainingMessages()  throws CommunicationException{
+        return localClient.readMessages(EMPTY_FILTER);
+    }
+
+    @Override
+    public void getRemainingMessages(FutureCallback<List<BulletinBoardMessage>> callback) {
+        localClient.readMessages(EMPTY_FILTER, callback);
+    }
+
+    @Override
+    public long getRemainingMessagesCount() throws CommunicationException {
+        return localClient.readMessages(EMPTY_FILTER).size();
+    }
+
+    @Override
+    public void subscribeToRemainingMessagesCount(FutureCallback<Integer> callback) {
+        messageCountCallbacks.add(callback);
+    }
+
+    @Override
+    public void run() {
+
+        if (syncStatus != SyncStatus.STOPPED) {
+
+            updateSyncStatus(SyncStatus.PENDING);
+            SyncCallback callback = new SyncCallback();
+
+            while (syncStatus != SyncStatus.STOPPED) {
+
+                try {
+
+                    do {
+                        localClient.readMessages(EMPTY_FILTER, callback);
+                    } while (syncStatus == SyncStatus.PENDING);
+
+                    synchronized (this) {
+                        this.wait(SLEEP_INTERVAL);
+                    }
+
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+
+            }
+
+        }
+
+    }
+
+    @Override
+    public void nudge() {
+        synchronized (this) {
+            this.notify();
+        }
+    }
+
+    @Override
+    public void stop() {
+
+        updateSyncStatus(SyncStatus.STOPPED);
+
+    }
+
+}

bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java

@@ -97,6 +97,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
 
         @Override
         public void onSuccess(T result) {
+            if (futureCallback != null)
                 futureCallback.onSuccess(result);
         }
 
@@ -115,6 +116,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
                 scheduleWorker(worker, this);
             } else {
                 // No more retries: notify caller about failure
+                if (futureCallback != null)
                     futureCallback.onFailure(t);
             }
 
@@ -150,6 +152,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
             }
 
             if (batchDataRemaining.decrementAndGet() == 0){
+                if (callback != null)
                     callback.onSuccess(this.aggregatedResult.get());
             }
         }
@@ -158,6 +161,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
         public void onFailure(Throwable t) {
 
             // Notify caller about failure
+            if (callback != null)
                 callback.onFailure(t);
 
         }
@@ -195,25 +199,15 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
 
             if (remainingQueries.decrementAndGet() == 0){
 
-                String batchIdStr = BulletinBoardUtils.findTagWithPrefix(batchMessage, BulletinBoardConstants.BATCH_ID_TAG_PREFIX);
+                if (callback != null)
-
+                    callback.onSuccess(new CompleteBatch(batchMessage, batchDataList));
-                if (batchIdStr == null){
-                    callback.onFailure(new CommunicationException("Server returned invalid message with no Batch ID tag"));
-                }
-
-                BeginBatchMessage beginBatchMessage =
-                        BeginBatchMessage.newBuilder()
-                                .setSignerId(batchMessage.getSig(0).getSignerId())
-                                .setBatchId(Integer.parseInt(batchIdStr))
-                                .addAllTag(BulletinBoardUtils.removePrefixTags(batchMessage, Arrays.asList(prefixes)))
-                                .build();
-                callback.onSuccess(new CompleteBatch(beginBatchMessage, batchDataList, batchMessage.getSig(0)));
             }
 
         }
 
         protected void fail(Throwable t) {
             if (failed.compareAndSet(false, true)) {
+                if (callback != null)
                     callback.onFailure(t);
             }
         }
@@ -289,6 +283,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
         public void onSuccess(List<BulletinBoardMessage> result) {
 
             // Report new messages to user
+            if (callback != null)
                 callback.onSuccess(result);
 
             // Remove last filter from list (MIN_ENTRY one)
@@ -315,6 +310,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
             fail();
 
             // Notify caller about failure and terminate subscription
+            if (callback != null)
                 callback.onFailure(t);
         }
     }
@@ -397,6 +393,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
 
         @Override
         public void onFailure(Throwable t) {
+            if (callback != null)
                 callback.onFailure(t);
         }
 
@@ -425,6 +422,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
 
         @Override
         public void onFailure(Throwable t) {
+            if (callback != null)
                 callback.onFailure(t);
         }
     }

bulletin-board-client/src/main/java/meerkat/bulletinboard/ThreadedBulletinBoardSubscriber.java

@@ -29,6 +29,8 @@ public class ThreadedBulletinBoardSubscriber implements BulletinBoardSubscriber
     private AtomicBoolean isSyncInProgress;
     private Semaphore rescheduleSemaphore;
 
+    private AtomicBoolean stopped;
+
     private static final Float[] BREAKPOINTS = {0.5f, 0.75f, 0.9f, 0.95f, 0.99f, 0.999f};
 
     public ThreadedBulletinBoardSubscriber(Collection<SubscriptionBulletinBoardClient> clients, BulletinBoardClient localClient) {
@@ -44,6 +46,8 @@ public class ThreadedBulletinBoardSubscriber implements BulletinBoardSubscriber
         isSyncInProgress = new AtomicBoolean(false);
         rescheduleSemaphore = new Semaphore(1);
 
+        stopped = new AtomicBoolean(false);
+
     }
 
     /**
@@ -131,6 +135,7 @@ public class ThreadedBulletinBoardSubscriber implements BulletinBoardSubscriber
 
                 //TODO: log
 
+                if (callback != null)
                     callback.onFailure(e); // Hard error: Cannot guarantee subscription safety
 
             }
@@ -217,6 +222,7 @@ public class ThreadedBulletinBoardSubscriber implements BulletinBoardSubscriber
         public void onSuccess(List<BulletinBoardMessage> result) {
 
             // Propagate result to caller
+            if (callback != null)
                 callback.onSuccess(result);
 
             // Renew subscription
@@ -249,6 +255,7 @@ public class ThreadedBulletinBoardSubscriber implements BulletinBoardSubscriber
         public void onSuccess(List<BulletinBoardMessage> result) {
 
             // Propagate result to caller
+            if (callback != null)
                 callback.onSuccess(result);
 
         }

bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java

@@ -679,6 +679,30 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
 
 	}
 
+	@Override
+	public IntMsg getMessageCount(MessageFilterList filterList) throws CommunicationException {
+
+		BulletinBoardMessageList.Builder resultListBuilder = BulletinBoardMessageList.newBuilder();
+
+		// SQL length is roughly 50 characters per filter + 50 for the query itself
+		StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1));
+
+		// Check if Tag/Signature tables are required for filtering purposes
+
+		sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.COUNT_MESSAGES));
+
+		// Get conditions
+
+		SQLAndParameters sqlAndParameters = getSQLFromFilters(filterList);
+		sqlBuilder.append(sqlAndParameters.sql);
+
+		// Run query and stream the output using a MessageCallbackHandler
+
+		List<Long> count = jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new LongMapper());
+		return IntMsg.newBuilder().setValue(count.get(0).intValue()).build();
+
+	}
+
 
 	/**
 	 * This method returns a string representation of the tag associated with a batch ID
@@ -713,23 +737,9 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
 						.build())
 				.build();
 
-		// SQL length is roughly 50 characters per filter + 50 for the query itself
+		int count = getMessageCount(filterList).getValue();
-		StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1));
 
-		// Check if Tag/Signature tables are required for filtering purposes
+		return count > 0;
-
-		sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.COUNT_MESSAGES));
-
-		// Get conditions
-
-		SQLAndParameters sqlAndParameters = getSQLFromFilters(filterList);
-		sqlBuilder.append(sqlAndParameters.sql);
-
-		// Run query and stream the output using a MessageCallbackHandler
-
-		List<Long> count = jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new LongMapper());
-
-		return (count.size() > 0) && (count.get(0) > 0);
 
 	}
 

bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java

@@ -15,14 +15,12 @@ import meerkat.bulletinboard.sqlserver.MySQLQueryProvider;
 import meerkat.bulletinboard.sqlserver.SQLiteQueryProvider;
 import meerkat.comm.CommunicationException;
 import meerkat.comm.MessageOutputStream;
-import meerkat.protobuf.BulletinBoardAPI;
 import meerkat.protobuf.BulletinBoardAPI.*;
 import static meerkat.bulletinboard.BulletinBoardConstants.*;
 import static meerkat.rest.Constants.*;
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.util.Collection;
 
 /**
  * An implementation of the BulletinBoardServer which functions as a WebApp
@@ -99,6 +97,16 @@ public class BulletinBoardWebApp implements BulletinBoardServer, ServletContextL
 		bulletinBoard.readMessages(filterList, out);
 	}
 
+	@Path(COUNT_MESSAGES_PATH)
+	@POST
+	@Consumes(MEDIATYPE_PROTOBUF)
+	@Produces(MEDIATYPE_PROTOBUF)
+	@Override
+	public IntMsg getMessageCount(MessageFilterList filterList) throws CommunicationException {
+		init();
+		return bulletinBoard.getMessageCount(filterList);
+	}
+
 
 	@Path(READ_MESSAGES_PATH)
 	@POST

meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardClient.java

@@ -42,7 +42,15 @@ public interface BulletinBoardClient {
      * @param filterList return only messages that match the filters (null means no filtering)
      * @return the list of messages
      */
-    List<BulletinBoardMessage> readMessages(MessageFilterList filterList);
+    List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException;
+
+    /**
+     * Read a given batch message from the bulletin board
+     * @param batchSpecificationMessage contains the data required to specify a single batch instance
+     * @return the complete batch
+     * @throws CommunicationException
+     */
+    CompleteBatch readBatch(BatchSpecificationMessage batchSpecificationMessage) throws CommunicationException;
 
     /**
      * Create a SyncQuery to test against that corresponds with the current server state for a specific filter list

meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardConstants.java

@@ -10,6 +10,7 @@ public interface BulletinBoardConstants {
     public static final String BULLETIN_BOARD_SERVER_PATH = "/bbserver";
     public static final String GENERATE_SYNC_QUERY_PATH = "/generatesyncquery";
     public static final String READ_MESSAGES_PATH = "/readmessages";
+    public static final String COUNT_MESSAGES_PATH = "/countmessages";
     public static final String READ_BATCH_PATH = "/readbatch";
     public static final String POST_MESSAGE_PATH = "/postmessage";
     public static final String BEGIN_BATCH_PATH = "/beginbatch";

meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java

@@ -32,13 +32,21 @@ public interface BulletinBoardServer{
 	public BoolMsg postMessage(BulletinBoardMessage msg) throws CommunicationException;
 
 	 /**
-	  * Read all messages posted matching the given filter
+	  * Read all posted messages matching the given filters
 	  * @param filterList return only messages that match the filters (empty list or null means no filtering)
 	  * @param out is an output stream into which the matching messages are written
 	  * @throws CommunicationException on DB connection error
 	  */
 	public void readMessages(MessageFilterList filterList, MessageOutputStream<BulletinBoardMessage> out) throws CommunicationException;
 
+	/**
+	 * Return the number of posted messages matching the given filters
+	 * @param filterList count only messages that match the filters (empty list or null means no filtering)
+	 * @return an IntMsg containing the number of messages that match the filter
+	 * @throws CommunicationException on DB connection error
+	 */
+	public IntMsg getMessageCount(MessageFilterList filterList) throws CommunicationException;
+
 	/**
 	 * Informs server about a new batch message
 	 * @param message contains the required data about the new batch

meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardSynchronizer.java

@@ -1,5 +1,12 @@
 package meerkat.bulletinboard;
 
+import meerkat.comm.CommunicationException;
+import meerkat.protobuf.BulletinBoardAPI.*;
+
+import com.google.common.util.concurrent.FutureCallback;
+
+import java.util.List;
+
 /**
  * Created by Arbel Deutsch Peled on 08-Mar-16.
  * This interface defines the behaviour of a bulletin board synchronizer
@@ -7,16 +14,71 @@ package meerkat.bulletinboard;
  */
 public interface BulletinBoardSynchronizer extends Runnable {
 
-    /**
+    public enum SyncStatus{
-     *
+        SYNCHRONIZED,   // No more messages to upload
-     * @param localClient is a client for the local DB instance
+        PENDING,        // Synchronizer is uploading data
-     * @param remoteClient is a client for the remote DBs
+        SERVER_ERROR,   // Synchronizer encountered an error while uploading
-     * @param minRedundancy
+        STOPPED         // Stopped/Not started by user
-     */
+    }
-    public void init(BulletinBoardClient localClient, AsyncBulletinBoardClient remoteClient, float minRedundancy);
 
+    /**
+     * Initializes the synchronizer with the required data to function properly
+     * @param localClient is a client for the temporary local storage server which contains only data to be uploaded
+     * @param remoteClient is a client for the remote servers into which the data needs to be uploaded
+     */
+    public void init(DeletableSubscriptionBulletinBoardClient localClient, AsyncBulletinBoardClient remoteClient);
+
+    /**
+     * Returns the current server synchronization status
+     * @return the current synchronization status
+     */
+    public SyncStatus getSyncStatus();
+
+    /**
+     * Creates a subscription to sync status changes
+     * @param callback is the handler for any status changes
+     */
+    public void subscribeToSyncStatus(FutureCallback<SyncStatus> callback);
+
+    /**
+     * Returns the messages which have not yet been synchronized
+     * @return the list of messages remaining to be synchronized
+     */
+    public List<BulletinBoardMessage> getRemainingMessages() throws CommunicationException;
+
+    /**
+     * Asynchronously returns the messages which have not yet been synchronized
+     * @param callback is the handler for the list of messages
+     */
+    public void getRemainingMessages(FutureCallback<List<BulletinBoardMessage>> callback);
+
+    /**
+     * Returns the current number of unsynchronized messages
+     * @return the current synchronization status
+     */
+    public long getRemainingMessagesCount() throws CommunicationException;
+
+    /**
+     * Creates a subscription to changes in the number of unsynchronized messages
+     * @param callback is the handler for any status changes
+     */
+    public void subscribeToRemainingMessagesCount(FutureCallback<Integer> callback);
+
+    /**
+     * Starts the synchronization
+     */
     @Override
     public void run();
 
+    /**
+     * Lets the Synchronizer know that there is new data to be uploaded
+     * This is used to reduce the latency between local data-writes and uploads to the remote servers
+     */
+    public void nudge();
+
+    /**
+     * Stops the synchronization
+     */
+    public void stop();
 
 }

meerkat-common/src/main/java/meerkat/bulletinboard/CompleteBatch.java

@@ -3,8 +3,9 @@ package meerkat.bulletinboard;
 import com.google.protobuf.Timestamp;
 import meerkat.protobuf.BulletinBoardAPI.*;
 import meerkat.protobuf.Crypto.*;
-import meerkat.util.BulletinBoardMessageComparator;
+import meerkat.util.BulletinBoardUtils;
 
+import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 
@@ -49,6 +50,36 @@ public class CompleteBatch {
         this.timestamp = timestamp;
     }
 
+    /**
+     * Combines the actual Bulletin board representation of a batch into a CompleteBatch object
+     * @param batchMessage is the BulletinBoard message that specifies the batch
+     * @param batchDataList is the (ordered) list of batch data
+     */
+    public CompleteBatch(BulletinBoardMessage batchMessage, List<BatchData> batchDataList) throws IllegalArgumentException{
+
+        final String[] PREFIXES = {
+                BulletinBoardConstants.BATCH_ID_TAG_PREFIX,
+                BulletinBoardConstants.BATCH_TAG};
+
+        String batchIdStr = BulletinBoardUtils.findTagWithPrefix(batchMessage, BulletinBoardConstants.BATCH_ID_TAG_PREFIX);
+
+        if (batchIdStr == null){
+            throw new IllegalArgumentException("");
+        }
+
+        this.beginBatchMessage =
+                BeginBatchMessage.newBuilder()
+                        .setSignerId(batchMessage.getSig(0).getSignerId())
+                        .setBatchId(Integer.parseInt(batchIdStr))
+                        .addAllTag(BulletinBoardUtils.removePrefixTags(batchMessage, Arrays.asList(PREFIXES)))
+                        .build();
+
+        this.batchDataList = batchDataList;
+        this.signature = batchMessage.getSig(0);
+        this.timestamp = batchMessage.getMsg().getTimestamp();
+
+    }
+
     public BeginBatchMessage getBeginBatchMessage() {
         return beginBatchMessage;
     }