mirror of
https://code.briarproject.org/briar/briar.git
synced 2026-02-13 19:29:06 +01:00
Removed TransportUpdate and TransportAck.
This commit is contained in:
akwizgran
@@ -9,13 +9,11 @@ import org.briarproject.api.event.Event;
|
||||
import org.briarproject.api.event.EventBus;
|
||||
import org.briarproject.api.event.EventListener;
|
||||
import org.briarproject.api.event.LocalSubscriptionsUpdatedEvent;
|
||||
import org.briarproject.api.event.LocalTransportsUpdatedEvent;
|
||||
import org.briarproject.api.event.MessageRequestedEvent;
|
||||
import org.briarproject.api.event.MessageToAckEvent;
|
||||
import org.briarproject.api.event.MessageToRequestEvent;
|
||||
import org.briarproject.api.event.MessageValidatedEvent;
|
||||
import org.briarproject.api.event.RemoteSubscriptionsUpdatedEvent;
|
||||
import org.briarproject.api.event.RemoteTransportsUpdatedEvent;
|
||||
import org.briarproject.api.event.ShutdownEvent;
|
||||
import org.briarproject.api.event.TransportRemovedEvent;
|
||||
import org.briarproject.api.sync.Ack;
|
||||
@@ -25,8 +23,6 @@ import org.briarproject.api.sync.Request;
|
||||
import org.briarproject.api.sync.SubscriptionAck;
|
||||
import org.briarproject.api.sync.SubscriptionUpdate;
|
||||
import org.briarproject.api.sync.SyncSession;
|
||||
import org.briarproject.api.sync.TransportAck;
|
||||
import org.briarproject.api.sync.TransportUpdate;
|
||||
import org.briarproject.api.system.Clock;
|
||||
|
||||
import java.io.IOException;
|
||||
@@ -91,8 +87,6 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
|
||||
eventBus.addListener(this);
|
||||
try {
|
||||
// Start a query for each type of packet, in order of urgency
|
||||
dbExecutor.execute(new GenerateTransportAcks());
|
||||
dbExecutor.execute(new GenerateTransportUpdates());
|
||||
dbExecutor.execute(new GenerateSubscriptionAck());
|
||||
dbExecutor.execute(new GenerateSubscriptionUpdate());
|
||||
dbExecutor.execute(new GenerateAck());
|
||||
@@ -123,7 +117,6 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
|
||||
now = clock.currentTimeMillis();
|
||||
if (now >= nextRetxQuery) {
|
||||
// Check for retransmittable packets
|
||||
dbExecutor.execute(new GenerateTransportUpdates());
|
||||
dbExecutor.execute(new GenerateSubscriptionUpdate());
|
||||
dbExecutor.execute(new GenerateBatch());
|
||||
dbExecutor.execute(new GenerateOffer());
|
||||
@@ -171,8 +164,6 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
|
||||
dbExecutor.execute(new GenerateSubscriptionUpdate());
|
||||
dbExecutor.execute(new GenerateOffer());
|
||||
}
|
||||
} else if (e instanceof LocalTransportsUpdatedEvent) {
|
||||
dbExecutor.execute(new GenerateTransportUpdates());
|
||||
} else if (e instanceof MessageRequestedEvent) {
|
||||
if (((MessageRequestedEvent) e).getContactId().equals(contactId))
|
||||
dbExecutor.execute(new GenerateBatch());
|
||||
@@ -189,11 +180,6 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
|
||||
dbExecutor.execute(new GenerateSubscriptionAck());
|
||||
dbExecutor.execute(new GenerateOffer());
|
||||
}
|
||||
} else if (e instanceof RemoteTransportsUpdatedEvent) {
|
||||
RemoteTransportsUpdatedEvent r =
|
||||
(RemoteTransportsUpdatedEvent) e;
|
||||
if (r.getContactId().equals(contactId))
|
||||
dbExecutor.execute(new GenerateTransportAcks());
|
||||
} else if (e instanceof ShutdownEvent) {
|
||||
interrupt();
|
||||
} else if (e instanceof TransportRemovedEvent) {
|
||||
@@ -414,76 +400,4 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
|
||||
dbExecutor.execute(new GenerateSubscriptionUpdate());
|
||||
}
|
||||
}
|
||||
|
||||
// This task runs on the database thread
|
||||
private class GenerateTransportAcks implements Runnable {
|
||||
|
||||
public void run() {
|
||||
if (interrupted) return;
|
||||
try {
|
||||
Collection<TransportAck> acks =
|
||||
db.generateTransportAcks(contactId);
|
||||
if (LOG.isLoggable(INFO))
|
||||
LOG.info("Generated transport acks: " + (acks != null));
|
||||
if (acks != null) writerTasks.add(new WriteTransportAcks(acks));
|
||||
} catch (DbException e) {
|
||||
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
|
||||
interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// This tasks runs on the writer thread
|
||||
private class WriteTransportAcks implements ThrowingRunnable<IOException> {
|
||||
|
||||
private final Collection<TransportAck> acks;
|
||||
|
||||
private WriteTransportAcks(Collection<TransportAck> acks) {
|
||||
this.acks = acks;
|
||||
}
|
||||
|
||||
public void run() throws IOException {
|
||||
if (interrupted) return;
|
||||
for (TransportAck a : acks) packetWriter.writeTransportAck(a);
|
||||
LOG.info("Sent transport acks");
|
||||
dbExecutor.execute(new GenerateTransportAcks());
|
||||
}
|
||||
}
|
||||
|
||||
// This task runs on the database thread
|
||||
private class GenerateTransportUpdates implements Runnable {
|
||||
|
||||
public void run() {
|
||||
if (interrupted) return;
|
||||
try {
|
||||
Collection<TransportUpdate> t =
|
||||
db.generateTransportUpdates(contactId, maxLatency);
|
||||
if (LOG.isLoggable(INFO))
|
||||
LOG.info("Generated transport updates: " + (t != null));
|
||||
if (t != null) writerTasks.add(new WriteTransportUpdates(t));
|
||||
} catch (DbException e) {
|
||||
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
|
||||
interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// This task runs on the writer thread
|
||||
private class WriteTransportUpdates
|
||||
implements ThrowingRunnable<IOException> {
|
||||
|
||||
private final Collection<TransportUpdate> updates;
|
||||
|
||||
private WriteTransportUpdates(Collection<TransportUpdate> updates) {
|
||||
this.updates = updates;
|
||||
}
|
||||
|
||||
public void run() throws IOException {
|
||||
if (interrupted) return;
|
||||
for (TransportUpdate u : updates)
|
||||
packetWriter.writeTransportUpdate(u);
|
||||
LOG.info("Sent transport updates");
|
||||
dbExecutor.execute(new GenerateTransportUpdates());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,8 +19,6 @@ import org.briarproject.api.sync.Request;
|
||||
import org.briarproject.api.sync.SubscriptionAck;
|
||||
import org.briarproject.api.sync.SubscriptionUpdate;
|
||||
import org.briarproject.api.sync.SyncSession;
|
||||
import org.briarproject.api.sync.TransportAck;
|
||||
import org.briarproject.api.sync.TransportUpdate;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.concurrent.Executor;
|
||||
@@ -77,12 +75,6 @@ class IncomingSession implements SyncSession, EventListener {
|
||||
} else if (packetReader.hasSubscriptionUpdate()) {
|
||||
SubscriptionUpdate u = packetReader.readSubscriptionUpdate();
|
||||
dbExecutor.execute(new ReceiveSubscriptionUpdate(u));
|
||||
} else if (packetReader.hasTransportAck()) {
|
||||
TransportAck a = packetReader.readTransportAck();
|
||||
dbExecutor.execute(new ReceiveTransportAck(a));
|
||||
} else if (packetReader.hasTransportUpdate()) {
|
||||
TransportUpdate u = packetReader.readTransportUpdate();
|
||||
dbExecutor.execute(new ReceiveTransportUpdate(u));
|
||||
} else {
|
||||
throw new FormatException();
|
||||
}
|
||||
@@ -216,40 +208,4 @@ class IncomingSession implements SyncSession, EventListener {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class ReceiveTransportAck implements Runnable {
|
||||
|
||||
private final TransportAck ack;
|
||||
|
||||
private ReceiveTransportAck(TransportAck ack) {
|
||||
this.ack = ack;
|
||||
}
|
||||
|
||||
public void run() {
|
||||
try {
|
||||
db.receiveTransportAck(contactId, ack);
|
||||
} catch (DbException e) {
|
||||
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
|
||||
interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class ReceiveTransportUpdate implements Runnable {
|
||||
|
||||
private final TransportUpdate update;
|
||||
|
||||
private ReceiveTransportUpdate(TransportUpdate update) {
|
||||
this.update = update;
|
||||
}
|
||||
|
||||
public void run() {
|
||||
try {
|
||||
db.receiveTransportUpdate(contactId, update);
|
||||
} catch (DbException e) {
|
||||
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
|
||||
interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
package org.briarproject.sync;
|
||||
|
||||
import org.briarproject.api.FormatException;
|
||||
import org.briarproject.api.TransportId;
|
||||
import org.briarproject.api.TransportProperties;
|
||||
import org.briarproject.api.UniqueId;
|
||||
import org.briarproject.api.crypto.CryptoComponent;
|
||||
import org.briarproject.api.data.BdfReader;
|
||||
@@ -17,8 +15,6 @@ import org.briarproject.api.sync.PacketReader;
|
||||
import org.briarproject.api.sync.Request;
|
||||
import org.briarproject.api.sync.SubscriptionAck;
|
||||
import org.briarproject.api.sync.SubscriptionUpdate;
|
||||
import org.briarproject.api.sync.TransportAck;
|
||||
import org.briarproject.api.sync.TransportUpdate;
|
||||
import org.briarproject.util.ByteUtils;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
@@ -26,21 +22,14 @@ import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.briarproject.api.TransportPropertyConstants.MAX_PROPERTIES_PER_TRANSPORT;
|
||||
import static org.briarproject.api.TransportPropertyConstants.MAX_PROPERTY_LENGTH;
|
||||
import static org.briarproject.api.TransportPropertyConstants.MAX_TRANSPORT_ID_LENGTH;
|
||||
import static org.briarproject.api.sync.PacketTypes.ACK;
|
||||
import static org.briarproject.api.sync.PacketTypes.MESSAGE;
|
||||
import static org.briarproject.api.sync.PacketTypes.OFFER;
|
||||
import static org.briarproject.api.sync.PacketTypes.REQUEST;
|
||||
import static org.briarproject.api.sync.PacketTypes.SUBSCRIPTION_ACK;
|
||||
import static org.briarproject.api.sync.PacketTypes.SUBSCRIPTION_UPDATE;
|
||||
import static org.briarproject.api.sync.PacketTypes.TRANSPORT_ACK;
|
||||
import static org.briarproject.api.sync.PacketTypes.TRANSPORT_UPDATE;
|
||||
import static org.briarproject.api.sync.SyncConstants.MAX_PACKET_PAYLOAD_LENGTH;
|
||||
import static org.briarproject.api.sync.SyncConstants.MESSAGE_HEADER_LENGTH;
|
||||
import static org.briarproject.api.sync.SyncConstants.PACKET_HEADER_LENGTH;
|
||||
@@ -205,66 +194,4 @@ class PacketReaderImpl implements PacketReader {
|
||||
state = State.BUFFER_EMPTY;
|
||||
return u;
|
||||
}
|
||||
|
||||
public boolean hasTransportAck() throws IOException {
|
||||
return !eof() && header[1] == TRANSPORT_ACK;
|
||||
}
|
||||
|
||||
public TransportAck readTransportAck() throws IOException {
|
||||
if (!hasTransportAck()) throw new FormatException();
|
||||
// Set up the reader
|
||||
InputStream bais = new ByteArrayInputStream(payload, 0, payloadLength);
|
||||
BdfReader r = bdfReaderFactory.createReader(bais);
|
||||
// Read the start of the payload
|
||||
r.readListStart();
|
||||
// Read the transport ID and version
|
||||
String idString = r.readString(MAX_TRANSPORT_ID_LENGTH);
|
||||
if (idString.length() == 0) throw new FormatException();
|
||||
TransportId id = new TransportId(idString);
|
||||
long version = r.readInteger();
|
||||
if (version < 0) throw new FormatException();
|
||||
// Read the end of the payload
|
||||
r.readListEnd();
|
||||
if (!r.eof()) throw new FormatException();
|
||||
state = State.BUFFER_EMPTY;
|
||||
// Build and return the transport ack
|
||||
return new TransportAck(id, version);
|
||||
}
|
||||
|
||||
public boolean hasTransportUpdate() throws IOException {
|
||||
return !eof() && header[1] == TRANSPORT_UPDATE;
|
||||
}
|
||||
|
||||
public TransportUpdate readTransportUpdate() throws IOException {
|
||||
if (!hasTransportUpdate()) throw new FormatException();
|
||||
// Set up the reader
|
||||
InputStream bais = new ByteArrayInputStream(payload, 0, payloadLength);
|
||||
BdfReader r = bdfReaderFactory.createReader(bais);
|
||||
// Read the start of the payload
|
||||
r.readListStart();
|
||||
// Read the transport ID
|
||||
String idString = r.readString(MAX_TRANSPORT_ID_LENGTH);
|
||||
if (idString.length() == 0) throw new FormatException();
|
||||
TransportId id = new TransportId(idString);
|
||||
// Read the transport properties
|
||||
Map<String, String> p = new HashMap<String, String>();
|
||||
r.readDictionaryStart();
|
||||
for (int i = 0; !r.hasDictionaryEnd(); i++) {
|
||||
if (i == MAX_PROPERTIES_PER_TRANSPORT)
|
||||
throw new FormatException();
|
||||
String key = r.readString(MAX_PROPERTY_LENGTH);
|
||||
String value = r.readString(MAX_PROPERTY_LENGTH);
|
||||
p.put(key, value);
|
||||
}
|
||||
r.readDictionaryEnd();
|
||||
// Read the version number
|
||||
long version = r.readInteger();
|
||||
if (version < 0) throw new FormatException();
|
||||
// Read the end of the payload
|
||||
r.readListEnd();
|
||||
if (!r.eof()) throw new FormatException();
|
||||
state = State.BUFFER_EMPTY;
|
||||
// Build and return the transport update
|
||||
return new TransportUpdate(id, new TransportProperties(p), version);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,8 +12,6 @@ import org.briarproject.api.sync.PacketWriter;
|
||||
import org.briarproject.api.sync.Request;
|
||||
import org.briarproject.api.sync.SubscriptionAck;
|
||||
import org.briarproject.api.sync.SubscriptionUpdate;
|
||||
import org.briarproject.api.sync.TransportAck;
|
||||
import org.briarproject.api.sync.TransportUpdate;
|
||||
import org.briarproject.util.ByteUtils;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
@@ -25,8 +23,6 @@ import static org.briarproject.api.sync.PacketTypes.OFFER;
|
||||
import static org.briarproject.api.sync.PacketTypes.REQUEST;
|
||||
import static org.briarproject.api.sync.PacketTypes.SUBSCRIPTION_ACK;
|
||||
import static org.briarproject.api.sync.PacketTypes.SUBSCRIPTION_UPDATE;
|
||||
import static org.briarproject.api.sync.PacketTypes.TRANSPORT_ACK;
|
||||
import static org.briarproject.api.sync.PacketTypes.TRANSPORT_UPDATE;
|
||||
import static org.briarproject.api.sync.SyncConstants.MAX_PACKET_PAYLOAD_LENGTH;
|
||||
import static org.briarproject.api.sync.SyncConstants.PACKET_HEADER_LENGTH;
|
||||
import static org.briarproject.api.sync.SyncConstants.PROTOCOL_VERSION;
|
||||
@@ -125,27 +121,6 @@ class PacketWriterImpl implements PacketWriter {
|
||||
writePacket(SUBSCRIPTION_UPDATE);
|
||||
}
|
||||
|
||||
public void writeTransportAck(TransportAck a) throws IOException {
|
||||
if (payload.size() != 0) throw new IllegalStateException();
|
||||
BdfWriter w = bdfWriterFactory.createWriter(payload);
|
||||
w.writeListStart();
|
||||
w.writeString(a.getId().getString());
|
||||
w.writeInteger(a.getVersion());
|
||||
w.writeListEnd();
|
||||
writePacket(TRANSPORT_ACK);
|
||||
}
|
||||
|
||||
public void writeTransportUpdate(TransportUpdate u) throws IOException {
|
||||
if (payload.size() != 0) throw new IllegalStateException();
|
||||
BdfWriter w = bdfWriterFactory.createWriter(payload);
|
||||
w.writeListStart();
|
||||
w.writeString(u.getId().getString());
|
||||
w.writeDictionary(u.getProperties());
|
||||
w.writeInteger(u.getVersion());
|
||||
w.writeListEnd();
|
||||
writePacket(TRANSPORT_UPDATE);
|
||||
}
|
||||
|
||||
public void flush() throws IOException {
|
||||
out.flush();
|
||||
}
|
||||
|
||||
@@ -15,8 +15,6 @@ import org.briarproject.api.sync.PacketWriter;
|
||||
import org.briarproject.api.sync.SubscriptionAck;
|
||||
import org.briarproject.api.sync.SubscriptionUpdate;
|
||||
import org.briarproject.api.sync.SyncSession;
|
||||
import org.briarproject.api.sync.TransportAck;
|
||||
import org.briarproject.api.sync.TransportUpdate;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collection;
|
||||
@@ -68,7 +66,7 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
|
||||
this.transportId = transportId;
|
||||
this.maxLatency = maxLatency;
|
||||
this.packetWriter = packetWriter;
|
||||
outstandingQueries = new AtomicInteger(6); // One per type of packet
|
||||
outstandingQueries = new AtomicInteger(4); // One per type of packet
|
||||
writerTasks = new LinkedBlockingQueue<ThrowingRunnable<IOException>>();
|
||||
}
|
||||
|
||||
@@ -76,8 +74,6 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
|
||||
eventBus.addListener(this);
|
||||
try {
|
||||
// Start a query for each type of packet, in order of urgency
|
||||
dbExecutor.execute(new GenerateTransportAcks());
|
||||
dbExecutor.execute(new GenerateTransportUpdates());
|
||||
dbExecutor.execute(new GenerateSubscriptionAck());
|
||||
dbExecutor.execute(new GenerateSubscriptionUpdate());
|
||||
dbExecutor.execute(new GenerateAck());
|
||||
@@ -264,78 +260,4 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
|
||||
dbExecutor.execute(new GenerateSubscriptionUpdate());
|
||||
}
|
||||
}
|
||||
|
||||
// This task runs on the database thread
|
||||
private class GenerateTransportAcks implements Runnable {
|
||||
|
||||
public void run() {
|
||||
if (interrupted) return;
|
||||
try {
|
||||
Collection<TransportAck> acks =
|
||||
db.generateTransportAcks(contactId);
|
||||
if (LOG.isLoggable(INFO))
|
||||
LOG.info("Generated transport acks: " + (acks != null));
|
||||
if (acks == null) decrementOutstandingQueries();
|
||||
else writerTasks.add(new WriteTransportAcks(acks));
|
||||
} catch (DbException e) {
|
||||
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
|
||||
interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// This tasks runs on the writer thread
|
||||
private class WriteTransportAcks implements ThrowingRunnable<IOException> {
|
||||
|
||||
private final Collection<TransportAck> acks;
|
||||
|
||||
private WriteTransportAcks(Collection<TransportAck> acks) {
|
||||
this.acks = acks;
|
||||
}
|
||||
|
||||
public void run() throws IOException {
|
||||
if (interrupted) return;
|
||||
for (TransportAck a : acks) packetWriter.writeTransportAck(a);
|
||||
LOG.info("Sent transport acks");
|
||||
dbExecutor.execute(new GenerateTransportAcks());
|
||||
}
|
||||
}
|
||||
|
||||
// This task runs on the database thread
|
||||
private class GenerateTransportUpdates implements Runnable {
|
||||
|
||||
public void run() {
|
||||
if (interrupted) return;
|
||||
try {
|
||||
Collection<TransportUpdate> t =
|
||||
db.generateTransportUpdates(contactId, maxLatency);
|
||||
if (LOG.isLoggable(INFO))
|
||||
LOG.info("Generated transport updates: " + (t != null));
|
||||
if (t == null) decrementOutstandingQueries();
|
||||
else writerTasks.add(new WriteTransportUpdates(t));
|
||||
} catch (DbException e) {
|
||||
if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
|
||||
interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// This task runs on the writer thread
|
||||
private class WriteTransportUpdates
|
||||
implements ThrowingRunnable<IOException> {
|
||||
|
||||
private final Collection<TransportUpdate> updates;
|
||||
|
||||
private WriteTransportUpdates(Collection<TransportUpdate> updates) {
|
||||
this.updates = updates;
|
||||
}
|
||||
|
||||
public void run() throws IOException {
|
||||
if (interrupted) return;
|
||||
for (TransportUpdate u : updates)
|
||||
packetWriter.writeTransportUpdate(u);
|
||||
LOG.info("Sent transport updates");
|
||||
dbExecutor.execute(new GenerateTransportUpdates());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user