Added support for registering listeners with the database that are

called when new messages are available, and a new method
hasSendableMessages(ContactId) that listeners can call to see whether
it's worth trying to create a batch.
This commit is contained in:
akwizgran
2011-07-27 20:27:43 +01:00
parent e93fbe0b20
commit adee3e121c
10 changed files with 364 additions and 99 deletions

View File

@@ -48,6 +48,12 @@ public interface DatabaseComponent {
/** Waits for any open transactions to finish and closes the database. */ /** Waits for any open transactions to finish and closes the database. */
void close() throws DbException; void close() throws DbException;
/** Adds a listener to be notified when new messages are available. */
void addListener(MessageListener m);
/** Removes a listener. */
void removeListener(MessageListener m);
/** /**
* Adds a new contact to the database with the given transport details and * Adds a new contact to the database with the given transport details and
* returns an ID for the contact. * returns an ID for the contact.
@@ -112,6 +118,9 @@ public interface DatabaseComponent {
/** Returns the contacts to which the given group is visible. */ /** Returns the contacts to which the given group is visible. */
Collection<ContactId> getVisibility(GroupId g) throws DbException; Collection<ContactId> getVisibility(GroupId g) throws DbException;
/** Returns true if any messages are sendable to the given contact. */
boolean hasSendableMessages(ContactId c) throws DbException;
/** Processes an acknowledgement from the given contact. */ /** Processes an acknowledgement from the given contact. */
void receiveAck(ContactId c, Ack a) throws DbException; void receiveAck(ContactId c, Ack a) throws DbException;

View File

@@ -0,0 +1,10 @@
package net.sf.briar.api.db;
/**
* An interface for receiving notifications when the database may have new
* messages available.
*/
public interface MessageListener {
void messagesAdded();
}

View File

@@ -109,28 +109,28 @@ interface Database<T> {
void addSubscription(T txn, Group g) throws DbException; void addSubscription(T txn, Group g) throws DbException;
/** /**
* Returns true iff the database contains the given contact. * Returns true if the database contains the given contact.
* <p> * <p>
* Locking: contacts read. * Locking: contacts read.
*/ */
boolean containsContact(T txn, ContactId c) throws DbException; boolean containsContact(T txn, ContactId c) throws DbException;
/** /**
* Returns true iff the database contains the given message. * Returns true if the database contains the given message.
* <p> * <p>
* Locking: messages read. * Locking: messages read.
*/ */
boolean containsMessage(T txn, MessageId m) throws DbException; boolean containsMessage(T txn, MessageId m) throws DbException;
/** /**
* Returns true iff the user is subscribed to the given group. * Returns true if the user is subscribed to the given group.
* <p> * <p>
* Locking: subscriptions read. * Locking: subscriptions read.
*/ */
boolean containsSubscription(T txn, GroupId g) throws DbException; boolean containsSubscription(T txn, GroupId g) throws DbException;
/** /**
* Returns true iff the user is subscribed to the given group and the * Returns true if the user is subscribed to the given group and the
* group is visible to the given contact. * group is visible to the given contact.
* <p> * <p>
* Locking: contacts read, subscriptions read. * Locking: contacts read, subscriptions read.
@@ -189,7 +189,8 @@ interface Database<T> {
* if the message is not present in the database or is not sendable to the * if the message is not present in the database or is not sendable to the
* given contact. * given contact.
* <p> * <p>
* Locking: contacts read, messages read, messageStatuses read. * Locking: contacts read, messages read, messageStatuses read,
* subscriptions read.
*/ */
byte[] getMessageIfSendable(T txn, ContactId c, MessageId m) byte[] getMessageIfSendable(T txn, ContactId c, MessageId m)
throws DbException; throws DbException;
@@ -246,7 +247,8 @@ interface Database<T> {
* Returns the IDs of some messages that are eligible to be sent to the * Returns the IDs of some messages that are eligible to be sent to the
* given contact, with a total size less than or equal to the given size. * given contact, with a total size less than or equal to the given size.
* <p> * <p>
* Locking: contacts read, messages read, messageStatuses read. * Locking: contacts read, messages read, messageStatuses read,
* subscriptions read.
*/ */
Collection<MessageId> getSendableMessages(T txn, ContactId c, int size) Collection<MessageId> getSendableMessages(T txn, ContactId c, int size)
throws DbException; throws DbException;
@@ -293,6 +295,13 @@ interface Database<T> {
Collection<Group> getVisibleSubscriptions(T txn, ContactId c) Collection<Group> getVisibleSubscriptions(T txn, ContactId c)
throws DbException; throws DbException;
/**
* Returns true if any messages are sendable to the given contact.
* <p>
* Locking: contacts read, messages read, messageStatuses read.
*/
boolean hasSendableMessages(T txn, ContactId c) throws DbException;
/** /**
* Removes an outstanding batch that has been acknowledged. Any messages in * Removes an outstanding batch that has been acknowledged. Any messages in
* the batch that are still considered outstanding (Status.SENT) with * the batch that are still considered outstanding (Status.SENT) with

View File

@@ -25,7 +25,7 @@ interface DatabaseCleaner {
void checkFreeSpaceAndClean() throws DbException; void checkFreeSpaceAndClean() throws DbException;
/** /**
* Returns true iff the amount of free storage space available to the * Returns true if the amount of free storage space available to the
* database should be checked. * database should be checked.
*/ */
boolean shouldCheckFreeSpace(); boolean shouldCheckFreeSpace();

View File

@@ -1,5 +1,8 @@
package net.sf.briar.db; package net.sf.briar.db;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.logging.Level; import java.util.logging.Level;
import java.util.logging.Logger; import java.util.logging.Logger;
@@ -7,6 +10,7 @@ import net.sf.briar.api.ContactId;
import net.sf.briar.api.Rating; import net.sf.briar.api.Rating;
import net.sf.briar.api.db.DatabaseComponent; import net.sf.briar.api.db.DatabaseComponent;
import net.sf.briar.api.db.DbException; import net.sf.briar.api.db.DbException;
import net.sf.briar.api.db.MessageListener;
import net.sf.briar.api.db.Status; import net.sf.briar.api.db.Status;
import net.sf.briar.api.protocol.AuthorId; import net.sf.briar.api.protocol.AuthorId;
import net.sf.briar.api.protocol.Message; import net.sf.briar.api.protocol.Message;
@@ -25,6 +29,8 @@ DatabaseCleaner.Callback {
protected final Database<Txn> db; protected final Database<Txn> db;
protected final DatabaseCleaner cleaner; protected final DatabaseCleaner cleaner;
private final List<MessageListener> listeners =
new ArrayList<MessageListener>(); // Locking: self
private final Object spaceLock = new Object(); private final Object spaceLock = new Object();
private final Object writeLock = new Object(); private final Object writeLock = new Object();
private long bytesStoredSinceLastCheck = 0L; // Locking: spaceLock private long bytesStoredSinceLastCheck = 0L; // Locking: spaceLock
@@ -41,6 +47,18 @@ DatabaseCleaner.Callback {
cleaner.startCleaning(); cleaner.startCleaning();
} }
public void addListener(MessageListener m) {
synchronized(listeners) {
listeners.add(m);
}
}
public void removeListener(MessageListener m) {
synchronized(listeners) {
listeners.remove(m);
}
}
/** /**
* Removes the oldest messages from the database, with a total size less * Removes the oldest messages from the database, with a total size less
* than or equal to the given size. * than or equal to the given size.
@@ -61,6 +79,18 @@ DatabaseCleaner.Callback {
return sendability; return sendability;
} }
/** Notifies all MessageListeners that new messages may be available. */
protected void callMessageListeners() {
synchronized(listeners) {
if(!listeners.isEmpty()) {
// Shuffle the listeners so we don't always send new messages
// to contacts in the same order
Collections.shuffle(listeners);
for(MessageListener m : listeners) m.messagesAdded();
}
}
}
public void checkFreeSpaceAndClean() throws DbException { public void checkFreeSpaceAndClean() throws DbException {
long freeSpace = db.getFreeSpace(); long freeSpace = db.getFreeSpace();
while(freeSpace < MIN_FREE_SPACE) { while(freeSpace < MIN_FREE_SPACE) {
@@ -85,7 +115,7 @@ DatabaseCleaner.Callback {
} }
/** /**
* Returns true iff the database contains the given contact. * Returns true if the database contains the given contact.
* <p> * <p>
* Locking: contacts read. * Locking: contacts read.
*/ */
@@ -121,7 +151,7 @@ DatabaseCleaner.Callback {
if(bytesStoredSinceLastCheck > MAX_BYTES_BETWEEN_SPACE_CHECKS) { if(bytesStoredSinceLastCheck > MAX_BYTES_BETWEEN_SPACE_CHECKS) {
if(LOG.isLoggable(Level.FINE)) if(LOG.isLoggable(Level.FINE))
LOG.fine(bytesStoredSinceLastCheck LOG.fine(bytesStoredSinceLastCheck
+ " bytes stored since last check"); + " bytes stored since last check");
bytesStoredSinceLastCheck = 0L; bytesStoredSinceLastCheck = 0L;
timeOfLastCheck = now; timeOfLastCheck = now;
return true; return true;
@@ -234,7 +264,7 @@ DatabaseCleaner.Callback {
} }
if(LOG.isLoggable(Level.FINE)) if(LOG.isLoggable(Level.FINE))
LOG.fine(direct + " messages affected directly, " LOG.fine(direct + " messages affected directly, "
+ indirect + " indirectly"); + indirect + " indirectly");
} }
/** /**

View File

@@ -1226,6 +1226,42 @@ abstract class JdbcDatabase implements Database<Connection> {
} }
} }
public boolean hasSendableMessages(Connection txn, ContactId c)
throws DbException {
PreparedStatement ps = null;
ResultSet rs = null;
try {
String sql = "SELECT messages.messageId FROM messages"
+ " JOIN contactSubscriptions"
+ " ON messages.groupId = contactSubscriptions.groupId"
+ " JOIN visibilities"
+ " ON messages.groupId = visibilities.groupId"
+ " JOIN statuses ON messages.messageId = statuses.messageId"
+ " WHERE contactSubscriptions.contactId = ?"
+ " AND visibilities.contactId = ?"
+ " AND statuses.contactId = ?"
+ " AND status = ? AND sendability > ZERO()"
+ " LIMIT ?";
ps = txn.prepareStatement(sql);
ps.setInt(1, c.getInt());
ps.setInt(2, c.getInt());
ps.setInt(3, c.getInt());
ps.setShort(4, (short) Status.NEW.ordinal());
ps.setInt(5, 1);
rs = ps.executeQuery();
boolean found = rs.next();
assert !rs.next();
rs.close();
ps.close();
return found;
} catch(SQLException e) {
tryToClose(rs);
tryToClose(ps);
tryToClose(txn);
throw new DbException(e);
}
}
public void removeAckedBatch(Connection txn, ContactId c, BatchId b) public void removeAckedBatch(Connection txn, ContactId c, BatchId b)
throws DbException { throws DbException {
PreparedStatement ps = null; PreparedStatement ps = null;

View File

@@ -151,6 +151,7 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
public void addLocallyGeneratedMessage(Message m) throws DbException { public void addLocallyGeneratedMessage(Message m) throws DbException {
boolean added = false;
waitForPermissionToWrite(); waitForPermissionToWrite();
contactLock.readLock().lock(); contactLock.readLock().lock();
try { try {
@@ -165,7 +166,7 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
// Don't store the message if the user has // Don't store the message if the user has
// unsubscribed from the group // unsubscribed from the group
if(db.containsSubscription(txn, m.getGroup())) { if(db.containsSubscription(txn, m.getGroup())) {
boolean added = storeMessage(txn, m, null); added = storeMessage(txn, m, null);
if(!added) { if(!added) {
if(LOG.isLoggable(Level.FINE)) if(LOG.isLoggable(Level.FINE))
LOG.fine("Duplicate local message"); LOG.fine("Duplicate local message");
@@ -191,6 +192,8 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} finally { } finally {
contactLock.readLock().unlock(); contactLock.readLock().unlock();
} }
// Call the listeners outside the lock
if(added) callMessageListeners();
} }
public void findLostBatches(ContactId c) throws DbException { public void findLostBatches(ContactId c) throws DbException {
@@ -293,26 +296,32 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
int bytesSent = 0; int bytesSent = 0;
messageStatusLock.readLock().lock(); messageStatusLock.readLock().lock();
try { try {
Txn txn = db.startTransaction(); subscriptionLock.readLock().lock();
try { try {
int capacity = b.getCapacity(); Txn txn = db.startTransaction();
Iterator<MessageId> it = try {
db.getSendableMessages(txn, c, capacity).iterator(); int capacity = b.getCapacity();
sent = new ArrayList<MessageId>(); Collection<MessageId> sendable =
while(it.hasNext()) { db.getSendableMessages(txn, c, capacity);
MessageId m = it.next(); Iterator<MessageId> it = sendable.iterator();
byte[] message = db.getMessage(txn, m); sent = new ArrayList<MessageId>();
if(!b.writeMessage(message)) break; while(it.hasNext()) {
bytesSent += message.length; MessageId m = it.next();
sent.add(m); byte[] raw = db.getMessage(txn, m);
if(!b.writeMessage(raw)) break;
bytesSent += raw.length;
sent.add(m);
}
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
db.commitTransaction(txn); } finally {
} catch(DbException e) { subscriptionLock.readLock().unlock();
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
} finally { } finally {
messageStatusLock.readLock().unlock(); messageStatusLock.readLock().unlock();
@@ -351,24 +360,29 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
Collection<MessageId> sent; Collection<MessageId> sent;
messageStatusLock.readLock().lock(); messageStatusLock.readLock().lock();
try{ try{
Txn txn = db.startTransaction(); subscriptionLock.readLock().lock();
try { try {
sent = new ArrayList<MessageId>(); Txn txn = db.startTransaction();
int bytesSent = 0; try {
for(MessageId m : requested) { sent = new ArrayList<MessageId>();
byte[] message = db.getMessageIfSendable(txn, c, m); int bytesSent = 0;
if(message == null) continue; for(MessageId m : requested) {
if(!b.writeMessage(message)) break; byte[] raw = db.getMessageIfSendable(txn, c, m);
bytesSent += message.length; if(raw == null) continue;
sent.add(m); if(!b.writeMessage(raw)) break;
bytesSent += raw.length;
sent.add(m);
}
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
db.commitTransaction(txn); } finally {
} catch(DbException e) { subscriptionLock.readLock().unlock();
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
} finally { } finally {
messageStatusLock.readLock().unlock(); messageStatusLock.readLock().unlock();
@@ -610,6 +624,39 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
} }
public boolean hasSendableMessages(ContactId c) throws DbException {
contactLock.readLock().lock();
try {
if(!containsContact(c)) throw new NoSuchContactException();
messageLock.readLock().lock();
try {
messageStatusLock.readLock().lock();
try {
subscriptionLock.readLock().lock();
try {
Txn txn = db.startTransaction();
try {
boolean has = db.hasSendableMessages(txn, c);
db.commitTransaction(txn);
return has;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
} finally {
subscriptionLock.readLock().unlock();
}
} finally {
messageStatusLock.readLock().unlock();
}
} finally {
messageLock.readLock().unlock();
}
} finally {
contactLock.readLock().unlock();
}
}
public void receiveAck(ContactId c, Ack a) throws DbException { public void receiveAck(ContactId c, Ack a) throws DbException {
// Mark all messages in acked batches as seen // Mark all messages in acked batches as seen
contactLock.readLock().lock(); contactLock.readLock().lock();
@@ -644,6 +691,7 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
public void receiveBatch(ContactId c, Batch b) throws DbException { public void receiveBatch(ContactId c, Batch b) throws DbException {
boolean anyAdded = false;
waitForPermissionToWrite(); waitForPermissionToWrite();
contactLock.readLock().lock(); contactLock.readLock().lock();
try { try {
@@ -661,7 +709,10 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
received++; received++;
GroupId g = m.getGroup(); GroupId g = m.getGroup();
if(db.containsVisibleSubscription(txn, g, c)) { if(db.containsVisibleSubscription(txn, g, c)) {
if(storeMessage(txn, m, c)) stored++; if(storeMessage(txn, m, c)) {
anyAdded = true;
stored++;
}
} }
} }
if(LOG.isLoggable(Level.FINE)) if(LOG.isLoggable(Level.FINE))
@@ -685,6 +736,8 @@ class ReadWriteLockDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} finally { } finally {
contactLock.readLock().unlock(); contactLock.readLock().unlock();
} }
// Call the listeners outside the lock
if(anyAdded) callMessageListeners();
} }
public void receiveOffer(ContactId c, Offer o, RequestWriter r) public void receiveOffer(ContactId c, Offer o, RequestWriter r)

View File

@@ -116,6 +116,7 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
public void addLocallyGeneratedMessage(Message m) throws DbException { public void addLocallyGeneratedMessage(Message m) throws DbException {
boolean added = false;
waitForPermissionToWrite(); waitForPermissionToWrite();
synchronized(contactLock) { synchronized(contactLock) {
synchronized(messageLock) { synchronized(messageLock) {
@@ -126,7 +127,7 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
// Don't store the message if the user has // Don't store the message if the user has
// unsubscribed from the group // unsubscribed from the group
if(db.containsSubscription(txn, m.getGroup())) { if(db.containsSubscription(txn, m.getGroup())) {
boolean added = storeMessage(txn, m, null); added = storeMessage(txn, m, null);
if(!added) { if(!added) {
if(LOG.isLoggable(Level.FINE)) if(LOG.isLoggable(Level.FINE))
LOG.fine("Duplicate local message"); LOG.fine("Duplicate local message");
@@ -144,6 +145,8 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
} }
} }
// Call the listeners outside the lock
if(added) callMessageListeners();
} }
public void findLostBatches(ContactId c) throws DbException { public void findLostBatches(ContactId c) throws DbException {
@@ -217,31 +220,34 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
if(!containsContact(c)) throw new NoSuchContactException(); if(!containsContact(c)) throw new NoSuchContactException();
synchronized(messageLock) { synchronized(messageLock) {
synchronized(messageStatusLock) { synchronized(messageStatusLock) {
Txn txn = db.startTransaction(); synchronized(subscriptionLock) {
try { Txn txn = db.startTransaction();
int capacity = b.getCapacity(); try {
Iterator<MessageId> it = int capacity = b.getCapacity();
db.getSendableMessages(txn, c, capacity).iterator(); Collection<MessageId> sendable =
Collection<MessageId> sent = new ArrayList<MessageId>(); db.getSendableMessages(txn, c, capacity);
int bytesSent = 0; Iterator<MessageId> it = sendable.iterator();
while(it.hasNext()) { Collection<MessageId> sent =
MessageId m = it.next(); new ArrayList<MessageId>();
byte[] message = db.getMessage(txn, m); int bytesSent = 0;
if(!b.writeMessage(message)) break; while(it.hasNext()) {
bytesSent += message.length; MessageId m = it.next();
sent.add(m); byte[] raw = db.getMessage(txn, m);
if(!b.writeMessage(raw)) break;
bytesSent += raw.length;
sent.add(m);
}
BatchId id = b.finish();
if(!sent.isEmpty())
db.addOutstandingBatch(txn, c, id, sent);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
BatchId id = b.finish();
// Record the contents of the batch, unless it's empty
if(!sent.isEmpty())
db.addOutstandingBatch(txn, c, id, sent);
db.commitTransaction(txn);
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
} }
} }
@@ -254,29 +260,31 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
if(!containsContact(c)) throw new NoSuchContactException(); if(!containsContact(c)) throw new NoSuchContactException();
synchronized(messageLock) { synchronized(messageLock) {
synchronized(messageStatusLock) { synchronized(messageStatusLock) {
Txn txn = db.startTransaction(); synchronized(subscriptionLock) {
try { Txn txn = db.startTransaction();
Collection<MessageId> sent = new ArrayList<MessageId>(); try {
int bytesSent = 0; Collection<MessageId> sent =
for(MessageId m : requested) { new ArrayList<MessageId>();
byte[] message = db.getMessageIfSendable(txn, c, m); int bytesSent = 0;
if(message == null) continue; for(MessageId m : requested) {
if(!b.writeMessage(message)) break; byte[] raw = db.getMessageIfSendable(txn, c, m);
bytesSent += message.length; if(raw == null) continue;
sent.add(m); if(!b.writeMessage(raw)) break;
bytesSent += raw.length;
sent.add(m);
}
BatchId id = b.finish();
if(!sent.isEmpty())
db.addOutstandingBatch(txn, c, id, sent);
db.commitTransaction(txn);
return sent;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
BatchId id = b.finish();
// Record the contents of the batch, unless it's empty
if(!sent.isEmpty())
db.addOutstandingBatch(txn, c, id, sent);
db.commitTransaction(txn);
return sent;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
} catch(IOException e) {
db.abortTransaction(txn);
throw e;
} }
} }
} }
@@ -450,6 +458,27 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
} }
public boolean hasSendableMessages(ContactId c) throws DbException {
synchronized(contactLock) {
if(!containsContact(c)) throw new NoSuchContactException();
synchronized(messageLock) {
synchronized(messageStatusLock) {
synchronized(subscriptionLock) {
Txn txn = db.startTransaction();
try {
boolean has = db.hasSendableMessages(txn, c);
db.commitTransaction(txn);
return has;
} catch(DbException e) {
db.abortTransaction(txn);
throw e;
}
}
}
}
}
}
public void receiveAck(ContactId c, Ack a) throws DbException { public void receiveAck(ContactId c, Ack a) throws DbException {
// Mark all messages in acked batches as seen // Mark all messages in acked batches as seen
synchronized(contactLock) { synchronized(contactLock) {
@@ -475,6 +504,7 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
public void receiveBatch(ContactId c, Batch b) throws DbException { public void receiveBatch(ContactId c, Batch b) throws DbException {
boolean anyAdded = false;
waitForPermissionToWrite(); waitForPermissionToWrite();
synchronized(contactLock) { synchronized(contactLock) {
if(!containsContact(c)) throw new NoSuchContactException(); if(!containsContact(c)) throw new NoSuchContactException();
@@ -488,7 +518,10 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
received++; received++;
GroupId g = m.getGroup(); GroupId g = m.getGroup();
if(db.containsVisibleSubscription(txn, g, c)) { if(db.containsVisibleSubscription(txn, g, c)) {
if(storeMessage(txn, m, c)) stored++; if(storeMessage(txn, m, c)) {
anyAdded = true;
stored++;
}
} }
} }
if(LOG.isLoggable(Level.FINE)) if(LOG.isLoggable(Level.FINE))
@@ -504,6 +537,8 @@ class SynchronizedDatabaseComponent<Txn> extends DatabaseComponentImpl<Txn> {
} }
} }
} }
// Call the listeners outside the lock
if(anyAdded) callMessageListeners();
} }
public void receiveOffer(ContactId c, Offer o, RequestWriter r) public void receiveOffer(ContactId c, Offer o, RequestWriter r)

View File

@@ -12,6 +12,7 @@ import net.sf.briar.api.ContactId;
import net.sf.briar.api.Rating; import net.sf.briar.api.Rating;
import net.sf.briar.api.db.DatabaseComponent; import net.sf.briar.api.db.DatabaseComponent;
import net.sf.briar.api.db.DbException; import net.sf.briar.api.db.DbException;
import net.sf.briar.api.db.MessageListener;
import net.sf.briar.api.db.NoSuchContactException; import net.sf.briar.api.db.NoSuchContactException;
import net.sf.briar.api.db.Status; import net.sf.briar.api.db.Status;
import net.sf.briar.api.protocol.Ack; import net.sf.briar.api.protocol.Ack;
@@ -80,6 +81,7 @@ public abstract class DatabaseComponentTest extends TestCase {
final Database<Object> database = context.mock(Database.class); final Database<Object> database = context.mock(Database.class);
final DatabaseCleaner cleaner = context.mock(DatabaseCleaner.class); final DatabaseCleaner cleaner = context.mock(DatabaseCleaner.class);
final Group group = context.mock(Group.class); final Group group = context.mock(Group.class);
final MessageListener listener = context.mock(MessageListener.class);
context.checking(new Expectations() {{ context.checking(new Expectations() {{
allowing(database).startTransaction(); allowing(database).startTransaction();
will(returnValue(txn)); will(returnValue(txn));
@@ -117,6 +119,7 @@ public abstract class DatabaseComponentTest extends TestCase {
DatabaseComponent db = createDatabaseComponent(database, cleaner); DatabaseComponent db = createDatabaseComponent(database, cleaner);
db.open(false); db.open(false);
db.addListener(listener);
assertEquals(Rating.UNRATED, db.getRating(authorId)); assertEquals(Rating.UNRATED, db.getRating(authorId));
assertEquals(contactId, db.addContact(transports)); assertEquals(contactId, db.addContact(transports));
assertEquals(Collections.singletonList(contactId), db.getContacts()); assertEquals(Collections.singletonList(contactId), db.getContacts());
@@ -125,6 +128,7 @@ public abstract class DatabaseComponentTest extends TestCase {
assertEquals(Collections.singletonList(groupId), db.getSubscriptions()); assertEquals(Collections.singletonList(groupId), db.getSubscriptions());
db.unsubscribe(groupId); db.unsubscribe(groupId);
db.removeContact(contactId); db.removeContact(contactId);
db.removeListener(listener);
db.close(); db.close();
context.assertIsSatisfied(); context.assertIsSatisfied();
@@ -388,7 +392,7 @@ public abstract class DatabaseComponentTest extends TestCase {
} }
@Test @Test
public void testAddingASendableMessageTriggersBackwardInclusion() public void testAddingSendableMessageTriggersBackwardInclusion()
throws DbException { throws DbException {
Mockery context = new Mockery(); Mockery context = new Mockery();
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@@ -446,11 +450,11 @@ public abstract class DatabaseComponentTest extends TestCase {
final Transports transportsUpdate = context.mock(Transports.class); final Transports transportsUpdate = context.mock(Transports.class);
context.checking(new Expectations() {{ context.checking(new Expectations() {{
// Check whether the contact is still in the DB - which it's not // Check whether the contact is still in the DB - which it's not
exactly(11).of(database).startTransaction(); exactly(12).of(database).startTransaction();
will(returnValue(txn)); will(returnValue(txn));
exactly(11).of(database).containsContact(txn, contactId); exactly(12).of(database).containsContact(txn, contactId);
will(returnValue(false)); will(returnValue(false));
exactly(11).of(database).commitTransaction(txn); exactly(12).of(database).commitTransaction(txn);
}}); }});
DatabaseComponent db = createDatabaseComponent(database, cleaner); DatabaseComponent db = createDatabaseComponent(database, cleaner);
@@ -485,6 +489,11 @@ public abstract class DatabaseComponentTest extends TestCase {
assertTrue(false); assertTrue(false);
} catch(NoSuchContactException expected) {} } catch(NoSuchContactException expected) {}
try {
db.hasSendableMessages(contactId);
assertTrue(false);
} catch(NoSuchContactException expected) {}
try { try {
db.receiveAck(contactId, ack); db.receiveAck(contactId, ack);
assertTrue(false); assertTrue(false);
@@ -1019,4 +1028,65 @@ public abstract class DatabaseComponentTest extends TestCase {
context.assertIsSatisfied(); context.assertIsSatisfied();
} }
@Test
public void testAddingMessageCallsListeners() throws Exception {
Mockery context = new Mockery();
@SuppressWarnings("unchecked")
final Database<Object> database = context.mock(Database.class);
final DatabaseCleaner cleaner = context.mock(DatabaseCleaner.class);
final MessageListener listener = context.mock(MessageListener.class);
context.checking(new Expectations() {{
// addLocallyGeneratedMessage(message)
oneOf(database).startTransaction();
will(returnValue(txn));
oneOf(database).containsSubscription(txn, groupId);
will(returnValue(true));
oneOf(database).addMessage(txn, message);
will(returnValue(true));
oneOf(database).getContacts(txn);
will(returnValue(Collections.singletonList(contactId)));
oneOf(database).setStatus(txn, contactId, messageId, Status.NEW);
oneOf(database).getRating(txn, authorId);
will(returnValue(Rating.UNRATED));
oneOf(database).getNumberOfSendableChildren(txn, messageId);
will(returnValue(0));
oneOf(database).setSendability(txn, messageId, 0);
oneOf(database).commitTransaction(txn);
// The message was added, so the listener should be called
oneOf(listener).messagesAdded();
}});
DatabaseComponent db = createDatabaseComponent(database, cleaner);
db.addListener(listener);
db.addLocallyGeneratedMessage(message);
context.assertIsSatisfied();
}
@Test
public void testDuplicateMessageDoesNotCallListeners() throws Exception {
Mockery context = new Mockery();
@SuppressWarnings("unchecked")
final Database<Object> database = context.mock(Database.class);
final DatabaseCleaner cleaner = context.mock(DatabaseCleaner.class);
final MessageListener listener = context.mock(MessageListener.class);
context.checking(new Expectations() {{
// addLocallyGeneratedMessage(message)
oneOf(database).startTransaction();
will(returnValue(txn));
oneOf(database).containsSubscription(txn, groupId);
will(returnValue(true));
oneOf(database).addMessage(txn, message);
will(returnValue(false));
oneOf(database).commitTransaction(txn);
// The message was not added, so the listener should not be called
}});
DatabaseComponent db = createDatabaseComponent(database, cleaner);
db.addListener(listener);
db.addLocallyGeneratedMessage(message);
context.assertIsSatisfied();
}
} }

View File

@@ -210,12 +210,14 @@ public class H2DatabaseTest extends TestCase {
// The message should not be sendable // The message should not be sendable
assertEquals(0, db.getSendability(txn, messageId)); assertEquals(0, db.getSendability(txn, messageId));
assertFalse(db.hasSendableMessages(txn, contactId));
Iterator<MessageId> it = Iterator<MessageId> it =
db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
// Changing the sendability to > 0 should make the message sendable // Changing the sendability to > 0 should make the message sendable
db.setSendability(txn, messageId, 1); db.setSendability(txn, messageId, 1);
assertTrue(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertTrue(it.hasNext()); assertTrue(it.hasNext());
assertEquals(messageId, it.next()); assertEquals(messageId, it.next());
@@ -223,6 +225,7 @@ public class H2DatabaseTest extends TestCase {
// Changing the sendability to 0 should make the message unsendable // Changing the sendability to 0 should make the message unsendable
db.setSendability(txn, messageId, 0); db.setSendability(txn, messageId, 0);
assertFalse(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
@@ -244,12 +247,14 @@ public class H2DatabaseTest extends TestCase {
db.setSendability(txn, messageId, 1); db.setSendability(txn, messageId, 1);
// The message has no status yet, so it should not be sendable // The message has no status yet, so it should not be sendable
assertFalse(db.hasSendableMessages(txn, contactId));
Iterator<MessageId> it = Iterator<MessageId> it =
db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
// Changing the status to Status.NEW should make the message sendable // Changing the status to Status.NEW should make the message sendable
db.setStatus(txn, contactId, messageId, Status.NEW); db.setStatus(txn, contactId, messageId, Status.NEW);
assertTrue(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertTrue(it.hasNext()); assertTrue(it.hasNext());
assertEquals(messageId, it.next()); assertEquals(messageId, it.next());
@@ -257,6 +262,7 @@ public class H2DatabaseTest extends TestCase {
// Changing the status to SENT should make the message unsendable // Changing the status to SENT should make the message unsendable
db.setStatus(txn, contactId, messageId, Status.SENT); db.setStatus(txn, contactId, messageId, Status.SENT);
assertFalse(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
@@ -283,12 +289,14 @@ public class H2DatabaseTest extends TestCase {
db.setStatus(txn, contactId, messageId, Status.NEW); db.setStatus(txn, contactId, messageId, Status.NEW);
// The contact is not subscribed, so the message should not be sendable // The contact is not subscribed, so the message should not be sendable
assertFalse(db.hasSendableMessages(txn, contactId));
Iterator<MessageId> it = Iterator<MessageId> it =
db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
// The contact subscribing should make the message sendable // The contact subscribing should make the message sendable
db.setSubscriptions(txn, contactId, Collections.singleton(group), 1); db.setSubscriptions(txn, contactId, Collections.singleton(group), 1);
assertTrue(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertTrue(it.hasNext()); assertTrue(it.hasNext());
assertEquals(messageId, it.next()); assertEquals(messageId, it.next());
@@ -296,6 +304,7 @@ public class H2DatabaseTest extends TestCase {
// The contact unsubscribing should make the message unsendable // The contact unsubscribing should make the message unsendable
db.setSubscriptions(txn, contactId, Collections.<Group>emptySet(), 2); db.setSubscriptions(txn, contactId, Collections.<Group>emptySet(), 2);
assertFalse(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
@@ -317,12 +326,14 @@ public class H2DatabaseTest extends TestCase {
db.setSendability(txn, messageId, 1); db.setSendability(txn, messageId, 1);
db.setStatus(txn, contactId, messageId, Status.NEW); db.setStatus(txn, contactId, messageId, Status.NEW);
// The message is too large to send // The message is sendable, but too large to send
assertTrue(db.hasSendableMessages(txn, contactId));
Iterator<MessageId> it = Iterator<MessageId> it =
db.getSendableMessages(txn, contactId, size - 1).iterator(); db.getSendableMessages(txn, contactId, size - 1).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
// The message is just the right size to send // The message is just the right size to send
assertTrue(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, size).iterator(); it = db.getSendableMessages(txn, contactId, size).iterator();
assertTrue(it.hasNext()); assertTrue(it.hasNext());
assertEquals(messageId, it.next()); assertEquals(messageId, it.next());
@@ -347,12 +358,14 @@ public class H2DatabaseTest extends TestCase {
// The subscription is not visible to the contact, so the message // The subscription is not visible to the contact, so the message
// should not be sendable // should not be sendable
assertFalse(db.hasSendableMessages(txn, contactId));
Iterator<MessageId> it = Iterator<MessageId> it =
db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertFalse(it.hasNext()); assertFalse(it.hasNext());
// Making the subscription visible should make the message sendable // Making the subscription visible should make the message sendable
db.setVisibility(txn, groupId, Collections.singleton(contactId)); db.setVisibility(txn, groupId, Collections.singleton(contactId));
assertTrue(db.hasSendableMessages(txn, contactId));
it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator(); it = db.getSendableMessages(txn, contactId, ONE_MEGABYTE).iterator();
assertTrue(it.hasNext()); assertTrue(it.hasNext());
assertEquals(messageId, it.next()); assertEquals(messageId, it.next());