From aeb7c13436af18dde9455c82c40221ca0f6ba442 Mon Sep 17 00:00:00 2001 From: Arbel Deutsch Peled Date: Mon, 22 Feb 2016 08:04:01 +0200 Subject: [PATCH] Made read operations stream the results. Removed dependency on large Protobufs (BulletinBoardMessageList and BatchDataList). Partial implementation of Sync Query. Current version supports only H2 and MySQL (no SQLite support). --- .../bulletinboard/BulletinClientWorker.java | 2 +- .../bulletinboard/MultiServerWorker.java | 4 +- .../SimpleBulletinBoardClient.java | 2 +- .../SingleServerBulletinBoardClient.java | 56 ++-- .../bulletinboard/SingleServerWorker.java | 2 +- .../ThreadedBulletinBoardClient.java | 9 +- .../MultiServerGenericPostWorker.java | 2 - .../MultiServerGenericReadWorker.java | 4 +- .../MultiServerGetRedundancyWorker.java | 6 +- .../SingleServerGenericPostWorker.java | 4 +- .../SingleServerGetRedundancyWorker.java | 4 +- ...dedBulletinBoardClientIntegrationTest.java | 15 +- bulletin-board-server/build.gradle | 4 + .../sqlserver/BulletinBoardSQLServer.java | 265 ++++++++++++++---- .../sqlserver/H2QueryProvider.java | 18 +- .../sqlserver/MySQLQueryProvider.java | 27 +- .../mappers/BatchDataCallbackHandler.java | 32 +++ .../mappers/MessageCallbackHandler.java | 80 ++++++ .../sqlserver/mappers/MessageStubMapper.java | 31 ++ .../webapp/BulletinBoardWebApp.java | 94 ++++++- ...BulletinBoardSQLServerIntegrationTest.java | 38 ++- .../GenericBulletinBoardServerTest.java | 138 ++++++--- meerkat-common/build.gradle | 1 + .../bulletinboard/BulletinBoardConstants.java | 1 + .../bulletinboard/BulletinBoardServer.java | 19 +- .../meerkat/bulletinboard/CompleteBatch.java | 28 ++ .../java/meerkat/comm/MessageInputStream.java | 64 +++++ .../meerkat/comm/MessageOutputStream.java | 24 ++ .../src/main/java/meerkat/comm/Timestamp.java | 7 - .../util/BulletinBoardMessageGenerator.java | 98 +++++++ .../java/meerkat/util/BulletinBoardUtils.java | 42 +++ .../meerkat/util/TimeStampComparator.java | 30 ++ .../main/proto/meerkat/BulletinBoardAPI.proto | 51 +++- .../java/meerkat/comm/MessageStreamTest.java | 98 +++++++ 34 files changed, 1106 insertions(+), 194 deletions(-) create mode 100644 bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/BatchDataCallbackHandler.java create mode 100644 bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageCallbackHandler.java create mode 100644 bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java create mode 100644 meerkat-common/src/main/java/meerkat/comm/MessageInputStream.java create mode 100644 meerkat-common/src/main/java/meerkat/comm/MessageOutputStream.java delete mode 100644 meerkat-common/src/main/java/meerkat/comm/Timestamp.java create mode 100644 meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageGenerator.java create mode 100644 meerkat-common/src/main/java/meerkat/util/TimeStampComparator.java create mode 100644 meerkat-common/src/test/java/meerkat/comm/MessageStreamTest.java diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/BulletinClientWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/BulletinClientWorker.java index dba596b..1a4b62f 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/BulletinClientWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/BulletinClientWorker.java @@ -8,7 +8,7 @@ package meerkat.bulletinboard; */ public abstract class BulletinClientWorker { - protected IN payload; // Payload of the job + protected final IN payload; // Payload of the job private int maxRetry; // Number of retries for this job; set to -1 for infinite retries diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java index 727a922..7347f47 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java @@ -18,7 +18,7 @@ import java.util.concurrent.atomic.AtomicInteger; */ public abstract class MultiServerWorker extends BulletinClientWorker implements Runnable, FutureCallback{ - private List clients; + private final List clients; protected AtomicInteger minServers; // The minimal number of servers the job must be successful on for the job to be completed @@ -26,7 +26,7 @@ public abstract class MultiServerWorker extends BulletinClientWorker futureCallback; + private final FutureCallback futureCallback; /** * Constructor diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java index 8cac04d..633e495 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java @@ -125,7 +125,7 @@ public class SimpleBulletinBoardClient implements BulletinBoardClient{ * If at the operation is successful for some DB: return the results and stop iterating * If no operation is successful: return null (NOT blank list) * @param filterList return only messages that match the filters (null means no filtering). - * @return + * @return the list of Bulletin Board messages that are returned from a server */ @Override public List readMessages(MessageFilterList filterList) { diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java index e6070a5..ea0e93d 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java @@ -6,10 +6,12 @@ import com.google.common.util.concurrent.ListeningScheduledExecutorService; import com.google.common.util.concurrent.MoreExecutors; import com.google.protobuf.ByteString; import meerkat.bulletinboard.workers.singleserver.*; +import meerkat.comm.CommunicationException; import meerkat.protobuf.BulletinBoardAPI.*; import meerkat.protobuf.Voting.BulletinBoardClientParams; import meerkat.util.BulletinBoardUtils; +import javax.ws.rs.NotFoundException; import java.util.Arrays; import java.util.Iterator; import java.util.LinkedList; @@ -31,15 +33,15 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i private final int MAX_RETRIES = 11; - protected ListeningScheduledExecutorService executorService; + private ListeningScheduledExecutorService executorService; protected BatchDigest batchDigest; private long lastServerErrorTime; - protected final long failDelayInMilliseconds; + private final long failDelayInMilliseconds; - protected final long subscriptionIntervalInMilliseconds; + private final long subscriptionIntervalInMilliseconds; /** * Notify the client that a job has failed @@ -86,8 +88,8 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i */ class RetryCallback implements FutureCallback { - private SingleServerWorker worker; - private FutureCallback futureCallback; + private final SingleServerWorker worker; + private final FutureCallback futureCallback; public RetryCallback(SingleServerWorker worker, FutureCallback futureCallback) { this.worker = worker; @@ -128,7 +130,8 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i */ class PostBatchDataListCallback implements FutureCallback { - private FutureCallback callback; + private final FutureCallback callback; + private AtomicInteger batchDataRemaining; private AtomicBoolean aggregatedResult; @@ -168,7 +171,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i */ class CompleteBatchReadCallback { - private FutureCallback callback; + private final FutureCallback callback; private List batchDataList; private BulletinBoardMessage batchMessage; @@ -176,7 +179,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i private AtomicInteger remainingQueries; private AtomicBoolean failed; - public CompleteBatchReadCallback(FutureCallback callback) { + public CompleteBatchReadCallback(FutureCallback callback) { this.callback = callback; @@ -193,11 +196,16 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i if (remainingQueries.decrementAndGet() == 0){ + String batchIdStr = BulletinBoardUtils.findTagWithPrefix(batchMessage, BulletinBoardConstants.BATCH_ID_TAG_PREFIX); + + 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( - BulletinBoardUtils.findTagWithPrefix(batchMessage, BulletinBoardConstants.BATCH_ID_TAG_PREFIX))) + .setBatchId(Integer.parseInt(batchIdStr)) .addAllTag(BulletinBoardUtils.removePrefixTags(batchMessage, Arrays.asList(prefixes))) .build(); callback.onSuccess(new CompleteBatch(beginBatchMessage, batchDataList, batchMessage.getSig(0))); @@ -267,7 +275,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i class SubscriptionCallback implements FutureCallback> { private SingleServerReadMessagesWorker worker; - private MessageHandler messageHandler; + private final MessageHandler messageHandler; private MessageFilterList.Builder filterBuilder; @@ -339,7 +347,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i // Remove all but first DB address String dbAddress = meerkatDBs.get(0); - meerkatDBs = new LinkedList(); + meerkatDBs = new LinkedList<>(); meerkatDBs.add(dbAddress); } @@ -351,7 +359,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i SingleServerPostMessageWorker worker = new SingleServerPostMessageWorker(meerkatDBs.get(0), msg, MAX_RETRIES); // Submit worker and create callback - scheduleWorker(worker, new RetryCallback(worker, callback)); + scheduleWorker(worker, new RetryCallback<>(worker, callback)); // Calculate the correct message ID and return it batchDigest.reset(); @@ -362,8 +370,8 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i private class PostBatchDataCallback implements FutureCallback { - private CompleteBatch completeBatch; - FutureCallback callback; + private final CompleteBatch completeBatch; + private final FutureCallback callback; public PostBatchDataCallback(CompleteBatch completeBatch, FutureCallback callback) { this.completeBatch = completeBatch; @@ -391,8 +399,8 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i private class BeginBatchCallback implements FutureCallback { - private CompleteBatch completeBatch; - FutureCallback callback; + private final CompleteBatch completeBatch; + private final FutureCallback callback; public BeginBatchCallback(CompleteBatch completeBatch, FutureCallback callback) { this.completeBatch = completeBatch; @@ -438,7 +446,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i new SingleServerBeginBatchWorker(meerkatDBs.get(0), beginBatchMessage, MAX_RETRIES); // Submit worker and create callback - scheduleWorker(worker, new RetryCallback(worker, callback)); + scheduleWorker(worker, new RetryCallback<>(worker, callback)); } @@ -464,7 +472,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i new SingleServerPostBatchWorker(meerkatDBs.get(0), builder.build(), MAX_RETRIES); // Create worker with redundancy 1 and MAX_RETRIES retries - scheduleWorker(worker, new RetryCallback(worker, listCallback)); + scheduleWorker(worker, new RetryCallback<>(worker, listCallback)); // Increment position in batch startPosition++; @@ -502,7 +510,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i new SingleServerCloseBatchWorker(meerkatDBs.get(0), closeBatchMessage, MAX_RETRIES); // Submit worker and create callback - scheduleWorker(worker, new RetryCallback(worker, callback)); + scheduleWorker(worker, new RetryCallback<>(worker, callback)); } @@ -513,7 +521,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i SingleServerGetRedundancyWorker worker = new SingleServerGetRedundancyWorker(meerkatDBs.get(0), id, 1); // Submit job and create callback - scheduleWorker(worker, new RetryCallback(worker, callback)); + scheduleWorker(worker, new RetryCallback<>(worker, callback)); } @@ -524,7 +532,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i SingleServerReadMessagesWorker worker = new SingleServerReadMessagesWorker(meerkatDBs.get(0), filterList, 1); // Submit job and create callback - scheduleWorker(worker, new RetryCallback(worker, callback)); + scheduleWorker(worker, new RetryCallback<>(worker, callback)); } @@ -557,8 +565,8 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i CompleteBatchReadCallback completeBatchReadCallback = new CompleteBatchReadCallback(callback); // Submit jobs with wrapped callbacks - scheduleWorker(messageWorker, new RetryCallback(messageWorker, completeBatchReadCallback.asBulletinBoardMessageListFutureCallback())); - scheduleWorker(batchWorker, new RetryCallback(batchWorker, completeBatchReadCallback.asBatchDataListFutureCallback())); + scheduleWorker(messageWorker, new RetryCallback<>(messageWorker, completeBatchReadCallback.asBulletinBoardMessageListFutureCallback())); + scheduleWorker(batchWorker, new RetryCallback<>(batchWorker, completeBatchReadCallback.asBatchDataListFutureCallback())); } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerWorker.java index 82ca886..ecebc12 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerWorker.java @@ -25,7 +25,7 @@ public abstract class SingleServerWorker extends BulletinClientWorker clients; + private List clients; - BatchDigest batchDigest; + private BatchDigest batchDigest; private final static int POST_MESSAGE_RETRY_NUM = 3; private final static int READ_MESSAGES_RETRY_NUM = 1; @@ -59,7 +59,7 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple executorService = Executors.newFixedThreadPool(JOBS_THREAD_NUM); - clients = new ArrayList(clientParams.getBulletinBoardAddressCount()); + clients = new ArrayList<>(clientParams.getBulletinBoardAddressCount()); for (String address : clientParams.getBulletinBoardAddressList()){ SingleServerBulletinBoardClient client = @@ -80,7 +80,6 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple * Retry failed DBs * @param msg is the message, * @return the message ID for later retrieval - * @throws CommunicationException */ @Override public MessageID postMessage(BulletinBoardMessage msg, FutureCallback callback){ diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java index 4ff96b1..8172e14 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java @@ -31,8 +31,6 @@ public abstract class MultiServerGenericPostWorker extends MultiServerWorker< * It accesses the servers one by one and tries to post the payload to each in turn * The method will only iterate once through the server list * Successful post to a server results in removing the server from the list - * @return The original job, but with a modified server list - * @throws CommunicationException */ public void run() { diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java index 68fc020..88b4ac1 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java @@ -14,7 +14,7 @@ import java.util.List; */ public abstract class MultiServerGenericReadWorker extends MultiServerWorker{ - protected Iterator clientIterator; + private final Iterator clientIterator; public MultiServerGenericReadWorker(List clients, int minServers, IN payload, int maxRetry, @@ -32,8 +32,6 @@ public abstract class MultiServerGenericReadWorker extends MultiServerW * This method carries out the actual communication with the servers via HTTP Post * It accesses the servers in a random order until one answers it * Successful retrieval from any server terminates the method and returns the received values; The list is not changed - * @return The original job and the list of messages found in the first server that answered the query - * @throws CommunicationException */ public void run(){ diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java index 517dbdf..748916b 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java @@ -33,8 +33,6 @@ public class MultiServerGetRedundancyWorker extends MultiServerWorker= getClientNumber()){ - succeed(new Float(((float) serversContainingMessage.get()) / ((float) getClientNumber()) )); + succeed(((float) serversContainingMessage.get()) / ((float) getClientNumber())); } } @Override public void onFailure(Throwable t) { - onSuccess(new Float(0.0)); + onSuccess(0.0f); } } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerGenericPostWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerGenericPostWorker.java index c56af05..621a828 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerGenericPostWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerGenericPostWorker.java @@ -19,7 +19,7 @@ import static meerkat.bulletinboard.BulletinBoardConstants.BULLETIN_BOARD_SERVER */ public class SingleServerGenericPostWorker extends SingleServerWorker { - private String subPath; + private final String subPath; public SingleServerGenericPostWorker(String serverAddress, String subPath, T payload, int maxRetry) { super(serverAddress, payload, maxRetry); @@ -37,7 +37,7 @@ public class SingleServerGenericPostWorker extends SingleServerWorker 0){ // Message exists in the server - return new Float(1.0); + return 1.0f; } else { // Message does not exist in the server - return new Float(0.0); + return 0.0f; } } catch (ProcessingException | IllegalStateException e) { diff --git a/bulletin-board-client/src/test/java/ThreadedBulletinBoardClientIntegrationTest.java b/bulletin-board-client/src/test/java/ThreadedBulletinBoardClientIntegrationTest.java index c266086..dfccebc 100644 --- a/bulletin-board-client/src/test/java/ThreadedBulletinBoardClientIntegrationTest.java +++ b/bulletin-board-client/src/test/java/ThreadedBulletinBoardClientIntegrationTest.java @@ -43,8 +43,8 @@ public class ThreadedBulletinBoardClientIntegrationTest { private static String KEYFILE_PASSWORD1 = "secret"; private static String KEYFILE_PASSWORD3 = "shh"; - public static String CERT1_PEM_EXAMPLE = "/certs/enduser-certs/user1.crt"; - public static String CERT3_PEM_EXAMPLE = "/certs/enduser-certs/user3.crt"; + private static String CERT1_PEM_EXAMPLE = "/certs/enduser-certs/user1.crt"; + private static String CERT3_PEM_EXAMPLE = "/certs/enduser-certs/user3.crt"; // Server data @@ -81,7 +81,7 @@ public class ThreadedBulletinBoardClientIntegrationTest { InputStream keyStream = getClass().getResourceAsStream(KEYFILE_EXAMPLE); char[] password = KEYFILE_PASSWORD1.toCharArray(); - KeyStore.Builder keyStoreBuilder = null; + KeyStore.Builder keyStoreBuilder; try { keyStoreBuilder = signers[0].getPKCS12KeyStoreBuilder(keyStream, password); @@ -304,11 +304,11 @@ public class ThreadedBulletinBoardClientIntegrationTest { random = new Random(0); // We use insecure randomness in tests for repeatability - List testDB = new LinkedList(); + List testDB = new LinkedList<>(); testDB.add(BASE_URL); bulletinBoardClient.init(BulletinBoardClientParams.newBuilder() - .addBulletinBoardAddress("http://localhost:8081") + .addAllBulletinBoardAddress(testDB) .setMinRedundancy((float) 1.0) .build()); @@ -344,7 +344,6 @@ public class ThreadedBulletinBoardClientIntegrationTest { byte[] b1 = {(byte) 1, (byte) 2, (byte) 3, (byte) 4}; byte[] b2 = {(byte) 11, (byte) 12, (byte) 13, (byte) 14}; byte[] b3 = {(byte) 21, (byte) 22, (byte) 23, (byte) 24}; - byte[] b4 = {(byte) 4, (byte) 5, (byte) 100, (byte) -50, (byte) 0}; BulletinBoardMessage msg; @@ -353,8 +352,6 @@ public class ThreadedBulletinBoardClientIntegrationTest { MessageID messageID; - Comparator msgComparator = new BulletinBoardMessageComparator(); - msg = BulletinBoardMessage.newBuilder() .setMsg(UnsignedBulletinBoardMessage.newBuilder() .addTag("Signature") @@ -398,7 +395,7 @@ public class ThreadedBulletinBoardClientIntegrationTest { ) .build(); - msgList = new LinkedList(); + msgList = new LinkedList<>(); msgList.add(msg); readCallback = new ReadCallback(msgList); diff --git a/bulletin-board-server/build.gradle b/bulletin-board-server/build.gradle index 59045a2..8d824e7 100644 --- a/bulletin-board-server/build.gradle +++ b/bulletin-board-server/build.gradle @@ -101,6 +101,10 @@ task dbTest(type: Test) { outputs.upToDateWhen { false } } +task manualIntegration(type: Test) { + include '**/*IntegrationTest*' +} + task integrationTest(type: Test) { include '**/*IntegrationTest*' // debug = true diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java index a6313b6..b4e2949 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java @@ -1,13 +1,9 @@ package meerkat.bulletinboard.sqlserver; -import java.security.InvalidKeyException; -import java.security.SignatureException; -import java.security.cert.CertificateException; import java.sql.*; import java.util.*; -import com.google.protobuf.ByteString; -import com.google.protobuf.ProtocolStringList; +import com.google.protobuf.*; import meerkat.bulletinboard.*; import meerkat.bulletinboard.sqlserver.mappers.*; @@ -15,6 +11,7 @@ import static meerkat.bulletinboard.BulletinBoardConstants.*; import meerkat.comm.CommunicationException; +import meerkat.comm.MessageOutputStream; import meerkat.crypto.concrete.ECDSASignature; import meerkat.crypto.concrete.SHA256Digest; @@ -27,6 +24,8 @@ import static meerkat.bulletinboard.sqlserver.BulletinBoardSQLServer.SQLQueryPro import javax.sql.DataSource; +import meerkat.util.BulletinBoardUtils; +import meerkat.util.TimestampComparator; import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.jdbc.support.GeneratedKeyHolder; @@ -62,8 +61,8 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ ), INSERT_MSG( - new String[] {"MsgId","Msg"}, - new int[] {Types.BLOB, Types.BLOB} + new String[] {"MsgId","TimeStamp","Msg"}, + new int[] {Types.BLOB, Types.TIMESTAMP, Types.BLOB} ), INSERT_NEW_TAG( @@ -91,6 +90,21 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ new int[] {} ), + COUNT_MESSAGES( + new String[] {}, + new int[] {} + ), + + GET_MESSAGE_STUBS( + new String[] {}, + new int[] {} + ), + + GET_LAST_MESSAGE_ENTRY( + new String[] {}, + new int[] {} + ), + GET_BATCH_MESSAGE_ENTRY( new String[] {"SignerId", "BatchId"}, new int[] {Types.BLOB, Types.INTEGER} @@ -161,7 +175,8 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ MSG_ID("MsgId", Types.BLOB), SIGNER_ID("SignerId", Types.BLOB), TAG("Tag", Types.VARCHAR), - LIMIT("Limit", Types.INTEGER); + LIMIT("Limit", Types.INTEGER), + TIMESTAMP("TimeStamp", Types.TIMESTAMP); private FilterTypeParam(String paramName, int paramType) { this.paramName = paramName; @@ -191,6 +206,10 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ case MAX_MESSAGES: return LIMIT; + case BEFORE_TIME: // Go through + case AFTER_TIME: + return TIMESTAMP; + default: return null; } @@ -269,6 +288,10 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ case MAX_MESSAGES: return messageFilter.getMaxMessages(); + case BEFORE_TIME: // Go through + case AFTER_TIME: + return BulletinBoardUtils.toSQLTimestamp(messageFilter.getTimestamp()); + default: // Unsupported filter type return null; } @@ -316,8 +339,6 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ */ private void createSchema() throws SQLException { - final int TIMEOUT = 20; - for (String command : sqlQueryProvider.getSchemaCreationCommands()) { jdbcTemplate.update(command,(Map) null); } @@ -423,7 +444,6 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ // Add message to table if needed and store entry number of message. - sql = sqlQueryProvider.getSQLString(QueryType.FIND_MSG_ID); Map namedParameters = new HashMap(); namedParameters.put(QueryType.FIND_MSG_ID.getParamName(0),msgID); @@ -437,7 +457,9 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ } else{ sql = sqlQueryProvider.getSQLString(QueryType.INSERT_MSG); - namedParameters.put(QueryType.INSERT_MSG.getParamName(1), msg.getMsg().toByteArray()); + + namedParameters.put(QueryType.INSERT_MSG.getParamName(1), BulletinBoardUtils.toSQLTimestamp(msg.getMsg().getTimestamp())); + namedParameters.put(QueryType.INSERT_MSG.getParamName(2), msg.getMsg().toByteArray()); KeyHolder keyHolder = new GeneratedKeyHolder(); jdbcTemplate.update(sql,new MapSqlParameterSource(namedParameters),keyHolder); @@ -504,37 +526,36 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ return postMessage(msg, true); // Perform a post and check the signature for authenticity } - @Override - public BulletinBoardMessageList readMessages(MessageFilterList filterList) throws CommunicationException { - BulletinBoardMessageList.Builder resultListBuilder = BulletinBoardMessageList.newBuilder(); + /** + * This is a container class for and SQL string builder and a MapSqlParameterSource to be used with it + */ + class SQLAndParameters { - // SQL length is roughly 50 characters per filter + 50 for the query itself - StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1)); + public StringBuilder sql; + public MapSqlParameterSource parameters; - MapSqlParameterSource namedParameters; - int paramNum; + public SQLAndParameters(int numOfFilters) { + sql = new StringBuilder(50 * numOfFilters); + parameters = new MapSqlParameterSource(); + } - MessageMapper messageMapper = new MessageMapper(); - SignatureMapper signatureMapper = new SignatureMapper(); + } + + SQLAndParameters getSQLFromFilters(MessageFilterList filterList) { + + SQLAndParameters result = new SQLAndParameters(filterList.getFilterCount()); List filters = new ArrayList(filterList.getFilterList()); - boolean isFirstFilter = true; - Collections.sort(filters, new FilterTypeComparator()); - // Check if Tag/Signature tables are required for filtering purposes - - sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.GET_MESSAGES)); - // Add conditions - - namedParameters = new MapSqlParameterSource(); + boolean isFirstFilter = true; if (!filters.isEmpty()) { - sqlBuilder.append(" WHERE "); + result.sql.append(" WHERE "); - for (paramNum = 0 ; paramNum < filters.size() ; paramNum++) { + for (int paramNum = 0 ; paramNum < filters.size() ; paramNum++) { MessageFilter filter = filters.get(paramNum); @@ -542,15 +563,15 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ if (isFirstFilter) { isFirstFilter = false; } else { - sqlBuilder.append(" AND "); + result.sql.append(" AND "); } } - sqlBuilder.append(sqlQueryProvider.getCondition(filter.getType(), paramNum)); + result.sql.append(sqlQueryProvider.getCondition(filter.getType(), paramNum)); FilterTypeParam filterTypeParam = FilterTypeParam.getFilterTypeParamName(filter.getType()); - namedParameters.addValue( + result.parameters.addValue( filterTypeParam.getParamName() + Integer.toString(paramNum), getParam(filter), filterTypeParam.getParamType(), @@ -560,36 +581,56 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ } + return result; + + } + + + /** + * Used to retrieve just basic information about messages to allow calculation of checksum + * @param filterList is a filter list that defines which messages the client is interested in + * @return a list of Bulletin Board Messages that contain just the entry number, timestamp and message ID for each message + * The message ID is returned inside the message data field + */ + protected List readMessageStubs(MessageFilterList filterList) { + + StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1)); + + sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.GET_MESSAGE_STUBS)); + + // Get Conditions + + SQLAndParameters sqlAndParameters = getSQLFromFilters(filterList); + + sqlBuilder.append(sqlAndParameters.sql); + // Run query - List msgBuilders = - jdbcTemplate.query(sqlBuilder.toString(), namedParameters, messageMapper); + return jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new MessageStubMapper()); - // Compile list of messages + } - for (BulletinBoardMessage.Builder msgBuilder : msgBuilders) { - // Retrieve signatures + @Override + public void readMessages(MessageFilterList filterList, MessageOutputStream out) throws CommunicationException { - namedParameters = new MapSqlParameterSource(); - namedParameters.addValue(QueryType.GET_SIGNATURES.getParamName(0), msgBuilder.getEntryNum()); + BulletinBoardMessageList.Builder resultListBuilder = BulletinBoardMessageList.newBuilder(); - List signatures = jdbcTemplate.query( - sqlQueryProvider.getSQLString(QueryType.GET_SIGNATURES), - namedParameters, - signatureMapper); + // SQL length is roughly 50 characters per filter + 50 for the query itself + StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1)); - // Append signatures - msgBuilder.addAllSig(signatures); + // Check if Tag/Signature tables are required for filtering purposes - // Finalize message and add to message list. + sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.GET_MESSAGES)); - resultListBuilder.addMessage(msgBuilder.build()); + // Get conditions - } + SQLAndParameters sqlAndParameters = getSQLFromFilters(filterList); + sqlBuilder.append(sqlAndParameters.sql); - //Combine results and return. - return resultListBuilder.build(); + // Run query and stream the output using a MessageCallbackHandler + + jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new MessageCallbackHandler(jdbcTemplate, sqlQueryProvider, out)); } @@ -625,9 +666,23 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ .build()) .build(); - BulletinBoardMessageList messageList = readMessages(filterList); + // SQL length is roughly 50 characters per filter + 50 for the query itself + StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1)); - return (messageList.getMessageList().size() > 0); + // 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 count = jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new LongMapper()); + + return (count.size() > 0) && (count.get(0) > 0); } @@ -767,6 +822,7 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ .addTag(BATCH_TAG) .addTag(batchIdToTag(batchId)) .setData(message.getSig().getSignerId()) + .setTimestamp(message.getTimestamp()) .build()) .build(); @@ -788,7 +844,7 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ } @Override - public BatchDataList readBatch(BatchSpecificationMessage message) throws CommunicationException, IllegalArgumentException{ + public void readBatch(BatchSpecificationMessage message, MessageOutputStream out) throws CommunicationException, IllegalArgumentException{ // Check that batch is closed if (!isBatchClosed(message.getSignerId(), message.getBatchId())) { @@ -802,9 +858,102 @@ public class BulletinBoardSQLServer implements BulletinBoardServer{ namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA.getParamName(1),message.getBatchId()); namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA.getParamName(2),message.getStartPosition()); - return BatchDataList.newBuilder() - .addAllData(jdbcTemplate.query(sql, namedParameters, new BatchDataMapper())) - .build(); + jdbcTemplate.query(sql, namedParameters, new BatchDataCallbackHandler(out)); + + } + + /** + * Finds the entry number of the last entry in the database + * @return the entry number, or -1 if no entries are found + */ + protected long getLastMessageEntry() { + + String sql = sqlQueryProvider.getSQLString(QueryType.GET_LAST_MESSAGE_ENTRY); + + List resultList = jdbcTemplate.query(sql, new LongMapper()); + + if (resultList.size() <= 0){ + return -1; + } + + return resultList.get(0); + + } + + /** + * Searches for the latest time of sync of the DB relative to a given query and returns the metadata needed to complete the sync + * The checksum up to (and including) each given timestamp is calculated using bitwise XOR on 8-byte sized blocks of the message IDs + * @param syncQuery contains a succinct representation of states to compare to + * @return the current last entry num and latest time of sync if there is one; -1 as last entry and empty timestamp otherwise + * @throws CommunicationException + */ + @Override + public SyncQueryResponse querySync(SyncQuery syncQuery) throws CommunicationException { + + if (syncQuery == null){ + return SyncQueryResponse.newBuilder() + .setLastEntryNum(-1) + .setLastTimeOfSync(com.google.protobuf.Timestamp.getDefaultInstance()) + .build(); + } + + com.google.protobuf.Timestamp lastTimeOfSync = null; + + TimestampComparator timestampComparator = new TimestampComparator(); + + long lastEntryNum = getLastMessageEntry(); + + long checksum = 0; + + Iterator queryIterator = syncQuery.getQueryList().iterator(); + + SingleSyncQuery currentQuery = queryIterator.next(); + + List messageStubs = readMessageStubs(syncQuery.getFilterList()); + + for (BulletinBoardMessage message : messageStubs){ + + // Check for end of current query + if (timestampComparator.compare(message.getMsg().getTimestamp(), currentQuery.getTimeOfSync()) > 0){ + + if (checksum == currentQuery.getChecksum()){ + lastTimeOfSync = currentQuery.getTimeOfSync(); + } else { + break; + } + + if (queryIterator.hasNext()){ + currentQuery = queryIterator.next(); + } else{ + break; + } + + } + + // Advance checksum + + ByteString messageID = message.getMsg().getData(); + + checksum &= messageID.byteAt(0) & messageID.byteAt(1) & messageID.byteAt(2) & messageID.byteAt(3); + + } + + if (checksum == currentQuery.getChecksum()){ + lastTimeOfSync = currentQuery.getTimeOfSync(); + } + + if (lastTimeOfSync == null){ + return SyncQueryResponse.newBuilder() + .setLastEntryNum(-1) + .setLastTimeOfSync(com.google.protobuf.Timestamp.getDefaultInstance()) + .build(); + } else{ + return SyncQueryResponse.newBuilder() + .setLastEntryNum(lastEntryNum) + .setLastTimeOfSync(lastTimeOfSync) + .build(); + } + } @Override diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java index 659d9c3..c390048 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java @@ -51,6 +51,12 @@ public class H2QueryProvider implements BulletinBoardSQLServer.SQLQueryProvider case GET_MESSAGES: return "SELECT MsgTable.EntryNum, MsgTable.Msg FROM MsgTable"; + case COUNT_MESSAGES: + return "SELECT COUNT(MsgTable.EntryNum) FROM MsgTable"; + + case GET_MESSAGE_STUBS: + return "SELECT MsgTable.EntryNum, MsgTable.MsgId, MsgTable.ExactTime FROM MsgTable"; + case GET_SIGNATURES: return "SELECT Signature FROM SignatureTable WHERE EntryNum = :EntryNum"; @@ -61,6 +67,9 @@ public class H2QueryProvider implements BulletinBoardSQLServer.SQLQueryProvider return "INSERT INTO TagTable(Tag) SELECT DISTINCT :Tag AS NewTag FROM UtilityTable WHERE" + " NOT EXISTS (SELECT 1 FROM TagTable AS SubTable WHERE SubTable.Tag = :Tag)"; + case GET_LAST_MESSAGE_ENTRY: + return "SELECT MAX(MsgTable.EntryNum) FROM MsgTable"; + case GET_BATCH_MESSAGE_ENTRY: return MessageFormat.format( "SELECT MsgTable.EntryNum, MsgTable.Msg FROM MsgTable" @@ -147,6 +156,13 @@ public class H2QueryProvider implements BulletinBoardSQLServer.SQLQueryProvider return "EXISTS (SELECT 1 FROM TagTable" + " INNER JOIN MsgTagTable ON TagTable.TagId = MsgTagTable.TagId" + " WHERE TagTable.Tag = :Tag" + serialString + " AND MsgTagTable.EntryNum = MsgTable.EntryNum)"; + + case BEFORE_TIME: + return "MsgTable.ExactTime <= :TimeStamp"; + + case AFTER_TIME: + return "MsgTable.ExactTime >= :TimeStamp"; + default: throw new IllegalArgumentException("Cannot serve a filter of type " + filterType); } @@ -190,7 +206,7 @@ public class H2QueryProvider implements BulletinBoardSQLServer.SQLQueryProvider public List getSchemaCreationCommands() { List list = new LinkedList(); - list.add("CREATE TABLE IF NOT EXISTS MsgTable (EntryNum INT NOT NULL AUTO_INCREMENT PRIMARY KEY, MsgId TINYBLOB UNIQUE, Msg BLOB)"); + list.add("CREATE TABLE IF NOT EXISTS MsgTable (EntryNum INT NOT NULL AUTO_INCREMENT PRIMARY KEY, MsgId TINYBLOB UNIQUE, ExactTime TIMESTAMP, Msg BLOB)"); list.add("CREATE TABLE IF NOT EXISTS TagTable (TagId INT NOT NULL AUTO_INCREMENT PRIMARY KEY, Tag VARCHAR(50) UNIQUE)"); diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java index e3bdef0..f99114e 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java @@ -62,6 +62,12 @@ public class MySQLQueryProvider implements SQLQueryProvider { case GET_MESSAGES: return "SELECT MsgTable.EntryNum, MsgTable.Msg FROM MsgTable"; + case COUNT_MESSAGES: + return "SELECT COUNT(MsgTable.EntryNum) FROM MsgTable"; + + case GET_MESSAGE_STUBS: + return "SELECT MsgTable.EntryNum, MsgTable.MsgId, MsgTable.ExactTime FROM MsgTable"; + case GET_SIGNATURES: return MessageFormat.format( "SELECT Signature FROM SignatureTable WHERE EntryNum = :{0}", @@ -69,15 +75,19 @@ public class MySQLQueryProvider implements SQLQueryProvider { case INSERT_MSG: return MessageFormat.format( - "INSERT INTO MsgTable (MsgId, Msg) VALUES(:{0}, :{1})", + "INSERT INTO MsgTable (MsgId, ExactTime, Msg) VALUES(:{0}, :{1}, :{2})", QueryType.INSERT_MSG.getParamName(0), - QueryType.INSERT_MSG.getParamName(1)); + QueryType.INSERT_MSG.getParamName(1), + QueryType.INSERT_MSG.getParamName(2)); case INSERT_NEW_TAG: return MessageFormat.format( "INSERT IGNORE INTO TagTable(Tag) VALUES (:{0})", QueryType.INSERT_NEW_TAG.getParamName(0)); + case GET_LAST_MESSAGE_ENTRY: + return "SELECT MAX(MsgTable.EntryNum) FROM MsgTable"; + case GET_BATCH_MESSAGE_ENTRY: return MessageFormat.format( "SELECT MsgTable.EntryNum, MsgTable.Msg FROM MsgTable" @@ -164,6 +174,13 @@ public class MySQLQueryProvider implements SQLQueryProvider { return "EXISTS (SELECT 1 FROM TagTable" + " INNER JOIN MsgTagTable ON TagTable.TagId = MsgTagTable.TagId" + " WHERE TagTable.Tag = :Tag" + serialString + " AND MsgTagTable.EntryNum = MsgTable.EntryNum)"; + + case BEFORE_TIME: + return "MsgTable.ExactTime <= :TimeStamp"; + + case AFTER_TIME: + return "MsgTable.ExactTime >= :TimeStamp"; + default: throw new IllegalArgumentException("Cannot serve a filter of type " + filterType); } @@ -187,6 +204,10 @@ public class MySQLQueryProvider implements SQLQueryProvider { case TAG: return "VARCHAR"; + case AFTER_TIME: // Go through + case BEFORE_TIME: + return "TIMESTAMP"; + default: throw new IllegalArgumentException("Cannot serve a filter of type " + filterType); } @@ -212,7 +233,7 @@ public class MySQLQueryProvider implements SQLQueryProvider { List list = new LinkedList(); list.add("CREATE TABLE IF NOT EXISTS MsgTable (EntryNum INT NOT NULL AUTO_INCREMENT PRIMARY KEY," - + " MsgId TINYBLOB, Msg BLOB, UNIQUE(MsgId(50)))"); + + " MsgId TINYBLOB, ExactTime TIMESTAMP, Msg BLOB, UNIQUE(MsgId(50)))"); list.add("CREATE TABLE IF NOT EXISTS TagTable (TagId INT NOT NULL AUTO_INCREMENT PRIMARY KEY, Tag VARCHAR(50), UNIQUE(Tag))"); diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/BatchDataCallbackHandler.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/BatchDataCallbackHandler.java new file mode 100644 index 0000000..9ad0dc7 --- /dev/null +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/BatchDataCallbackHandler.java @@ -0,0 +1,32 @@ +package meerkat.bulletinboard.sqlserver.mappers; + +import com.google.protobuf.InvalidProtocolBufferException; +import meerkat.comm.MessageOutputStream; +import meerkat.protobuf.BulletinBoardAPI.BatchData; +import org.springframework.jdbc.core.RowCallbackHandler; +import org.springframework.jdbc.core.RowMapper; + +import java.io.IOException; +import java.sql.ResultSet; +import java.sql.SQLException; + +/** + * Created by Arbel Deutsch Peled on 19-Dec-15. + */ +public class BatchDataCallbackHandler implements RowCallbackHandler { + + private final MessageOutputStream out; + + public BatchDataCallbackHandler(MessageOutputStream out) { + this.out = out; + } + + @Override + public void processRow(ResultSet rs) throws SQLException { + try { + out.writeMessage(BatchData.parseFrom(rs.getBytes(1))); + } catch (IOException e) { + //TODO: Log + } + } +} diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageCallbackHandler.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageCallbackHandler.java new file mode 100644 index 0000000..bdba241 --- /dev/null +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageCallbackHandler.java @@ -0,0 +1,80 @@ +package meerkat.bulletinboard.sqlserver.mappers; + +import com.google.protobuf.InvalidProtocolBufferException; +import meerkat.bulletinboard.sqlserver.BulletinBoardSQLServer.*; +import meerkat.bulletinboard.sqlserver.BulletinBoardSQLServer.SQLQueryProvider.*; +import meerkat.comm.MessageOutputStream; +import meerkat.protobuf.BulletinBoardAPI.*; +import meerkat.protobuf.Crypto; +import org.springframework.jdbc.core.RowCallbackHandler; +import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; +import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; + +import java.io.IOException; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.List; + +/** + * Created by Arbel Deutsch Peled on 21-Feb-16. + */ +public class MessageCallbackHandler implements RowCallbackHandler { + + NamedParameterJdbcTemplate jdbcTemplate; + SQLQueryProvider sqlQueryProvider; + MessageOutputStream out; + + public MessageCallbackHandler(NamedParameterJdbcTemplate jdbcTemplate, SQLQueryProvider sqlQueryProvider, MessageOutputStream out) { + + this.jdbcTemplate = jdbcTemplate; + this.sqlQueryProvider = sqlQueryProvider; + this.out = out; + + } + + @Override + public void processRow(ResultSet rs) throws SQLException { + + BulletinBoardMessage.Builder result; + + try { + + result = BulletinBoardMessage.newBuilder() + .setEntryNum(rs.getLong(1)) + .setMsg(UnsignedBulletinBoardMessage.parseFrom(rs.getBytes(2))); + + + } catch (InvalidProtocolBufferException e) { + //TODO: log + return; + } + + // Retrieve signatures + + MapSqlParameterSource sqlParameterSource = new MapSqlParameterSource(); + sqlParameterSource.addValue(QueryType.GET_SIGNATURES.getParamName(0), result.getEntryNum()); + + List signatures = jdbcTemplate.query( + sqlQueryProvider.getSQLString(QueryType.GET_SIGNATURES), + sqlParameterSource, + new SignatureMapper()); + + // Append signatures + result.addAllSig(signatures); + + // Finalize message and add to message list. + + try { + + out.writeMessage(result.build()); + + } catch (IOException e) { + + //TODO: log + e.printStackTrace(); + + } + + } + +} diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java new file mode 100644 index 0000000..1f9c459 --- /dev/null +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java @@ -0,0 +1,31 @@ +package meerkat.bulletinboard.sqlserver.mappers; + +import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; +import meerkat.protobuf.BulletinBoardAPI.BulletinBoardMessage; +import meerkat.protobuf.BulletinBoardAPI.UnsignedBulletinBoardMessage; +import meerkat.util.BulletinBoardUtils; +import org.springframework.jdbc.core.RowMapper; + +import java.sql.ResultSet; +import java.sql.SQLException; + +/** + * Created by Arbel Deutsch Peled on 11-Dec-15. + */ +public class MessageStubMapper implements RowMapper { + + @Override + public BulletinBoardMessage mapRow(ResultSet rs, int rowNum) throws SQLException { + + return BulletinBoardMessage.newBuilder() + .setEntryNum(rs.getLong(1)) + .setMsg(UnsignedBulletinBoardMessage.newBuilder() + .setData(ByteString.copyFrom(rs.getBytes(2))) + .setTimestamp(BulletinBoardUtils.toTimestampProto(rs.getTimestamp(3))) + .build()) + .build(); + + } + +} diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java index 766af19..2e4782f 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java @@ -3,13 +3,10 @@ package meerkat.bulletinboard.webapp; import javax.servlet.ServletContext; import javax.servlet.ServletContextEvent; import javax.servlet.ServletContextListener; -import javax.ws.rs.Consumes; -import javax.ws.rs.GET; -import javax.ws.rs.POST; -import javax.ws.rs.Path; -import javax.ws.rs.Produces; +import javax.ws.rs.*; import javax.ws.rs.core.Context; import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.StreamingOutput; import meerkat.bulletinboard.BulletinBoardServer; import meerkat.bulletinboard.sqlserver.BulletinBoardSQLServer; @@ -17,12 +14,19 @@ import meerkat.bulletinboard.sqlserver.H2QueryProvider; 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.List; +/** + * An implementation of the BulletinBoardServer which functions as a WebApp + */ @Path(BULLETIN_BOARD_SERVER_PATH) public class BulletinBoardWebApp implements BulletinBoardServer, ServletContextListener{ @@ -88,15 +92,39 @@ public class BulletinBoardWebApp implements BulletinBoardServer, ServletContextL init(); return bulletinBoard.postMessage(msg); } - + + @Override + public void readMessages(MessageFilterList filterList, MessageOutputStream out) throws CommunicationException { + init(); + bulletinBoard.readMessages(filterList, out); + } + + @Path(READ_MESSAGES_PATH) @POST @Consumes(MEDIATYPE_PROTOBUF) - @Produces(MEDIATYPE_PROTOBUF) - @Override - public BulletinBoardMessageList readMessages(MessageFilterList filterList) throws CommunicationException { - init(); - return bulletinBoard.readMessages(filterList); + /** + * Wrapper for the readMessages method which streams the output into the response + */ + public StreamingOutput readMessages(final MessageFilterList filterList) { + + return new StreamingOutput() { + + @Override + public void write(OutputStream output) throws IOException, WebApplicationException { + MessageOutputStream out = new MessageOutputStream<>(output); + + try { + init(); + bulletinBoard.readMessages(filterList, out); + } catch (CommunicationException e) { + //TODO: Log + out.writeMessage(null); + } + } + + }; + } @Path(BEGIN_BATCH_PATH) @@ -144,15 +172,53 @@ public class BulletinBoardWebApp implements BulletinBoardServer, ServletContextL } } + + @Override + public void readBatch(BatchSpecificationMessage message, MessageOutputStream out) { + try { + init(); + bulletinBoard.readBatch(message, out); + } catch (CommunicationException | IllegalArgumentException e) { + System.err.println(e.getMessage()); + } + } + @Path(READ_BATCH_PATH) @POST @Consumes(MEDIATYPE_PROTOBUF) + /** + * Wrapper for the readBatch method which streams the output into the response + */ + public StreamingOutput readBatch(final BatchSpecificationMessage message) { + + return new StreamingOutput() { + + @Override + public void write(OutputStream output) throws IOException, WebApplicationException { + MessageOutputStream out = new MessageOutputStream<>(output); + + try { + init(); + bulletinBoard.readBatch(message, out); + } catch (CommunicationException e) { + //TODO: Log + out.writeMessage(null); + } + } + + }; + + } + + @Path(SYNC_QUERY_PATH) + @POST + @Consumes(MEDIATYPE_PROTOBUF) @Produces(MEDIATYPE_PROTOBUF) @Override - public BatchDataList readBatch(BatchSpecificationMessage message) { - try { + public SyncQueryResponse querySync(SyncQuery syncQuery) throws CommunicationException { + try{ init(); - return bulletinBoard.readBatch(message); + return bulletinBoard.querySync(syncQuery); } catch (CommunicationException | IllegalArgumentException e) { System.err.println(e.getMessage()); return null; diff --git a/bulletin-board-server/src/test/java/meerkat/bulletinboard/BulletinBoardSQLServerIntegrationTest.java b/bulletin-board-server/src/test/java/meerkat/bulletinboard/BulletinBoardSQLServerIntegrationTest.java index 4b8b586..744a086 100644 --- a/bulletin-board-server/src/test/java/meerkat/bulletinboard/BulletinBoardSQLServerIntegrationTest.java +++ b/bulletin-board-server/src/test/java/meerkat/bulletinboard/BulletinBoardSQLServerIntegrationTest.java @@ -4,6 +4,8 @@ package meerkat.bulletinboard; import com.google.protobuf.ByteString; import com.google.protobuf.TextFormat; +import com.google.protobuf.Timestamp; +import meerkat.comm.MessageInputStream; import meerkat.protobuf.Crypto.*; import meerkat.protobuf.BulletinBoardAPI.*; import static meerkat.bulletinboard.BulletinBoardConstants.*; @@ -18,7 +20,10 @@ import javax.ws.rs.client.Client; import javax.ws.rs.client.ClientBuilder; import javax.ws.rs.client.Entity; import javax.ws.rs.client.WebTarget; +import javax.ws.rs.core.MediaType; import javax.ws.rs.core.Response; +import java.io.InputStream; +import java.util.List; public class BulletinBoardSQLServerIntegrationTest { @@ -44,6 +49,16 @@ public class BulletinBoardSQLServerIntegrationTest { byte[] b3 = {(byte) 21, (byte) 22, (byte) 23, (byte) 24}; byte[] b4 = {(byte) 4, (byte) 5, (byte) 100, (byte) -50, (byte) 0}; + Timestamp t1 = Timestamp.newBuilder() + .setSeconds(8276482) + .setNanos(4314) + .build(); + + Timestamp t2 = Timestamp.newBuilder() + .setSeconds(987591) + .setNanos(1513) + .build(); + WebTarget webTarget; Response response; BoolMsg bool; @@ -51,7 +66,7 @@ public class BulletinBoardSQLServerIntegrationTest { BulletinBoardMessage msg; MessageFilterList filterList; - BulletinBoardMessageList msgList; + List msgList; // Test writing mechanism @@ -64,6 +79,7 @@ public class BulletinBoardSQLServerIntegrationTest { .addTag("Signature") .addTag("Trustee") .setData(ByteString.copyFrom(b1)) + .setTimestamp(t1) .build()) .addSig(Signature.newBuilder() .setType(SignatureType.DSA) @@ -87,6 +103,7 @@ public class BulletinBoardSQLServerIntegrationTest { .addTag("Vote") .addTag("Trustee") .setData(ByteString.copyFrom(b4)) + .setTimestamp(t2) .build()) .addSig(Signature.newBuilder() .setType(SignatureType.ECDSA) @@ -113,13 +130,20 @@ public class BulletinBoardSQLServerIntegrationTest { ) .build(); - response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(filterList, Constants.MEDIATYPE_PROTOBUF)); - System.err.println(response); - msgList = response.readEntity(BulletinBoardMessageList.class); - System.err.println("List size: " + msgList.getMessageCount()); + InputStream in = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(filterList, Constants.MEDIATYPE_PROTOBUF), InputStream.class); + + MessageInputStream inputStream = + MessageInputStream.MessageInputStreamFactory.createMessageInputStream(in, BulletinBoardMessage.class); + + msgList = inputStream.asList(); + System.err.println("List size: " + msgList.size()); System.err.println("This is the list:"); - System.err.println(TextFormat.printToString(msgList)); - assert msgList.getMessageCount() == 1; + + for (BulletinBoardMessage message : msgList) { + System.err.println(TextFormat.printToString(message)); + } + + assert msgList.size() == 1; } } diff --git a/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java b/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java index 86a32de..e33c739 100644 --- a/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java +++ b/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java @@ -1,9 +1,12 @@ package meerkat.bulletinboard; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; import java.lang.management.ManagementFactory; import java.lang.management.ThreadMXBean; +import java.lang.reflect.InvocationTargetException; import java.math.BigInteger; import java.security.InvalidKeyException; import java.security.KeyStore; @@ -16,17 +19,23 @@ import java.util.*; import com.google.protobuf.ByteString; +import com.google.protobuf.Timestamp; import meerkat.comm.CommunicationException; +import meerkat.comm.MessageInputStream; +import meerkat.comm.MessageOutputStream; +import meerkat.comm.MessageInputStream.MessageInputStreamFactory; import meerkat.crypto.concrete.ECDSASignature; import meerkat.protobuf.BulletinBoardAPI.*; +import meerkat.util.BulletinBoardUtils; import static org.junit.Assert.*; import static org.hamcrest.CoreMatchers.*; +import static org.hamcrest.MatcherAssert.assertThat; public class GenericBulletinBoardServerTest { protected BulletinBoardServer bulletinBoardServer; - private GenericBatchDigitalSignature signers[]; + private GenericBatchDigitalSignature[] signers; private ByteString[] signerIDs; private Random random; @@ -172,7 +181,8 @@ public class GenericBulletinBoardServerTest { for (i = 1; i <= MESSAGE_NUM; i++) { unsignedMsgBuilder = UnsignedBulletinBoardMessage.newBuilder() - .setData(ByteString.copyFrom(data[i - 1])); + .setData(ByteString.copyFrom(data[i - 1])) + .setTimestamp(BulletinBoardUtils.toTimestampProto()); // Add tags based on bit-representation of message number. @@ -232,28 +242,39 @@ public class GenericBulletinBoardServerTest { System.err.println("Starting to test tag and signature mechanism"); long start = threadBean.getCurrentThreadCpuTime(); - List messages; + List messages = new LinkedList<>(); // Check tag mechanism - + for (int i = 0 ; i < TAG_NUM ; i++){ // Retrieve messages having tag i try { - messages = bulletinBoardServer.readMessages( - MessageFilterList.newBuilder() - .addFilter(MessageFilter.newBuilder() - .setType(FilterType.TAG) - .setTag(tags[i]) - .build() - ) - .build() - ) - .getMessageList(); + MessageFilterList filterList = MessageFilterList.newBuilder() + .addFilter(MessageFilter.newBuilder() + .setType(FilterType.TAG) + .setTag(tags[i]) + .build() + ) + .build(); - } catch (CommunicationException e) { + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + + bulletinBoardServer.readMessages(filterList, new MessageOutputStream(outputStream)); + + MessageInputStream inputStream = + MessageInputStreamFactory.createMessageInputStream(new ByteArrayInputStream( + outputStream.toByteArray()), + BulletinBoardMessage.class); + + messages = inputStream.asList(); + + } catch (CommunicationException | IOException e) { + fail(e.getMessage()); + return; + } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { fail(e.getMessage()); return; } @@ -330,11 +351,26 @@ public class GenericBulletinBoardServerTest { ); try { - messages = bulletinBoardServer.readMessages(filterListBuilder.build()).getMessageList(); + + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + + bulletinBoardServer.readMessages(filterListBuilder.build(), new MessageOutputStream(outputStream)); + + MessageInputStream inputStream = + MessageInputStreamFactory.createMessageInputStream(new ByteArrayInputStream( + outputStream.toByteArray()), + BulletinBoardMessage.class); + + messages = inputStream.asList(); + } catch (CommunicationException e) { System.err.println("Failed retrieving multi-tag messages from DB: " + e.getMessage()); fail("Failed retrieving multi-tag messages from DB: " + e.getMessage()); return; + } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | IOException e) { + System.err.println("Falied to read from stream while retrieving multi-tag messages: " + e.getMessage()); + fail("Falied to read from stream while retrieving multi-tag messages: " + e.getMessage()); + return; } expectedMsgCount /= 2; @@ -361,11 +397,26 @@ public class GenericBulletinBoardServerTest { .build()); try { - messages = bulletinBoardServer.readMessages(filterListBuilder.build()).getMessageList(); + + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + + bulletinBoardServer.readMessages(filterListBuilder.build(), new MessageOutputStream(outputStream)); + + MessageInputStream inputStream = + MessageInputStreamFactory.createMessageInputStream(new ByteArrayInputStream( + outputStream.toByteArray()), + BulletinBoardMessage.class); + + messages = inputStream.asList(); + } catch (CommunicationException e) { System.err.println("Failed retrieving multi-signature message from DB: " + e.getMessage()); fail("Failed retrieving multi-signature message from DB: " + e.getMessage()); return; + } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | IOException e) { + System.err.println("Falied to read from stream while retrieving multi-signature message: " + e.getMessage()); + fail("Falied to read from stream while retrieving multi-signature message: " + e.getMessage()); + return; } assertThat(messages.size(), is(MESSAGE_NUM / 4)); @@ -388,7 +439,10 @@ public class GenericBulletinBoardServerTest { final int BATCH_ID = 100; - CompleteBatch completeBatch = new CompleteBatch(); + CompleteBatch completeBatch = new CompleteBatch(Timestamp.newBuilder() + .setSeconds(978325) + .setNanos(8097234) + .build()); BoolMsg result; // Create data @@ -429,11 +483,7 @@ public class GenericBulletinBoardServerTest { // Close batch - result = bulletinBoardServer.closeBatchMessage(CloseBatchMessage.newBuilder() - .setBatchId(BATCH_ID) - .setBatchLength(1) - .setSig(completeBatch.getSignature()) - .build()); + result = bulletinBoardServer.closeBatchMessage(completeBatch.getCloseBatchMessage()); assertThat("Was not able to close batch", result.getValue(), is(true)); @@ -457,7 +507,10 @@ public class GenericBulletinBoardServerTest { */ public void testPostBatch() throws CommunicationException, SignatureException { - CompleteBatch completeBatch = new CompleteBatch(); + CompleteBatch completeBatch = new CompleteBatch(Timestamp.newBuilder() + .setSeconds(12345) + .setNanos(1111) + .build()); int currentBatch = completeBatches.size(); BoolMsg result; @@ -523,11 +576,7 @@ public class GenericBulletinBoardServerTest { signers[0].updateContent(completeBatch); completeBatch.setSignature(signers[0].sign()); - result = bulletinBoardServer.closeBatchMessage(CloseBatchMessage.newBuilder() - .setBatchId(currentBatch) - .setBatchLength(tempBatchData.length) - .setSig(completeBatch.getSignature()) - .build()); + result = bulletinBoardServer.closeBatchMessage(completeBatch.getCloseBatchMessage()); assertThat("Could not close batch " + currentBatch, result.getValue(), is(true)); @@ -540,17 +589,32 @@ public class GenericBulletinBoardServerTest { for (CompleteBatch completeBatch : completeBatches) { - List batchDataList = - bulletinBoardServer.readBatch(BatchSpecificationMessage.newBuilder() - .setSignerId(completeBatch.getBeginBatchMessage().getSignerId()) - .setBatchId(completeBatch.getBeginBatchMessage().getBatchId()) - .setStartPosition(0) - .build()); + try { - assertThat("Non-matching batch data for batch " + completeBatch.getBeginBatchMessage().getBatchId(), - completeBatch.getBatchDataList().equals(batchDataList), is(true)); + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + BatchSpecificationMessage batchSpecificationMessage = + BatchSpecificationMessage.newBuilder() + .setSignerId(completeBatch.getBeginBatchMessage().getSignerId()) + .setBatchId(completeBatch.getBeginBatchMessage().getBatchId()) + .setStartPosition(0) + .build(); + bulletinBoardServer.readBatch(batchSpecificationMessage, new MessageOutputStream(outputStream)); + + MessageInputStream inputStream = + MessageInputStreamFactory.createMessageInputStream(new ByteArrayInputStream( + outputStream.toByteArray()), + BatchData.class); + + List batchDataList = inputStream.asList(); + + assertThat("Non-matching batch data for batch " + completeBatch.getBeginBatchMessage().getBatchId(), + completeBatch.getBatchDataList().equals(batchDataList), is(true)); + + } catch (IOException | NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { + assertThat("Error reading batch data list from input stream", false); + } } diff --git a/meerkat-common/build.gradle b/meerkat-common/build.gradle index 3783531..a9035b0 100644 --- a/meerkat-common/build.gradle +++ b/meerkat-common/build.gradle @@ -39,6 +39,7 @@ version += "${isSnapshot ? '-SNAPSHOT' : ''}" dependencies { // Logging compile 'org.slf4j:slf4j-api:1.7.7' + compile 'javax.ws.rs:javax.ws.rs-api:2.0.+' runtime 'ch.qos.logback:logback-classic:1.1.2' runtime 'ch.qos.logback:logback-core:1.1.2' diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardConstants.java b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardConstants.java index 6bfc06f..66652f8 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardConstants.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardConstants.java @@ -14,6 +14,7 @@ public interface BulletinBoardConstants { public static final String BEGIN_BATCH_PATH = "/beginbatch"; public static final String POST_BATCH_PATH = "/postbatch"; public static final String CLOSE_BATCH_PATH = "/closebatch"; + public static final String SYNC_QUERY_PATH = "/syncquery"; // Other Constants diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java index cbf06ff..0b279c2 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java @@ -1,9 +1,9 @@ package meerkat.bulletinboard; import meerkat.comm.CommunicationException; +import meerkat.comm.MessageOutputStream; import meerkat.protobuf.BulletinBoardAPI.*; -import java.util.List; /** * Created by Arbel on 07/11/15. @@ -32,10 +32,10 @@ public interface BulletinBoardServer{ /** * Read all messages posted matching the given filter * @param filterList return only messages that match the filters (empty list or null means no filtering) - * @return + * @param out is an output stream into which the matching messages are written * @throws CommunicationException on DB connection error */ - public BulletinBoardMessageList readMessages(MessageFilterList filterList) throws CommunicationException; + public void readMessages(MessageFilterList filterList, MessageOutputStream out) throws CommunicationException; /** * Informs server about a new batch message @@ -71,11 +71,20 @@ public interface BulletinBoardServer{ /** * Reads a batch message from the server (starting with the supplied position) * @param message specifies the signer ID and the batch ID to read as well as an (optional) start position - * @return an ordered list of batch messages starting from the specified start position (if given) or from the beginning (if omitted) + * @param out is a stream of the ordered batch messages starting from the specified start position (if given) or from the beginning (if omitted) * @throws CommunicationException on DB connection error * @throws IllegalArgumentException if message does not specify a batch */ - public BatchDataList readBatch(BatchSpecificationMessage message) throws CommunicationException, IllegalArgumentException; + public void readBatch(BatchSpecificationMessage message, MessageOutputStream out) throws CommunicationException, IllegalArgumentException; + + + /** + * Queries the database for sync status with respect to a given sync query + * @param syncQuery contains a succinct representation of states to compare to + * @return a SyncQueryResponse object containing the representation of the most recent state the database matches + * @throws CommunicationException + */ + public SyncQueryResponse querySync(SyncQuery syncQuery) throws CommunicationException; /** * This method closes the connection to the DB diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/CompleteBatch.java b/meerkat-common/src/main/java/meerkat/bulletinboard/CompleteBatch.java index 227c7ca..14e87e7 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/CompleteBatch.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/CompleteBatch.java @@ -1,5 +1,6 @@ package meerkat.bulletinboard; +import com.google.protobuf.Timestamp; import meerkat.protobuf.BulletinBoardAPI.*; import meerkat.protobuf.Crypto.*; import meerkat.util.BulletinBoardMessageComparator; @@ -17,6 +18,7 @@ public class CompleteBatch { private BeginBatchMessage beginBatchMessage; private List batchDataList; private Signature signature; + private Timestamp timestamp; public CompleteBatch() { batchDataList = new LinkedList(); @@ -37,6 +39,16 @@ public class CompleteBatch { signature = newSignature; } + public CompleteBatch(BeginBatchMessage newBeginBatchMessage, List newDataList, Signature newSignature, Timestamp timestamp) { + this(newBeginBatchMessage, newDataList, newSignature); + this.timestamp = timestamp; + } + + public CompleteBatch(Timestamp timestamp) { + this(); + this.timestamp = timestamp; + } + public BeginBatchMessage getBeginBatchMessage() { return beginBatchMessage; } @@ -49,11 +61,16 @@ public class CompleteBatch { return signature; } + public Timestamp getTimestamp() { + return timestamp; + } + public CloseBatchMessage getCloseBatchMessage() { return CloseBatchMessage.newBuilder() .setBatchId(getBeginBatchMessage().getBatchId()) .setBatchLength(getBatchDataList().size()) .setSig(getSignature()) + .setTimestamp(getTimestamp()) .build(); } @@ -73,6 +90,10 @@ public class CompleteBatch { signature = newSignature; } + public void setTimestamp(Timestamp timestamp) { + this.timestamp = timestamp; + } + @Override public boolean equals(Object other) { @@ -105,6 +126,13 @@ public class CompleteBatch { result = result && signature.equals(otherBatch.getSignature()); } + if (timestamp == null) { + if (otherBatch.getTimestamp() != null) + return false; + } else { + result = result && timestamp.equals(otherBatch.getTimestamp()); + } + return result; } diff --git a/meerkat-common/src/main/java/meerkat/comm/MessageInputStream.java b/meerkat-common/src/main/java/meerkat/comm/MessageInputStream.java new file mode 100644 index 0000000..33deb3f --- /dev/null +++ b/meerkat-common/src/main/java/meerkat/comm/MessageInputStream.java @@ -0,0 +1,64 @@ +package meerkat.comm; + +import com.google.protobuf.Message; + +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.InvocationTargetException; +import java.util.LinkedList; +import java.util.List; + +/** + * Created by Arbel Deutsch Peled on 21-Feb-16. + * A input stream of Protobuf messages + */ +public class MessageInputStream{ + + private T.Builder builder; + + private InputStream in; + + MessageInputStream(InputStream in, Class type) throws IOException, NoSuchMethodException, InvocationTargetException, IllegalAccessException { + this.in = in; + this.builder = (T.Builder) type.getMethod("newBuilder").invoke(type); + } + + /** + * Factory class for actually creating a MessageInputStream + */ + public static class MessageInputStreamFactory { + + public static MessageInputStream createMessageInputStream(InputStream in, Class type) + throws IOException, NoSuchMethodException, IllegalAccessException, InvocationTargetException { + + return new MessageInputStream<>(in, type); + + } + + } + + public T readMessage() throws IOException{ + + builder.clear(); + builder.mergeDelimitedFrom(in); + return (T) builder.build(); + + } + + public boolean isAvailable() throws IOException { + return (in.available() > 0); + } + + public List asList() throws IOException{ + + List list = new LinkedList<>(); + + while (isAvailable()){ + list.add(readMessage()); + } + + return list; + + } + +} diff --git a/meerkat-common/src/main/java/meerkat/comm/MessageOutputStream.java b/meerkat-common/src/main/java/meerkat/comm/MessageOutputStream.java new file mode 100644 index 0000000..b55bc8e --- /dev/null +++ b/meerkat-common/src/main/java/meerkat/comm/MessageOutputStream.java @@ -0,0 +1,24 @@ +package meerkat.comm; + +import com.google.protobuf.Message; + +import java.io.IOException; +import java.io.OutputStream; + +/** + * Created by Arbel Deutsch Peled on 21-Feb-16. + * An output stream of Protobuf messages + */ +public class MessageOutputStream { + + private OutputStream out; + + public MessageOutputStream(OutputStream out) throws IOException { + this.out = out; + } + + public void writeMessage(T message) throws IOException { + message.writeDelimitedTo(out); + } + +} diff --git a/meerkat-common/src/main/java/meerkat/comm/Timestamp.java b/meerkat-common/src/main/java/meerkat/comm/Timestamp.java deleted file mode 100644 index 6c63854..0000000 --- a/meerkat-common/src/main/java/meerkat/comm/Timestamp.java +++ /dev/null @@ -1,7 +0,0 @@ -package meerkat.comm; - -/** - * Created by talm on 24/10/15. - */ -public class Timestamp { -} diff --git a/meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageGenerator.java b/meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageGenerator.java new file mode 100644 index 0000000..5ca3e0b --- /dev/null +++ b/meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageGenerator.java @@ -0,0 +1,98 @@ +package meerkat.util; + +import com.google.protobuf.ByteString; +import meerkat.crypto.DigitalSignature; +import meerkat.protobuf.BulletinBoardAPI.*; +import com.google.protobuf.Timestamp; + +import java.math.BigInteger; +import java.security.SignatureException; +import java.util.Arrays; +import java.util.Random; + +/** + * Created by Arbel Deutsch Peled on 21-Feb-16. + * This class contains methods used to generate random Bulletin Board Messages + */ +public class BulletinBoardMessageGenerator { + + private Random random; + + public BulletinBoardMessageGenerator(Random random) { + this.random = random; + } + + private byte randomByte(){ + return (byte) random.nextInt(); + } + + private String randomString(){ + return new BigInteger(130, random).toString(32); + } + + /** + * Generates a complete instance of a BulletinBoardMessage + * @param signers contains the (possibly multiple) credentials required to sign the message + * @param timestamp contains the time used in the message + * @param dataSize is the length of the data contained in the message + * @param tagNumber is the number of tags to generate + * @return a random, signed Bulletin Board Message containing random data and tags and the given timestamp + */ + + public BulletinBoardMessage generateRandomMessage(DigitalSignature[] signers, Timestamp timestamp, int dataSize, int tagNumber) + throws SignatureException { + + // Generate random data. + + byte[] data = new byte[dataSize]; + String[] tags = new String[tagNumber]; + + for (int i = 0; i < dataSize; i++) { + data[i] = randomByte(); + } + + for (int i = 0; i < tagNumber; i++) { + tags[i] = randomString(); + } + + UnsignedBulletinBoardMessage unsignedMessage = + UnsignedBulletinBoardMessage.newBuilder() + .setData(ByteString.copyFrom(data)) + .setTimestamp(timestamp) + .addAllTag(Arrays.asList(tags)) + .build(); + + BulletinBoardMessage.Builder messageBuilder = + BulletinBoardMessage.newBuilder() + .setMsg(unsignedMessage); + + for (int i = 0 ; i < signers.length ; i++) { + signers[i].updateContent(unsignedMessage); + messageBuilder.addSig(signers[i].sign()); + } + + return messageBuilder.build(); + + } + + /** + * Generates a complete instance of a BulletinBoardMessage + * @param signers contains the (possibly multiple) credentials required to sign the message + * @param dataSize is the length of the data contained in the message + * @param tagNumber is the number of tags to generate + * @return a random, signed Bulletin Board Message containing random data, tags and timestamp + */ + + public BulletinBoardMessage generateRandomMessage(DigitalSignature[] signers, int dataSize, int tagNumber) + throws SignatureException { + + Timestamp timestamp = Timestamp.newBuilder() + .setSeconds(random.nextLong()) + .setNanos(random.nextInt()) + .build(); + + return generateRandomMessage(signers, timestamp, dataSize, tagNumber); + + } + +} diff --git a/meerkat-common/src/main/java/meerkat/util/BulletinBoardUtils.java b/meerkat-common/src/main/java/meerkat/util/BulletinBoardUtils.java index b19bab3..d8b362c 100644 --- a/meerkat-common/src/main/java/meerkat/util/BulletinBoardUtils.java +++ b/meerkat-common/src/main/java/meerkat/util/BulletinBoardUtils.java @@ -62,4 +62,46 @@ public class BulletinBoardUtils { } + /** + * This method creates a Timestamp Protobuf from a time specification + * @param timeInMillis is the time to encode since the Epoch time in milliseconds + * @return a Timestamp Protobuf encoding of the given time + */ + public static com.google.protobuf.Timestamp toTimestampProto(long timeInMillis) { + + return com.google.protobuf.Timestamp.newBuilder() + .setSeconds(timeInMillis / 1000) + .setNanos((int) ((timeInMillis % 1000) * 1000000)) + .build(); + + } + + /** + * This method creates a Timestamp Protobuf from the current system time + * @return a Timestamp Protobuf encoding of the current system time + */ + public static com.google.protobuf.Timestamp toTimestampProto() { + + return toTimestampProto(System.currentTimeMillis()); + + } + + /** + * This method converts an SQL Timestamp object into a Protobuf Timestamp object + * @param sqlTimestamp is the SQL Timestamp + * @return an equivalent Protobuf Timestamp + */ + public static com.google.protobuf.Timestamp toTimestampProto(java.sql.Timestamp sqlTimestamp) { + return toTimestampProto(sqlTimestamp.getTime()); + } + + /** + * This method converts a Protobuf Timestamp object into an SQL Timestamp object + * @param protoTimestamp is the Protobuf Timestamp + * @return an equivalent SQL Timestamp + */ + public static java.sql.Timestamp toSQLTimestamp(com.google.protobuf.Timestamp protoTimestamp) { + return new java.sql.Timestamp(protoTimestamp.getSeconds() * 1000 + protoTimestamp.getNanos() / 1000000); + } + } diff --git a/meerkat-common/src/main/java/meerkat/util/TimeStampComparator.java b/meerkat-common/src/main/java/meerkat/util/TimeStampComparator.java new file mode 100644 index 0000000..1dc207b --- /dev/null +++ b/meerkat-common/src/main/java/meerkat/util/TimeStampComparator.java @@ -0,0 +1,30 @@ +package meerkat.util; + +import com.google.protobuf.Timestamp; + +import java.util.Comparator; + +/** + * Created by Arbel Deutsch Peled on 20-Feb-16. + */ +public class TimestampComparator implements Comparator { + + @Override + public int compare(Timestamp o1, Timestamp o2) { + + if (o1.getSeconds() != o2.getSeconds()){ + + return o1.getSeconds() > o2.getSeconds() ? 2 : -2; + + } else if (o1.getNanos() != o2.getNanos()){ + + return o1.getNanos() > o2.getNanos() ? 1 : -1; + + } else{ + + return 0; + + } + + } +} diff --git a/meerkat-common/src/main/proto/meerkat/BulletinBoardAPI.proto b/meerkat-common/src/main/proto/meerkat/BulletinBoardAPI.proto index b86debd..fd95503 100644 --- a/meerkat-common/src/main/proto/meerkat/BulletinBoardAPI.proto +++ b/meerkat-common/src/main/proto/meerkat/BulletinBoardAPI.proto @@ -5,6 +5,7 @@ package meerkat; option java_package = "meerkat.protobuf"; import 'meerkat/crypto.proto'; +import 'google/protobuf/timestamp.proto'; message BoolMsg { bool value = 1; @@ -21,11 +22,14 @@ message MessageID { } message UnsignedBulletinBoardMessage { - // Optional tags describing message + // Optional tags describing message; Used for message retrieval repeated string tag = 1; + // Timestamp of the message (as defined by client) + google.protobuf.Timestamp timestamp = 2; + // The actual content of the message - bytes data = 2; + bytes data = 3; } message BulletinBoardMessage { @@ -38,6 +42,7 @@ message BulletinBoardMessage { // Signature of message (and tags), excluding the entry number. repeated meerkat.Signature sig = 3; + } message BulletinBoardMessageList { @@ -53,11 +58,13 @@ enum FilterType { MIN_ENTRY = 3; // Find all entries in database starting from specified entry number (chronological) SIGNER_ID = 4; // Find all entries in database that correspond to specific signature (signer) TAG = 5; // Find all entries in database that have a specific tag + AFTER_TIME = 6; // Find all entries in database that occurred on or after a given timestamp + BEFORE_TIME = 7; // Find all entries in database that occurred on or before a given timestamp // NOTE: The MAX_MESSAGES filter must remain the last filter type // This is because the condition it specifies in an SQL statement must come last in the statement // Keeping it last here allows for easily sorting the filters and keeping the code general - MAX_MESSAGES = 6; // Return at most some specified number of messages + MAX_MESSAGES = 8; // Return at most some specified number of messages } message MessageFilter { @@ -69,6 +76,7 @@ message MessageFilter { int64 entry = 3; string tag = 4; int64 maxMessages = 5; + google.protobuf.Timestamp timestamp = 6; } } @@ -89,9 +97,10 @@ message BeginBatchMessage { // This message is used to finalize and sign a batch transfer to the Bulletin Board Server message CloseBatchMessage { - int32 batchId = 1; // Unique identifier for the batch (unique per signer) - int32 batchLength = 2; // Number of messages in the batch - meerkat.Signature sig = 3; // Signature on the (ordered) batch messages + int32 batchId = 1; // Unique identifier for the batch (unique per signer) + int32 batchLength = 2; // Number of messages in the batch + google.protobuf.Timestamp timestamp = 3; // Timestamp of the batch (as defined by client) + meerkat.Signature sig = 4; // Signature on the (ordered) batch messages } // Container for single batch message data @@ -117,4 +126,34 @@ message BatchSpecificationMessage { bytes signerId = 1; // Unique signer identifier int32 batchId = 2; // Unique identifier for the batch (unique per signer) int32 startPosition = 3; // Position in batch to start reading from +} + +// This message is used to define a single query to the server to ascertain whether or not the server is synched with the client +// up till a specified timestamp +message SingleSyncQuery { + + google.protobuf.Timestamp timeOfSync = 1; + int64 checksum = 2; + +} + +// This message defines a complete server sync query +message SyncQuery { + + MessageFilterList filterList = 1; + + repeated SingleSyncQuery query = 2; + +} + +// This message defines the server's response format to a sync query +message SyncQueryResponse { + + // Serial entry number of current last entry in database + // Set to zero (0) in case no query checksums match + int64 lastEntryNum = 1; + + // Largest value of timestamp for which the checksums match + google.protobuf.Timestamp lastTimeOfSync = 2; + } \ No newline at end of file diff --git a/meerkat-common/src/test/java/meerkat/comm/MessageStreamTest.java b/meerkat-common/src/test/java/meerkat/comm/MessageStreamTest.java new file mode 100644 index 0000000..58dc7c1 --- /dev/null +++ b/meerkat-common/src/test/java/meerkat/comm/MessageStreamTest.java @@ -0,0 +1,98 @@ +package meerkat.comm; + +import com.google.protobuf.*; +import meerkat.comm.MessageInputStream.MessageInputStreamFactory; +import meerkat.protobuf.BulletinBoardAPI.*; +import meerkat.protobuf.Crypto; +import meerkat.util.BulletinBoardMessageComparator; +import org.junit.Test; + +import java.io.*; +import java.lang.reflect.InvocationTargetException; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; +import static org.hamcrest.core.IsEqual.equalTo; + +/** + * Created by Arbel Deutsch Peled on 21-Feb-16. + * Tests for MessageInputStream and MessageOutputStream classes + */ +public class MessageStreamTest { + + @Test + public void testWithBulletinBoardMessages() { + + MessageOutputStream out; + MessageInputStream in; + ByteArrayOutputStream stream = new ByteArrayOutputStream(); + + BulletinBoardMessageComparator comparator = new BulletinBoardMessageComparator(); + + byte[] b1 = {(byte) 1, (byte) 2, (byte) 3, (byte) 4}; + byte[] b2 = {(byte) 11, (byte) 12, (byte) 13, (byte) 14}; + byte[] b3 = {(byte) 21, (byte) 22, (byte) 23, (byte) 24}; + + try { + + out = new MessageOutputStream<>(stream); + + } catch (IOException e) { + + System.err.println(e.getMessage()); + assertThat("Error creating streams: " + e.getMessage(), false); + return; + + } + + + + BulletinBoardMessage message = BulletinBoardMessage.newBuilder() + .setEntryNum(1) + .setMsg(UnsignedBulletinBoardMessage.newBuilder() + .setData(ByteString.copyFrom(b1)) + .addTag("Test") + .addTag("1234") + .setTimestamp(com.google.protobuf.Timestamp.newBuilder() + .setSeconds(19823451) + .setNanos(2134) + .build()) + .build()) + .addSig(Crypto.Signature.newBuilder() + .setSignerId(ByteString.copyFrom(b2)) + .setData(ByteString.copyFrom(b3)) + .build()) + .build(); + + try { + + out.writeMessage(message); + + } catch (IOException e) { + + System.err.println(e.getMessage()); + assertThat("Error writing message: " + e.getMessage(), false); + + } + + try { + + in = MessageInputStreamFactory.createMessageInputStream( + new ByteArrayInputStream(stream.toByteArray()), + BulletinBoardMessage.class); + + assertThat("Retrieved message was not identical to send message", comparator.compare(message, in.readMessage()), is(equalTo(0))); + + } catch (IOException e) { + + System.err.println(e.getMessage()); + assertThat("Error reading message: " + e.getMessage(), false); + + } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { + System.err.println(e.getMessage()); + assertThat("Error creating input stream " + e.getMessage(), false); + } + + } + +}