Commit e8ad1121 authored by Torsten Grote's avatar Torsten Grote
Browse files

Merge branch '266-message-queues' into 'master'

Message queues. #266

A message queue is a group shared by two devices that delivers messages from each device to the other in order. The first 64 bits of the message body contain a sequence number that's incremented for each message sent in a given direction. The incoming and outgoing sequence numbers and information about any messages received out of order and waiting to be delivered are stored in the group metadata.

See merge request !121
parents e458853f 02eca8ec
package org.briarproject.api.clients;
import org.briarproject.api.db.DbException;
import org.briarproject.api.db.Metadata;
import org.briarproject.api.db.Transaction;
import org.briarproject.api.sync.ClientId;
import org.briarproject.api.sync.Group;
public interface MessageQueueManager {
/**
* The key used for storing the queue's state in the group metadata.
*/
String QUEUE_STATE_KEY = "queueState";
/**
* Sends a message using the given queue.
*/
QueueMessage sendMessage(Transaction txn, Group queue, long timestamp,
byte[] body, Metadata meta) throws DbException;
/**
* Sets the message validator for the given client.
*/
void registerMessageValidator(ClientId c, QueueMessageValidator v);
/**
* Sets the incoming message hook for the given client. The hook will be
* called once for each incoming message that passes validation. Messages
* are passed to the hook in order.
*/
void registerIncomingMessageHook(ClientId c, IncomingQueueMessageHook hook);
interface QueueMessageValidator {
/**
* Validates the given message and returns its metadata if the message
* is valid, or null if the message is invalid.
*/
Metadata validateMessage(QueueMessage q, Group g);
}
interface IncomingQueueMessageHook {
/**
* Called once for each incoming message that passes validation.
* Messages are passed to the hook in order.
*/
void incomingMessage(Transaction txn, QueueMessage q, Metadata meta)
throws DbException;
}
}
package org.briarproject.api.clients;
import org.briarproject.api.sync.GroupId;
import org.briarproject.api.sync.Message;
import org.briarproject.api.sync.MessageId;
import static org.briarproject.api.sync.SyncConstants.MAX_MESSAGE_BODY_LENGTH;
import static org.briarproject.api.sync.SyncConstants.MESSAGE_HEADER_LENGTH;
public class QueueMessage extends Message {
public static final int QUEUE_MESSAGE_HEADER_LENGTH =
MESSAGE_HEADER_LENGTH + 8;
public static final int MAX_QUEUE_MESSAGE_BODY_LENGTH =
MAX_MESSAGE_BODY_LENGTH - 8;
private final long queuePosition;
public QueueMessage(MessageId id, GroupId groupId, long timestamp,
long queuePosition, byte[] raw) {
super(id, groupId, timestamp, raw);
this.queuePosition = queuePosition;
}
public long getQueuePosition() {
return queuePosition;
}
}
package org.briarproject.api.clients;
import org.briarproject.api.sync.GroupId;
import org.briarproject.api.sync.MessageId;
public interface QueueMessageFactory {
QueueMessage createMessage(GroupId groupId, long timestamp,
long queuePosition, byte[] body);
QueueMessage createMessage(MessageId id, byte[] raw);
}
package org.briarproject.api.sync;
import org.briarproject.api.db.Metadata;
public interface MessageValidator {
/**
* Validates the given message and returns its metadata if the message
* is valid, or null if the message is invalid.
*/
Metadata validateMessage(Message m, Group g);
}
......@@ -30,14 +30,32 @@ public interface ValidationManager {
}
}
/** Sets the message validator for the given client. */
/**
* Sets the message validator for the given client.
*/
void registerMessageValidator(ClientId c, MessageValidator v);
/** Registers a hook to be called whenever a message is validated. */
void registerValidationHook(ValidationHook hook);
/**
* Sets the incoming message hook for the given client. The hook will be
* called once for each incoming message that passes validation.
*/
void registerIncomingMessageHook(ClientId c, IncomingMessageHook hook);
interface ValidationHook {
void validatingMessage(Transaction txn, Message m, ClientId c,
Metadata meta) throws DbException;
interface MessageValidator {
/**
* Validates the given message and returns its metadata if the message
* is valid, or null if the message is invalid.
*/
Metadata validateMessage(Message m, Group g);
}
interface IncomingMessageHook {
/**
* Called once for each incoming message that passes validation.
*/
void incomingMessage(Transaction txn, Message m, Metadata meta)
throws DbException;
}
}
package org.briarproject.clients;
import org.briarproject.api.FormatException;
import org.briarproject.api.clients.ClientHelper;
import org.briarproject.api.clients.MessageQueueManager.IncomingQueueMessageHook;
import org.briarproject.api.clients.QueueMessage;
import org.briarproject.api.data.BdfDictionary;
import org.briarproject.api.data.BdfList;
import org.briarproject.api.data.MetadataParser;
import org.briarproject.api.db.DbException;
import org.briarproject.api.db.Metadata;
import org.briarproject.api.db.Transaction;
import org.briarproject.api.sync.Message;
import org.briarproject.api.sync.ValidationManager.IncomingMessageHook;
import org.briarproject.api.system.Clock;
import static org.briarproject.api.clients.QueueMessage.QUEUE_MESSAGE_HEADER_LENGTH;
import static org.briarproject.api.sync.SyncConstants.MESSAGE_HEADER_LENGTH;
public abstract class BdfIncomingMessageHook implements IncomingMessageHook,
IncomingQueueMessageHook {
protected final ClientHelper clientHelper;
protected final MetadataParser metadataParser;
protected BdfIncomingMessageHook(ClientHelper clientHelper,
MetadataParser metadataParser, Clock clock) {
this.clientHelper = clientHelper;
this.metadataParser = metadataParser;
}
protected abstract void incomingMessage(Transaction txn, Message m,
BdfList body, BdfDictionary meta) throws DbException,
FormatException;
@Override
public void incomingMessage(Transaction txn, Message m, Metadata meta)
throws DbException {
incomingMessage(txn, m, meta, MESSAGE_HEADER_LENGTH);
}
@Override
public void incomingMessage(Transaction txn, QueueMessage q, Metadata meta)
throws DbException {
incomingMessage(txn, q, meta, QUEUE_MESSAGE_HEADER_LENGTH);
}
private void incomingMessage(Transaction txn, Message m, Metadata meta,
int headerLength) throws DbException {
try {
byte[] raw = m.getRaw();
BdfList body = clientHelper.toList(raw, headerLength,
raw.length - headerLength);
BdfDictionary metaDictionary = metadataParser.parse(meta);
incomingMessage(txn, m, body, metaDictionary);
} catch (FormatException e) {
throw new DbException(e);
}
}
}
......@@ -2,22 +2,26 @@ package org.briarproject.clients;
import org.briarproject.api.FormatException;
import org.briarproject.api.clients.ClientHelper;
import org.briarproject.api.clients.MessageQueueManager.QueueMessageValidator;
import org.briarproject.api.clients.QueueMessage;
import org.briarproject.api.data.BdfDictionary;
import org.briarproject.api.data.BdfList;
import org.briarproject.api.data.MetadataEncoder;
import org.briarproject.api.db.Metadata;
import org.briarproject.api.sync.Group;
import org.briarproject.api.sync.Message;
import org.briarproject.api.sync.MessageValidator;
import org.briarproject.api.sync.ValidationManager.MessageValidator;
import org.briarproject.api.system.Clock;
import org.briarproject.util.StringUtils;
import java.util.logging.Logger;
import static org.briarproject.api.clients.QueueMessage.QUEUE_MESSAGE_HEADER_LENGTH;
import static org.briarproject.api.sync.SyncConstants.MESSAGE_HEADER_LENGTH;
import static org.briarproject.api.transport.TransportConstants.MAX_CLOCK_DIFFERENCE;
public abstract class BdfMessageValidator implements MessageValidator {
public abstract class BdfMessageValidator implements MessageValidator,
QueueMessageValidator {
protected static final Logger LOG =
Logger.getLogger(BdfMessageValidator.class.getName());
......@@ -33,11 +37,20 @@ public abstract class BdfMessageValidator implements MessageValidator {
this.clock = clock;
}
protected abstract BdfDictionary validateMessage(BdfList message, Group g,
long timestamp) throws FormatException;
protected abstract BdfDictionary validateMessage(Message m, Group g,
BdfList body) throws FormatException;
@Override
public Metadata validateMessage(Message m, Group g) {
return validateMessage(m, g, MESSAGE_HEADER_LENGTH);
}
@Override
public Metadata validateMessage(QueueMessage q, Group g) {
return validateMessage(q, g, QUEUE_MESSAGE_HEADER_LENGTH);
}
private Metadata validateMessage(Message m, Group g, int headerLength) {
// Reject the message if it's too far in the future
long now = clock.currentTimeMillis();
if (m.getTimestamp() - now > MAX_CLOCK_DIFFERENCE) {
......@@ -45,10 +58,14 @@ public abstract class BdfMessageValidator implements MessageValidator {
return null;
}
byte[] raw = m.getRaw();
if (raw.length <= headerLength) {
LOG.info("Message is too short");
return null;
}
try {
BdfList message = clientHelper.toList(raw, MESSAGE_HEADER_LENGTH,
raw.length - MESSAGE_HEADER_LENGTH);
BdfDictionary meta = validateMessage(message, g, m.getTimestamp());
BdfList body = clientHelper.toList(raw, headerLength,
raw.length - headerLength);
BdfDictionary meta = validateMessage(m, g, body);
if (meta == null) {
LOG.info("Invalid message");
return null;
......@@ -87,7 +104,7 @@ public abstract class BdfMessageValidator implements MessageValidator {
}
protected void checkSize(BdfList list, int minSize, int maxSize)
throws FormatException {
throws FormatException {
if (list != null) {
if (list.size() < minSize) throw new FormatException();
if (list.size() > maxSize) throw new FormatException();
......
......@@ -3,13 +3,17 @@ package org.briarproject.clients;
import com.google.inject.AbstractModule;
import org.briarproject.api.clients.ClientHelper;
import org.briarproject.api.clients.MessageQueueManager;
import org.briarproject.api.clients.PrivateGroupFactory;
import org.briarproject.api.clients.QueueMessageFactory;
public class ClientsModule extends AbstractModule {
@Override
protected void configure() {
bind(ClientHelper.class).to(ClientHelperImpl.class);
bind(MessageQueueManager.class).to(MessageQueueManagerImpl.class);
bind(PrivateGroupFactory.class).to(PrivateGroupFactoryImpl.class);
bind(QueueMessageFactory.class).to(QueueMessageFactoryImpl.class);
}
}
package org.briarproject.clients;
import org.briarproject.api.FormatException;
import org.briarproject.api.clients.ClientHelper;
import org.briarproject.api.clients.MessageQueueManager;
import org.briarproject.api.clients.QueueMessage;
import org.briarproject.api.clients.QueueMessageFactory;
import org.briarproject.api.data.BdfDictionary;
import org.briarproject.api.data.BdfList;
import org.briarproject.api.db.DatabaseComponent;
import org.briarproject.api.db.DbException;
import org.briarproject.api.db.Metadata;
import org.briarproject.api.db.Transaction;
import org.briarproject.api.sync.ClientId;
import org.briarproject.api.sync.Group;
import org.briarproject.api.sync.GroupId;
import org.briarproject.api.sync.Message;
import org.briarproject.api.sync.MessageId;
import org.briarproject.api.sync.ValidationManager;
import org.briarproject.api.sync.ValidationManager.IncomingMessageHook;
import org.briarproject.util.ByteUtils;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.TreeMap;
import java.util.logging.Logger;
import javax.inject.Inject;
import static java.util.logging.Level.INFO;
import static org.briarproject.api.clients.QueueMessage.QUEUE_MESSAGE_HEADER_LENGTH;
import static org.briarproject.api.sync.SyncConstants.MESSAGE_HEADER_LENGTH;
class MessageQueueManagerImpl implements MessageQueueManager {
private static final String OUTGOING_POSITION_KEY = "nextOut";
private static final String INCOMING_POSITION_KEY = "nextIn";
private static final String PENDING_MESSAGES_KEY = "pending";
private static final Logger LOG =
Logger.getLogger(MessageQueueManagerImpl.class.getName());
private final DatabaseComponent db;
private final ClientHelper clientHelper;
private final QueueMessageFactory queueMessageFactory;
private final ValidationManager validationManager;
@Inject
MessageQueueManagerImpl(DatabaseComponent db, ClientHelper clientHelper,
QueueMessageFactory queueMessageFactory,
ValidationManager validationManager) {
this.db = db;
this.clientHelper = clientHelper;
this.queueMessageFactory = queueMessageFactory;
this.validationManager = validationManager;
}
@Override
public QueueMessage sendMessage(Transaction txn, Group queue,
long timestamp, byte[] body, Metadata meta) throws DbException {
QueueState queueState = loadQueueState(txn, queue.getId());
long queuePosition = queueState.outgoingPosition;
queueState.outgoingPosition++;
saveQueueState(txn, queue.getId(), queueState);
QueueMessage q = queueMessageFactory.createMessage(queue.getId(),
timestamp, queuePosition, body);
db.addLocalMessage(txn, q, queue.getClientId(), meta, true);
return q;
}
@Override
public void registerMessageValidator(ClientId c, QueueMessageValidator v) {
validationManager.registerMessageValidator(c,
new DelegatingMessageValidator(v));
}
@Override
public void registerIncomingMessageHook(ClientId c,
IncomingQueueMessageHook hook) {
validationManager.registerIncomingMessageHook(c,
new DelegatingIncomingMessageHook(hook));
}
private QueueState loadQueueState(Transaction txn, GroupId g)
throws DbException {
try {
TreeMap<Long, MessageId> pending = new TreeMap<Long, MessageId>();
Metadata groupMeta = db.getGroupMetadata(txn, g);
byte[] raw = groupMeta.get(QUEUE_STATE_KEY);
if (raw == null) return new QueueState(0, 0, pending);
BdfDictionary d = clientHelper.toDictionary(raw, 0, raw.length);
long outgoingPosition = d.getLong(OUTGOING_POSITION_KEY);
long incomingPosition = d.getLong(INCOMING_POSITION_KEY);
BdfList pendingList = d.getList(PENDING_MESSAGES_KEY);
for (int i = 0; i < pendingList.size(); i++) {
BdfList item = pendingList.getList(i);
if (item.size() != 2) throw new FormatException();
pending.put(item.getLong(0), new MessageId(item.getRaw(1)));
}
return new QueueState(outgoingPosition, incomingPosition, pending);
} catch (FormatException e) {
throw new DbException(e);
}
}
private void saveQueueState(Transaction txn, GroupId g,
QueueState queueState) throws DbException {
try {
BdfDictionary d = new BdfDictionary();
d.put(OUTGOING_POSITION_KEY, queueState.outgoingPosition);
d.put(INCOMING_POSITION_KEY, queueState.incomingPosition);
BdfList pendingList = new BdfList();
for (Entry<Long, MessageId> e : queueState.pending.entrySet())
pendingList.add(BdfList.of(e.getKey(), e.getValue()));
d.put(PENDING_MESSAGES_KEY, pendingList);
Metadata groupMeta = new Metadata();
groupMeta.put(QUEUE_STATE_KEY, clientHelper.toByteArray(d));
db.mergeGroupMetadata(txn, g, groupMeta);
} catch (FormatException e) {
throw new RuntimeException(e);
}
}
private static class QueueState {
private long outgoingPosition, incomingPosition;
private final TreeMap<Long, MessageId> pending;
QueueState(long outgoingPosition, long incomingPosition,
TreeMap<Long, MessageId> pending) {
this.outgoingPosition = outgoingPosition;
this.incomingPosition = incomingPosition;
this.pending = pending;
}
MessageId popIncomingMessageId() {
Iterator<Entry<Long, MessageId>> it = pending.entrySet().iterator();
if (!it.hasNext()) return null;
Entry<Long, MessageId> e = it.next();
if (!e.getKey().equals(incomingPosition)) return null;
it.remove();
incomingPosition++;
return e.getValue();
}
}
private static class DelegatingMessageValidator
implements ValidationManager.MessageValidator {
private final QueueMessageValidator delegate;
DelegatingMessageValidator(QueueMessageValidator delegate) {
this.delegate = delegate;
}
@Override
public Metadata validateMessage(Message m, Group g) {
byte[] raw = m.getRaw();
if (raw.length < QUEUE_MESSAGE_HEADER_LENGTH) return null;
long queuePosition = ByteUtils.readUint64(raw,
MESSAGE_HEADER_LENGTH);
if (queuePosition < 0) return null;
QueueMessage q = new QueueMessage(m.getId(), m.getGroupId(),
m.getTimestamp(), queuePosition, raw);
return delegate.validateMessage(q, g);
}
}
private class DelegatingIncomingMessageHook implements IncomingMessageHook {
private final IncomingQueueMessageHook delegate;
DelegatingIncomingMessageHook(IncomingQueueMessageHook delegate) {
this.delegate = delegate;
}
@Override
public void incomingMessage(Transaction txn, Message m, Metadata meta)
throws DbException {
long queuePosition = ByteUtils.readUint64(m.getRaw(),
MESSAGE_HEADER_LENGTH);
QueueState queueState = loadQueueState(txn, m.getGroupId());
if (LOG.isLoggable(INFO)) {
LOG.info("Received message with position "
+ queuePosition + ", expecting "
+ queueState.incomingPosition);
}
if (queuePosition < queueState.incomingPosition) {
// A message with this queue position has already been seen
LOG.warning("Deleting message with duplicate position");
db.deleteMessage(txn, m.getId());
db.deleteMessageMetadata(txn, m.getId());
} else if (queuePosition > queueState.incomingPosition) {
// The message is out of order, add it to the pending list
LOG.info("Message is out of order, adding to pending list");
queueState.pending.put(queuePosition, m.getId());
saveQueueState(txn, m.getGroupId(), queueState);
} else {
// The message is in order, pass it to the delegate
LOG.info("Message is in order, delivering");
QueueMessage q = new QueueMessage(m.getId(), m.getGroupId(),
m.getTimestamp(), queuePosition, m.getRaw());
delegate.incomingMessage(txn, q, meta);
queueState.incomingPosition++;
// Pass any consecutive messages to the delegate
MessageId id;
while ((id = queueState.popIncomingMessageId()) != null) {
byte[] raw = db.getRawMessage(txn, id);
meta = db.getMessageMetadata(txn, id);
q = queueMessageFactory.createMessage(id, raw);
if (LOG.isLoggable(INFO)) {
LOG.info("Delivering pending message with position "
+ q.getQueuePosition());
}
delegate.incomingMessage(txn, q, meta);
}
saveQueueState(txn, m.getGroupId(), queueState);
}
}
}
}
......@@ -3,28 +3,26 @@ package org.briarproject.clients;
import com.google.inject.Inject;
import org.briarproject.api.Bytes;
import org.briarproject.api.FormatException;
import org.briarproject.api.clients.ClientHelper;
import org.briarproject.api.clients.PrivateGroupFactory;
import org.briarproject.api.contact.Contact;
import org.briarproject.api.data.BdfWriter;
import org.briarproject.api.data.BdfWriterFactory;
import org.briarproject.api.data.BdfList;
import org.briarproject.api.identity.AuthorId;
import org.briarproject.api.sync.ClientId;
import org.briarproject.api.sync.Group;
import org.briarproject.api.sync.GroupFactory;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
class PrivateGroupFactoryImpl implements PrivateGroupFactory {
private final GroupFactory groupFactory;
private final BdfWriterFactory bdfWriterFactory;
private final ClientHelper clientHelper;
@Inject
PrivateGroupFactoryImpl(GroupFactory groupFactory,
BdfWriterFactory bdfWriterFactory) {
ClientHelper clientHelper) {
this.groupFactory = groupFactory;
this.bdfWriterFactory = bdfWriterFactory;
this.clientHelper = clientHelper;
}
@Override
......@@ -36,22 +34,12 @@ class PrivateGroupFactoryImpl implements PrivateGroupFactory {
}
private byte[] createGroupDescriptor(AuthorId local, AuthorId remote) {
ByteArrayOutputStream out = new ByteArrayOutputStream();
BdfWriter w = bdfWriterFactory.createWriter(out);
try {
w.writeListStart();
if (Bytes.COMPARATOR.compare(local, remote) < 0) {
w.writeRaw(local.getBytes());
w.writeRaw(remote.getBytes());
} else {
w.writeRaw(remote.getBytes());