meerkat
/
meerkat-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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).
Bulletin-Board-Batch
Arbel Deutsch Peled 2016-02-22 08:04:01 +02:00
parent 9a78330e29
commit aeb7c13436
34 changed files with 1106 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
bulletin-board-client/src/main/java/meerkat/bulletinboard/BulletinClientWorker.java
View File

@ -8,7 +8,7 @@ package meerkat.bulletinboard;
*/
public abstract class BulletinClientWorker<IN> {
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

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

@ -18,7 +18,7 @@ import java.util.concurrent.atomic.AtomicInteger;
*/
public abstract class MultiServerWorker<IN, OUT> extends BulletinClientWorker<IN> implements Runnable, FutureCallback<OUT>{
private List<SingleServerBulletinBoardClient> clients;
private final List<SingleServerBulletinBoardClient> 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<IN, OUT> extends BulletinClientWorker<IN
private AtomicBoolean returnedResult;
private FutureCallback<OUT> futureCallback;
private final FutureCallback<OUT> futureCallback;
/**
* Constructor

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

@ -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<BulletinBoardMessage> readMessages(MessageFilterList filterList) {

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

@ -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<T> implements FutureCallback<T> {
private SingleServerWorker worker;
private FutureCallback<T> futureCallback;
private final SingleServerWorker worker;
private final FutureCallback<T> futureCallback;
public RetryCallback(SingleServerWorker worker, FutureCallback<T> futureCallback) {
this.worker = worker;
@ -128,7 +130,8 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
*/
class PostBatchDataListCallback implements FutureCallback<Boolean> {
private FutureCallback<Boolean> callback;
private final FutureCallback<Boolean> 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<CompleteBatch> callback;
private List<BatchData> 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<CompleteBatch> 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<List<BulletinBoardMessage>> {
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<String>();
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<Boolean> {
private CompleteBatch completeBatch;
FutureCallback<Boolean> callback;
private final CompleteBatch completeBatch;
private final FutureCallback<Boolean> callback;
public PostBatchDataCallback(CompleteBatch completeBatch, FutureCallback<Boolean> callback) {
this.completeBatch = completeBatch;
@ -391,8 +399,8 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
private class BeginBatchCallback implements FutureCallback<Boolean> {
private CompleteBatch completeBatch;
FutureCallback<Boolean> callback;
private final CompleteBatch completeBatch;
private final FutureCallback<Boolean> callback;
public BeginBatchCallback(CompleteBatch completeBatch, FutureCallback<Boolean> 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()));
}

2
bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerWorker.java
View File

@ -25,7 +25,7 @@ public abstract class SingleServerWorker<IN, OUT> extends BulletinClientWorker<I
}
};
protected String serverAddress;
protected final String serverAddress;
public SingleServerWorker(String serverAddress, IN payload, int maxRetry) {
super(payload, maxRetry);

9
bulletin-board-client/src/main/java/meerkat/bulletinboard/ThreadedBulletinBoardClient.java
View File

@ -25,12 +25,12 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple
// Executor service for handling jobs
private final static int JOBS_THREAD_NUM = 5;
ExecutorService executorService;
private ExecutorService executorService;
// Per-server clients
List<SingleServerBulletinBoardClient> clients;
private List<SingleServerBulletinBoardClient> 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<SingleServerBulletinBoardClient>(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<Boolean> callback){

2
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java
View File

@ -31,8 +31,6 @@ public abstract class MultiServerGenericPostWorker<T> 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() {

4
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java
View File

@ -14,7 +14,7 @@ import java.util.List;
*/
public abstract class MultiServerGenericReadWorker<IN, OUT> extends MultiServerWorker<IN, OUT>{
protected Iterator<SingleServerBulletinBoardClient> clientIterator;
private final Iterator<SingleServerBulletinBoardClient> clientIterator;
public MultiServerGenericReadWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, IN payload, int maxRetry,
@ -32,8 +32,6 @@ public abstract class MultiServerGenericReadWorker<IN, OUT> 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(){

6
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java
View File

@ -33,8 +33,6 @@ public class MultiServerGetRedundancyWorker extends MultiServerWorker<MessageID,
* 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(){
@ -61,14 +59,14 @@ public class MultiServerGetRedundancyWorker extends MultiServerWorker<MessageID,
}
if (totalContactedServers.incrementAndGet() >= 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);
}
}

4
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerGenericPostWorker.java
View File

@ -19,7 +19,7 @@ import static meerkat.bulletinboard.BulletinBoardConstants.BULLETIN_BOARD_SERVER
*/
public class SingleServerGenericPostWorker<T> extends SingleServerWorker<T, Boolean> {
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<T> extends SingleServerWorker<T, Bool
Client client = clientLocal.get();
WebTarget webTarget = client.target(serverAddress).path(BULLETIN_BOARD_SERVER_PATH).path(subPath);;
WebTarget webTarget = client.target(serverAddress).path(BULLETIN_BOARD_SERVER_PATH).path(subPath);
Response response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(
Entity.entity(payload, Constants.MEDIATYPE_PROTOBUF));

4
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerGetRedundancyWorker.java
View File

@ -57,11 +57,11 @@ public class SingleServerGetRedundancyWorker extends SingleServerWorker<MessageI
if (msgList.getMessageList().size() > 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) {

15
bulletin-board-client/src/test/java/ThreadedBulletinBoardClientIntegrationTest.java
View File

@ -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<String> testDB = new LinkedList<String>();
List<String> 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<BulletinBoardMessage> msgComparator = new BulletinBoardMessageComparator();
msg = BulletinBoardMessage.newBuilder()
.setMsg(UnsignedBulletinBoardMessage.newBuilder()
.addTag("Signature")
@ -398,7 +395,7 @@ public class ThreadedBulletinBoardClientIntegrationTest {
)
.build();
msgList = new LinkedList<BulletinBoardMessage>();
msgList = new LinkedList<>();
msgList.add(msg);
readCallback = new ReadCallback(msgList);

4
bulletin-board-server/build.gradle
View File

@ -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

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

@ -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<MessageFilter> filters = new ArrayList<MessageFilter>(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<BulletinBoardMessage> 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<BulletinBoardMessage.Builder> 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<BulletinBoardMessage> out) throws CommunicationException {
namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.GET_SIGNATURES.getParamName(0), msgBuilder.getEntryNum());
BulletinBoardMessageList.Builder resultListBuilder = BulletinBoardMessageList.newBuilder();
List<Signature> 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<Long> 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<BatchData> 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<Long> 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<SingleSyncQuery> queryIterator = syncQuery.getQueryList().iterator();
SingleSyncQuery currentQuery = queryIterator.next();
List<BulletinBoardMessage> 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

18
bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java
View File

@ -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<String> getSchemaCreationCommands() {
List<String> list = new LinkedList<String>();
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)");

27
bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java
View File

@ -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<String> list = new LinkedList<String>();
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))");

32
bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/BatchDataCallbackHandler.java Normal file
View File

@ -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<BatchData> out;
public BatchDataCallbackHandler(MessageOutputStream<BatchData> 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
}
}
}

80
bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageCallbackHandler.java Normal file
View File

@ -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<BulletinBoardMessage> out;
public MessageCallbackHandler(NamedParameterJdbcTemplate jdbcTemplate, SQLQueryProvider sqlQueryProvider, MessageOutputStream<BulletinBoardMessage> 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<Crypto.Signature> 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();
}
}
}

31
bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java Normal file
View File

@ -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<BulletinBoardMessage> {
@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();
}
}

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

@ -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<BulletinBoardMessage> 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<BulletinBoardMessage> 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<BatchData> 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<BatchData> 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;

38
bulletin-board-server/src/test/java/meerkat/bulletinboard/BulletinBoardSQLServerIntegrationTest.java
View File

@ -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<BulletinBoardMessage> 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<BulletinBoardMessage> 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;
}
}

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

