Bulletin Board Client support for streaming and Timestamps

Created standard Checksum interface and implementation for Sync Query mechanism
Added the Timestamp into the Batch Digest and Signature logic

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

@@ -381,11 +381,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
         @Override
         public void onSuccess(Boolean msg) {
             closeBatch(
-                    CloseBatchMessage.newBuilder()
+                    completeBatch.getCloseBatchMessage(),
-                            .setBatchId(completeBatch.getBeginBatchMessage().getBatchId())
-                            .setSig(completeBatch.getSignature())
-                            .setBatchLength(completeBatch.getBatchDataList().size())
-                            .build(),
                     callback
             );
         }

bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerReadBatchWorker.java

@@ -3,6 +3,7 @@ package meerkat.bulletinboard.workers.singleserver;
 import meerkat.bulletinboard.CompleteBatch;
 import meerkat.bulletinboard.SingleServerWorker;
 import meerkat.comm.CommunicationException;
+import meerkat.comm.MessageInputStream;
 import meerkat.protobuf.BulletinBoardAPI.*;
 import meerkat.rest.Constants;
 
@@ -12,6 +13,9 @@ import javax.ws.rs.client.Entity;
 import javax.ws.rs.client.WebTarget;
 import javax.ws.rs.core.GenericType;
 import javax.ws.rs.core.Response;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.InvocationTargetException;
 import java.util.List;
 
 import static meerkat.bulletinboard.BulletinBoardConstants.BULLETIN_BOARD_SERVER_PATH;
@@ -40,29 +44,33 @@ public class SingleServerReadBatchWorker extends SingleServerWorker<BatchSpecifi
         Client client = clientLocal.get();
 
         WebTarget webTarget;
-        Response response;
 
         // Get the batch data
 
         webTarget = client.target(serverAddress).path(BULLETIN_BOARD_SERVER_PATH).path(READ_BATCH_PATH);
-        response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(
+        InputStream in = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(payload, Constants.MEDIATYPE_PROTOBUF), InputStream.class);
-                Entity.entity(payload, Constants.MEDIATYPE_PROTOBUF));
 
-        // Retrieve answer
+        MessageInputStream<BatchData> inputStream = null;
 
         try {
 
-            // If a BatchDataList is returned: the read was successful
+            inputStream = MessageInputStream.MessageInputStreamFactory.createMessageInputStream(in, BatchData.class);
-            return response.readEntity(BatchDataList.class).getDataList();
 
-        } catch (ProcessingException | IllegalStateException e) {
+            return inputStream.asList();
+
+        } catch (IOException | InvocationTargetException e) {
 
             // Read failed
-            throw new CommunicationException("Could not contact the server");
+            throw new CommunicationException("Could not contact the server or server returned illegal result");
 
-        }
+        } catch (NoSuchMethodException | IllegalAccessException e) {
-        finally {
+
-            response.close();
+            throw new CommunicationException("MessageInputStream error");
+
+        } finally {
+            try {
+                inputStream.close();
+            } catch (IOException ignored) {}
         }
 
     }

bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerReadMessagesWorker.java

@@ -2,6 +2,8 @@ package meerkat.bulletinboard.workers.singleserver;
 
 import meerkat.bulletinboard.SingleServerWorker;
 import meerkat.comm.CommunicationException;
+import meerkat.comm.MessageInputStream;
+import meerkat.protobuf.BulletinBoardAPI;
 import meerkat.protobuf.BulletinBoardAPI.BulletinBoardMessageList;
 import meerkat.protobuf.BulletinBoardAPI.MessageFilterList;
 import meerkat.protobuf.BulletinBoardAPI.BulletinBoardMessage;
@@ -13,6 +15,9 @@ import javax.ws.rs.client.Entity;
 import javax.ws.rs.client.WebTarget;
 import javax.ws.rs.core.Response;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.InvocationTargetException;
 import java.util.List;
 
 import static meerkat.bulletinboard.BulletinBoardConstants.BULLETIN_BOARD_SERVER_PATH;
@@ -38,30 +43,33 @@ public class SingleServerReadMessagesWorker extends SingleServerWorker<MessageFi
         Client client = clientLocal.get();
 
         WebTarget webTarget;
