Kaladaran/briar
0
0
Fork 0
mirror of https://code.briarproject.org/briar/briar.git synced 2026-02-18 05:39:53 +01:00
Code Issues Projects Releases Wiki Activity

Check periodically for retransmittable packets. Bug #46.

Browse Source
This commit is contained in:
akwizgran
2014-12-14 20:26:41 +00:00
parent 29a6596ee3
commit 388b36b6be
51 changed files with 351 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
briar-core/src/org/briarproject/messaging/DuplexOutgoingSession.java
View File

@@ -6,7 +6,6 @@ import static java.util.logging.Level.WARNING;
import static org.briarproject.api.messaging.MessagingConstants.MAX_PACKET_LENGTH;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
@@ -37,7 +36,6 @@ import org.briarproject.api.messaging.Ack;
import org.briarproject.api.messaging.MessagingSession;
import org.briarproject.api.messaging.Offer;
import org.briarproject.api.messaging.PacketWriter;
import org.briarproject.api.messaging.PacketWriterFactory;
import org.briarproject.api.messaging.Request;
import org.briarproject.api.messaging.RetentionAck;
import org.briarproject.api.messaging.RetentionUpdate;
@@ -45,16 +43,18 @@ import org.briarproject.api.messaging.SubscriptionAck;
import org.briarproject.api.messaging.SubscriptionUpdate;
import org.briarproject.api.messaging.TransportAck;
import org.briarproject.api.messaging.TransportUpdate;
import org.briarproject.api.system.Clock;
/**
* An outgoing {@link org.briarproject.api.messaging.MessagingSession
* MessagingSession} suitable for duplex transports. The session offers
* messages before sending them, keeps its output stream open when there are no
* more packets to send, and reacts to events that make packets available to
* send.
* packets to send, and reacts to events that make packets available to send.
*/
class DuplexOutgoingSession implements MessagingSession, EventListener {
// Check for retransmittable packets once every 60 seconds
private static final int RETX_QUERY_INTERVAL = 60 * 1000;
private static final Logger LOG =
Logger.getLogger(DuplexOutgoingSession.class.getName());
@@ -66,28 +66,32 @@ class DuplexOutgoingSession implements MessagingSession, EventListener {
private final DatabaseComponent db;
private final Executor dbExecutor;
private final EventBus eventBus;
private final Clock clock;
private final ContactId contactId;
private final TransportId transportId;
private final long maxLatency, maxIdleTime;
private final OutputStream out;
private final int maxLatency, maxIdleTime;
private final PacketWriter packetWriter;
private final BlockingQueue<ThrowingRunnable<IOException>> writerTasks;
// The following must only be accessed on the writer thread
private long nextKeepalive = 0, nextRetxQuery = 0;
private boolean dataToFlush = true;
private volatile boolean interrupted = false;
DuplexOutgoingSession(DatabaseComponent db, Executor dbExecutor,
EventBus eventBus, PacketWriterFactory packetWriterFactory,
ContactId contactId, TransportId transportId, long maxLatency,
long maxIdleTime, OutputStream out) {
EventBus eventBus, Clock clock, ContactId contactId,
TransportId transportId, int maxLatency, int maxIdleTime,
PacketWriter packetWriter) {
this.db = db;
this.dbExecutor = dbExecutor;
this.eventBus = eventBus;
this.clock = clock;
this.contactId = contactId;
this.transportId = transportId;
this.maxLatency = maxLatency;
this.maxIdleTime = maxIdleTime;
this.out = out;
packetWriter = packetWriterFactory.createPacketWriter(out);
this.packetWriter = packetWriter;
writerTasks = new LinkedBlockingQueue<ThrowingRunnable<IOException>>();
}
@@ -105,21 +109,50 @@ class DuplexOutgoingSession implements MessagingSession, EventListener {
dbExecutor.execute(new GenerateBatch());
dbExecutor.execute(new GenerateOffer());
dbExecutor.execute(new GenerateRequest());
long now = clock.currentTimeMillis();
nextKeepalive = now + maxIdleTime;
nextRetxQuery = now + RETX_QUERY_INTERVAL;
// Write packets until interrupted
try {
while(!interrupted) {
// Flush the stream if it's going to be idle
if(writerTasks.isEmpty()) out.flush();
ThrowingRunnable<IOException> task =
writerTasks.poll(maxIdleTime, MILLISECONDS);
if(task == null) {
LOG.info("Idle timeout");
continue; // Flush and wait again
// Work out how long we should wait for a packet
now = clock.currentTimeMillis();
long wait = Math.min(nextKeepalive, nextRetxQuery) - now;
if(wait < 0) wait = 0;
// Flush any unflushed data if we're going to wait
if(wait > 0 && dataToFlush && writerTasks.isEmpty()) {
packetWriter.flush();
dataToFlush = false;
nextKeepalive = now + maxIdleTime;
}
// Wait for a packet
ThrowingRunnable<IOException> task = writerTasks.poll(wait,
MILLISECONDS);
if(task == null) {
now = clock.currentTimeMillis();
if(now >= nextRetxQuery) {
// Check for retransmittable packets
dbExecutor.execute(new GenerateTransportUpdates());
dbExecutor.execute(new GenerateSubscriptionUpdate());
dbExecutor.execute(new GenerateRetentionUpdate());
dbExecutor.execute(new GenerateBatch());
dbExecutor.execute(new GenerateOffer());
nextRetxQuery = now + RETX_QUERY_INTERVAL;
}
if(now >= nextKeepalive) {
// Flush the stream to keep it alive
packetWriter.flush();
dataToFlush = false;
nextKeepalive = now + maxIdleTime;
}
} else if(task == CLOSE) {
break;
} else {
task.run();
dataToFlush = true;
}
if(task == CLOSE) break;
task.run();
}
out.flush();
if(dataToFlush) packetWriter.flush();
} catch(InterruptedException e) {
LOG.info("Interrupted while waiting for a packet to write");
Thread.currentThread().interrupt();

9
briar-core/src/org/briarproject/messaging/IncomingSession.java
View File

@@ -3,7 +3,6 @@ package org.briarproject.messaging;
import static java.util.logging.Level.WARNING;
import java.io.IOException;
import java.io.InputStream;
import java.security.GeneralSecurityException;
import java.util.concurrent.Executor;
import java.util.logging.Logger;
@@ -25,7 +24,6 @@ import org.briarproject.api.messaging.MessageVerifier;
import org.briarproject.api.messaging.MessagingSession;
import org.briarproject.api.messaging.Offer;
import org.briarproject.api.messaging.PacketReader;
import org.briarproject.api.messaging.PacketReaderFactory;
import org.briarproject.api.messaging.Request;
import org.briarproject.api.messaging.RetentionAck;
import org.briarproject.api.messaging.RetentionUpdate;
@@ -56,9 +54,8 @@ class IncomingSession implements MessagingSession, EventListener {
IncomingSession(DatabaseComponent db, Executor dbExecutor,
Executor cryptoExecutor, EventBus eventBus,
MessageVerifier messageVerifier,
PacketReaderFactory packetReaderFactory, ContactId contactId,
TransportId transportId, InputStream in) {
MessageVerifier messageVerifier, ContactId contactId,
TransportId transportId, PacketReader packetReader) {
this.db = db;
this.dbExecutor = dbExecutor;
this.cryptoExecutor = cryptoExecutor;
@@ -66,7 +63,7 @@ class IncomingSession implements MessagingSession, EventListener {
this.messageVerifier = messageVerifier;
this.contactId = contactId;
this.transportId = transportId;
packetReader = packetReaderFactory.createPacketReader(in);
this.packetReader = packetReader;
}
public void run() throws IOException {

25
briar-core/src/org/briarproject/messaging/MessagingSessionFactoryImpl.java
View File

@@ -15,8 +15,11 @@ import org.briarproject.api.event.EventBus;
import org.briarproject.api.messaging.MessageVerifier;
import org.briarproject.api.messaging.MessagingSession;
import org.briarproject.api.messaging.MessagingSessionFactory;
import org.briarproject.api.messaging.PacketReader;
import org.briarproject.api.messaging.PacketReaderFactory;
import org.briarproject.api.messaging.PacketWriter;
import org.briarproject.api.messaging.PacketWriterFactory;
import org.briarproject.api.system.Clock;
class MessagingSessionFactoryImpl implements MessagingSessionFactory {
@@ -24,6 +27,7 @@ class MessagingSessionFactoryImpl implements MessagingSessionFactory {
private final Executor dbExecutor, cryptoExecutor;
private final MessageVerifier messageVerifier;
private final EventBus eventBus;
private final Clock clock;
private final PacketReaderFactory packetReaderFactory;
private final PacketWriterFactory packetWriterFactory;
@@ -31,7 +35,7 @@ class MessagingSessionFactoryImpl implements MessagingSessionFactory {
MessagingSessionFactoryImpl(DatabaseComponent db,
@DatabaseExecutor Executor dbExecutor,
@CryptoExecutor Executor cryptoExecutor,
MessageVerifier messageVerifier, EventBus eventBus,
MessageVerifier messageVerifier, EventBus eventBus, Clock clock,
PacketReaderFactory packetReaderFactory,
PacketWriterFactory packetWriterFactory) {
this.db = db;
@@ -39,26 +43,29 @@ class MessagingSessionFactoryImpl implements MessagingSessionFactory {
this.cryptoExecutor = cryptoExecutor;
this.messageVerifier = messageVerifier;
this.eventBus = eventBus;
this.clock = clock;
this.packetReaderFactory = packetReaderFactory;
this.packetWriterFactory = packetWriterFactory;
}
public MessagingSession createIncomingSession(ContactId c, TransportId t,
InputStream in) {
PacketReader packetReader = packetReaderFactory.createPacketReader(in);
return new IncomingSession(db, dbExecutor, cryptoExecutor, eventBus,
messageVerifier, packetReaderFactory, c, t, in);
messageVerifier, c, t, packetReader);
}
public MessagingSession createSimplexOutgoingSession(ContactId c,
TransportId t, long maxLatency, OutputStream out) {
return new SimplexOutgoingSession(db, dbExecutor, eventBus,
packetWriterFactory, c, t, maxLatency, out);
TransportId t, int maxLatency, OutputStream out) {
PacketWriter packetWriter = packetWriterFactory.createPacketWriter(out);
return new SimplexOutgoingSession(db, dbExecutor, eventBus, c, t,
maxLatency, packetWriter);
}
public MessagingSession createDuplexOutgoingSession(ContactId c,
TransportId t, long maxLatency, long maxIdleTime,
OutputStream out) {
return new DuplexOutgoingSession(db, dbExecutor, eventBus,
packetWriterFactory, c, t, maxLatency, maxIdleTime, out);
TransportId t, int maxLatency, int maxIdleTime, OutputStream out) {
PacketWriter packetWriter = packetWriterFactory.createPacketWriter(out);
return new DuplexOutgoingSession(db, dbExecutor, eventBus, clock, c, t,
maxLatency, maxIdleTime, packetWriter);
}
}

4
briar-core/src/org/briarproject/messaging/PacketWriterImpl.java
View File

@@ -143,4 +143,8 @@ class PacketWriterImpl implements PacketWriter {
w.writeInteger(u.getVersion());
w.writeStructEnd();
}
public void flush() throws IOException {
out.flush();
}
}

15
briar-core/src/org/briarproject/messaging/SimplexOutgoingSession.java
View File

@@ -5,7 +5,6 @@ import static java.util.logging.Level.WARNING;
import static org.briarproject.api.messaging.MessagingConstants.MAX_PACKET_LENGTH;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
@@ -26,7 +25,6 @@ import org.briarproject.api.event.TransportRemovedEvent;
import org.briarproject.api.messaging.Ack;
import org.briarproject.api.messaging.MessagingSession;
import org.briarproject.api.messaging.PacketWriter;
import org.briarproject.api.messaging.PacketWriterFactory;
import org.briarproject.api.messaging.RetentionAck;
import org.briarproject.api.messaging.RetentionUpdate;
import org.briarproject.api.messaging.SubscriptionAck;
@@ -55,8 +53,7 @@ class SimplexOutgoingSession implements MessagingSession, EventListener {
private final EventBus eventBus;
private final ContactId contactId;
private final TransportId transportId;
private final long maxLatency;
private final OutputStream out;
private final int maxLatency;
private final PacketWriter packetWriter;
private final AtomicInteger outstandingQueries;
private final BlockingQueue<ThrowingRunnable<IOException>> writerTasks;
@@ -64,17 +61,15 @@ class SimplexOutgoingSession implements MessagingSession, EventListener {
private volatile boolean interrupted = false;
SimplexOutgoingSession(DatabaseComponent db, Executor dbExecutor,
EventBus eventBus, PacketWriterFactory packetWriterFactory,
ContactId contactId, TransportId transportId, long maxLatency,
OutputStream out) {
EventBus eventBus, ContactId contactId, TransportId transportId,
int maxLatency, PacketWriter packetWriter) {
this.db = db;
this.dbExecutor = dbExecutor;
this.eventBus = eventBus;
this.contactId = contactId;
this.transportId = transportId;
this.maxLatency = maxLatency;
this.out = out;
packetWriter = packetWriterFactory.createPacketWriter(out);
this.packetWriter = packetWriter;
outstandingQueries = new AtomicInteger(8); // One per type of packet
writerTasks = new LinkedBlockingQueue<ThrowingRunnable<IOException>>();
}
@@ -98,7 +93,7 @@ class SimplexOutgoingSession implements MessagingSession, EventListener {
if(task == CLOSE) break;
task.run();
}
out.flush();
packetWriter.flush();
} catch(InterruptedException e) {
LOG.info("Interrupted while waiting for a packet to write");
Thread.currentThread().interrupt();