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

Changed Bulletin Board Message payload to either data or message ID 

Added server-generated unique batch identifiers
Changed Client-side interfaces
Refactored Client-side code for new batch mechanisms

Not tested on client-side yet
Cached-Client
Arbel Deutsch Peled 2016-06-19 22:00:43 +03:00
parent b501992643
commit 1951db546d
40 changed files with 1274 additions and 884 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
bulletin-board-client/src/main/java/meerkat/bulletinboard/BatchDataContainer.java
View File

@ -1,6 +1,7 @@
package meerkat.bulletinboard;
import meerkat.protobuf.BulletinBoardAPI.BatchChunk;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import java.util.List;
@ -10,13 +11,11 @@ import java.util.List;
*/
public class BatchDataContainer {
public final byte[] signerId;
public final int batchId;
public final MultiServerBatchIdentifier batchId;
public final List<BatchChunk> batchChunkList;
public final int startPosition;
public BatchDataContainer(byte[] signerId, int batchId, List<BatchChunk> batchChunkList, int startPosition) {
this.signerId = signerId;
public BatchDataContainer(MultiServerBatchIdentifier batchId, List<BatchChunk> batchChunkList, int startPosition) {
this.batchId = batchId;
this.batchChunkList = batchChunkList;
this.startPosition = startPosition;

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

@ -1,11 +1,13 @@
package meerkat.bulletinboard;
import com.google.common.util.concurrent.FutureCallback;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import meerkat.comm.CommunicationException;
import meerkat.protobuf.BulletinBoardAPI;
import meerkat.protobuf.BulletinBoardAPI.*;
import meerkat.protobuf.Comm;
import meerkat.protobuf.Crypto.Signature;
import meerkat.protobuf.Voting.*;
import meerkat.util.BulletinBoardUtils;
import java.util.List;
@ -46,21 +48,12 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
for (BulletinBoardMessage msg : result) {
try {
if (msg.getMsg().getTagList().contains(BulletinBoardConstants.BATCH_TAG)) {
if (msg.getMsg().getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {
// This is a batch message: need to upload batch data as well as the message itself
ByteString signerID = msg.getSig(0).getSignerId();
int batchID = Integer.parseInt(BulletinBoardUtils.findTagWithPrefix(msg, BulletinBoardConstants.BATCH_ID_TAG_PREFIX));
BulletinBoardMessage completeMessage = localClient.readBatchData(msg);
BatchSpecificationMessage batchSpecificationMessage = BatchSpecificationMessage.newBuilder()
.setSignerId(signerID)
.setBatchId(batchID)
.setStartPosition(0)
.build();
CompleteBatch completeBatch = localClient.readBatch(batchSpecificationMessage);
localClient.postBatch(completeBatch);
localClient.postMessage(completeMessage);
} else {
@ -126,12 +119,12 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public MessageID postBatch(final CompleteBatch completeBatch, final FutureCallback<Boolean> callback) {
public MessageID postAsBatch(final BulletinBoardMessage msg, final int chunkSize, final FutureCallback<Boolean> callback) {
return localClient.postBatch(completeBatch, new FutureCallback<Boolean>() {
return localClient.postAsBatch(msg, chunkSize, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
remoteClient.postBatch(completeBatch, callback);
remoteClient.postAsBatch(msg, chunkSize, callback);
}
@Override
@ -144,12 +137,31 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public void beginBatch(final BeginBatchMessage beginBatchMessage, final FutureCallback<Boolean> callback) {
public void beginBatch(final Iterable<String> tags, final FutureCallback<BatchIdentifier> callback) {
localClient.beginBatch(tags, new FutureCallback<BatchIdentifier>() {
private BatchIdentifier localIdentifier;
localClient.beginBatch(beginBatchMessage, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
remoteClient.beginBatch(beginBatchMessage, callback);
public void onSuccess(BatchIdentifier result) {
localIdentifier = result;
remoteClient.beginBatch(tags, new FutureCallback<BatchIdentifier>() {
@Override
public void onSuccess(BatchIdentifier result) {
if (callback != null)
callback.onSuccess(new CachedClientBatchIdentifier(localIdentifier, result));
}
@Override
public void onFailure(Throwable t) {
if (callback != null)
callback.onFailure(t);
}
});
}
@Override
@ -162,13 +174,19 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public void postBatchData(final byte[] signerId, final int batchId, final List<BatchChunk> batchChunkList,
final int startPosition, final FutureCallback<Boolean> callback) {
public void postBatchData(final BatchIdentifier batchIdentifier, final List<BatchChunk> batchChunkList,
final int startPosition, final FutureCallback<Boolean> callback) throws IllegalArgumentException{
localClient.postBatchData(signerId, batchId, batchChunkList, startPosition, new FutureCallback<Boolean>() {
if (!(batchIdentifier instanceof CachedClientBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
final CachedClientBatchIdentifier identifier = (CachedClientBatchIdentifier) batchIdentifier;
localClient.postBatchData(identifier.getLocalIdentifier(), batchChunkList, startPosition, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
remoteClient.postBatchData(signerId, batchId, batchChunkList, startPosition, callback);
remoteClient.postBatchData(identifier.getRemoteIdentifier(), batchChunkList, startPosition, callback);
}
@Override
@ -181,12 +199,19 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public void postBatchData(final byte[] signerId, final int batchId, final List<BatchChunk> batchChunkList, final FutureCallback<Boolean> callback) {
public void postBatchData(final BatchIdentifier batchIdentifier, final List<BatchChunk> batchChunkList, final FutureCallback<Boolean> callback)
throws IllegalArgumentException{
localClient.postBatchData(signerId, batchId, batchChunkList, new FutureCallback<Boolean>() {
if (!(batchIdentifier instanceof CachedClientBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
final CachedClientBatchIdentifier identifier = (CachedClientBatchIdentifier) batchIdentifier;
localClient.postBatchData(identifier.getLocalIdentifier(), batchChunkList, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
remoteClient.postBatchData(signerId, batchId, batchChunkList, callback);
remoteClient.postBatchData(identifier.getRemoteIdentifier(), batchChunkList, callback);
}
@Override
@ -199,49 +224,21 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public void postBatchData(final ByteString signerId, final int batchId, final List<BatchChunk> batchChunkList,
final int startPosition, final FutureCallback<Boolean> callback) {
public void closeBatch(final BatchIdentifier batchIdentifier, final Timestamp timestamp, final Iterable<Signature> signatures,
final FutureCallback<Boolean> callback) {
localClient.postBatchData(signerId, batchId, batchChunkList, startPosition, new FutureCallback<Boolean>() {
if (!(batchIdentifier instanceof CachedClientBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
final CachedClientBatchIdentifier identifier = (CachedClientBatchIdentifier) batchIdentifier;
localClient.closeBatch(identifier.getLocalIdentifier(), timestamp, signatures, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
remoteClient.postBatchData(signerId, batchId, batchChunkList, startPosition, callback);
}
@Override
public void onFailure(Throwable t) {
if (callback != null)
callback.onFailure(t);
}
});
remoteClient.closeBatch(identifier.getRemoteIdentifier(), timestamp, signatures, callback);
}
@Override
public void postBatchData(final ByteString signerId, final int batchId, final List<BatchChunk> batchChunkList, final FutureCallback<Boolean> callback) {
localClient.postBatchData(signerId, batchId, batchChunkList, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
remoteClient.postBatchData(signerId, batchId, batchChunkList, callback);
}
@Override
public void onFailure(Throwable t) {
if (callback != null)
callback.onFailure(t);
}
});
}
@Override
public void closeBatch(final CloseBatchMessage closeBatchMessage, final FutureCallback<Boolean> callback) {
localClient.closeBatch(closeBatchMessage, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
remoteClient.closeBatch(closeBatchMessage, callback);
}
@Override
@ -270,11 +267,12 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public void readBatch(final BatchSpecificationMessage batchSpecificationMessage, final FutureCallback<CompleteBatch> callback) {
public void readBatch(final MessageID msgID, final FutureCallback<BulletinBoardMessage> callback) {
localClient.readBatch(msgID, new FutureCallback<BulletinBoardMessage>() {
localClient.readBatch(batchSpecificationMessage, new FutureCallback<CompleteBatch>() {
@Override
public void onSuccess(CompleteBatch result) {
public void onSuccess(BulletinBoardMessage result) {
if (callback != null)
callback.onSuccess(result); // Read from local client was successful
}
@ -284,19 +282,61 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
// Read from local unsuccessful: try to read from remote
remoteClient.readBatch(batchSpecificationMessage, new FutureCallback<CompleteBatch>() {
remoteClient.readBatch(msgID, new FutureCallback<BulletinBoardMessage>() {
@Override
public void onSuccess(CompleteBatch result) {
public void onSuccess(BulletinBoardMessage result) {
// Read from remote was successful: store in local and return result
localClient.postBatch(result, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {}
@Override
public void onFailure(Throwable t) {}
});
localClient.postMessage(result, null);
if (callback != null)
callback.onSuccess(result);
}
@Override
public void onFailure(Throwable t) {
// Read from remote was unsuccessful: report error
if (callback != null)
callback.onFailure(t);
}
});
}
});
}
@Override
public void readBatchData(final BulletinBoardMessage stub, final FutureCallback<BulletinBoardMessage> callback) throws IllegalArgumentException {
localClient.readBatchData(stub, new FutureCallback<BulletinBoardMessage>() {
@Override
public void onSuccess(BulletinBoardMessage result) {
if (callback != null)
callback.onSuccess(result); // Read from local client was successful
}
@Override
public void onFailure(Throwable t) {
// Read from local unsuccessful: try to read from remote
remoteClient.readBatchData(stub, new FutureCallback<BulletinBoardMessage>() {
@Override
public void onSuccess(BulletinBoardMessage result) {
// Read from remote was successful: store in local and return result
localClient.postMessage(result, null);
if (callback != null)
callback.onSuccess(result);
@ -340,8 +380,10 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public MessageID postBatch(CompleteBatch completeBatch) throws CommunicationException {
return localClient.postBatch(completeBatch);
public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize) throws CommunicationException {
MessageID result = localClient.postAsBatch(msg, chunkSize);
remoteClient.postAsBatch(msg, chunkSize);
return result;
}
@Override
@ -356,8 +398,49 @@ public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClien
}
@Override
public CompleteBatch readBatch(BatchSpecificationMessage batchSpecificationMessage) throws CommunicationException {
return localClient.readBatch(batchSpecificationMessage);
public BulletinBoardMessage readBatch(MessageID msgID) throws CommunicationException {
BulletinBoardMessage result = null;
try {
result = localClient.readBatch(msgID);
} catch (CommunicationException e) {
//TODO: log
}
if (result == null){
result = remoteClient.readBatch(msgID);
if (result != null){
localClient.postMessage(result);
}
}
return result;
}
@Override
public BulletinBoardMessage readBatchData(BulletinBoardMessage stub) throws CommunicationException, IllegalArgumentException {
BulletinBoardMessage result = null;
try {
result = localClient.readBatchData(stub);
} catch (CommunicationException e) {
//TODO: log
}
if (result == null){
result = remoteClient.readBatchData(stub);
if (result != null){
localClient.postMessage(result);
}
}
return result;
}
@Override

29
bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedClientBatchIdentifier.java Normal file
View File

@ -0,0 +1,29 @@
package meerkat.bulletinboard;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import java.util.Arrays;
/**
* Created by Arbel Deutsch Peled on 17-Jun-16.
*/
public final class CachedClientBatchIdentifier implements BatchIdentifier {
// Per-server identifiers
private final BatchIdentifier localIdentifier;
private final BatchIdentifier remoteIdentifier;
public CachedClientBatchIdentifier(BatchIdentifier localIdentifier, BatchIdentifier remoteIdentifier) {
this.localIdentifier = localIdentifier;
this.remoteIdentifier = remoteIdentifier;
}
public BatchIdentifier getLocalIdentifier() {
return localIdentifier;
}
public BatchIdentifier getRemoteIdentifier() {
return remoteIdentifier;
}
}

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

@ -1,20 +1,21 @@
package meerkat.bulletinboard;
import com.google.common.util.concurrent.*;
import com.google.protobuf.ByteString;
import com.google.protobuf.Int64Value;
import com.google.protobuf.Timestamp;
import meerkat.comm.CommunicationException;
import meerkat.comm.MessageInputStream;
import meerkat.comm.MessageInputStream.MessageInputStreamFactory;
import meerkat.comm.MessageOutputStream;
import meerkat.crypto.concrete.SHA256Digest;
import meerkat.protobuf.BulletinBoardAPI.*;
import meerkat.protobuf.Crypto.Signature;
import meerkat.protobuf.Voting.*;
import meerkat.util.BulletinBoardUtils;
import javax.ws.rs.NotFoundException;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
@ -74,48 +75,57 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
private class CompleteBatchPoster implements Callable<Boolean> {
private final BulletinBoardMessage completeBatch;
private final BulletinBoardMessage msg;
private final int chunkSize;
public CompleteBatchPoster(BulletinBoardMessage completeBatch) {
this.completeBatch = completeBatch;
public CompleteBatchPoster(BulletinBoardMessage msg, int chunkSize) {
this.msg = msg;
this.chunkSize = chunkSize;
}
@Override
public Boolean call() throws CommunicationException {
server.beginBatch(BeginBatchMessage.newBuilder().setSignerId(completeBatch.getSig(0))
completeBatch.getBeginBatchMessage());
BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder()
.addAllTag(msg.getMsg().getTagList())
.build();
Int64Value batchId = server.beginBatch(beginBatchMessage);
BatchMessage.Builder builder = BatchMessage.newBuilder()
.setSignerId(completeBatch.getSignature().getSignerId())
.setBatchId(completeBatch.getBeginBatchMessage().getBatchId());
.setBatchId(batchId.getValue());
List<BatchChunk> batchChunkList = BulletinBoardUtils.breakToBatch(msg, chunkSize);
int i=0;
for (BatchChunk data : completeBatch.getBatchDataList()){
for (BatchChunk chunk : batchChunkList){
server.postBatchMessage(builder.setSerialNum(i).setData(data).build());
server.postBatchMessage(builder.setSerialNum(i).setData(chunk).build());
i++;
}
return server.closeBatch(completeBatch.getCloseBatchMessage()).getValue();
CloseBatchMessage closeBatchMessage = BulletinBoardUtils.generateCloseBatchMessage(batchId, batchChunkList.size(), msg);
return server.closeBatch(closeBatchMessage).getValue();
}
}
@Override
public MessageID postBatch(ByteString signerId, int batchId, BulletinBoardMessage completeBatch, FutureCallback<Boolean> callback) {
public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback<Boolean> callback) {
Futures.addCallback(executorService.submit(new CompleteBatchPoster(completeBatch)), callback);
Futures.addCallback(executorService.submit(new CompleteBatchPoster(msg, chunkSize)), callback);
digest.update(completeBatch);
digest.reset();
digest.update(msg);
return digest.digestAsMessageID();
}
private class BatchBeginner implements Callable<Boolean> {
private class BatchBeginner implements Callable<SingleServerBatchIdentifier> {
private final BeginBatchMessage msg;
@ -125,26 +135,29 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
@Override
public Boolean call() throws Exception {
return server.beginBatch(msg).getValue();
public SingleServerBatchIdentifier call() throws Exception {
return new SingleServerBatchIdentifier(server.beginBatch(msg));
}
}
@Override
public void beginBatch(BeginBatchMessage beginBatchMessage, FutureCallback<Boolean> callback) {
public void beginBatch(Iterable<String> tags, FutureCallback<BatchIdentifier> callback) {
BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder()
.addAllTag(tags)
.build();
Futures.addCallback(executorService.submit(new BatchBeginner(beginBatchMessage)), callback);
}
private class BatchDataPoster implements Callable<Boolean> {
private final ByteString signerId;
private final int batchId;
private final SingleServerBatchIdentifier batchId;
private final List<BatchChunk> batchChunkList;
private final int startPosition;
public BatchDataPoster(ByteString signerId, int batchId, List<BatchChunk> batchChunkList, int startPosition) {
this.signerId = signerId;
public BatchDataPoster(SingleServerBatchIdentifier batchId, List<BatchChunk> batchChunkList, int startPosition) {
this.batchId = batchId;
this.batchChunkList = batchChunkList;
this.startPosition = startPosition;
@ -155,8 +168,7 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
public Boolean call() throws Exception {
BatchMessage.Builder msgBuilder = BatchMessage.newBuilder()
.setSignerId(signerId)
.setBatchId(batchId);
.setBatchId(batchId.getBatchId().getValue());
int i = startPosition;
for (BatchChunk data : batchChunkList){
@ -178,24 +190,28 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
}
@Override
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList, int startPosition, FutureCallback<Boolean> callback) {
postBatchData(ByteString.copyFrom(signerId), batchId, batchChunkList, startPosition, callback);
public void postBatchData(BatchIdentifier batchId, List<BatchChunk> batchChunkList, int startPosition, FutureCallback<Boolean> callback)
throws IllegalArgumentException{
// Cast identifier to usable form
if (!(batchId instanceof SingleServerBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchId;
// Add worker
Futures.addCallback(executorService.submit(new BatchDataPoster(identifier, batchChunkList, startPosition)), callback);
}
@Override
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) {
postBatchData(signerId, batchId, batchChunkList, 0, callback);
public void postBatchData(BatchIdentifier batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) throws IllegalArgumentException{
postBatchData(batchId, batchChunkList, 0, callback);
}
@Override
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList, int startPosition, FutureCallback<Boolean> callback) {
Futures.addCallback(executorService.submit(new BatchDataPoster(signerId, batchId, batchChunkList, startPosition)), callback);
}
@Override
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) {
postBatchData(signerId, batchId, batchChunkList, 0, callback);
}
private class BatchCloser implements Callable<Boolean> {
@ -214,8 +230,26 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
}
@Override
public void closeBatch(CloseBatchMessage closeBatchMessage, FutureCallback<Boolean> callback) {
public void closeBatch(BatchIdentifier batchId, Timestamp timestamp, Iterable<Signature> signatures, FutureCallback<Boolean> callback) {
// Cast identifier to usable form
if (!(batchId instanceof SingleServerBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchId;
// Add worker
CloseBatchMessage closeBatchMessage = CloseBatchMessage.newBuilder()
.setBatchId(identifier.getBatchId().getValue())
.setTimestamp(timestamp)
.addAllSig(signatures)
.build();
Futures.addCallback(executorService.submit(new BatchCloser(closeBatchMessage)), callback);
}
private class RedundancyGetter implements Callable<Float> {
@ -287,31 +321,6 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
}
private class BatchDataReader implements Callable<List<BatchChunk>> {
private final BatchSpecificationMessage batchSpecificationMessage;
public BatchDataReader(BatchSpecificationMessage batchSpecificationMessage) {
this.batchSpecificationMessage = batchSpecificationMessage;
}
@Override
public List<BatchChunk> call() throws Exception {
ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
MessageOutputStream<BatchChunk> outputStream = new MessageOutputStream<>(byteOutputStream);
server.readBatch(batchSpecificationMessage, outputStream);
MessageInputStream<BatchChunk> inputStream =
MessageInputStreamFactory.createMessageInputStream(
new ByteArrayInputStream(byteOutputStream.toByteArray()),
BatchChunk.class);
return inputStream.asList();
}
}
@Override
public void readMessages(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {
Futures.addCallback(executorService.submit(new MessageReader(filterList)), callback);
@ -387,83 +396,116 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
subscribe(filterList, 0, callback);
}
private class BatchDataReader implements Callable<List<BatchChunk>> {
private final MessageID msgID;
public BatchDataReader(MessageID msgID) {
this.msgID = msgID;
}
@Override
public List<BatchChunk> call() throws Exception {
BatchQuery batchQuery = BatchQuery.newBuilder()
.setMsgID(msgID)
.setStartPosition(0)
.build();
ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
MessageOutputStream<BatchChunk> batchOutputStream = new MessageOutputStream<>(byteOutputStream);
server.readBatch(batchQuery,batchOutputStream);
MessageInputStream<BatchChunk> inputStream =
MessageInputStreamFactory.createMessageInputStream(
new ByteArrayInputStream(byteOutputStream.toByteArray()),
BatchChunk.class);
return inputStream.asList();
}
}
private class CompleteBatchReader implements Callable<BulletinBoardMessage> {
private final BatchSpecificationMessage batchSpecificationMessage;
private final MessageID msgID;
public CompleteBatchReader(BatchSpecificationMessage batchSpecificationMessage) {
this.batchSpecificationMessage = batchSpecificationMessage;
public CompleteBatchReader(MessageID msgID) {
this.msgID = msgID;
}
@Override
public BulletinBoardMessage call() throws Exception {
final String[] TAGS_TO_REMOVE = {BulletinBoardConstants.BATCH_TAG, BulletinBoardConstants.BATCH_ID_TAG_PREFIX};
CompleteBatch completeBatch = new CompleteBatch(BeginBatchMessage.newBuilder()
.setSignerId(batchSpecificationMessage.getSignerId())
.setBatchId(batchSpecificationMessage.getBatchId())
.build());
ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
MessageOutputStream<BatchChunk> batchOutputStream = new MessageOutputStream<>(byteOutputStream);
server.readBatch(batchSpecificationMessage,batchOutputStream);
MessageInputStream<BatchChunk> batchInputStream =
MessageInputStreamFactory.createMessageInputStream(
new ByteArrayInputStream(byteOutputStream.toByteArray()),
BatchChunk.class);
completeBatch.appendBatchData(batchInputStream.asList());
// Read message stub
MessageFilterList filterList = MessageFilterList.newBuilder()
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_TAG)
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_ID_TAG_PREFIX + completeBatch.getBeginBatchMessage().getBatchId())
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.SIGNER_ID)
.setId(completeBatch.getBeginBatchMessage().getSignerId())
.setType(FilterType.MSG_ID)
.setId(msgID.getID())
.build())
.build();
byteOutputStream = new ByteArrayOutputStream();
MessageOutputStream<BulletinBoardMessage> messageOutputStream = new MessageOutputStream<>(byteOutputStream);
server.readMessages(filterList,messageOutputStream);
MessageReader messageReader = new MessageReader(filterList);
List<BulletinBoardMessage> bulletinBoardMessages = messageReader.call();
MessageInputStream<BulletinBoardMessage> messageInputStream =
MessageInputStreamFactory.createMessageInputStream(
new ByteArrayInputStream(byteOutputStream.toByteArray()),
BulletinBoardMessage.class);
if (!messageInputStream.isAvailable())
if (bulletinBoardMessages.size() <= 0) {
throw new NotFoundException("Batch does not exist");
}
BulletinBoardMessage message = messageInputStream.readMessage();
BulletinBoardMessage stub = bulletinBoardMessages.get(0);
completeBatch.setBeginBatchMessage(BeginBatchMessage.newBuilder()
.addAllTag(BulletinBoardUtils.removePrefixTags(message, Arrays.asList(TAGS_TO_REMOVE)))
.setSignerId(message.getSig(0).getSignerId())
.setBatchId(Integer.parseInt(BulletinBoardUtils.findTagWithPrefix(message, BulletinBoardConstants.BATCH_ID_TAG_PREFIX)))
.build());
// Read data
completeBatch.setSignature(message.getSig(0));
completeBatch.setTimestamp(message.getMsg().getTimestamp());
BatchDataReader batchDataReader = new BatchDataReader(msgID);
List<BatchChunk> batchChunkList = batchDataReader.call();
return completeBatch;
// Combine and return
return BulletinBoardUtils.gatherBatch(stub, batchChunkList);
}
}
private class BatchDataCombiner implements Callable<BulletinBoardMessage> {
private final BulletinBoardMessage stub;
public BatchDataCombiner(BulletinBoardMessage stub) {
this.stub = stub;
}
@Override
public BulletinBoardMessage call() throws Exception {
MessageID msgID = MessageID.newBuilder().setID(stub.getMsg().getMsgId()).build();
BatchDataReader batchDataReader = new BatchDataReader(msgID);
List<BatchChunk> batchChunkList = batchDataReader.call();
return BulletinBoardUtils.gatherBatch(stub, batchChunkList);
}
}
@Override
public void readBatch(BatchSpecificationMessage batchSpecificationMessage, FutureCallback<BulletinBoardMessage> callback) {
Futures.addCallback(executorService.submit(new CompleteBatchReader(batchSpecificationMessage)), callback);
public void readBatch(MessageID msgID, FutureCallback<BulletinBoardMessage> callback) {
Futures.addCallback(executorService.submit(new CompleteBatchReader(msgID)), callback);
}
@Override
public void readBatchData(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) throws IllegalArgumentException {
if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID){
throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
}
Futures.addCallback(executorService.submit(new BatchDataCombiner(stub)),callback);
}
private class SyncQueryHandler implements Callable<SyncQueryResponse> {
@ -506,12 +548,16 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
}
@Override
public MessageID postBatch(CompleteBatch completeBatch) throws CommunicationException {
public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize) throws CommunicationException {
CompleteBatchPoster poster = new CompleteBatchPoster(completeBatch);
poster.call();
CompleteBatchPoster poster = new CompleteBatchPoster(msg, chunkSize);
Boolean result = poster.call();
digest.update(completeBatch);
if (!result)
throw new CommunicationException("Batch post failed");
digest.reset();
digest.update(msg);
return digest.digestAsMessageID();
}
@ -545,38 +591,48 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo
}
@Override
public CompleteBatch readBatch(BatchSpecificationMessage batchSpecificationMessage) throws CommunicationException {
public BulletinBoardMessage readBatch(MessageID msgID) throws CommunicationException {
MessageFilterList filterList = MessageFilterList.newBuilder()
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_TAG)
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_ID_TAG_PREFIX + batchSpecificationMessage.getBatchId())
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.SIGNER_ID)
.setId(batchSpecificationMessage.getSignerId())
.setType(FilterType.MSG_ID)
.setId(msgID.getID())
.build())
.build();
BulletinBoardMessage batchMessage = readMessages(filterList).get(0);
BatchDataReader batchDataReader = new BatchDataReader(batchSpecificationMessage);
CompleteBatchReader completeBatchReader = new CompleteBatchReader(msgID);
try {
List<BatchChunk> batchChunkList = batchDataReader.call();
return new CompleteBatch(batchMessage, batchChunkList);
return completeBatchReader.call();
} catch (Exception e) {
throw new CommunicationException("Error reading batch");
throw new CommunicationException(e.getMessage() + " " + e.getMessage());
}
}
@Override
public BulletinBoardMessage readBatchData(BulletinBoardMessage stub) throws CommunicationException, IllegalArgumentException {
if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID){
throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
}
MessageID msgID = MessageID.newBuilder().setID(stub.getMsg().getMsgId()).build();
BatchDataReader batchDataReader = new BatchDataReader(msgID);
List<BatchChunk> batchChunkList = null;
try {
batchChunkList = batchDataReader.call();
} catch (Exception e) {
throw new CommunicationException(e.getCause() + " " + e.getMessage());
}
return BulletinBoardUtils.gatherBatch(stub, batchChunkList);
}
@Override
public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException {
return server.generateSyncQuery(generateSyncQueryParams);

27
bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerBatchIdentifier.java Normal file
View File

@ -0,0 +1,27 @@
package meerkat.bulletinboard;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import java.util.Arrays;
/**
* Created by Arbel Deutsch Peled on 17-Jun-16.
*/
public final class MultiServerBatchIdentifier implements AsyncBulletinBoardClient.BatchIdentifier {
// Per-server identifiers
private final Iterable<BatchIdentifier> identifiers;
public MultiServerBatchIdentifier(Iterable<BatchIdentifier> identifiers) {
this.identifiers = identifiers;
}
public MultiServerBatchIdentifier(BatchIdentifier[] identifiers) {
this.identifiers = Arrays.asList(identifiers);
}
public Iterable<BatchIdentifier> getIdentifiers() {
return identifiers;
}
}

10
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 final List<SingleServerBulletinBoardClient> clients;
protected final List<SingleServerBulletinBoardClient> clients;
protected AtomicInteger minServers; // The minimal number of servers the job must be successful on for the job to be completed
@ -91,14 +91,6 @@ public abstract class MultiServerWorker<IN, OUT> extends BulletinClientWorker<IN
}
}
/**
* Used by implementations to get a Single Server Client iterator
* @return the requested iterator
*/
protected Iterator<SingleServerBulletinBoardClient> getClientIterator() {
return clients.iterator();
}
protected int getClientNumber() {
return clients.size();
}

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

@ -2,12 +2,15 @@ package meerkat.bulletinboard;
import com.google.protobuf.BoolValue;
import com.google.protobuf.ByteString;
import com.google.protobuf.Int64Value;
import meerkat.bulletinboard.workers.singleserver.*;
import meerkat.comm.CommunicationException;
import meerkat.crypto.concrete.SHA256Digest;
import meerkat.protobuf.BulletinBoardAPI;
import meerkat.protobuf.BulletinBoardAPI.*;
import meerkat.protobuf.Voting.*;
import meerkat.rest.*;
import meerkat.util.BulletinBoardUtils;
import java.util.List;
@ -86,47 +89,6 @@ public class SimpleBulletinBoardClient implements BulletinBoardClient{
return MessageID.newBuilder().setID(ByteString.copyFrom(digest.digest())).build();
}
@Override
public MessageID postBatch(CompleteBatch completeBatch) throws CommunicationException {
int pos = 0;
ByteString signerID = completeBatch.getSignature().getSignerId();
int batchID = completeBatch.getBeginBatchMessage().getBatchId();
// Post message to all databases
for (String db : meerkatDBs) {
SingleServerBeginBatchWorker beginBatchWorker = new SingleServerBeginBatchWorker(db, completeBatch.getBeginBatchMessage(), 0);
beginBatchWorker.call();
BatchMessage.Builder builder = BatchMessage.newBuilder().setSignerId(signerID).setBatchId(batchID);
for (BatchChunk batchChunk : completeBatch.getBatchDataList()) {
SingleServerPostBatchWorker postBatchWorker =
new SingleServerPostBatchWorker(
db,
builder.setData(batchChunk).setSerialNum(pos).build(),
0);
postBatchWorker.call();
pos++;
}
SingleServerCloseBatchWorker closeBatchWorker = new SingleServerCloseBatchWorker(db, completeBatch.getCloseBatchMessage(), 0);
closeBatchWorker.call();
}
digest.update(completeBatch);
return digest.digestAsMessageID();
}
/**
* Access each database and search for a given message ID
* Return the number of databases in which the message was found
@ -173,17 +135,18 @@ public class SimpleBulletinBoardClient implements BulletinBoardClient{
* @return the list of Bulletin Board messages that are returned from a server
*/
@Override
public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) {
public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException{
// Replace null filter list with blank one.
if (filterList == null){
filterList = MessageFilterList.getDefaultInstance();
}
String exceptionString = "";
for (String db : meerkatDBs) {
try {
webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(READ_MESSAGES_PATH);
SingleServerReadMessagesWorker worker = new SingleServerReadMessagesWorker(db, filterList, 0);
@ -191,33 +154,92 @@ public class SimpleBulletinBoardClient implements BulletinBoardClient{
return result;
} catch (Exception ignored) {}
} catch (Exception e) {
//TODO: log
exceptionString += e.getMessage() + "\n";
}
}
return null;
throw new CommunicationException("Could not find message in any DB. Errors follow:\n" + exceptionString);
}
@Override
public CompleteBatch readBatch(BatchSpecificationMessage batchSpecificationMessage) throws CommunicationException {
public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize) throws CommunicationException {
// Create job with no retries for retrieval of the Bulletin Board Message that defines the batch
List<BatchChunk> chunkList = BulletinBoardUtils.breakToBatch(msg, chunkSize);
BeginBatchMessage beginBatchMessage = BulletinBoardUtils.generateBeginBatchMessage(msg);
boolean posted = false;
// Post message to all databases
for (String db : meerkatDBs) {
try {
int pos = 0;
SingleServerBeginBatchWorker beginBatchWorker = new SingleServerBeginBatchWorker(db, beginBatchMessage, 0);
Int64Value batchId = beginBatchWorker.call();
BatchMessage.Builder builder = BatchMessage.newBuilder().setBatchId(batchId.getValue());
for (BatchChunk batchChunk : chunkList) {
SingleServerPostBatchWorker postBatchWorker =
new SingleServerPostBatchWorker(
db,
builder.setData(batchChunk).setSerialNum(pos).build(),
0);
postBatchWorker.call();
pos++;
}
CloseBatchMessage closeBatchMessage = BulletinBoardUtils.generateCloseBatchMessage(batchId, chunkList.size(), msg);
SingleServerCloseBatchWorker closeBatchWorker = new SingleServerCloseBatchWorker(db, closeBatchMessage, 0);
closeBatchWorker.call();
posted = true;
} catch(Exception ignored) {}
}
if (!posted){
throw new CommunicationException("Could not post to any server");
}
digest.reset();
digest.update(msg);
return digest.digestAsMessageID();
}
@Override
public BulletinBoardMessage readBatch(MessageID msgID) throws CommunicationException {
MessageFilterList filterList = MessageFilterList.newBuilder()
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_TAG)
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_ID_TAG_PREFIX + batchSpecificationMessage.getBatchId())
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.SIGNER_ID)
.setId(batchSpecificationMessage.getSignerId())
.setType(FilterType.MSG_ID)
.setId(msgID.getID())
.build())
.build();
BatchQuery batchQuery = BatchQuery.newBuilder()
.setMsgID(msgID)
.setStartPosition(0)
.build();
String exceptionString = "";
for (String db : meerkatDBs) {
try {
@ -228,20 +250,57 @@ public class SimpleBulletinBoardClient implements BulletinBoardClient{
if (messages == null || messages.size() < 1)
continue;
BulletinBoardMessage batchMessage = messages.get(0);
BulletinBoardMessage stub = messages.get(0);
SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(db, batchSpecificationMessage, 0);
SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(db, batchQuery, 0);
List<BatchChunk> batchChunkList = batchWorker.call();
CompleteBatch result = new CompleteBatch(batchMessage, batchChunkList);
return BulletinBoardUtils.gatherBatch(stub, batchChunkList);
return result;
} catch (Exception ignored) {}
} catch (Exception e) {
//TODO: log
exceptionString += e.getMessage() + "\n";
}
}
return null;
throw new CommunicationException("Could not find message in any DB. Errors follow:\n" + exceptionString);
}
@Override
public BulletinBoardMessage readBatchData(BulletinBoardMessage stub) throws CommunicationException, IllegalArgumentException {
if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID){
throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
}
BatchQuery batchQuery = BatchQuery.newBuilder()
.setMsgID(MessageID.newBuilder()
.setID(stub.getMsg().getMsgId())
.build())
.setStartPosition(0)
.build();
String exceptionString = "";
for (String db : meerkatDBs) {
try {
SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(db, batchQuery, 0);
List<BatchChunk> batchChunkList = batchWorker.call();
return BulletinBoardUtils.gatherBatch(stub, batchChunkList);
} catch (Exception e) {
//TODO: log
exceptionString += e.getMessage() + "\n";
}
}
throw new CommunicationException("Could not find message in any DB. Errors follow:\n" + exceptionString);
}

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

@ -137,22 +137,13 @@ public class SimpleBulletinBoardSynchronizer implements BulletinBoardSynchronize
try {
if (message.getMsg().getTagList().contains(BulletinBoardConstants.BATCH_TAG)) {
if (message.getMsg().getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {
// This is a batch message: need to upload batch data as well as the message itself
ByteString signerID = message.getSig(0).getSignerId();
int batchID = Integer.parseInt(BulletinBoardUtils.findTagWithPrefix(message, BulletinBoardConstants.BATCH_ID_TAG_PREFIX));
BatchSpecificationMessage batchSpecificationMessage = BatchSpecificationMessage.newBuilder()
.setSignerId(signerID)
.setBatchId(batchID)
.setStartPosition(0)
.build();
BulletinBoardMessage completeMsg = localClient.readBatchData(message);
CompleteBatch completeBatch = localClient.readBatch(batchSpecificationMessage);
remoteClient.postBatch(completeBatch, new MessageDeleteCallback(message.getEntryNum(), syncStatusUpdateCallback));
remoteClient.postMessage(completeMsg, new MessageDeleteCallback(message.getEntryNum(), syncStatusUpdateCallback));
} else {

42
bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBatchIdentifier.java Normal file
View File

@ -0,0 +1,42 @@
package meerkat.bulletinboard;
import com.google.protobuf.Int64Value;
/**
* Created by Arbel Deutsch Peled on 16-Jun-16.
* Single-server implementation of the BatchIdentifier interface
*/
final class SingleServerBatchIdentifier implements AsyncBulletinBoardClient.BatchIdentifier {
private final Int64Value batchId;
private int length;
public SingleServerBatchIdentifier(Int64Value batchId) {
this.batchId = batchId;
length = 0;
}
public SingleServerBatchIdentifier(long batchId) {
this(Int64Value.newBuilder().setValue(batchId).build());
}
public Int64Value getBatchId() {
return batchId;
}
/**
* Overrides the existing length with the new one only if the new length is longer
* @param newLength
*/
public void setLength(int newLength) {
if (newLength > length) {
length = newLength;
}
}
public int getLength() {
return length;
}
}

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

@ -4,11 +4,15 @@ import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.ByteString;
import com.google.protobuf.Int64Value;
import com.google.protobuf.Timestamp;
import meerkat.bulletinboard.workers.singleserver.*;
import meerkat.protobuf.BulletinBoardAPI.*;
import meerkat.protobuf.Crypto;
import meerkat.protobuf.Voting.BulletinBoardClientParams;
import meerkat.util.BulletinBoardUtils;
import java.lang.Iterable;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@ -124,14 +128,14 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
* It reports success back to the user only if all of the batch-data were successfully posted
* If any batch-data fails to post: this callback reports failure
*/
class PostBatchDataListCallback implements FutureCallback<Boolean> {
class PostBatchChunkListCallback implements FutureCallback<Boolean> {
private final FutureCallback<Boolean> callback;
private AtomicInteger batchDataRemaining;
private AtomicBoolean aggregatedResult;
public PostBatchDataListCallback(int batchDataLength, FutureCallback<Boolean> callback) {
public PostBatchChunkListCallback(int batchDataLength, FutureCallback<Boolean> callback) {
this.callback = callback;
this.batchDataRemaining = new AtomicInteger(batchDataLength);
@ -169,15 +173,15 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
*/
class CompleteBatchReadCallback {
private final FutureCallback<CompleteBatch> callback;
private final FutureCallback<BulletinBoardMessage> callback;
private List<BatchChunk> batchChunkList;
private BulletinBoardMessage batchMessage;
private BulletinBoardMessage stub;
private AtomicInteger remainingQueries;
private AtomicBoolean failed;
public CompleteBatchReadCallback(FutureCallback<CompleteBatch> callback) {
public CompleteBatchReadCallback(FutureCallback<BulletinBoardMessage> callback) {
this.callback = callback;
@ -188,14 +192,10 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
protected void combineAndReturn() {
final String[] prefixes = {
BulletinBoardConstants.BATCH_ID_TAG_PREFIX,
BulletinBoardConstants.BATCH_TAG};
if (remainingQueries.decrementAndGet() == 0){
if (callback != null)
callback.onSuccess(new CompleteBatch(batchMessage, batchChunkList));
callback.onSuccess(BulletinBoardUtils.gatherBatch(stub, batchChunkList));
}
}
@ -241,7 +241,7 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
return;
}
batchMessage = result.get(0);
stub = result.get(0);
combineAndReturn();
}
@ -367,18 +367,22 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
private class PostBatchDataCallback implements FutureCallback<Boolean> {
private final CompleteBatch completeBatch;
private final BulletinBoardMessage msg;
private final BatchIdentifier identifier;
private final FutureCallback<Boolean> callback;
public PostBatchDataCallback(CompleteBatch completeBatch, FutureCallback<Boolean> callback) {
this.completeBatch = completeBatch;
public PostBatchDataCallback(BulletinBoardMessage msg, BatchIdentifier identifier, FutureCallback<Boolean> callback) {
this.msg = msg;
this.identifier = identifier;
this.callback = callback;
}
@Override
public void onSuccess(Boolean msg) {
public void onSuccess(Boolean result) {
closeBatch(
completeBatch.getCloseBatchMessage(),
identifier,
msg.getMsg().getTimestamp(),
msg.getSigList(),
callback
);
}
@ -391,25 +395,28 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
}
private class BeginBatchCallback implements FutureCallback<Boolean> {
private class ContinueBatchCallback implements FutureCallback<BatchIdentifier> {
private final CompleteBatch completeBatch;
private final BulletinBoardMessage msg;
private final int chunkSize;
private final FutureCallback<Boolean> callback;
public BeginBatchCallback(CompleteBatch completeBatch, FutureCallback<Boolean> callback) {
this.completeBatch = completeBatch;
public ContinueBatchCallback(BulletinBoardMessage msg, int chunkSize, FutureCallback<Boolean> callback) {
this.msg = msg;
this.chunkSize = chunkSize;
this.callback = callback;
}
@Override
public void onSuccess(Boolean msg) {
public void onSuccess(BatchIdentifier identifier) {
List<BatchChunk> batchChunkList = BulletinBoardUtils.breakToBatch(msg, chunkSize);
postBatchData(
completeBatch.getBeginBatchMessage().getSignerId(),
completeBatch.getBeginBatchMessage().getBatchId(),
completeBatch.getBatchDataList(),
identifier,
batchChunkList,
0,
new PostBatchDataCallback(completeBatch,callback));
new PostBatchDataCallback(msg, identifier, callback));
}
@Override
@ -420,45 +427,80 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
}
@Override
public MessageID postBatch(CompleteBatch completeBatch, FutureCallback<Boolean> callback) {
public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback<Boolean> callback) {
beginBatch(
completeBatch.getBeginBatchMessage(),
new BeginBatchCallback(completeBatch, callback)
msg.getMsg().getTagList(),
new ContinueBatchCallback(msg, chunkSize, callback)
);
digest.update(completeBatch);
digest.update(msg);
return digest.digestAsMessageID();
}
private class BeginBatchCallback implements FutureCallback<Int64Value> {
private final FutureCallback<BatchIdentifier> callback;
public BeginBatchCallback(FutureCallback<BatchIdentifier> callback) {
this.callback = callback;
}
@Override
public void onSuccess(Int64Value result) {
callback.onSuccess(new SingleServerBatchIdentifier(result));
}
@Override
public void onFailure(Throwable t) {
callback.onFailure(t);
}
}
@Override
public void beginBatch(BeginBatchMessage beginBatchMessage, FutureCallback<Boolean> callback) {
public void beginBatch(Iterable<String> tags, FutureCallback<BatchIdentifier> callback) {
BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder()
.addAllTag(tags)
.build();
// Create worker with redundancy 1 and MAX_RETRIES retries
SingleServerBeginBatchWorker worker =
new SingleServerBeginBatchWorker(meerkatDBs.get(0), beginBatchMessage, MAX_RETRIES);
// Submit worker and create callback
scheduleWorker(worker, new RetryCallback<>(worker, callback));
scheduleWorker(worker, new RetryCallback<>(worker, new BeginBatchCallback(callback)));
}
@Override
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback) {
BatchMessage.Builder builder = BatchMessage.newBuilder()
.setSignerId(signerId)
.setBatchId(batchId);
@Override
public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback) throws IllegalArgumentException{
// Cast identifier to usable form
if (!(batchIdentifier instanceof SingleServerBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchIdentifier;
// Update batch size
identifier.setLength(startPosition + batchChunkList.size());
// Create a unified callback to aggregate successful posts
PostBatchDataListCallback listCallback = new PostBatchDataListCallback(batchChunkList.size(), callback);
PostBatchChunkListCallback listCallback = new PostBatchChunkListCallback(batchChunkList.size(), callback);
// Iterate through data list
BatchMessage.Builder builder = BatchMessage.newBuilder()
.setBatchId(identifier.getBatchId().getValue());
for (BatchChunk data : batchChunkList) {
builder.setSerialNum(startPosition).setData(data);
@ -476,29 +518,29 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
}
@Override
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) {
public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback)
throws IllegalArgumentException {
postBatchData(signerId, batchId, batchChunkList, 0, callback);
postBatchData(batchIdentifier, batchChunkList, 0, callback);
}
@Override
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback) {
public void closeBatch(BatchIdentifier batchIdentifier, Timestamp timestamp, Iterable<Crypto.Signature> signatures, FutureCallback<Boolean> callback)
throws IllegalArgumentException {
postBatchData(ByteString.copyFrom(signerId), batchId, batchChunkList, startPosition, callback);
if (!(batchIdentifier instanceof SingleServerBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
}
SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchIdentifier;
@Override
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) {
postBatchData(signerId, batchId, batchChunkList, 0, callback);
}
@Override
public void closeBatch(CloseBatchMessage closeBatchMessage, FutureCallback<Boolean> callback) {
CloseBatchMessage closeBatchMessage = CloseBatchMessage.newBuilder()
.setBatchId(identifier.getBatchId().getValue())
.setBatchLength(identifier.getLength())
.setTimestamp(timestamp)
.addAllSig(signatures)
.build();
// Create worker with redundancy 1 and MAX_RETRIES retries
SingleServerCloseBatchWorker worker =
@ -532,29 +574,26 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
}
@Override
public void readBatch(BatchSpecificationMessage batchSpecificationMessage, FutureCallback<CompleteBatch> callback) {
public void readBatch(MessageID msgID, FutureCallback<BulletinBoardMessage> callback) {
// Create job with no retries for retrieval of the Bulletin Board Message that defines the batch
// Create job with MAX retries for retrieval of the Bulletin Board Message that defines the batch
MessageFilterList filterList = MessageFilterList.newBuilder()
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_TAG)
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.TAG)
.setTag(BulletinBoardConstants.BATCH_ID_TAG_PREFIX + batchSpecificationMessage.getBatchId())
.build())
.addFilter(MessageFilter.newBuilder()
.setType(FilterType.SIGNER_ID)
.setId(batchSpecificationMessage.getSignerId())
.setType(FilterType.MSG_ID)
.setId(msgID.getID())
.build())
.build();
SingleServerReadMessagesWorker messageWorker = new SingleServerReadMessagesWorker(meerkatDBs.get(0), filterList, 1);
BatchQuery batchQuery = BatchQuery.newBuilder()
.setMsgID(msgID)
.setStartPosition(0)
.build();
// Create job with no retries for retrieval of the Batch Data List
SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(meerkatDBs.get(0), batchSpecificationMessage, 1);
SingleServerReadMessagesWorker messageWorker = new SingleServerReadMessagesWorker(meerkatDBs.get(0), filterList, MAX_RETRIES);
// Create job with MAX retries for retrieval of the Batch Data List
SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(meerkatDBs.get(0), batchQuery, MAX_RETRIES);
// Create callback that will combine the two worker products
CompleteBatchReadCallback completeBatchReadCallback = new CompleteBatchReadCallback(callback);
@ -565,6 +604,49 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i
}
private class ReadBatchCallback implements FutureCallback<List<BatchChunk>> {
private final BulletinBoardMessage stub;
private final FutureCallback<BulletinBoardMessage> callback;
public ReadBatchCallback(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) {
this.stub = stub;
this.callback = callback;
}
@Override
public void onSuccess(List<BatchChunk> result) {
callback.onSuccess(BulletinBoardUtils.gatherBatch(stub, result));
}
@Override
public void onFailure(Throwable t) {
}
}
@Override
public void readBatchData(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) throws IllegalArgumentException{
if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {
throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
}
// Create job with MAX retries for retrieval of the Batch Data List
BatchQuery batchQuery = BatchQuery.newBuilder()
.setMsgID(MessageID.newBuilder()
.setID(stub.getMsg().getMsgId())
.build())
.setStartPosition(0)
.build();
SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(meerkatDBs.get(0), batchQuery, MAX_RETRIES);
scheduleWorker(batchWorker, new RetryCallback<>(batchWorker, new ReadBatchCallback(stub, callback)));
}
@Override
public void querySync(SyncQuery syncQuery, FutureCallback<SyncQueryResponse> callback) {

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

@ -1,10 +1,12 @@
package meerkat.bulletinboard;
import com.google.common.util.concurrent.FutureCallback;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import meerkat.bulletinboard.workers.multiserver.*;
import meerkat.protobuf.BulletinBoardAPI;
import meerkat.protobuf.BulletinBoardAPI.*;
import meerkat.protobuf.Crypto.Signature;
import meerkat.protobuf.Voting.*;
import java.util.ArrayList;
@ -41,6 +43,7 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple
private int minAbsoluteRedundancy;
/**
* Stores database locations and initializes the web Client
* Stores the required minimum redundancy.
@ -98,28 +101,28 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple
}
@Override
public MessageID postBatch(BulletinBoardMessage completeBatch, FutureCallback<Boolean> callback) {
public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback<Boolean> callback) {
// Create job
MultiServerPostBatchWorker worker =
new MultiServerPostBatchWorker(clients, minAbsoluteRedundancy, completeBatch, POST_MESSAGE_RETRY_NUM, callback);
new MultiServerPostBatchWorker(clients, minAbsoluteRedundancy, msg, chunkSize, POST_MESSAGE_RETRY_NUM, callback);
// Submit job
executorService.submit(worker);
// Calculate the correct message ID and return it
batchDigest.reset();
batchDigest.update(completeBatch);
batchDigest.update(msg);
return batchDigest.digestAsMessageID();
}
@Override
public void beginBatch(BeginBatchMessage beginBatchMessage, FutureCallback<Boolean> callback) {
public void beginBatch(Iterable<String> tags, FutureCallback<BatchIdentifier> callback) {
// Create job
MultiServerBeginBatchWorker worker =
new MultiServerBeginBatchWorker(clients, minAbsoluteRedundancy, beginBatchMessage, POST_MESSAGE_RETRY_NUM, callback);
new MultiServerBeginBatchWorker(clients, minAbsoluteRedundancy, tags, POST_MESSAGE_RETRY_NUM, callback);
// Submit job
executorService.submit(worker);
@ -127,10 +130,18 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple
}
@Override
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback) {
public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback) throws IllegalArgumentException {
BatchDataContainer batchDataContainer = new BatchDataContainer(signerId, batchId, batchChunkList, startPosition);
// Cast identifier to usable form
if (!(batchIdentifier instanceof MultiServerBatchIdentifier)){
throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
}
MultiServerBatchIdentifier identifier = (MultiServerBatchIdentifier) batchIdentifier;
BatchDataContainer batchDataContainer = new BatchDataContainer(identifier, batchChunkList, startPosition);
// Create job
MultiServerPostBatchDataWorker worker =
@ -142,33 +153,19 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple
}
@Override
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) {
public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback)
throws IllegalArgumentException {
postBatchData(signerId, batchId, batchChunkList, 0, callback);
postBatchData(batchIdentifier, batchChunkList, 0, callback);
}
@Override
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback) {
postBatchData(signerId.toByteArray(), batchId, batchChunkList, startPosition, callback);
}
@Override
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) {
postBatchData(signerId, batchId, batchChunkList, 0, callback);
}
@Override
public void closeBatch(CloseBatchMessage closeBatchMessage, FutureCallback<Boolean> callback) {
public void closeBatch(BatchIdentifier payload, Timestamp timestamp, Iterable<Signature> signatures, FutureCallback<Boolean> callback) {
// Create job
MultiServerCloseBatchWorker worker =
new MultiServerCloseBatchWorker(clients, minAbsoluteRedundancy, closeBatchMessage, POST_MESSAGE_RETRY_NUM, callback);
new MultiServerCloseBatchWorker(clients, minAbsoluteRedundancy, payload, timestamp, signatures, POST_MESSAGE_RETRY_NUM, callback);
// Submit job
executorService.submit(worker);
@ -211,11 +208,29 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple
}
@Override
public void readBatch(BatchSpecificationMessage batchSpecificationMessage, FutureCallback<BulletinBoardMessage> callback) {
public void readBatch(MessageID msgID, FutureCallback<BulletinBoardMessage> callback) {
//Create job
MultiServerReadBatchWorker worker =
new MultiServerReadBatchWorker(clients, minAbsoluteRedundancy, msgID, READ_MESSAGES_RETRY_NUM, callback);
// Submit job
executorService.submit(worker);
}
@Override
public void readBatchData(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) throws IllegalArgumentException {
if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {
throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
}
MessageID msgID = MessageID.newBuilder().setID(stub.getMsg().getMsgId()).build();
// Create job
MultiServerReadBatchWorker worker =
new MultiServerReadBatchWorker(clients, minAbsoluteRedundancy, batchSpecificationMessage, READ_MESSAGES_RETRY_NUM, callback);
MultiServerReadBatchDataWorker worker =
new MultiServerReadBatchDataWorker(clients, minAbsoluteRedundancy, msgID, READ_MESSAGES_RETRY_NUM, callback);
// Submit job
executorService.submit(worker);
@ -251,4 +266,3 @@ public class ThreadedBulletinBoardClient extends SimpleBulletinBoardClient imple
}
}
==== BASE ====

80
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerBeginBatchWorker.java
View File

@ -1,28 +1,96 @@
package meerkat.bulletinboard.workers.multiserver;
import com.google.common.util.concurrent.FutureCallback;
import meerkat.bulletinboard.MultiServerBatchIdentifier;
import meerkat.bulletinboard.MultiServerWorker;
import meerkat.bulletinboard.SingleServerBulletinBoardClient;
import meerkat.protobuf.BulletinBoardAPI.BeginBatchMessage;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import meerkat.comm.CommunicationException;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Created by Arbel Deutsch Peled on 27-Dec-15.
*/
public class MultiServerBeginBatchWorker extends MultiServerGenericPostWorker<BeginBatchMessage> {
public class MultiServerBeginBatchWorker extends MultiServerWorker<Iterable<String>, BatchIdentifier> {
private BatchIdentifier[] identifiers;
private AtomicInteger remainingServers;
public MultiServerBeginBatchWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, BeginBatchMessage payload, int maxRetry,
FutureCallback<Boolean> futureCallback) {
int minServers, Iterable<String> payload, int maxRetry,
FutureCallback<BatchIdentifier> futureCallback) {
super(clients, minServers, payload, maxRetry, futureCallback);
identifiers = new BatchIdentifier[clients.size()];
for (int i = 0 ; i < identifiers.length ; i++) {
identifiers[i] = null;
}
remainingServers = new AtomicInteger(clients.size());
}
private class BeginBatchCallback implements FutureCallback<BatchIdentifier> {
private final int clientNum;
public BeginBatchCallback(int clientNum) {
this.clientNum = clientNum;
}
private void finishPost() {
if (remainingServers.decrementAndGet() <= 0){
if (minServers.get() <= 0) {
MultiServerBeginBatchWorker.this.onSuccess(new MultiServerBatchIdentifier(identifiers));
} else {
MultiServerBeginBatchWorker.this.onFailure(new CommunicationException("Could not open batch in enough servers"));
}
}
}
@Override
public void onSuccess(BatchIdentifier result) {
identifiers[clientNum] = result;
finishPost();
}
@Override
public void onFailure(Throwable t) {
finishPost();
}
}
@Override
protected void doPost(SingleServerBulletinBoardClient client, BeginBatchMessage payload) {
client.beginBatch(payload, this);
public void onSuccess(BatchIdentifier result) {
succeed(result);
}
@Override
public void onFailure(Throwable t) {
fail(t);
}
@Override
public void run() {
int clientNum = 0;
for (SingleServerBulletinBoardClient client : clients){
client.beginBatch(payload, new BeginBatchCallback(clientNum));
clientNum++;
}
}
}

22
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerCloseBatchWorker.java
View File

@ -1,27 +1,35 @@
package meerkat.bulletinboard.workers.multiserver;
import com.google.common.util.concurrent.FutureCallback;
import com.google.protobuf.Timestamp;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import meerkat.bulletinboard.SingleServerBulletinBoardClient;
import meerkat.protobuf.BulletinBoardAPI.CloseBatchMessage;
import meerkat.protobuf.Crypto.Signature;
import java.util.List;
/**
* Created by Arbel Deutsch Peled on 27-Dec-15.
*/
public class MultiServerCloseBatchWorker extends MultiServerGenericPostWorker<CloseBatchMessage> {
public class MultiServerCloseBatchWorker extends MultiServerGenericPostWorker<BatchIdentifier> {
public MultiServerCloseBatchWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, CloseBatchMessage payload, int maxRetry,
FutureCallback<Boolean> futureCallback) {
private final Timestamp timestamp;
private final Iterable<Signature> signatures;
public MultiServerCloseBatchWorker(List<SingleServerBulletinBoardClient> clients, int minServers,
BatchIdentifier payload, Timestamp timestamp, Iterable<Signature> signatures,
int maxRetry, FutureCallback<Boolean> futureCallback) {
super(clients, minServers, payload, maxRetry, futureCallback);
this.timestamp = timestamp;
this.signatures = signatures;
}
@Override
protected void doPost(SingleServerBulletinBoardClient client, CloseBatchMessage payload) {
client.closeBatch(payload, this);
protected void doPost(SingleServerBulletinBoardClient client, BatchIdentifier payload) {
client.closeBatch(payload, timestamp, signatures, this);
}

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

@ -35,14 +35,9 @@ public abstract class MultiServerGenericPostWorker<T> extends MultiServerWorker<
public void run() {
// Iterate through servers
Iterator<SingleServerBulletinBoardClient> clientIterator = getClientIterator();
while (clientIterator.hasNext()) {
for (SingleServerBulletinBoardClient client : clients) {
// Send request to Server
SingleServerBulletinBoardClient client = clientIterator.next();
doPost(client, payload);
}

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

@ -14,7 +14,9 @@ import java.util.List;
*/
public abstract class MultiServerGenericReadWorker<IN, OUT> extends MultiServerWorker<IN, OUT>{
private final Iterator<SingleServerBulletinBoardClient> clientIterator;
private Iterator<SingleServerBulletinBoardClient> clientIterator;
private String errorString;
public MultiServerGenericReadWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, IN payload, int maxRetry,
@ -22,7 +24,8 @@ public abstract class MultiServerGenericReadWorker<IN, OUT> extends MultiServerW
super(clients, true, minServers, payload, maxRetry, futureCallback); // Shuffle clients on creation to balance load
clientIterator = getClientIterator();
clientIterator = clients.iterator();
errorString = "";
}
@ -46,7 +49,7 @@ public abstract class MultiServerGenericReadWorker<IN, OUT> extends MultiServerW
doRead(payload, client);
} else {
fail(new CommunicationException("Could not contact any server"));
fail(new CommunicationException("Could not contact any server. Errors follow:\n" + errorString));
}
}
@ -58,6 +61,8 @@ public abstract class MultiServerGenericReadWorker<IN, OUT> extends MultiServerW
@Override
public void onFailure(Throwable t) {
//TODO: log
errorString += t.getCause() + " " + t.getMessage() + "\n";
run(); // Retry with next server
}

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

@ -36,13 +36,8 @@ public class MultiServerGetRedundancyWorker extends MultiServerWorker<MessageID,
*/
public void run(){
Iterator<SingleServerBulletinBoardClient> clientIterator = getClientIterator();
// Iterate through clients
while (clientIterator.hasNext()) {
SingleServerBulletinBoardClient client = clientIterator.next();
for (SingleServerBulletinBoardClient client : clients) {
// Send request to client
client.getRedundancy(payload,this);

50
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchDataWorker.java
View File

@ -1,15 +1,18 @@
package meerkat.bulletinboard.workers.multiserver;
import com.google.common.util.concurrent.FutureCallback;
import meerkat.bulletinboard.AsyncBulletinBoardClient.BatchIdentifier;
import meerkat.bulletinboard.MultiServerWorker;
import meerkat.bulletinboard.SingleServerBulletinBoardClient;
import meerkat.bulletinboard.BatchDataContainer;
import java.util.Iterator;
import java.util.List;
/**
* Created by Arbel Deutsch Peled on 27-Dec-15.
*/
public class MultiServerPostBatchDataWorker extends MultiServerGenericPostWorker<BatchDataContainer> {
public class MultiServerPostBatchDataWorker extends MultiServerWorker<BatchDataContainer, Boolean> {
public MultiServerPostBatchDataWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, BatchDataContainer payload, int maxRetry,
@ -20,9 +23,50 @@ public class MultiServerPostBatchDataWorker extends MultiServerGenericPostWorker
}
@Override
protected void doPost(SingleServerBulletinBoardClient client, BatchDataContainer payload) {
client.postBatchData(payload.signerId, payload.batchId, payload.batchChunkList, payload.startPosition, this);
public void run() {
Iterator<BatchIdentifier> identifierIterator = payload.batchId.getIdentifiers().iterator();
// Iterate through client
for (SingleServerBulletinBoardClient client : clients) {
if (identifierIterator.hasNext()) {
// Fetch the batch identifier supplied by the specific client (may be null if batch open failed on client
BatchIdentifier identifier = identifierIterator.next();
if (identifier != null) {
// Post the data with the matching identifier to the client
client.postBatchData(identifierIterator.next(), payload.batchChunkList, payload.startPosition, this);
} else {
// Count servers with no batch identifier as failed
maxFailedServers.decrementAndGet();
}
}
}
}
@Override
public void onSuccess(Boolean result) {
if (minServers.decrementAndGet() <= 0){
succeed(result);
}
}
@Override
public void onFailure(Throwable t) {
if (maxFailedServers.decrementAndGet() <= 0){
fail(t);
}
}
}

10
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchWorker.java
View File

@ -11,17 +11,23 @@ import java.util.List;
*/
public class MultiServerPostBatchWorker extends MultiServerGenericPostWorker<BulletinBoardMessage> {
private final int chunkSize;
public MultiServerPostBatchWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, BulletinBoardMessage payload, int maxRetry,
int minServers, BulletinBoardMessage payload, int chunkSize, int maxRetry,
FutureCallback<Boolean> futureCallback) {
super(clients, minServers, payload, maxRetry, futureCallback);
this.chunkSize = chunkSize;
}
@Override
protected void doPost(SingleServerBulletinBoardClient client, BulletinBoardMessage payload) {
client.postBatch(payload, this);
client.postAsBatch(payload, chunkSize, this);
}

29
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchDataWorker.java Normal file
View File

@ -0,0 +1,29 @@
package meerkat.bulletinboard.workers.multiserver;
import com.google.common.util.concurrent.FutureCallback;
import meerkat.bulletinboard.SingleServerBulletinBoardClient;
import meerkat.protobuf.BulletinBoardAPI.*;
import java.util.List;
/**
* Created by Arbel Deutsch Peled on 27-Dec-15.
*/
public class MultiServerReadBatchDataWorker extends MultiServerGenericReadWorker<MessageID, BulletinBoardMessage> {
public MultiServerReadBatchDataWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, MessageID payload, int maxRetry,
FutureCallback<BulletinBoardMessage> futureCallback) {
super(clients, minServers, payload, maxRetry, futureCallback);
}
@Override
protected void doRead(MessageID payload, SingleServerBulletinBoardClient client) {
client.readBatch(payload, this);
}
}

8
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchWorker.java
View File

@ -3,7 +3,7 @@ package meerkat.bulletinboard.workers.multiserver;
import com.google.common.util.concurrent.FutureCallback;
import meerkat.bulletinboard.SingleServerBulletinBoardClient;
import meerkat.protobuf.BulletinBoardAPI.BulletinBoardMessage;
import meerkat.protobuf.BulletinBoardAPI.BatchSpecificationMessage;
import meerkat.protobuf.BulletinBoardAPI.MessageID;
import java.util.List;
@ -11,10 +11,10 @@ import java.util.List;
/**
* Created by Arbel Deutsch Peled on 27-Dec-15.
*/
public class MultiServerReadBatchWorker extends MultiServerGenericReadWorker<BatchSpecificationMessage, BulletinBoardMessage> {
public class MultiServerReadBatchWorker extends MultiServerGenericReadWorker<MessageID, BulletinBoardMessage> {
public MultiServerReadBatchWorker(List<SingleServerBulletinBoardClient> clients,
int minServers, BatchSpecificationMessage payload, int maxRetry,
int minServers, MessageID payload, int maxRetry,
FutureCallback<BulletinBoardMessage> futureCallback) {
super(clients, minServers, payload, maxRetry, futureCallback);
@ -22,7 +22,7 @@ public class MultiServerReadBatchWorker extends MultiServerGenericReadWorker<Bat
}
@Override
protected void doRead(BatchSpecificationMessage payload, SingleServerBulletinBoardClient client) {
protected void doRead(MessageID payload, SingleServerBulletinBoardClient client) {
client.readBatch(payload, this);
}

37
bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerBeginBatchWorker.java
View File

@ -1,17 +1,50 @@
package meerkat.bulletinboard.workers.singleserver;
import com.google.protobuf.Int64Value;
import meerkat.bulletinboard.SingleServerWorker;
import meerkat.comm.CommunicationException;
import meerkat.protobuf.BulletinBoardAPI.BeginBatchMessage;
import meerkat.rest.Constants;
import javax.ws.rs.ProcessingException;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Response;
import static meerkat.bulletinboard.BulletinBoardConstants.BEGIN_BATCH_PATH;
import static meerkat.bulletinboard.BulletinBoardConstants.BULLETIN_BOARD_SERVER_PATH;
/**
* Created by Arbel Deutsch Peled on 27-Dec-15.
* Tries to contact server once and perform a post operation
*/
public class SingleServerBeginBatchWorker extends SingleServerGenericPostWorker<BeginBatchMessage> {
public class SingleServerBeginBatchWorker extends SingleServerWorker<BeginBatchMessage,Int64Value> {
public SingleServerBeginBatchWorker(String serverAddress, BeginBatchMessage payload, int maxRetry) {
super(serverAddress, BEGIN_BATCH_PATH, payload, maxRetry);
super(serverAddress, payload, maxRetry);
}
@Override
public Int64Value call() throws Exception {
Client client = clientLocal.get();
WebTarget webTarget = client.target(serverAddress).path(BULLETIN_BOARD_SERVER_PATH).path(BEGIN_BATCH_PATH);
Response response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(
Entity.entity(payload, Constants.MEDIATYPE_PROTOBUF));
try {
return response.readEntity(Int64Value.class);
} catch (ProcessingException | IllegalStateException e) {
// Post to this server failed
throw new CommunicationException("Could not contact the server");
}
finally {
response.close();
}
}
}

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

@ -17,14 +17,12 @@ import java.util.List;
import static meerkat.bulletinboard.BulletinBoardConstants.BULLETIN_BOARD_SERVER_PATH;
import static meerkat.bulletinboard.BulletinBoardConstants.READ_BATCH_PATH;
import static meerkat.bulletinboard.BulletinBoardConstants.BATCH_ID_TAG_PREFIX;
/**
* Created by Arbel Deutsch Peled on 27-Dec-15.
*/
public class SingleServerReadBatchWorker extends SingleServerWorker<BatchSpecificationMessage, List<BatchChunk>> {
public class SingleServerReadBatchWorker extends SingleServerWorker<BatchQuery, List<BatchChunk>> {
public SingleServerReadBatchWorker(String serverAddress, BatchSpecificationMessage payload, int maxRetry) {
public SingleServerReadBatchWorker(String serverAddress, BatchQuery payload, int maxRetry) {
super(serverAddress, payload, maxRetry);
}

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

@ -118,58 +118,43 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
new int[] {}
),
GET_BATCH_MESSAGE_ENTRY(
new String[] {"SignerId", "BatchId"},
new int[] {Types.BLOB, Types.INTEGER}
),
CHECK_BATCH_LENGTH(
new String[] {"SignerId", "BatchId"},
new String[] {"BatchId"},
new int[] {Types.BLOB, Types.INTEGER}
),
CHECK_BATCH_OPEN(
new String[] {"SignerId", "BatchId"},
new String[] {"BatchId"},
new int[] {Types.BLOB, Types.INTEGER}
),
GET_BATCH_MESSAGE_DATA(
GET_BATCH_MESSAGE_DATA_BY_MSG_ID(
new String[] {"MsgId", "StartPosition"},
new int[] {Types.BLOB, Types.INTEGER}
),
GET_BATCH_MESSAGE_DATA_BY_IDS(
new String[] {"SignerId", "BatchId", "StartPosition"},
new int[] {Types.BLOB, Types.INTEGER, Types.INTEGER}
GET_BATCH_MESSAGE_DATA_BY_BATCH_ID(
new String[] {"BatchId", "StartPosition"},
new int[] {Types.INTEGER, Types.INTEGER}
),
INSERT_BATCH_DATA(
new String[] {"SignerId", "BatchId", "SerialNum", "Data"},
new int[] {Types.BLOB, Types.INTEGER, Types.INTEGER, Types.BLOB}
new String[] {"BatchId", "SerialNum", "Data"},
new int[] {Types.INTEGER, Types.INTEGER, Types.BLOB}
),
STORE_BATCH_TAGS(
new String[] {"SignerId", "BatchId", "Tags"},
new int[] {Types.BLOB, Types.INTEGER, Types.BLOB}
new String[] {"Tags"},
new int[] {Types.BLOB}
),
GET_BATCH_TAGS(
new String[] {"SignerId", "BatchId"},
new int[] {Types.BLOB, Types.INTEGER}
),
REMOVE_BATCH_TAGS(
new String[] {"SignerId", "BatchId"},
new int[] {Types.BLOB, Types.INTEGER}
),
REMOVE_BATCH_IDS(
new String[] {"SignerId", "BatchId"},
new String[] {"BatchId"},
new int[] {Types.BLOB, Types.INTEGER}
),
ADD_ENTRY_NUM_TO_BATCH(
new String[] {"SignerId", "BatchId", "EntryNum"},
new String[] {"BatchId", "EntryNum"},
new int[] {Types.BLOB, Types.INTEGER, Types.INTEGER}
);
@ -783,23 +768,21 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
return false;
}
return messages.get(0).getMsg().getIsStub();
return (messages.get(0).getMsg().getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.MSGID);
}
/**
* This method checks if a specified batch exists and is still open
* @param signerId is the signer ID
* @param batchId is the batch ID
* @param batchId is the temporary batch ID
* @return TRUE if the batch is closed and FALSE if it is still open or doesn't exist at all
*/
private boolean isBatchOpen(ByteString signerId, int batchId) throws CommunicationException {
private boolean isBatchOpen(long batchId) throws CommunicationException {
String sql = sqlQueryProvider.getSQLString(QueryType.CHECK_BATCH_OPEN);
MapSqlParameterSource namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.CHECK_BATCH_OPEN.getParamName(0),signerId.toByteArray());
namedParameters.addValue(QueryType.CHECK_BATCH_OPEN.getParamName(1),batchId);
namedParameters.addValue(QueryType.CHECK_BATCH_OPEN.getParamName(0),batchId);
List<Long> result = jdbcTemplate.query(sql, namedParameters, new LongMapper());
@ -808,24 +791,22 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
}
@Override
public BoolValue beginBatch(BeginBatchMessage message) throws CommunicationException {
// Check if batch is already open
if (isBatchOpen(message.getSignerId(), message.getBatchId())) {
return BoolValue.newBuilder().setValue(false).build();
}
public Int64Value beginBatch(BeginBatchMessage message) throws CommunicationException {
// Store tags
String sql = sqlQueryProvider.getSQLString(QueryType.STORE_BATCH_TAGS);
MapSqlParameterSource namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.STORE_BATCH_TAGS.getParamName(0),message.getSignerId().toByteArray());
namedParameters.addValue(QueryType.STORE_BATCH_TAGS.getParamName(1),message.getBatchId());
namedParameters.addValue(QueryType.STORE_BATCH_TAGS.getParamName(2),message.toByteArray());
namedParameters.addValue(QueryType.STORE_BATCH_TAGS.getParamName(0),message.toByteArray());
jdbcTemplate.update(sql,namedParameters);
return BoolValue.newBuilder().setValue(true).build();
KeyHolder keyHolder = new GeneratedKeyHolder();
jdbcTemplate.update(sql, namedParameters, keyHolder);
long entryNum = keyHolder.getKey().longValue();
return Int64Value.newBuilder().setValue(entryNum).build();
}
@ -834,7 +815,7 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
public BoolValue postBatchMessage(BatchMessage batchMessage) throws CommunicationException{
// Make sure batch is open
if (!isBatchOpen(batchMessage.getSignerId(), batchMessage.getBatchId())) {
if (!isBatchOpen(batchMessage.getBatchId())) {
return BoolValue.newBuilder().setValue(false).build();
}
@ -842,10 +823,9 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
String sql = sqlQueryProvider.getSQLString(QueryType.INSERT_BATCH_DATA);
MapSqlParameterSource namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(0),batchMessage.getSignerId().toByteArray());
namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(1),batchMessage.getBatchId());
namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(2),batchMessage.getSerialNum());
namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(3),batchMessage.getData().toByteArray());
namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(0),batchMessage.getBatchId());
namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(1),batchMessage.getSerialNum());
namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(2),batchMessage.getData().toByteArray());
jdbcTemplate.update(sql, namedParameters);
@ -857,8 +837,6 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
@Override
public BoolValue closeBatch(CloseBatchMessage message) throws CommunicationException {
ByteString signerId = message.getSig(0).getSignerId();
int batchId = message.getBatchId();
// Check batch size
@ -866,8 +844,7 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
MapSqlParameterSource namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.CHECK_BATCH_LENGTH.getParamName(0),signerId.toByteArray());
namedParameters.addValue(QueryType.CHECK_BATCH_LENGTH.getParamName(1),batchId);
namedParameters.addValue(QueryType.CHECK_BATCH_LENGTH.getParamName(0),message.getBatchId());
List<Long> lengthResult = jdbcTemplate.query(sql, namedParameters, new LongMapper());
@ -880,8 +857,7 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_TAGS);
namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.GET_BATCH_TAGS.getParamName(0),signerId.toByteArray());
namedParameters.addValue(QueryType.GET_BATCH_TAGS.getParamName(1),batchId);
namedParameters.addValue(QueryType.GET_BATCH_TAGS.getParamName(0),message.getBatchId());
List<BeginBatchMessage> beginBatchMessages = jdbcTemplate.query(sql, namedParameters, new BeginBatchMessageMapper());
@ -892,7 +868,6 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
UnsignedBulletinBoardMessage unsignedMessage = UnsignedBulletinBoardMessage.newBuilder()
.addAllTag(beginBatchMessages.get(0).getTagList())
.setTimestamp(message.getTimestamp())
.setIsStub(true)
.build();
// Digest the data
@ -900,12 +875,11 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
digest.reset();
digest.update(unsignedMessage);
sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS);
sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID);
namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(0),signerId.toByteArray());
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(1),batchId);
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(2),0); // Read from the beginning
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(0),message.getBatchId());
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(1),0); // Read from the beginning
jdbcTemplate.query(sql, namedParameters, new BatchDataDigestHandler(digest));
byte[] msgID = digest.digest();
@ -914,33 +888,21 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
// Create Bulletin Board message
BulletinBoardMessage bulletinBoardMessage = BulletinBoardMessage.newBuilder()
.setMsg(unsignedMessage)
.setMsg(UnsignedBulletinBoardMessage.newBuilder()
.mergeFrom(unsignedMessage)
.setMsgId(ByteString.copyFrom(msgID)))
.addAllSig(message.getSigList())
.build();
// Post message with pre-calculated ID and without checking signature validity
long entryNum = postMessage(bulletinBoardMessage, msgID);
// Remove signer ID and batch ID from batch data table
// This allows reuse of the batch ID in subsequent batches
sql = sqlQueryProvider.getSQLString(QueryType.REMOVE_BATCH_IDS);
namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.REMOVE_BATCH_IDS.getParamName(0), signerId.toByteArray());
namedParameters.addValue(QueryType.REMOVE_BATCH_IDS.getParamName(1), batchId);
jdbcTemplate.update(sql, namedParameters);
// Remove tags from temporary table
sql = sqlQueryProvider.getSQLString(QueryType.REMOVE_BATCH_TAGS);
jdbcTemplate.update(sql, namedParameters);
// Add entry num to tag data table
sql = sqlQueryProvider.getSQLString(QueryType.ADD_ENTRY_NUM_TO_BATCH);
namedParameters.addValue(QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(2), entryNum);
namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(0), message.getBatchId());
namedParameters.addValue(QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1), entryNum);
jdbcTemplate.update(sql, namedParameters);
@ -959,11 +921,11 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
throw new IllegalArgumentException("No such batch");
}
String sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_MESSAGE_DATA);
String sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID);
MapSqlParameterSource namedParameters = new MapSqlParameterSource();
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA.getParamName(0),batchQuery.getMsgID().getID().toByteArray());
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA.getParamName(1),batchQuery.getStartPosition());
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(0),batchQuery.getMsgID().getID().toByteArray());
namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(1),batchQuery.getStartPosition());
jdbcTemplate.query(sql, namedParameters, new BatchDataCallbackHandler(out));
@ -1031,7 +993,7 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
checksumChanged = true;
checksum.update(message.getMsg().getData());
checksum.update(message.getMsg().getMsgId());
lastTimestamp = message.getMsg().getTimestamp();
message = messageIterator.next();
@ -1109,7 +1071,7 @@ public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{
// Advance checksum
ByteString messageID = message.getMsg().getData(); // The data field contains the message ID
ByteString messageID = message.getMsg().getMsgId(); // The data field contains the message ID
checksum.update(messageID);

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

@ -94,90 +94,55 @@ public class H2QueryProvider implements BulletinBoardSQLServer.SQLQueryProvider
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"
+ " INNER JOIN SignatureTable ON MsgTable.EntryNum = SignatureTable.EntryNum"
+ " INNER JOIN MsgTagTable ON MsgTable.EntryNum = MsgTagTable.EntryNum"
+ " INNER JOIN TagTable ON MsgTagTable.TagId = TagTable.TagId"
+ " WHERE SignatureTable.SignerId = :{0}"
+ " AND TagTable.Tag = :{1}",
QueryType.GET_BATCH_MESSAGE_ENTRY.getParamName(0),
QueryType.GET_BATCH_MESSAGE_ENTRY.getParamName(1));
case GET_BATCH_MESSAGE_DATA:
case GET_BATCH_MESSAGE_DATA_BY_MSG_ID:
return MessageFormat.format(
"SELECT Data FROM BatchTable"
+ " INNER JOIN MsgTable ON MsgTable.EntryNum = BatchTable.EntryNum"
+ " WHERE MsgTable.MsgId = :{0} AND BatchTable.SerialNum >= :{1}"
+ " ORDER BY BatchTable.SerialNum ASC",
QueryType.GET_BATCH_MESSAGE_DATA.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA.getParamName(1));
QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(1));
case GET_BATCH_MESSAGE_DATA_BY_IDS:
case GET_BATCH_MESSAGE_DATA_BY_BATCH_ID:
return MessageFormat.format(
"SELECT Data FROM BatchTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1} AND SerialNum >= :{2}"
+ " WHERE BatchId = :{0} AND SerialNum >= :{1}"
+ " ORDER BY BatchTable.SerialNum ASC",
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(1),
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(2));
QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(1));
case INSERT_BATCH_DATA:
return MessageFormat.format(
"INSERT INTO BatchTable (SignerId, BatchId, SerialNum, Data)"
+ " VALUES (:{0}, :{1}, :{2}, :{3})",
"INSERT INTO BatchTable (BatchId, SerialNum, Data) VALUES (:{0}, :{1}, :{2})",
QueryType.INSERT_BATCH_DATA.getParamName(0),
QueryType.INSERT_BATCH_DATA.getParamName(1),
QueryType.INSERT_BATCH_DATA.getParamName(2),
QueryType.INSERT_BATCH_DATA.getParamName(3));
QueryType.INSERT_BATCH_DATA.getParamName(2));
case CHECK_BATCH_LENGTH:
return MessageFormat.format(
"SELECT COUNT(Data) AS BatchLength FROM BatchTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.CHECK_BATCH_LENGTH.getParamName(0),
QueryType.CHECK_BATCH_LENGTH.getParamName(1));
"SELECT COUNT(Data) AS BatchLength FROM BatchTable WHERE BatchId = :{0}",
QueryType.CHECK_BATCH_LENGTH.getParamName(0));
case CHECK_BATCH_OPEN:
return MessageFormat.format(
"SELECT COUNT(SignerId) AS signCount FROM BatchTagTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.CHECK_BATCH_OPEN.getParamName(0),
QueryType.CHECK_BATCH_OPEN.getParamName(1));
"SELECT COUNT(BatchId) AS batchCount FROM BatchTagTable WHERE BatchId = :{0}",
QueryType.CHECK_BATCH_OPEN.getParamName(0));
case STORE_BATCH_TAGS:
return MessageFormat.format(
"INSERT INTO BatchTagTable (SignerId, BatchId, Tags) VALUES (:{0}, :{1}, :{2})",
QueryType.STORE_BATCH_TAGS.getParamName(0),
QueryType.STORE_BATCH_TAGS.getParamName(1),
QueryType.STORE_BATCH_TAGS.getParamName(2));
"INSERT INTO BatchTagTable (Tags) VALUES (:{0})",
QueryType.STORE_BATCH_TAGS.getParamName(0));
case GET_BATCH_TAGS:
return MessageFormat.format(
"SELECT Tags FROM BatchTagTable WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.GET_BATCH_TAGS.getParamName(0),
QueryType.GET_BATCH_TAGS.getParamName(1));
case REMOVE_BATCH_TAGS:
return MessageFormat.format(
"DELETE FROM BatchTagTable WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.REMOVE_BATCH_TAGS.getParamName(0),
QueryType.REMOVE_BATCH_TAGS.getParamName(1));
case REMOVE_BATCH_IDS:
return MessageFormat.format(
"UPDATE BatchTable Set (SignerId, BatchId) = (NULL, NULL)" +
" WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.REMOVE_BATCH_IDS.getParamName(0),
QueryType.REMOVE_BATCH_IDS.getParamName(1));
"SELECT Tags FROM BatchTagTable WHERE BatchId = :{0}",
QueryType.GET_BATCH_TAGS.getParamName(0));
case ADD_ENTRY_NUM_TO_BATCH:
return MessageFormat.format(
"UPDATE BatchTable SET EntryNum = :{2} WHERE SignerId = :{0} AND BatchId = :{1}",
"UPDATE BatchTable SET EntryNum = :{1} WHERE BatchId = :{0}",
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(0),
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1),
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(2));
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1));
default:
throw new IllegalArgumentException("Cannot serve a query of type " + queryType);
@ -266,7 +231,8 @@ 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, ExactTime TIMESTAMP, 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)");
@ -281,12 +247,14 @@ public class H2QueryProvider implements BulletinBoardSQLServer.SQLQueryProvider
list.add("CREATE INDEX IF NOT EXISTS SignerIndex ON SignatureTable(SignerId)");
list.add("CREATE UNIQUE INDEX IF NOT EXISTS SignatureIndex ON SignatureTable(SignerId, EntryNum)");
list.add("CREATE TABLE IF NOT EXISTS BatchTable (EntryNum INT, SignerId TINYBLOB, BatchId INT, SerialNum INT, Data BLOB,"
+ " UNIQUE(SignerId, BatchId, SerialNum))");
list.add("CREATE TABLE IF NOT EXISTS BatchTagTable (BatchId INT AUTO_INCREMENT PRIMARY KEY, Tags BLOB)");
list.add("CREATE TABLE IF NOT EXISTS BatchTable (BatchId INT, EntryNum INT, SerialNum INT, Data BLOB,"
+ " UNIQUE(BatchId, SerialNum),"
+ " FOREIGN KEY (BatchId) REFERENCES BatchTagTable(BatchId) ON DELETE CASCADE)");
list.add("CREATE INDEX IF NOT EXISTS BatchDataIndex ON BatchTable(EntryNum, SerialNum)");
list.add("CREATE TABLE IF NOT EXISTS BatchTagTable (SignerId TINYBLOB, BatchId INT, Tags BLOB)");
list.add("CREATE UNIQUE INDEX IF NOT EXISTS BatchTagIndex ON BatchTagTable(SignerId, BatchId)");
// This is used to create a simple table with one entry.
// It is used for implementing a workaround for the missing INSERT IGNORE syntax
@ -302,12 +270,12 @@ public class H2QueryProvider implements BulletinBoardSQLServer.SQLQueryProvider
list.add("DROP TABLE IF EXISTS UtilityTable");
list.add("DROP INDEX IF EXISTS BatchTagIndex");
list.add("DROP TABLE IF EXISTS BatchTagTable");
list.add("DROP INDEX IF EXISTS BatchDataIndex");
list.add("DROP TABLE IF EXISTS BatchTable");
list.add("DROP INDEX IF EXISTS BatchTagIndex");
list.add("DROP TABLE IF EXISTS BatchTagTable");
list.add("DROP TABLE IF EXISTS MsgTagTable");
list.add("DROP INDEX IF EXISTS SignerIdIndex");

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

@ -97,90 +97,55 @@ public class MySQLQueryProvider implements SQLQueryProvider {
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"
+ " INNER JOIN SignatureTable ON MsgTable.EntryNum = SignatureTable.EntryNum"
+ " INNER JOIN MsgTagTable ON MsgTable.EntryNum = MsgTagTable.EntryNum"
+ " INNER JOIN TagTable ON MsgTagTable.TagId = TagTable.TagId"
+ " WHERE SignatureTable.SignerId = :{0}"
+ " AND TagTable.Tag = :{1}",
QueryType.GET_BATCH_MESSAGE_ENTRY.getParamName(0),
QueryType.GET_BATCH_MESSAGE_ENTRY.getParamName(1));
case GET_BATCH_MESSAGE_DATA:
case GET_BATCH_MESSAGE_DATA_BY_MSG_ID:
return MessageFormat.format(
"SELECT Data FROM BatchTable"
+ " INNER JOIN MsgTable ON MsgTable.EntryNum = BatchTable.EntryNum"
+ " WHERE MsgTable.MsgId = :{0} AND BatchTable.SerialNum >= :{1}"
+ " ORDER BY BatchTable.SerialNum ASC",
QueryType.GET_BATCH_MESSAGE_DATA.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA.getParamName(1));
QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(1));
case GET_BATCH_MESSAGE_DATA_BY_IDS:
case GET_BATCH_MESSAGE_DATA_BY_BATCH_ID:
return MessageFormat.format(
"SELECT Data FROM BatchTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1} AND SerialNum >= :{2}"
+ " WHERE BatchId = :{0} AND SerialNum >= :{1}"
+ " ORDER BY BatchTable.SerialNum ASC",
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(1),
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(2));
QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(1));
case INSERT_BATCH_DATA:
return MessageFormat.format(
"INSERT INTO BatchTable (SignerId, BatchId, SerialNum, Data)"
+ " VALUES (:{0}, :{1}, :{2}, :{3})",
"INSERT INTO BatchTable (BatchId, SerialNum, Data) VALUES (:{0}, :{1}, :{2})",
QueryType.INSERT_BATCH_DATA.getParamName(0),
QueryType.INSERT_BATCH_DATA.getParamName(1),
QueryType.INSERT_BATCH_DATA.getParamName(2),
QueryType.INSERT_BATCH_DATA.getParamName(3));
QueryType.INSERT_BATCH_DATA.getParamName(2));
case CHECK_BATCH_LENGTH:
return MessageFormat.format(
"SELECT COUNT(Data) AS BatchLength FROM BatchTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.CHECK_BATCH_LENGTH.getParamName(0),
QueryType.CHECK_BATCH_LENGTH.getParamName(1));
"SELECT COUNT(Data) AS BatchLength FROM BatchTable WHERE BatchId = :{0}",
QueryType.CHECK_BATCH_LENGTH.getParamName(0));
case CHECK_BATCH_OPEN:
return MessageFormat.format(
"SELECT COUNT(Tags) AS signCount FROM BatchTagTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.CHECK_BATCH_OPEN.getParamName(0),
QueryType.CHECK_BATCH_OPEN.getParamName(1));
"SELECT COUNT(BatchId) AS batchCount FROM BatchTagTable WHERE BatchId = :{0}",
QueryType.CHECK_BATCH_OPEN.getParamName(0));
case STORE_BATCH_TAGS:
return MessageFormat.format(
"INSERT INTO BatchTagTable (SignerId, BatchId, Tags) VALUES (:{0}, :{1}, :{2})",
QueryType.STORE_BATCH_TAGS.getParamName(0),
QueryType.STORE_BATCH_TAGS.getParamName(1),
QueryType.STORE_BATCH_TAGS.getParamName(2));
"INSERT INTO BatchTagTable (Tags) VALUES (:{0})",
QueryType.STORE_BATCH_TAGS.getParamName(0));
case GET_BATCH_TAGS:
return MessageFormat.format(
"SELECT Tags FROM BatchTagTable WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.GET_BATCH_TAGS.getParamName(0),
QueryType.GET_BATCH_TAGS.getParamName(1));
case REMOVE_BATCH_TAGS:
return MessageFormat.format(
"DELETE FROM BatchTagTable WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.REMOVE_BATCH_TAGS.getParamName(0),
QueryType.REMOVE_BATCH_TAGS.getParamName(1));
case REMOVE_BATCH_IDS:
return MessageFormat.format(
"UPDATE BatchTable Set SignerId = NULL, BatchId = NULL" +
" WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.REMOVE_BATCH_IDS.getParamName(0),
QueryType.REMOVE_BATCH_IDS.getParamName(1));
"SELECT Tags FROM BatchTagTable WHERE BatchId = :{0}",
QueryType.GET_BATCH_TAGS.getParamName(0));
case ADD_ENTRY_NUM_TO_BATCH:
return MessageFormat.format(
"UPDATE BatchTable SET EntryNum = :{2} WHERE SignerId = :{0} AND BatchId = :{1}",
"UPDATE BatchTable SET EntryNum = :{1} WHERE BatchId = :{0}",
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(0),
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1),
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(2));
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1));
default:
throw new IllegalArgumentException("Cannot serve a query of type " + queryType);
@ -276,19 +241,21 @@ public class MySQLQueryProvider implements SQLQueryProvider {
list.add("CREATE TABLE IF NOT EXISTS TagTable (TagId INT NOT NULL AUTO_INCREMENT PRIMARY KEY, Tag VARCHAR(50), UNIQUE(Tag))");
list.add("CREATE TABLE IF NOT EXISTS MsgTagTable (EntryNum INT, TagId INT,"
+ " CONSTRAINT FOREIGN KEY (EntryNum) REFERENCES MsgTable(EntryNum),"
+ " CONSTRAINT FOREIGN KEY (TagId) REFERENCES TagTable(TagId),"
+ " CONSTRAINT FOREIGN KEY (EntryNum) REFERENCES MsgTable(EntryNum) ON DELETE CASCADE,"
+ " CONSTRAINT FOREIGN KEY (TagId) REFERENCES TagTable(TagId) ON DELETE CASCADE,"
+ " CONSTRAINT UNIQUE (EntryNum, TagID))");
list.add("CREATE TABLE IF NOT EXISTS SignatureTable (EntryNum INT, SignerId TINYBLOB, Signature TINYBLOB,"
+ " INDEX(SignerId(32)), CONSTRAINT Unique_Signature UNIQUE(SignerId(32), EntryNum),"
+ " CONSTRAINT FOREIGN KEY (EntryNum) REFERENCES MsgTable(EntryNum))");
+ " CONSTRAINT FOREIGN KEY (EntryNum) REFERENCES MsgTable(EntryNum) ON DELETE CASCADE)");
list.add("CREATE TABLE IF NOT EXISTS BatchTagTable (BatchId INT AUTO_INCREMENT PRIMARY KEY, Tags BLOB)");
list.add("CREATE TABLE IF NOT EXISTS BatchTable (BatchId INT, EntryNum INT, SerialNum INT, Data BLOB,"
+ " CONSTRAINT UNIQUE(BatchId, SerialNum),"
+ " CONSTRAINT FOREIGN KEY (BatchId) REFERENCES BatchTagTable(BatchId) ON DELETE CASCADE)");
list.add("CREATE TABLE IF NOT EXISTS BatchTable (EntryNum INT, SignerId TINYBLOB, BatchId INT, SerialNum INT, Data BLOB,"
+ " CONSTRAINT Unique_Batch UNIQUE(SignerId(32), BatchId, SerialNum))");
list.add("CREATE TABLE IF NOT EXISTS BatchTagTable (SignerId TINYBLOB, BatchId INT, Tags BLOB,"
+ " INDEX(SignerId(32), BatchId))");
return list;
}
@ -297,8 +264,8 @@ public class MySQLQueryProvider implements SQLQueryProvider {
public List<String> getSchemaDeletionCommands() {
List<String> list = new LinkedList<String>();
list.add("DROP TABLE IF EXISTS BatchTagTable");
list.add("DROP TABLE IF EXISTS BatchTable");
list.add("DROP TABLE IF EXISTS BatchTagTable");
list.add("DROP TABLE IF EXISTS MsgTagTable");
list.add("DROP TABLE IF EXISTS SignatureTable");
list.add("DROP TABLE IF EXISTS TagTable");

93
bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/SQLiteQueryProvider.java
View File

@ -1,6 +1,7 @@
package meerkat.bulletinboard.sqlserver;
import meerkat.protobuf.BulletinBoardAPI.*;
import org.apache.commons.dbcp2.BasicDataSource;
import org.sqlite.SQLiteDataSource;
import javax.sql.DataSource;
@ -60,91 +61,55 @@ public class SQLiteQueryProvider implements BulletinBoardSQLServer.SQLQueryProvi
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"
+ " INNER JOIN SignatureTable ON MsgTable.EntryNum = SignatureTable.EntryNum"
+ " INNER JOIN MsgTagTable ON MsgTable.EntryNum = MsgTagTable.EntryNum"
+ " INNER JOIN TagTable ON MsgTagTable.TagId = TagTable.TagId"
+ " WHERE SignatureTable.SignerId = :{0}"
+ " AND TagTable.Tag = :{1}",
QueryType.GET_BATCH_MESSAGE_ENTRY.getParamName(0),
QueryType.GET_BATCH_MESSAGE_ENTRY.getParamName(1));
case GET_BATCH_MESSAGE_DATA:
case GET_BATCH_MESSAGE_DATA_BY_MSG_ID:
return MessageFormat.format(
"SELECT Data FROM BatchTable"
+ " INNER JOIN MsgTable ON MsgTable.EntryNum = BatchTable.EntryNum"
+ " WHERE MsgTable.MsgId = :{0} AND BatchTable.SerialNum >= :{1}"
+ " ORDER BY BatchTable.SerialNum ASC",
QueryType.GET_BATCH_MESSAGE_DATA.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA.getParamName(1));
QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(1));
case GET_BATCH_MESSAGE_DATA_BY_IDS:
case GET_BATCH_MESSAGE_DATA_BY_BATCH_ID:
return MessageFormat.format(
"SELECT Data FROM BatchTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1} AND SerialNum >= :{2}"
+ " WHERE BatchId = :{0} AND SerialNum >= :{1}"
+ " ORDER BY BatchTable.SerialNum ASC",
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(1),
QueryType.GET_BATCH_MESSAGE_DATA_BY_IDS.getParamName(2));
QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(0),
QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(1));
case INSERT_BATCH_DATA:
return MessageFormat.format(
"INSERT INTO BatchTable (SignerId, BatchId, SerialNum, Data)"
+ " VALUES (:{0}, :{1}, :{2}, :{3})",
"INSERT INTO BatchTable (BatchId, SerialNum, Data) VALUES (:{0}, :{1}, :{2})",
QueryType.INSERT_BATCH_DATA.getParamName(0),
QueryType.INSERT_BATCH_DATA.getParamName(1),
QueryType.INSERT_BATCH_DATA.getParamName(2),
QueryType.INSERT_BATCH_DATA.getParamName(3));
QueryType.INSERT_BATCH_DATA.getParamName(2));
case CHECK_BATCH_LENGTH:
return MessageFormat.format(
"SELECT COUNT(Data) AS BatchLength FROM BatchTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.CHECK_BATCH_LENGTH.getParamName(0),
QueryType.CHECK_BATCH_LENGTH.getParamName(1));
"SELECT COUNT(Data) AS BatchLength FROM BatchTable WHERE BatchId = :{0}",
QueryType.CHECK_BATCH_LENGTH.getParamName(0));
case CHECK_BATCH_OPEN:
return MessageFormat.format(
"SELECT COUNT(Tags) AS signCount FROM BatchTagTable"
+ " WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.CHECK_BATCH_OPEN.getParamName(0),
QueryType.CHECK_BATCH_OPEN.getParamName(1));
"SELECT COUNT(BatchId) AS batchCount FROM BatchTagTable WHERE BatchId = :{0}",
QueryType.CHECK_BATCH_OPEN.getParamName(0));
case STORE_BATCH_TAGS:
return MessageFormat.format(
"INSERT INTO BatchTagTable (SignerId, BatchId, Tags) VALUES (:{0}, :{1}, :{2})",
QueryType.STORE_BATCH_TAGS.getParamName(0),
QueryType.STORE_BATCH_TAGS.getParamName(1),
QueryType.STORE_BATCH_TAGS.getParamName(2));
"INSERT INTO BatchTagTable (Tags) VALUES (:{0})",
QueryType.STORE_BATCH_TAGS.getParamName(0));
case GET_BATCH_TAGS:
return MessageFormat.format(
"SELECT Tag FROM TagTable INNER JOIN BatchTagTable ON TagTable.TagId = BatchTagTable.TagId"
+ " WHERE SignerId = :{0} AND BatchId = :{1} ORDER BY Tag ASC",
QueryType.GET_BATCH_TAGS.getParamName(0),
QueryType.GET_BATCH_TAGS.getParamName(1));
case REMOVE_BATCH_TAGS:
return MessageFormat.format(
"DELETE FROM BatchTagTable WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.REMOVE_BATCH_TAGS.getParamName(0),
QueryType.REMOVE_BATCH_TAGS.getParamName(1));
case REMOVE_BATCH_IDS:
return MessageFormat.format(
"UPDATE BatchTable Set (SignerId, BatchId) = (NULL, NULL)" +
" WHERE SignerId = :{0} AND BatchId = :{1}",
QueryType.REMOVE_BATCH_IDS.getParamName(0),
QueryType.REMOVE_BATCH_IDS.getParamName(1));
"SELECT Tags FROM BatchTagTable WHERE BatchId = :{0}",
QueryType.GET_BATCH_TAGS.getParamName(0));
case ADD_ENTRY_NUM_TO_BATCH:
return MessageFormat.format(
"UPDATE BatchTable SET EntryNum = :{2} WHERE SignerId = :{0} AND BatchId = :{1}",
"UPDATE BatchTable SET EntryNum = :{1} WHERE BatchId = :{0}",
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(0),
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1),
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(2));
QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1));
default:
throw new IllegalArgumentException("Cannot serve a query of type " + queryType);
@ -220,7 +185,8 @@ public class SQLiteQueryProvider implements BulletinBoardSQLServer.SQLQueryProvi
@Override
public DataSource getDataSource() {
SQLiteDataSource dataSource = new SQLiteDataSource();
BasicDataSource dataSource = new BasicDataSource();
dataSource.setDriverClassName("org.sqlite.JDBC");
dataSource.setUrl("jdbc:sqlite:" + dbName);
return dataSource;
@ -234,14 +200,23 @@ public class SQLiteQueryProvider implements BulletinBoardSQLServer.SQLQueryProvi
list.add("CREATE TABLE IF NOT EXISTS MsgTable (EntryNum INTEGER PRIMARY KEY, MsgId BLOB UNIQUE, Msg BLOB)");
list.add("CREATE TABLE IF NOT EXISTS TagTable (TagId INTEGER PRIMARY KEY, Tag varchar(50) UNIQUE)");
list.add("CREATE TABLE IF NOT EXISTS MsgTagTable (EntryNum BLOB, TagId INTEGER, FOREIGN KEY (EntryNum)"
+ " REFERENCES MsgTable(EntryNum), FOREIGN KEY (TagId) REFERENCES TagTable(TagId), UNIQUE (EntryNum, TagID))");
list.add("CREATE TABLE IF NOT EXISTS MsgTagTable (EntryNum BLOB, TagId INTEGER,"
+ " FOREIGN KEY (EntryNum) REFERENCES MsgTable(EntryNum) ON DELETE CASCADE,"
+ " FOREIGN KEY (TagId) REFERENCES TagTable(TagId) ON DELETE CASCADE,"
+ " UNIQUE (EntryNum, TagID))");
list.add("CREATE TABLE IF NOT EXISTS SignatureTable (EntryNum INTEGER, SignerId BLOB, Signature BLOB,"
+ " FOREIGN KEY (EntryNum) REFERENCES MsgTable(EntryNum), UNIQUE(SignerId, EntryNum))");
+ " FOREIGN KEY (EntryNum) REFERENCES MsgTable(EntryNum) ON DELETE CASCADE,"
+ " UNIQUE(SignerId, EntryNum))");
list.add("CREATE INDEX IF NOT EXISTS SignerIndex ON SignatureTable(SignerId)");
list.add("CREATE TABLE IF NOT EXISTS BatchTagTable (BatchId INTEGER PRIMARY KEY, Tags BLOB)");
list.add("CREATE TABLE IF NOT EXISTS BatchTable (BatchId INTEGER, EntryNum INTEGER, SerialNum INTEGER, Data BLOB,"
+ " UNIQUE(BatchId, SerialNum),"
+ " FOREIGN KEY (BatchId) REFERENCES BatchTagTable(BatchId) ON DELETE CASCADE)");
return list;
}

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

@ -21,7 +21,7 @@ public class MessageStubMapper implements RowMapper<BulletinBoardMessage> {
return BulletinBoardMessage.newBuilder()
.setEntryNum(rs.getLong(1))
.setMsg(UnsignedBulletinBoardMessage.newBuilder()
.setData(ByteString.copyFrom(rs.getBytes(2)))
.setMsgId(ByteString.copyFrom(rs.getBytes(2)))
.setTimestamp(BulletinBoardUtils.toTimestampProto(rs.getTimestamp(3)))
.build())
.build();

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

@ -10,6 +10,7 @@ import javax.ws.rs.core.StreamingOutput;
import com.google.protobuf.BoolValue;
import com.google.protobuf.Int32Value;
import com.google.protobuf.Int64Value;
import meerkat.bulletinboard.BulletinBoardServer;
import meerkat.bulletinboard.sqlserver.BulletinBoardSQLServer;
import meerkat.bulletinboard.sqlserver.H2QueryProvider;
@ -142,7 +143,7 @@ public class BulletinBoardWebApp implements BulletinBoardServer, ServletContextL
@Consumes(MEDIATYPE_PROTOBUF)
@Produces(MEDIATYPE_PROTOBUF)
@Override
public BoolValue beginBatch(BeginBatchMessage message) {
public Int64Value beginBatch(BeginBatchMessage message) {
try {
init();
return bulletinBoard.beginBatch(message);

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

@ -20,6 +20,7 @@ import java.util.*;
import com.google.protobuf.BoolValue;
import com.google.protobuf.ByteString;
import com.google.protobuf.Int64Value;
import com.google.protobuf.Timestamp;
import meerkat.comm.CommunicationException;
import meerkat.comm.MessageInputStream;
@ -443,24 +444,18 @@ public class GenericBulletinBoardServerTest {
}
private void postAsBatch(BulletinBoardMessage message, ByteString signerId, int batchId, int chunkSize, boolean close) throws CommunicationException {
private void postAsBatch(BulletinBoardMessage message, int chunkSize, boolean close) throws CommunicationException {
List<BatchChunk> batchChunks = BulletinBoardUtils.breakToBatch(message, chunkSize);
BeginBatchMessage beginBatchMessage = BulletinBoardUtils.generateBeginBatchMessage(signerId, batchId, message);
BeginBatchMessage beginBatchMessage = BulletinBoardUtils.generateBeginBatchMessage(message);
BoolValue result;
// Begin batch
result = bulletinBoardServer.beginBatch(beginBatchMessage);
Int64Value batchId = bulletinBoardServer.beginBatch(beginBatchMessage);
assertThat("Was not able to open batch", result.getValue(), is(true));
// Attempt to open batch again
result = bulletinBoardServer.beginBatch(beginBatchMessage);
assertThat("Was able to open a closed batch", result.getValue(), is(false));
assertThat("Was not able to open batch", batchId.getValue() != -1);
// Post data
@ -469,8 +464,7 @@ public class GenericBulletinBoardServerTest {
for (int i = 0 ; i < batchChunks.size() ; i++){
batchMessage = BatchMessage.newBuilder()
.setSignerId(signerId)
.setBatchId(batchId)
.setBatchId(batchId.getValue())
.setSerialNum(i)
.setData(batchChunks.get(i))
.build();
@ -494,45 +488,6 @@ public class GenericBulletinBoardServerTest {
}
/**
* Tests that opening the same batch ID while one is open results in failure
* @throws CommunicationException
*/
public void testReopen() throws CommunicationException, SignatureException {
// Create data
final int BATCH_ID = 100;
final int DATA_SIZE = 1;
final int CHUNK_SIZE = 1;
final int TAG_NUMBER = 1;
Timestamp timestamp = Timestamp.newBuilder()
.setSeconds(141510)
.setNanos(48015)
.build();
BulletinBoardMessage batch = bulletinBoardMessageGenerator.generateRandomMessage(signers, timestamp, DATA_SIZE, TAG_NUMBER);
// Post batch but do not close
postAsBatch(batch, signerIDs[0], BATCH_ID, CHUNK_SIZE, false);
// Attempt to open batch again
BoolValue result;
result = bulletinBoardServer.beginBatch(BulletinBoardUtils.generateBeginBatchMessage(signerIDs[0], BATCH_ID, batch));
assertThat("Was able to open a closed batch", result.getValue(), is(false));
// Close batch
result = bulletinBoardServer.closeBatch(BulletinBoardUtils.generateCloseBatchMessage(BATCH_ID, DATA_SIZE / CHUNK_SIZE, batch));
assertThat("Was not able to close batch", result.getValue(), is(true));
}
/**
* Posts a complete batch message
* @throws CommunicationException
@ -554,7 +509,7 @@ public class GenericBulletinBoardServerTest {
// Post batch
postAsBatch(batch, signerIDs[0], BATCH_ID, CHUNK_SIZE, true);
postAsBatch(batch, CHUNK_SIZE, true);
// Update locally stored batches
batches.add(batch);

10
bulletin-board-server/src/test/java/meerkat/bulletinboard/H2BulletinBoardServerTest.java
View File

@ -106,16 +106,6 @@ public class H2BulletinBoardServerTest {
System.err.println("Time of operation: " + (end - start));
}
@Test
public void testBatchReopen() {
try{
serverTest.testReopen();
} catch (Exception e) {
System.err.println(e.getMessage());
fail(e.getMessage());
}
}
@Test
public void testBatch() {

10
bulletin-board-server/src/test/java/meerkat/bulletinboard/MySQLBulletinBoardServerTest.java
View File

@ -110,16 +110,6 @@ public class MySQLBulletinBoardServerTest {
System.err.println("Time of operation: " + (end - start));
}
@Test
public void testBatchReopen() {
try{
serverTest.testReopen();
} catch (Exception e) {
System.err.println(e.getMessage());
fail(e.getMessage());
}
}
@Test
public void testBatch() {

25
bulletin-board-server/src/test/java/meerkat/bulletinboard/SQLiteBulletinBoardServerTest.java
View File

@ -60,7 +60,7 @@ public class SQLiteBulletinBoardServerTest{
System.err.println("Time of operation: " + (end - start));
}
@Test
// @Test
public void bulkTest() {
System.err.println("Starting bulkTest of SQLiteBulletinBoardServerTest");
long start = threadBean.getCurrentThreadCpuTime();
@ -91,6 +91,29 @@ public class SQLiteBulletinBoardServerTest{
System.err.println("Time of operation: " + (end - start));
}
// @Test
public void testBatch() {
final int BATCH_NUM = 20;
try{
for (int i = 0 ; i < BATCH_NUM ; i++) {
serverTest.testPostBatch();
}
} catch (Exception e) {
System.err.println(e.getMessage());
fail(e.getMessage());
}
try{
serverTest.testReadBatch();
} catch (Exception e) {
System.err.println(e.getMessage());
fail(e.getMessage());
}
}
@After
public void close() {
System.err.println("Starting to close SQLiteBulletinBoardServerTest");

70
meerkat-common/src/main/java/meerkat/bulletinboard/AsyncBulletinBoardClient.java
View File

@ -1,8 +1,9 @@
package meerkat.bulletinboard;
import com.google.common.util.concurrent.FutureCallback;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import meerkat.protobuf.BulletinBoardAPI.*;
import meerkat.protobuf.Crypto.Signature;
import java.util.List;
@ -21,68 +22,58 @@ public interface AsyncBulletinBoardClient extends BulletinBoardClient {
public MessageID postMessage(BulletinBoardMessage msg, FutureCallback<Boolean> callback);
/**
* Perform an end-to-end post of a signed batch message
* @param signerId is the canonical form for the ID of the sender of this batch
* @param batchId is a unique (per signer) ID for this batch
* @param completeBatch contains all the data of the batch including the meta-data and the signature
* Perform an end-to-end post of a message in batch form
* @param completeBatch contains all the data of the batch
* @param chunkSize is the maximum size of each chunk of the message in bytes
* @param callback is a class containing methods to handle the result of the operation
* @return a unique identifier for the batch message
*/
public MessageID postBatch(byte[] signerId, int batchId, BulletinBoardMessage completeBatch, FutureCallback<Boolean> callback);
public MessageID postAsBatch(BulletinBoardMessage completeBatch, int chunkSize, FutureCallback<Boolean> callback);
/**
* Perform an end-to-end post of a signed batch message
* @param signerId is the canonical form for the ID of the sender of this batch
* @param batchId is a unique (per signer) ID for this batch
* @param completeBatch contains all the data of the batch including the meta-data and the signature
* @param callback is a class containing methods to handle the result of the operation
* @return a unique identifier for the batch message
* An interface for returning an opaque identifier for a batch message
* This identifier is used to uniquely identify the batch until it is completely posted and signed
* After the batch is fully posted: it is identified by its digest (like any message)
* This can be implementation-specific (and not necessarily interchangeable between different implementations)
*/
public MessageID postBatch(ByteString signerId, int batchId, BulletinBoardMessage completeBatch, FutureCallback<Boolean> callback);
public interface BatchIdentifier {}
/**
* This message informs the server about the existence of a new batch message and supplies it with the tags associated with it
* @param beginBatchMessage contains the data required to begin the batch
* @param tags contains the tags used in the batch
* @param callback is a callback function class for handling results of the operation
* it receives a BatchIdentifier for use in subsequent batch post operations
*/
public void beginBatch(BeginBatchMessage beginBatchMessage, FutureCallback<Boolean> callback);
public void beginBatch(Iterable<String> tags, FutureCallback<BatchIdentifier> callback);
/**
* This method posts batch data into an (assumed to be open) batch
* It does not close the batch
* @param signerId is the canonical form for the ID of the sender of this batch
* @param batchId is a unique (per signer) ID for this batch
* @param batchIdentifier is the temporary batch identifier
* @param batchChunkList is the (canonically ordered) list of data comprising the portion of the batch to be posted
* @param startPosition is the location (in the batch) of the first entry in batchDataList
* (optionally used to continue interrupted post operations)
* The first position in the batch is position 0
* (optionally used to continue interrupted post operations)
* The first position in the batch is position 0
* @param callback is a callback function class for handling results of the operation
* @throws IllegalArgumentException if the batch identifier given was of an illegal format
*/
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback);
public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback) throws IllegalArgumentException;
/**
* Overloading of the postBatchData method which starts at the first position in the batch
*/
public void postBatchData(byte[] signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback);
/**
* Overloading of the postBatchData method which uses ByteString
*/
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList,
int startPosition, FutureCallback<Boolean> callback);
/**
* Overloading of the postBatchData method which uses ByteString and starts at the first position in the batch
*/
public void postBatchData(ByteString signerId, int batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback);
public void postBatchData(BatchIdentifier batchIdentifier, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback)
throws IllegalArgumentException;
/**
* Attempts to close a batch message
* @param closeBatchMessage contains the data required to close the batch
* @param batchIdentifier is the temporary batch identifier
* @param callback is a callback function class for handling results of the operation
* @throws IllegalArgumentException if the batch identifier given was of an illegal format
*/
public void closeBatch(CloseBatchMessage closeBatchMessage, FutureCallback<Boolean> callback);
public void closeBatch(BatchIdentifier batchIdentifier, Timestamp timestamp, Iterable<Signature> signatures, FutureCallback<Boolean> callback)
throws IllegalArgumentException;
/**
* Check how "safe" a given message is in an asynchronous manner
@ -110,6 +101,15 @@ public interface AsyncBulletinBoardClient extends BulletinBoardClient {
*/
public void readBatch(MessageID msgID, FutureCallback<BulletinBoardMessage> callback);
/**
* Read batch data for a specific stub message
* @param stub is a batch message stub
* @param callback is a callback class for handling the result of the operation
* @return a new BulletinBoardMessage containing both metadata from the stub and actual data from the server
* @throws IllegalArgumentException if the received message is not a stub
*/
public void readBatchData(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) throws IllegalArgumentException;
/**
* Perform a Sync Query on the bulletin board

20
meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardClient.java
View File

@ -45,14 +45,32 @@ public interface BulletinBoardClient {
*/
List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException;
/**
* Breaks up a bulletin board message into chunks and posts it as a batch message
* @param msg is the message to post
* @param chunkSize is the maximal chunk size in bytes
* @return the unique message ID
* @throws CommunicationException if operation is unsuccessful
*/
MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize) throws CommunicationException;
/**
* Read a given batch message from the bulletin board
* @param msgID is the batch message ID to be read
* @return the complete batch
* @throws CommunicationException
* @throws CommunicationException if operation is unsuccessful
*/
BulletinBoardMessage readBatch(MessageID msgID) throws CommunicationException;
/**
* Read batch data for a specific stub message
* @param stub is a batch message stub
* @return a new BulletinBoardMessage containing both metadata from the stub and actual data from the server
* @throws CommunicationException if operation is unsuccessful
* @throws IllegalArgumentException if the received message is not a stub
*/
BulletinBoardMessage readBatchData(BulletinBoardMessage stub) throws CommunicationException, IllegalArgumentException;
/**
* Create a SyncQuery to test against that corresponds with the current server state for a specific filter list
* Should only be called on instances for which the actual server contacted is known (i.e. there is only one server)

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

@ -2,6 +2,7 @@ package meerkat.bulletinboard;
import com.google.protobuf.BoolValue;
import com.google.protobuf.Int32Value;
import com.google.protobuf.Int64Value;
import meerkat.comm.CommunicationException;
import meerkat.comm.MessageOutputStream;
import meerkat.protobuf.BulletinBoardAPI.*;
@ -50,12 +51,10 @@ public interface BulletinBoardServer{
/**
* Informs server about a new batch message
* @param message contains the required data about the new batch
* @return TRUE if the batch request is accepted amd FALSE otherwise
* Specifically, if such a batch already exists and is not yet closed: the value returned will be TRUE
* However, if such a batch exists and is already closed: the value returned will be FALSE
* @return a unique batch identifier for the new batch ; -1 if batch creation was unsuccessful
* @throws CommunicationException on DB connection error
*/
public BoolValue beginBatch(BeginBatchMessage message) throws CommunicationException;
public Int64Value beginBatch(BeginBatchMessage message) throws CommunicationException;
/**
* Posts a chunk of a batch message to the bulletin board
@ -93,7 +92,7 @@ public interface BulletinBoardServer{
* @return The generated SyncQuery
* @throws CommunicationException on DB connection error
*/
SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException;
public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException;
/**
* Queries the database for sync status with respect to a given sync query

2
meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardDigest.java
View File

@ -65,7 +65,7 @@ public class GenericBulletinBoardDigest implements BulletinBoardDigest {
update(msg.getTimestamp());
if (!msg.getIsStub()){
if (msg.getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.DATA){
update(msg.getData().toByteArray());
}

2
meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardSignature.java
View File

@ -42,7 +42,7 @@ public class GenericBulletinBoardSignature implements BulletinBoardSignature {
updateContent(msg.getTimestamp());
if (!msg.getIsStub()){
if (msg.getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.DATA){
updateContent(msg.getData().toByteArray());
}

10
meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageComparator.java
View File

@ -47,16 +47,20 @@ public class BulletinBoardMessageComparator implements Comparator<BulletinBoardM
}
}
// Compare data (only for non-stub messages)
// Compare data
if (msg1.getMsg().getIsStub() != msg2.getMsg().getIsStub()){
if (msg1.getMsg().getDataTypeCase() != msg2.getMsg().getDataTypeCase()){
return -1;
}
if (msg1.getMsg().getIsStub()){
if (msg1.getMsg().getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.DATA){
if (!msg1.getMsg().getData().equals(msg2.getMsg().getData())){
return -1;
}
} else if (msg1.getMsg().getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.MSGID){
if (!msg1.getMsg().getMsgId().equals(msg2.getMsg().getMsgId())){
return -1;
}
}
// Compare signatures (do not enforce order)

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

@ -1,6 +1,7 @@
package meerkat.util;
import com.google.protobuf.ByteString;
import com.google.protobuf.Int64Value;
import meerkat.protobuf.BulletinBoardAPI.*;
import java.util.ArrayList;
@ -150,6 +151,8 @@ public class BulletinBoardUtils {
/**
* Removes concrete data from the message and turns it into a stub
* Note that the stub does not contain the message ID
* Therefore, it cannot be used to retrieve the message from a server
* @param msg is the original message
* @return the message stub
*/
@ -157,9 +160,9 @@ public class BulletinBoardUtils {
return BulletinBoardMessage.newBuilder()
.mergeFrom(msg)
.setMsg(UnsignedBulletinBoardMessage.newBuilder().
mergeFrom(msg.getMsg())
.setIsStub(true)
.setMsg(UnsignedBulletinBoardMessage.newBuilder()
.mergeFrom(msg.getMsg())
.clearDataType()
.clearData()
.build())
.build();
@ -182,9 +185,8 @@ public class BulletinBoardUtils {
return BulletinBoardMessage.newBuilder()
.mergeFrom(msgStub)
.setMsg(UnsignedBulletinBoardMessage.newBuilder().
mergeFrom(msgStub.getMsg())
.setIsStub(false)
.setMsg(UnsignedBulletinBoardMessage.newBuilder()
.mergeFrom(msgStub.getMsg())
.setData(ByteString.copyFrom(dataList))
.build())
.build();
@ -193,51 +195,29 @@ public class BulletinBoardUtils {
/**
* Gerenates a BeginBatchMessage Protobuf which is used to begin uploading a message as a batch
* @param signerId is the ID of the message generator
* @param batchId is the identifier for the message
* @param msg is the Bulletin Board message to be uploaded, which can be a stub or a complete message
* @return the required BeginBatchMessage
*/
public static BeginBatchMessage generateBeginBatchMessage(ByteString signerId, int batchId, BulletinBoardMessage msg) {
public static BeginBatchMessage generateBeginBatchMessage(BulletinBoardMessage msg) {
if (msg.getSigCount() <= 0){
throw new IllegalArgumentException("No signatures found");
}
return BeginBatchMessage.newBuilder()
.setSignerId(signerId)
.setBatchId(batchId)
.addAllTag(msg.getMsg().getTagList())
.build();
}
/**
* Gerenates a BeginBatchMessage Protobuf which is used to begin uploading a message as a batch
* @param batchId is the identifier for the message
* @param msg is the Bulletin Board message to be uploaded, which can be a stub or a complete message
* The signer ID used will be the one from the first signature in the message
* @return the required BeginBatchMessage
* @throws IllegalArgumentException if the message contains no signatures
*/
public static BeginBatchMessage generateBeginBatchMessage(int batchId, BulletinBoardMessage msg) throws IllegalArgumentException {
if (msg.getSigCount() <= 0){
throw new IllegalArgumentException("No signatures found");
}
return generateBeginBatchMessage(msg.getSig(0).getSignerId(), batchId, msg);
}
/**
* Gerenates a CloseBatchMessage Protobuf which is used to finalize a batch message
* @param batchId is the identifier for the message
* @param batchId is the temporary identifier for the message
* @param batchLength is the number of chunks in the batch
* @param msg is the Bulletin Board message that was uploaded (and can also be a stub of said message)
* @throws IllegalArgumentException if the message contains no signatures
*/
public static CloseBatchMessage generateCloseBatchMessage(int batchId, int batchLength, BulletinBoardMessage msg) {
public static CloseBatchMessage generateCloseBatchMessage(Int64Value batchId, int batchLength, BulletinBoardMessage msg) {
if (msg.getSigCount() <= 0){
throw new IllegalArgumentException("No signatures found");
@ -246,7 +226,7 @@ public class BulletinBoardUtils {
return CloseBatchMessage.newBuilder()
.setTimestamp(msg.getMsg().getTimestamp())
.setBatchLength(batchLength)
.setBatchId(batchId)
.setBatchId(batchId.getValue())
.addAllSig(msg.getSigList())
.build();

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

@ -20,12 +20,18 @@ message UnsignedBulletinBoardMessage {
// Timestamp of the message (as defined by client)
google.protobuf.Timestamp timestamp = 2;
// Defines whether or not the message is a stub
// If the message is a stub: the data is ignored
bool isStub = 3;
// The payload of the message
oneof dataType{
// A unique message identifier
bytes msgId = 3;
// The actual content of the message
bytes data = 4;
}
// The actual content of the message
bytes data = 4;
}
message BulletinBoardMessage {
@ -86,17 +92,15 @@ message MessageFilterList {
// This message is used to start a batch transfer to the Bulletin Board Server
message BeginBatchMessage {
bytes signerId = 1; // Unique identifier of the main signer of the message
int32 batchId = 2; // Unique identifier for the batch (unique per signer)
repeated string tag = 3; // Tags for the batch message
repeated string tag = 1; // Tags for the batch message
}
// 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)
int64 batchId = 1; // Unique temporary identifier for the batch
int32 batchLength = 2; // Number of messages in the batch
google.protobuf.Timestamp timestamp = 3; // Timestamp of the batch (as defined by client)
repeated meerkat.Signature sig = 4; // Signature on the (ordered) batch messages ; First signature belongs to main signer
repeated meerkat.Signature sig = 4; // Signatures on the (ordered) batch messages
}
// Container for single chunk of abatch message
@ -111,10 +115,9 @@ message BatchChunkList {
// These messages comprise a batch message
message BatchMessage {
bytes signerId = 1; // Unique signer identifier
int32 batchId = 2; // Unique identifier for the batch (unique per signer)
int32 serialNum = 3; // Location of the message in the batch: starting from 0
BatchChunk data = 4; // Actual data
int64 batchId = 1; // Unique temporary identifier for the batch
int32 serialNum = 2; // Location of the message in the batch: starting from 0
BatchChunk data = 3; // Actual data
}
// This message is used to define a single query to the server to ascertain whether or not the server is synched with the client