-        Response response;
 
         // Send request to Server
         webTarget = client.target(serverAddress).path(BULLETIN_BOARD_SERVER_PATH).path(READ_MESSAGES_PATH);
-        response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(
+        InputStream in = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(payload, Constants.MEDIATYPE_PROTOBUF), InputStream.class);
-                Entity.entity(payload, Constants.MEDIATYPE_PROTOBUF));
 
-        // Retrieve answer
+        MessageInputStream<BulletinBoardMessage> inputStream = null;
 
         try {
 
-            // If a BulletinBoardMessageList is returned: the read was successful
+            inputStream = MessageInputStream.MessageInputStreamFactory.createMessageInputStream(in, BulletinBoardMessage.class);
-            return response.readEntity(BulletinBoardMessageList.class).getMessageList();
 
-        } catch (ProcessingException | IllegalStateException e) {
+            return inputStream.asList();
+
+        } catch (IOException | InvocationTargetException e) {
 
             // Read failed
-            throw new CommunicationException("Could not contact the server");
+            throw new CommunicationException("Could not contact the server or server returned illegal result");
 
-        }
+        } catch (NoSuchMethodException | IllegalAccessException e) {
-        finally {
-            response.close();
-        }
 
+            throw new CommunicationException("MessageInputStream error");
+
+        } finally {
+            try {
+                inputStream.close();
+            } catch (IOException ignored) {}
+        }
 
     }
 

bulletin-board-client/src/test/java/ThreadedBulletinBoardClientIntegrationTest.java

@@ -1,5 +1,6 @@
 import com.google.common.util.concurrent.FutureCallback;
-import com.google.protobuf.ByteString;
+import com.google.protobuf.*;
+import com.google.protobuf.Timestamp;
 import meerkat.bulletinboard.AsyncBulletinBoardClient;
 import meerkat.bulletinboard.CompleteBatch;
 import meerkat.bulletinboard.GenericBatchDigitalSignature;
@@ -284,6 +285,11 @@ public class ThreadedBulletinBoardClientIntegrationTest {
 
         }
 
+        completeBatch.setTimestamp(Timestamp.newBuilder()
+                .setSeconds(Math.abs(90))
+                .setNanos(50)
+                .build());
+
         signers[signer].updateContent(completeBatch);
 
         completeBatch.setSignature(signers[signer].sign());
