diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/BatchDataContainer.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/BatchDataContainer.java index 884aad7..1026529 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/BatchDataContainer.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/BatchDataContainer.java @@ -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 batchChunkList; public final int startPosition; - public BatchDataContainer(byte[] signerId, int batchId, List batchChunkList, int startPosition) { - this.signerId = signerId; + public BatchDataContainer(MultiServerBatchIdentifier batchId, List batchChunkList, int startPosition) { this.batchId = batchId; this.batchChunkList = batchChunkList; this.startPosition = startPosition; diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedBulletinBoardClient.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedBulletinBoardClient.java index 3e1ed48..31521e1 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedBulletinBoardClient.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedBulletinBoardClient.java @@ -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 callback) { + public MessageID postAsBatch(final BulletinBoardMessage msg, final int chunkSize, final FutureCallback callback) { - return localClient.postBatch(completeBatch, new FutureCallback() { + return localClient.postAsBatch(msg, chunkSize, new FutureCallback() { @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 callback) { + public void beginBatch(final Iterable tags, final FutureCallback callback) { + + localClient.beginBatch(tags, new FutureCallback() { + + private BatchIdentifier localIdentifier; - localClient.beginBatch(beginBatchMessage, new FutureCallback() { @Override - public void onSuccess(Boolean result) { - remoteClient.beginBatch(beginBatchMessage, callback); + public void onSuccess(BatchIdentifier result) { + + localIdentifier = result; + + remoteClient.beginBatch(tags, new FutureCallback() { + @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 batchChunkList, - final int startPosition, final FutureCallback callback) { + public void postBatchData(final BatchIdentifier batchIdentifier, final List batchChunkList, + final int startPosition, final FutureCallback callback) throws IllegalArgumentException{ - localClient.postBatchData(signerId, batchId, batchChunkList, startPosition, new FutureCallback() { + 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() { @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 batchChunkList, final FutureCallback callback) { + public void postBatchData(final BatchIdentifier batchIdentifier, final List batchChunkList, final FutureCallback callback) + throws IllegalArgumentException{ - localClient.postBatchData(signerId, batchId, batchChunkList, new FutureCallback() { + 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() { @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 batchChunkList, - final int startPosition, final FutureCallback callback) { + public void closeBatch(final BatchIdentifier batchIdentifier, final Timestamp timestamp, final Iterable signatures, + final FutureCallback callback) { - localClient.postBatchData(signerId, batchId, batchChunkList, startPosition, new FutureCallback() { + 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() { @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 batchChunkList, final FutureCallback callback) { - - localClient.postBatchData(signerId, batchId, batchChunkList, new FutureCallback() { - @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 callback) { - - localClient.closeBatch(closeBatchMessage, new FutureCallback() { - @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 callback) { + public void readBatch(final MessageID msgID, final FutureCallback callback) { + + localClient.readBatch(msgID, new FutureCallback() { - localClient.readBatch(batchSpecificationMessage, new FutureCallback() { @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() { + remoteClient.readBatch(msgID, new FutureCallback() { @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() { - @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 callback) throws IllegalArgumentException { + + localClient.readBatchData(stub, new FutureCallback() { + + @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() { + + @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 diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedClientBatchIdentifier.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedClientBatchIdentifier.java new file mode 100644 index 0000000..322473d --- /dev/null +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/CachedClientBatchIdentifier.java @@ -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; + } + +} diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/LocalBulletinBoardClient.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/LocalBulletinBoardClient.java index d982dd2..f09b2d3 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/LocalBulletinBoardClient.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/LocalBulletinBoardClient.java @@ -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 { - 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 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 callback) { + public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback 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 { + private class BatchBeginner implements Callable { 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 callback) { + public void beginBatch(Iterable tags, FutureCallback callback) { + + BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder() + .addAllTag(tags) + .build(); + Futures.addCallback(executorService.submit(new BatchBeginner(beginBatchMessage)), callback); } private class BatchDataPoster implements Callable { - private final ByteString signerId; - private final int batchId; + private final SingleServerBatchIdentifier batchId; private final List batchChunkList; private final int startPosition; - public BatchDataPoster(ByteString signerId, int batchId, List batchChunkList, int startPosition) { - this.signerId = signerId; + public BatchDataPoster(SingleServerBatchIdentifier batchId, List 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 batchChunkList, int startPosition, FutureCallback callback) { - postBatchData(ByteString.copyFrom(signerId), batchId, batchChunkList, startPosition, callback); + public void postBatchData(BatchIdentifier batchId, List batchChunkList, int startPosition, FutureCallback 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 batchChunkList, FutureCallback callback) { - postBatchData(signerId, batchId, batchChunkList, 0, callback); + public void postBatchData(BatchIdentifier batchId, List batchChunkList, FutureCallback callback) throws IllegalArgumentException{ + postBatchData(batchId, batchChunkList, 0, callback); } - @Override - public void postBatchData(ByteString signerId, int batchId, List batchChunkList, int startPosition, FutureCallback callback) { - Futures.addCallback(executorService.submit(new BatchDataPoster(signerId, batchId, batchChunkList, startPosition)), callback); - } - - @Override - public void postBatchData(ByteString signerId, int batchId, List batchChunkList, FutureCallback callback) { - postBatchData(signerId, batchId, batchChunkList, 0, callback); - } private class BatchCloser implements Callable { @@ -214,8 +230,26 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo } @Override - public void closeBatch(CloseBatchMessage closeBatchMessage, FutureCallback callback) { + public void closeBatch(BatchIdentifier batchId, Timestamp timestamp, Iterable signatures, FutureCallback 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 { @@ -287,31 +321,6 @@ public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBo } - private class BatchDataReader implements Callable> { - - private final BatchSpecificationMessage batchSpecificationMessage; - - public BatchDataReader(BatchSpecificationMessage batchSpecificationMessage) { - this.batchSpecificationMessage = batchSpecificationMessage; - } - - @Override - public List call() throws Exception { - - ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream(); - MessageOutputStream outputStream = new MessageOutputStream<>(byteOutputStream); - server.readBatch(batchSpecificationMessage, outputStream); - - MessageInputStream inputStream = - MessageInputStreamFactory.createMessageInputStream( - new ByteArrayInputStream(byteOutputStream.toByteArray()), - BatchChunk.class); - - return inputStream.asList(); - - } - } - @Override public void readMessages(MessageFilterList filterList, FutureCallback> 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> { + + private final MessageID msgID; + + public BatchDataReader(MessageID msgID) { + this.msgID = msgID; + } + + @Override + public List call() throws Exception { + + BatchQuery batchQuery = BatchQuery.newBuilder() + .setMsgID(msgID) + .setStartPosition(0) + .build(); + + ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream(); + MessageOutputStream batchOutputStream = new MessageOutputStream<>(byteOutputStream); + server.readBatch(batchQuery,batchOutputStream); + + MessageInputStream inputStream = + MessageInputStreamFactory.createMessageInputStream( + new ByteArrayInputStream(byteOutputStream.toByteArray()), + BatchChunk.class); + + return inputStream.asList(); + + } + } + private class CompleteBatchReader implements Callable { - 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 batchOutputStream = new MessageOutputStream<>(byteOutputStream); - server.readBatch(batchSpecificationMessage,batchOutputStream); - - MessageInputStream 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 messageOutputStream = new MessageOutputStream<>(byteOutputStream); - server.readMessages(filterList,messageOutputStream); + MessageReader messageReader = new MessageReader(filterList); + List bulletinBoardMessages = messageReader.call(); - MessageInputStream 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 batchChunkList = batchDataReader.call(); - return completeBatch; + // Combine and return + + return BulletinBoardUtils.gatherBatch(stub, batchChunkList); + + } + + } + + private class BatchDataCombiner implements Callable { + + 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 batchChunkList = batchDataReader.call(); + + return BulletinBoardUtils.gatherBatch(stub, batchChunkList); } } @Override - public void readBatch(BatchSpecificationMessage batchSpecificationMessage, FutureCallback callback) { - Futures.addCallback(executorService.submit(new CompleteBatchReader(batchSpecificationMessage)), callback); + public void readBatch(MessageID msgID, FutureCallback callback) { + Futures.addCallback(executorService.submit(new CompleteBatchReader(msgID)), callback); + } + + @Override + public void readBatchData(BulletinBoardMessage stub, FutureCallback 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 { @@ -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 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 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); diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerBatchIdentifier.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerBatchIdentifier.java new file mode 100644 index 0000000..4934ee6 --- /dev/null +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerBatchIdentifier.java @@ -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 identifiers; + + public MultiServerBatchIdentifier(Iterable identifiers) { + this.identifiers = identifiers; + } + + public MultiServerBatchIdentifier(BatchIdentifier[] identifiers) { + this.identifiers = Arrays.asList(identifiers); + } + + public Iterable getIdentifiers() { + return identifiers; + } + +} diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java index 60327fa..0073be2 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/MultiServerWorker.java @@ -18,7 +18,7 @@ import java.util.concurrent.atomic.AtomicInteger; */ public abstract class MultiServerWorker extends BulletinClientWorker implements Runnable, FutureCallback{ - private final List clients; + protected final List 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 extends BulletinClientWorker getClientIterator() { - return clients.iterator(); - } - protected int getClientNumber() { return clients.size(); } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java index a93a0d2..6250784 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java @@ -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 readMessages(MessageFilterList filterList) { + public List 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 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 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 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); } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardSynchronizer.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardSynchronizer.java index 7700fd6..e8abf6b 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardSynchronizer.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardSynchronizer.java @@ -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 { diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBatchIdentifier.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBatchIdentifier.java new file mode 100644 index 0000000..9c4ad40 --- /dev/null +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBatchIdentifier.java @@ -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; + } + +} diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java index 6ef013f..b985a8a 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/SingleServerBulletinBoardClient.java @@ -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 { + class PostBatchChunkListCallback implements FutureCallback { private final FutureCallback callback; private AtomicInteger batchDataRemaining; private AtomicBoolean aggregatedResult; - public PostBatchDataListCallback(int batchDataLength, FutureCallback callback) { + public PostBatchChunkListCallback(int batchDataLength, FutureCallback callback) { this.callback = callback; this.batchDataRemaining = new AtomicInteger(batchDataLength); @@ -169,15 +173,15 @@ public class SingleServerBulletinBoardClient extends SimpleBulletinBoardClient i */ class CompleteBatchReadCallback { - private final FutureCallback callback; + private final FutureCallback callback; private List batchChunkList; - private BulletinBoardMessage batchMessage; + private BulletinBoardMessage stub; private AtomicInteger remainingQueries; private AtomicBoolean failed; - public CompleteBatchReadCallback(FutureCallback callback) { + public CompleteBatchReadCallback(FutureCallback 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 { - private final CompleteBatch completeBatch; + private final BulletinBoardMessage msg; + private final BatchIdentifier identifier; private final FutureCallback callback; - public PostBatchDataCallback(CompleteBatch completeBatch, FutureCallback callback) { - this.completeBatch = completeBatch; + public PostBatchDataCallback(BulletinBoardMessage msg, BatchIdentifier identifier, FutureCallback 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 { + private class ContinueBatchCallback implements FutureCallback { - private final CompleteBatch completeBatch; + private final BulletinBoardMessage msg; + private final int chunkSize; private final FutureCallback callback; - public BeginBatchCallback(CompleteBatch completeBatch, FutureCallback callback) { - this.completeBatch = completeBatch; + public ContinueBatchCallback(BulletinBoardMessage msg, int chunkSize, FutureCallback callback) { + this.msg = msg; + this.chunkSize = chunkSize; this.callback = callback; } @Override - public void onSuccess(Boolean msg) { + public void onSuccess(BatchIdentifier identifier) { + + List 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 callback) { + public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback 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 { + + private final FutureCallback callback; + + public BeginBatchCallback(FutureCallback 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 callback) { + public void beginBatch(Iterable tags, FutureCallback 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 batchChunkList, - int startPosition, FutureCallback callback) { - BatchMessage.Builder builder = BatchMessage.newBuilder() - .setSignerId(signerId) - .setBatchId(batchId); + @Override + public void postBatchData(BatchIdentifier batchIdentifier, List batchChunkList, + int startPosition, FutureCallback 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 batchChunkList, FutureCallback callback) { + public void postBatchData(BatchIdentifier batchIdentifier, List batchChunkList, FutureCallback callback) + throws IllegalArgumentException { - postBatchData(signerId, batchId, batchChunkList, 0, callback); + postBatchData(batchIdentifier, batchChunkList, 0, callback); } @Override - public void postBatchData(byte[] signerId, int batchId, List batchChunkList, - int startPosition, FutureCallback callback) { + public void closeBatch(BatchIdentifier batchIdentifier, Timestamp timestamp, Iterable signatures, FutureCallback 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 batchChunkList, FutureCallback callback) { - - postBatchData(signerId, batchId, batchChunkList, 0, callback); - - } - - @Override - public void closeBatch(CloseBatchMessage closeBatchMessage, FutureCallback 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 callback) { + public void readBatch(MessageID msgID, FutureCallback 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> { + + private final BulletinBoardMessage stub; + private final FutureCallback callback; + + public ReadBatchCallback(BulletinBoardMessage stub, FutureCallback callback) { + this.stub = stub; + this.callback = callback; + } + + @Override + public void onSuccess(List result) { + callback.onSuccess(BulletinBoardUtils.gatherBatch(stub, result)); + } + + @Override + public void onFailure(Throwable t) { + + } + } + + @Override + public void readBatchData(BulletinBoardMessage stub, FutureCallback 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 callback) { diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/ThreadedBulletinBoardClient.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/ThreadedBulletinBoardClient.java index 289ffcf..7f8232f 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/ThreadedBulletinBoardClient.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/ThreadedBulletinBoardClient.java @@ -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 callback) { + public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback 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 callback) { + public void beginBatch(Iterable tags, FutureCallback 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 batchChunkList, - int startPosition, FutureCallback callback) { + public void postBatchData(BatchIdentifier batchIdentifier, List batchChunkList, + int startPosition, FutureCallback 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 batchChunkList, FutureCallback callback) { + public void postBatchData(BatchIdentifier batchIdentifier, List batchChunkList, FutureCallback callback) + throws IllegalArgumentException { - postBatchData(signerId, batchId, batchChunkList, 0, callback); + postBatchData(batchIdentifier, batchChunkList, 0, callback); } @Override - public void postBatchData(ByteString signerId, int batchId, List batchChunkList, - int startPosition, FutureCallback callback) { - - postBatchData(signerId.toByteArray(), batchId, batchChunkList, startPosition, callback); - - } - - @Override - public void postBatchData(ByteString signerId, int batchId, List batchChunkList, FutureCallback callback) { - - postBatchData(signerId, batchId, batchChunkList, 0, callback); - - } - - @Override - public void closeBatch(CloseBatchMessage closeBatchMessage, FutureCallback callback) { + public void closeBatch(BatchIdentifier payload, Timestamp timestamp, Iterable signatures, FutureCallback 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 callback) { + public void readBatch(MessageID msgID, FutureCallback 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 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 ==== diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerBeginBatchWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerBeginBatchWorker.java index e0e92bb..793fc6e 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerBeginBatchWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerBeginBatchWorker.java @@ -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 { +public class MultiServerBeginBatchWorker extends MultiServerWorker, BatchIdentifier> { + + private BatchIdentifier[] identifiers; + private AtomicInteger remainingServers; public MultiServerBeginBatchWorker(List clients, - int minServers, BeginBatchMessage payload, int maxRetry, - FutureCallback futureCallback) { + int minServers, Iterable payload, int maxRetry, + FutureCallback 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 { + + 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++; + + } + + } } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerCloseBatchWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerCloseBatchWorker.java index 300440f..68aa8ef 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerCloseBatchWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerCloseBatchWorker.java @@ -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 { +public class MultiServerCloseBatchWorker extends MultiServerGenericPostWorker { - public MultiServerCloseBatchWorker(List clients, - int minServers, CloseBatchMessage payload, int maxRetry, - FutureCallback futureCallback) { + private final Timestamp timestamp; + private final Iterable signatures; + + public MultiServerCloseBatchWorker(List clients, int minServers, + BatchIdentifier payload, Timestamp timestamp, Iterable signatures, + int maxRetry, FutureCallback 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); } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java index 8172e14..a720eda 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericPostWorker.java @@ -35,14 +35,9 @@ public abstract class MultiServerGenericPostWorker extends MultiServerWorker< public void run() { // Iterate through servers - - Iterator clientIterator = getClientIterator(); - - while (clientIterator.hasNext()) { + for (SingleServerBulletinBoardClient client : clients) { // Send request to Server - SingleServerBulletinBoardClient client = clientIterator.next(); - doPost(client, payload); } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java index 88b4ac1..f17708b 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGenericReadWorker.java @@ -14,7 +14,9 @@ import java.util.List; */ public abstract class MultiServerGenericReadWorker extends MultiServerWorker{ - private final Iterator clientIterator; + private Iterator clientIterator; + + private String errorString; public MultiServerGenericReadWorker(List clients, int minServers, IN payload, int maxRetry, @@ -22,7 +24,8 @@ public abstract class MultiServerGenericReadWorker 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 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 extends MultiServerW @Override public void onFailure(Throwable t) { + //TODO: log + errorString += t.getCause() + " " + t.getMessage() + "\n"; run(); // Retry with next server } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java index 748916b..b76a94f 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerGetRedundancyWorker.java @@ -36,13 +36,8 @@ public class MultiServerGetRedundancyWorker extends MultiServerWorker clientIterator = getClientIterator(); - // Iterate through clients - - while (clientIterator.hasNext()) { - - SingleServerBulletinBoardClient client = clientIterator.next(); + for (SingleServerBulletinBoardClient client : clients) { // Send request to client client.getRedundancy(payload,this); diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchDataWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchDataWorker.java index ca7b4fd..052b625 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchDataWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchDataWorker.java @@ -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 { +public class MultiServerPostBatchDataWorker extends MultiServerWorker { public MultiServerPostBatchDataWorker(List 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 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); + } + } } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchWorker.java index 5b5198c..b938f52 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerPostBatchWorker.java @@ -11,17 +11,23 @@ import java.util.List; */ public class MultiServerPostBatchWorker extends MultiServerGenericPostWorker { + private final int chunkSize; + public MultiServerPostBatchWorker(List clients, - int minServers, BulletinBoardMessage payload, int maxRetry, + int minServers, BulletinBoardMessage payload, int chunkSize, int maxRetry, FutureCallback 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); + } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchDataWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchDataWorker.java new file mode 100644 index 0000000..959c60f --- /dev/null +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchDataWorker.java @@ -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 { + + public MultiServerReadBatchDataWorker(List clients, + int minServers, MessageID payload, int maxRetry, + FutureCallback futureCallback) { + + super(clients, minServers, payload, maxRetry, futureCallback); + + } + + @Override + protected void doRead(MessageID payload, SingleServerBulletinBoardClient client) { + client.readBatch(payload, this); + } + + +} diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchWorker.java index db58247..59b4ce5 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/multiserver/MultiServerReadBatchWorker.java @@ -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 { +public class MultiServerReadBatchWorker extends MultiServerGenericReadWorker { public MultiServerReadBatchWorker(List clients, - int minServers, BatchSpecificationMessage payload, int maxRetry, + int minServers, MessageID payload, int maxRetry, FutureCallback futureCallback) { super(clients, minServers, payload, maxRetry, futureCallback); @@ -22,7 +22,7 @@ public class MultiServerReadBatchWorker extends MultiServerGenericReadWorker { +public class SingleServerBeginBatchWorker extends SingleServerWorker { 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(); + } + } } diff --git a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerReadBatchWorker.java b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerReadBatchWorker.java index ee2e193..8bc4bcd 100644 --- a/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerReadBatchWorker.java +++ b/bulletin-board-client/src/main/java/meerkat/bulletinboard/workers/singleserver/SingleServerReadBatchWorker.java @@ -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> { +public class SingleServerReadBatchWorker extends SingleServerWorker> { - public SingleServerReadBatchWorker(String serverAddress, BatchSpecificationMessage payload, int maxRetry) { + public SingleServerReadBatchWorker(String serverAddress, BatchQuery payload, int maxRetry) { super(serverAddress, payload, maxRetry); } diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java index 6661491..dcea3e7 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/BulletinBoardSQLServer.java @@ -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 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 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 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); diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java index 2f31e9e..872e226 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/H2QueryProvider.java @@ -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 getSchemaCreationCommands() { List list = new LinkedList(); - 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"); diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java index 2931a34..097095f 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/MySQLQueryProvider.java @@ -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 getSchemaDeletionCommands() { List list = new LinkedList(); - 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"); diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/SQLiteQueryProvider.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/SQLiteQueryProvider.java index 0462543..9f68955 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/SQLiteQueryProvider.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/SQLiteQueryProvider.java @@ -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; } diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java index 1f9c459..e9174e5 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/sqlserver/mappers/MessageStubMapper.java @@ -21,7 +21,7 @@ public class MessageStubMapper implements RowMapper { 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(); diff --git a/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java b/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java index 7389090..f9d3a94 100644 --- a/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java +++ b/bulletin-board-server/src/main/java/meerkat/bulletinboard/webapp/BulletinBoardWebApp.java @@ -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); diff --git a/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java b/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java index a5997a3..eafd80c 100644 --- a/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java +++ b/bulletin-board-server/src/test/java/meerkat/bulletinboard/GenericBulletinBoardServerTest.java @@ -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 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); diff --git a/bulletin-board-server/src/test/java/meerkat/bulletinboard/H2BulletinBoardServerTest.java b/bulletin-board-server/src/test/java/meerkat/bulletinboard/H2BulletinBoardServerTest.java index 4ee2282..fa96635 100644 --- a/bulletin-board-server/src/test/java/meerkat/bulletinboard/H2BulletinBoardServerTest.java +++ b/bulletin-board-server/src/test/java/meerkat/bulletinboard/H2BulletinBoardServerTest.java @@ -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() { diff --git a/bulletin-board-server/src/test/java/meerkat/bulletinboard/MySQLBulletinBoardServerTest.java b/bulletin-board-server/src/test/java/meerkat/bulletinboard/MySQLBulletinBoardServerTest.java index c4bd7fa..334620c 100644 --- a/bulletin-board-server/src/test/java/meerkat/bulletinboard/MySQLBulletinBoardServerTest.java +++ b/bulletin-board-server/src/test/java/meerkat/bulletinboard/MySQLBulletinBoardServerTest.java @@ -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() { diff --git a/bulletin-board-server/src/test/java/meerkat/bulletinboard/SQLiteBulletinBoardServerTest.java b/bulletin-board-server/src/test/java/meerkat/bulletinboard/SQLiteBulletinBoardServerTest.java index 1d7aae0..4dcd97b 100644 --- a/bulletin-board-server/src/test/java/meerkat/bulletinboard/SQLiteBulletinBoardServerTest.java +++ b/bulletin-board-server/src/test/java/meerkat/bulletinboard/SQLiteBulletinBoardServerTest.java @@ -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"); diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/AsyncBulletinBoardClient.java b/meerkat-common/src/main/java/meerkat/bulletinboard/AsyncBulletinBoardClient.java index fcc6394..42af8ef 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/AsyncBulletinBoardClient.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/AsyncBulletinBoardClient.java @@ -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 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 callback); + public MessageID postAsBatch(BulletinBoardMessage completeBatch, int chunkSize, FutureCallback 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 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 callback); + public void beginBatch(Iterable tags, FutureCallback 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 batchChunkList, - int startPosition, FutureCallback callback); + public void postBatchData(BatchIdentifier batchIdentifier, List batchChunkList, + int startPosition, FutureCallback 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 batchChunkList, FutureCallback callback); - - /** - * Overloading of the postBatchData method which uses ByteString - */ - public void postBatchData(ByteString signerId, int batchId, List batchChunkList, - int startPosition, FutureCallback 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 batchChunkList, FutureCallback callback); + public void postBatchData(BatchIdentifier batchIdentifier, List batchChunkList, FutureCallback 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 callback); + public void closeBatch(BatchIdentifier batchIdentifier, Timestamp timestamp, Iterable signatures, FutureCallback 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 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 callback) throws IllegalArgumentException; + /** * Perform a Sync Query on the bulletin board diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardClient.java b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardClient.java index 9ce3943..142bb35 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardClient.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardClient.java @@ -45,14 +45,32 @@ public interface BulletinBoardClient { */ List 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) diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java index 3d33704..8121b85 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/BulletinBoardServer.java @@ -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 diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardDigest.java b/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardDigest.java index 9930ab3..fb7bd3f 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardDigest.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardDigest.java @@ -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()); } diff --git a/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardSignature.java b/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardSignature.java index e818071..aad6466 100644 --- a/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardSignature.java +++ b/meerkat-common/src/main/java/meerkat/bulletinboard/GenericBulletinBoardSignature.java @@ -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()); } diff --git a/meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageComparator.java b/meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageComparator.java index ec9b2e0..7005568 100644 --- a/meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageComparator.java +++ b/meerkat-common/src/main/java/meerkat/util/BulletinBoardMessageComparator.java @@ -47,16 +47,20 @@ public class BulletinBoardMessageComparator implements Comparator