@ -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<BulletinBoardMessage> messages;
List<BulletinBoardMessage> 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<BulletinBoardMessage>(outputStream));
MessageInputStream<BulletinBoardMessage> 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<BulletinBoardMessage>(outputStream));
MessageInputStream<BulletinBoardMessage> 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<BulletinBoardMessage>(outputStream));
MessageInputStream<BulletinBoardMessage> 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<BatchData> 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<BatchData>(outputStream));
MessageInputStream<BatchData> inputStream =
MessageInputStreamFactory.createMessageInputStream(new ByteArrayInputStream(
outputStream.toByteArray()),
BatchData.class);
List<BatchData> 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);
}
}

1
meerkat-common/build.gradle
View File

@ -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'

1
meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardConstants.java
View File

@ -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

19
meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java
View File

@ -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<BulletinBoardMessage> 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<BatchData> 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

28
meerkat-common/src/main/java/meerkat/bulletinboard/CompleteBatch.java
View File

@ -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<BatchData> batchDataList;
private Signature signature;
private Timestamp timestamp;
public CompleteBatch() {
batchDataList = new LinkedList<BatchData>();
@ -37,6 +39,16 @@ public class CompleteBatch {
signature = newSignature;
}
public CompleteBatch(BeginBatchMessage newBeginBatchMessage, List<BatchData> 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;
}

64
meerkat-common/src/main/java/meerkat/comm/MessageInputStream.java Normal file
View File

@ -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<T extends Message>{
private T.Builder builder;
private InputStream in;
MessageInputStream(InputStream in, Class<T> 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 <T extends Message> MessageInputStream<T> createMessageInputStream(InputStream in, Class<T> 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<T> asList() throws IOException{
List<T> list = new LinkedList<>();
while (isAvailable()){
list.add(readMessage());
}
return list;
}
}

24
meerkat-common/src/main/java/meerkat/comm/MessageOutputStream.java Normal file
View File

@ -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<T extends Message> {
private OutputStream out;
public MessageOutputStream(OutputStream out) throws IOException {
this.out = out;
}
public void writeMessage(T message) throws IOException {
message.writeDelimitedTo(out);
}
}

7
meerkat-common/src/main/java/meerkat/comm/Timestamp.java
View File

@ -1,7 +0,0 @@
package meerkat.comm;
/**
* Created by talm on 24/10/15.
*/
public class Timestamp {
}

98
meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageGenerator.java Normal file
View File

@ -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);
}
}

42
meerkat-common/src/main/java/meerkat/util/BulletinBoardUtils.java
View File

@ -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);
}
}

30
meerkat-common/src/main/java/meerkat/util/TimeStampComparator.java Normal file
View File

@ -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<com.google.protobuf.Timestamp> {
@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;
}
}
}

51
meerkat-common/src/main/proto/meerkat/BulletinBoardAPI.proto
View File

@ -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;
}

98
meerkat-common/src/test/java/meerkat/comm/MessageStreamTest.java Normal file
View File

@ -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<BulletinBoardMessage> out;
MessageInputStream<BulletinBoardMessage> 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);
}
}
}