@@ -357,6 +363,10 @@ public class ThreadedBulletinBoardClientIntegrationTest {
                         .addTag("Signature")
                         .addTag("Trustee")
                         .setData(ByteString.copyFrom(b1))
+                        .setTimestamp(Timestamp.newBuilder()
+                                .setSeconds(20)
+                                .setNanos(30)
+                                .build())
                         .build())
                 .addSig(Crypto.Signature.newBuilder()
                         .setType(Crypto.SignatureType.DSA)
@@ -440,6 +450,10 @@ public class ThreadedBulletinBoardClientIntegrationTest {
         CloseBatchMessage closeBatchMessage = CloseBatchMessage.newBuilder()
                 .setBatchId(BATCH_ID)
                 .setBatchLength(BATCH_LENGTH)
+                .setTimestamp(Timestamp.newBuilder()
+                        .setSeconds(50)
+                        .setNanos(80)
+                        .build())
                 .setSig(completeBatch.getSignature())
                 .build();
 
@@ -525,6 +539,10 @@ public class ThreadedBulletinBoardClientIntegrationTest {
                         .setBatchId(NON_EXISTENT_BATCH_ID)
                         .setBatchLength(1)
                         .setSig(Crypto.Signature.getDefaultInstance())
+                        .setTimestamp(Timestamp.newBuilder()
+                                .setSeconds(9)
+                                .setNanos(12)
+                                .build())
                         .build();
 
         // Try to close the (unopened) batch;

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

@@ -446,9 +446,10 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{
 
 		sql = sqlQueryProvider.getSQLString(QueryType.FIND_MSG_ID);
 		Map namedParameters = new HashMap();
+
 		namedParameters.put(QueryType.FIND_MSG_ID.getParamName(0),msgID);
 
-		List<Long> entryNums = jdbcTemplate.query(sql, new MapSqlParameterSource(namedParameters), new LongMapper());
+		List<Long> entryNums = jdbcTemplate.query(sql, namedParameters, new LongMapper());
 
 		if (entryNums.size() > 0){
 
@@ -462,7 +463,7 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{
 			namedParameters.put(QueryType.INSERT_MSG.getParamName(2), msg.getMsg().toByteArray());
 
 			KeyHolder keyHolder = new GeneratedKeyHolder();
-			jdbcTemplate.update(sql,new MapSqlParameterSource(namedParameters),keyHolder);
+			jdbcTemplate.update(sql, new MapSqlParameterSource(namedParameters), keyHolder);
 
 			entryNum = keyHolder.getKey().longValue();
 
@@ -785,6 +786,9 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{
 						.build()
 		);
 
+		// Add timestamp to CompleteBatch
+		completeBatch.setTimestamp(message.getTimestamp());
+
 		// Add actual batch data to CompleteBatch
 
 		sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_MESSAGE_DATA);
@@ -903,20 +907,20 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{
 
 		long lastEntryNum = getLastMessageEntry();
 
-		long checksum = 0;
-
 		Iterator<SingleSyncQuery> queryIterator = syncQuery.getQueryList().iterator();
 
 		SingleSyncQuery currentQuery = queryIterator.next();
 
 		List<BulletinBoardMessage> messageStubs = readMessageStubs(syncQuery.getFilterList());
 
+		Checksum checksum = new SimpleChecksum();
+
 		for (BulletinBoardMessage message : messageStubs){
 
 			// Check for end of current query
 			if (timestampComparator.compare(message.getMsg().getTimestamp(), currentQuery.getTimeOfSync()) > 0){
 
-				if (checksum == currentQuery.getChecksum()){
+				if (checksum.getChecksum() == currentQuery.getChecksum()){
 					lastTimeOfSync = currentQuery.getTimeOfSync();
 				} else {
 					break;
@@ -932,13 +936,13 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{
 
 			// Advance checksum
 
-			ByteString messageID = message.getMsg().getData();
+			ByteString messageID = message.getMsg().getData();	// The data field contains the message ID
 
-			checksum &= messageID.byteAt(0) & messageID.byteAt(1) & messageID.byteAt(2) & messageID.byteAt(3);
+			checksum.update(messageID);
 
 		}
 
-		if (checksum == currentQuery.getChecksum()){
+		if (checksum.getChecksum() == currentQuery.getChecksum()){
 			lastTimeOfSync = currentQuery.getTimeOfSync();
 		}
 

bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java

@@ -637,6 +637,8 @@ public class GenericBulletinBoardServerTest {
 	public void testSyncQuery()
 			throws SignatureException, CommunicationException, IOException,NoSuchMethodException, IllegalAccessException, InvocationTargetException {
 
+		Checksum checksum = new SimpleChecksum();
+
 		Timestamp timestamp = Timestamp.newBuilder()
 				.setSeconds(1)
 				.setNanos(0)
@@ -687,7 +689,7 @@ public class GenericBulletinBoardServerTest {
 		syncQuery = SyncQuery.newBuilder()
 				.setFilterList(MessageFilterList.getDefaultInstance())
 				.addQuery(SingleSyncQuery.newBuilder()
-						.setChecksum(messageID.byteAt(0) & messageID.byteAt(1) & messageID.byteAt(2) & messageID.byteAt(3))
+						.setChecksum(checksum.getChecksum(messageID))
 						.setTimeOfSync(timestamp)
 						.build())
 				.build();

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

@@ -0,0 +1,122 @@
+package meerkat.bulletinboard;
+
+import com.google.protobuf.ByteString;
+import meerkat.crypto.Digest;
+import meerkat.protobuf.BulletinBoardAPI.BulletinBoardMessage;
+import meerkat.protobuf.BulletinBoardAPI.MessageID;
+
+import java.util.Collection;
+
+/**
+ * Created by Arbel Deutsch Peled on 01-Mar-16.
+ * This interface is used to create checksums of Bulletin Board messages IDs
+ * This is useful in comparing database states
+ */
+public interface Checksum {
+
+    /**
+     * Sets the Digest method which is used in creating message IDs from the messages themselves
+     * This method must be called with an initialized Digest before calling any methods that receive a parameter of type BulletinBoardMessage
+     * @param digest is the Digest that will be used to create message IDs from Bulletin Board Messages
+     */
+    public void setDigest(Digest digest);
+
+    /**
+     * Used to reset the current checksum state
+     */
+    public void reset();
+
+    /**
+     * Update the current checksum with the given ID
+     * @param messageID is the message ID to be added
+     */
+    public void update(MessageID messageID);
+
+    /**
+     * Update the current checksum with the given collection of IDs
+     * @param messageIDs contains the message IDs
+     */
+    public void update(Collection<MessageID> messageIDs);
+
+    /**
+     * Update the current checksum with the given ID
+     * @param messageID is the message ID to be added
+     */
+    public void update(ByteString messageID);
+
+    /**
+     * Update the current checksum with the given ID
+     * @param messageID is the message ID to be added
+     */
+    public void update(byte[] messageID);
+
+    /**
+     * Update the current checksum with the message ID of the given message
+     * @param bulletinBoardMessage is the message whose ID should be added to the checksum
+     * @throws IllegalStateException if a Digest has not been set before calling this method
+     */
+    public void digestAndUpdate(BulletinBoardMessage bulletinBoardMessage) throws IllegalStateException;
+
+
+    /**
+     * Update the current checksum with the message IDs of the given messages
+     * @param bulletinBoardMessages contains the messages whose IDs should be added to the checksum
+     * @throws IllegalStateException if a Digest has not been set before calling this method
+     */
+    public void digestAndUpdate(Collection<BulletinBoardMessage> bulletinBoardMessages) throws IllegalStateException;
+
+    /**
+     * Returns the current checksum without changing the checksum state
+     * @return the current checksum
+     */
+    public long getChecksum();
+
+    /**
+     * Updates the current checksum with the given ID and returns the resulting checksum
+     * The checksum is not reset afterwards
+     * @param messageID is the message ID to be added
+     * @return the updated checksum
+     */
+    public long getChecksum(MessageID messageID);
+
+    /**
+     * Updates the current checksum with the given ID and returns the resulting checksum
+     * The checksum is not reset afterwards
+     * @param messageID is the message ID to be added
+     * @return the updated checksum
+     */
+    public long getChecksum(ByteString messageID);
+
+    /**
+     * Updates the current checksum with the given ID and returns the resulting checksum
+     * The checksum is not reset afterwards
+     * @param messageID is the message ID to be added
+     * @return the updated checksum
+     */
+    public long getChecksum(byte[] messageID);
+
+    /**
+     * Updates the current checksum with the given IDs and returns the resulting checksum
+     * The checksum is not reset afterwards
+     * @param messageIDs contains the message IDs to be added
+     * @return the updated checksum
+     */
+    public long getChecksum(Collection<MessageID> messageIDs);
+
+    /**
+     * Updates the current checksum with the message ID of the given message
+     * The checksum is not reset afterwards
+     * @param bulletinBoardMessage is the message whose ID should be added to the checksum
+     * @return the updated checksum
+     */
+    public long digestAndGetChecksum(BulletinBoardMessage bulletinBoardMessage) throws IllegalStateException;
+
+    /**
+     * Updates the current checksum with the message IDs of the given messages
+     * The checksum is not reset afterwards
+     * @param bulletinBoardMessages contains the messages whose IDs should be added to the checksum
+     * @return the updated checksum
+     */
+    public long digestAndGetChecksum(Collection<BulletinBoardMessage> bulletinBoardMessages) throws IllegalStateException;
+
+}

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

@@ -29,6 +29,8 @@ public class GenericBatchDigest implements BatchDigest{
             update(batchData);
         }
 
+        update(completeBatch.getTimestamp());
+
     }
 
     @Override

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

@@ -32,10 +32,13 @@ public class GenericBatchDigitalSignature implements BatchDigitalSignature{
     public void updateContent(CompleteBatch completeBatch) throws SignatureException {
 
         digitalSignature.updateContent(completeBatch.getBeginBatchMessage());
+
         for (BatchData batchData : completeBatch.getBatchDataList()) {
             digitalSignature.updateContent(batchData);
         }
 
+        digitalSignature.updateContent(completeBatch.getTimestamp());
+
     }
 
     @Override

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

@@ -0,0 +1,131 @@
+package meerkat.bulletinboard;
+
+import com.google.protobuf.ByteString;
+import meerkat.crypto.Digest;
+import meerkat.protobuf.BulletinBoardAPI.MessageID;
+import meerkat.protobuf.BulletinBoardAPI.BulletinBoardMessage;
+
+import java.util.Collection;
+
+/**
+ * Created by Arbel Deutsch Peled on 01-Mar-16.
+ * Implementation of Checksum via bitwise XOR of the bytes of message IDs
+ */
+public class SimpleChecksum implements Checksum{
+
+    private Digest digest;
+    private long checksum;
+
+    public SimpleChecksum() {
+        digest = null;
+        reset();
+    }
+
+    @Override
+    public void setDigest(Digest digest) {
+        this.digest = digest;
+    }
+
+    @Override
+    public void reset() {
+        checksum = 0;
+    }
+
+    @Override
+    public void update(MessageID messageID) {
+        ByteString messageIDByteString = messageID.getID();
+        update(messageIDByteString);
+    }
+
+    @Override
+    public void update(Collection<MessageID> messageIDs) {
+
+        for (MessageID messageID : messageIDs){
+            update(messageID);
+        }
+
+    }
+
+    @Override
+    public void update(ByteString messageID) {
+        for (int i = 0 ; i < messageID.size() ; i++){
+            checksum &= messageID.byteAt(i);
+        }
+    }
+
+    @Override
+    public void update(byte[] messageID) {
+        for (int i = 0 ; i < messageID.length ; i++){
+            checksum &= messageID[i];
+        }
+    }
+
+    private void checkDigest() throws IllegalStateException {
+
+        if (digest == null){
+            throw new IllegalStateException("Digest method not set. Use setDigest method before calling digestAndUpdate.");
+        }
+
+    }
+
+    @Override
+    public void digestAndUpdate(BulletinBoardMessage bulletinBoardMessage) throws IllegalStateException {
+
+        checkDigest();
+
+        digest.reset();
+        digest.update(bulletinBoardMessage);
+        update(digest.digest());
+
+    }
+
+    @Override
+    public void digestAndUpdate(Collection<BulletinBoardMessage> bulletinBoardMessages) throws IllegalStateException {
+
+        for (BulletinBoardMessage bulletinBoardMessage : bulletinBoardMessages){
+            digestAndUpdate(bulletinBoardMessage);
+        }
+
+    }
+
+    @Override
+    public long getChecksum() {
+        return checksum;
+    }
+
+    @Override
+    public long getChecksum(MessageID messageID) {
+        update(messageID);
+        return getChecksum();
+    }
+
+    @Override
+    public long getChecksum(ByteString messageID) {
+        update(messageID);
+        return getChecksum();
+    }
+
+    @Override
+    public long getChecksum(byte[] messageID) {
+        update(messageID);
+        return getChecksum();
+    }
+
+    @Override
+    public long getChecksum(Collection<MessageID> messageIDs) {
+        update(messageIDs);
+        return getChecksum();
+    }
+
+    @Override
+    public long digestAndGetChecksum(BulletinBoardMessage bulletinBoardMessage) throws IllegalStateException {
+        digestAndUpdate(bulletinBoardMessage);
+        return getChecksum();
+    }
+
+    @Override
+    public long digestAndGetChecksum(Collection<BulletinBoardMessage> bulletinBoardMessages) throws IllegalStateException {
+        digestAndUpdate(bulletinBoardMessages);
+        return getChecksum();
+    }
+}

meerkat-common/src/main/java/meerkat/comm/MessageInputStream.java

@@ -89,4 +89,10 @@ public class MessageInputStream<T extends Message> implements Iterable<T>{
 
     }
 
+    public void close() throws IOException {
+
+        in.close();
+
+    }
+
 }

meerkat-common/src/main/java/meerkat/comm/MessageOutputStream.java

@@ -21,4 +21,8 @@ public class MessageOutputStream<T extends Message> {
         message.writeDelimitedTo(out);
     }
 
+    public void close() throws IOException {
+        out.close();
+    }
+
 }