Reliability layer for the modem plugin (untested).

src/net/sf/briar/plugins/modem/Ack.java

@@ -0,0 +1,26 @@
+package net.sf.briar.plugins.modem;
+
+import net.sf.briar.util.ByteUtils;
+
+class Ack extends Frame {
+
+	static final int LENGTH = 12;
+
+	Ack() {
+		super(new byte[LENGTH], LENGTH);
+		b[0] = (byte) Frame.ACK_FLAG;
+	}
+
+	Ack(byte[] b) {
+		super(b, LENGTH);
+		b[0] = (byte) Frame.ACK_FLAG;
+	}
+
+	int getWindowSize() {
+		return ByteUtils.readUint24(b, 5);
+	}
+
+	void setWindowSize(int windowSize) {
+		ByteUtils.writeUint24(windowSize, b, 5);
+	}
+}

src/net/sf/briar/plugins/modem/Crc32.java

@@ -0,0 +1,28 @@
+package net.sf.briar.plugins.modem;
+
+class Crc32 {
+
+	private static final long[] TABLE = new long[256];
+
+	static {
+		for(int i = 0; i < 256; i++) {
+			long c = i;
+			for(int j = 0; j < 8; j++) {
+				if((c & 1) != 0) c = 0xedb88320L ^ (c >> 1);
+				else c >>= 1;
+			}
+			TABLE[i] = c;
+		}
+	}
+
+	static long update(long c, byte[] b, int off, int len) {
+		for(int i = off; i < off + len; i++) {
+			c = TABLE[(int) ((c ^ b[i]) & 0xff)] ^ (c >> 8);
+		}
+		return c;
+	}
+
+	static long crc(byte[] b, int off, int len) {
+		return update(0xffffffffL, b, off, len) ^ 0xffffffffL;
+	}
+}

src/net/sf/briar/plugins/modem/Data.java

@@ -0,0 +1,27 @@
+package net.sf.briar.plugins.modem;
+
+class Data extends Frame {
+
+	static final int HEADER_LENGTH = 5, FOOTER_LENGTH = 4;
+	static final int MIN_LENGTH = HEADER_LENGTH + FOOTER_LENGTH;
+	static final int MAX_PAYLOAD_LENGTH = 1024;
+	static final int MAX_LENGTH = MIN_LENGTH + MAX_PAYLOAD_LENGTH;
+
+	Data(byte[] b, int length) {
+		super(b, length);
+		if(length < MIN_LENGTH || length > MAX_LENGTH)
+			throw new IllegalArgumentException();
+	}
+
+	boolean isLastFrame() {
+		return b[0] == Frame.FIN_FLAG;
+	}
+
+	void setLastFrame(boolean lastFrame) {
+		if(lastFrame) b[0] = (byte) Frame.FIN_FLAG;
+	}
+
+	int getPayloadLength() {
+		return length - MIN_LENGTH;
+	}
+}

src/net/sf/briar/plugins/modem/Frame.java

@@ -0,0 +1,59 @@
+package net.sf.briar.plugins.modem;
+
+import net.sf.briar.util.ByteUtils;
+
+abstract class Frame {
+
+	static final byte ACK_FLAG = (byte) 128, FIN_FLAG = 64;
+
+	protected final byte[] b;
+	protected final int length;
+
+	Frame(byte[] b, int length) {
+		this.b = b;
+		this.length = length;
+	}
+
+	byte[] getBuffer() {
+		return b;
+	}
+
+	int getLength() {
+		return length;
+	}
+
+	long getChecksum() {
+		return ByteUtils.readUint32(b, length - 4);
+	}
+
+	void setChecksum(long checksum) {
+		ByteUtils.writeUint32(checksum, b, length - 4);
+	}
+
+	long calculateChecksum() {
+		return Crc32.crc(b, 0, length - 4);
+	}
+
+	long getSequenceNumber() {
+		return ByteUtils.readUint32(b, 1);
+	}
+
+	void setSequenceNumber(long sequenceNumber) {
+		ByteUtils.writeUint32(sequenceNumber, b, 1);
+	}
+
+	@Override
+	public int hashCode() {
+		return (int) getSequenceNumber();
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if(o instanceof Frame) {
+			Frame f = (Frame) o;
+			if(b[0] != f.b[0]) return false;
+			return getSequenceNumber() == f.getSequenceNumber();
+		}
+		return false;
+	}
+}

src/net/sf/briar/plugins/modem/ModemImpl.java

@@ -8,9 +8,7 @@ import static jssc.SerialPort.PURGE_TXCLEAR;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Executor;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Logger;
@@ -20,7 +18,7 @@ import jssc.SerialPortEvent;
 import jssc.SerialPortEventListener;
 import jssc.SerialPortException;
 
-class ModemImpl implements Modem, SerialPortEventListener {
+class ModemImpl implements Modem, WriteHandler, SerialPortEventListener {
 
 	private static final Logger LOG =
 			Logger.getLogger(ModemImpl.class.getName());
@@ -40,8 +38,7 @@ class ModemImpl implements Modem, SerialPortEventListener {
 
 	private int lineLen = 0;
 
-	private volatile ModemInputStream inputStream = null;
-	private volatile ModemOutputStream outputStream = null;
+	private volatile ReliabilityLayer reliabilityLayer = null;
 
 	ModemImpl(Executor executor, Callback callback, String portName) {
 		this.executor = executor;
@@ -80,8 +77,6 @@ class ModemImpl implements Modem, SerialPortEventListener {
 				if(!initialised.get()) initialised.wait(OK_TIMEOUT);
 			}
 		} catch(InterruptedException e) {
-			if(LOG.isLoggable(WARNING))
-				LOG.warning("Interrupted while initialising modem");
 			tryToClose(port);
 			Thread.currentThread().interrupt();
 			throw new IOException("Interrupted while initialising modem");
@@ -108,8 +103,6 @@ class ModemImpl implements Modem, SerialPortEventListener {
 				if(!connected.get()) connected.wait(CONNECT_TIMEOUT);
 			}
 		} catch(InterruptedException e) {
-			if(LOG.isLoggable(WARNING))
-				LOG.warning("Interrupted while connecting outgoing call");
 			tryToClose(port);
 			Thread.currentThread().interrupt();
 			throw new IOException("Interrupted while connecting outgoing call");
@@ -120,11 +113,11 @@ class ModemImpl implements Modem, SerialPortEventListener {
 	}
 
 	public InputStream getInputStream() {
-		return inputStream;
+		return reliabilityLayer.getInputStream();
 	}
 
 	public OutputStream getOutputStream() {
-		return outputStream;
+		return reliabilityLayer.getOutputStream();
 	}
 
 	public void hangUp() throws IOException {
@@ -135,18 +128,30 @@ class ModemImpl implements Modem, SerialPortEventListener {
 			tryToClose(port);
 			throw new IOException(e.toString());
 		}
-		inputStream.closed = true;
-		inputStream.received.add(new byte[0]); // Poison pill
-		outputStream.closed = true;
+		reliabilityLayer.invalidate();
 		connected.set(false);
 		offHook.release();
 	}
 
+	public void handleWrite(byte[] b, int length) throws IOException {
+		if(length < b.length) {
+			byte[] copy = new byte[length];
+			System.arraycopy(b, 0, copy, 0, length);
+			b = copy;
+		}
+		try {
+			port.writeBytes(b);
+		} catch(SerialPortException e) {
+			tryToClose(port);
+			throw new IOException(e.toString());
+		}
+	}
+
 	public void serialEvent(SerialPortEvent ev) {
 		try {
 			if(ev.isRXCHAR()) {
 				byte[] b = port.readBytes();
-				if(connected.get()) inputStream.received.add(b);
+				if(connected.get()) reliabilityLayer.handleRead(b, b.length);
 				else handleText(b);
 			} else if(ev.isDSR() && ev.getEventValue() == 0) {
 				if(LOG.isLoggable(INFO)) LOG.info("Remote end hung up");
@@ -171,8 +176,7 @@ class ModemImpl implements Modem, SerialPortEventListener {
 				lineLen = 0;
 				if(LOG.isLoggable(INFO)) LOG.info("Modem status: " + s);
 				if(s.startsWith("CONNECT")) {
-					inputStream = new ModemInputStream();
-					outputStream = new ModemOutputStream();
+					reliabilityLayer = new ReliabilityLayer(this);
 					synchronized(connected) {
 						if(connected.getAndSet(true))
 							throw new IOException("Connected twice");
@@ -183,7 +187,7 @@ class ModemImpl implements Modem, SerialPortEventListener {
 					if(off < b.length) {
 						byte[] data = new byte[b.length - off];
 						System.arraycopy(b, off, data, 0, data.length);
-						inputStream.received.add(data);
+						reliabilityLayer.handleRead(data, data.length);
 					}
 					return;
 				} else if(s.equals("OK")) {
@@ -228,8 +232,6 @@ class ModemImpl implements Modem, SerialPortEventListener {
 				if(!connected.get()) connected.wait(CONNECT_TIMEOUT);
 			}
 		} catch(InterruptedException e) {
-			if(LOG.isLoggable(WARNING))
-				LOG.warning("Interrupted while connecting incoming call");
 			tryToClose(port);
 			Thread.currentThread().interrupt();
 			throw new IOException("Interrupted while connecting incoming call");
@@ -245,100 +247,4 @@ class ModemImpl implements Modem, SerialPortEventListener {
 			if(LOG.isLoggable(WARNING)) LOG.warning(e.toString());
 		}
 	}
-
-	private class ModemInputStream extends InputStream {
-
-		private final BlockingQueue<byte[]> received;
-
-		private byte[] buf = null;
-		private int offset = 0;
-
-		private volatile boolean closed = false;
-
-		private ModemInputStream() {
-			this.received = new LinkedBlockingQueue<byte[]>();
-		}
-
-		@Override
-		public int read() throws IOException {
-			if(closed) throw new IOException("Connection closed");
-			getBufferIfNecessary();
-			return buf[offset++];
-		}
-
-		@Override
-		public int read(byte[] b) throws IOException {
-			if(closed) throw new IOException("Connection closed");
-			getBufferIfNecessary();
-			int len = Math.min(b.length, buf.length - offset);
-			System.arraycopy(buf, offset, b, 0, len);
-			offset += len;
-			return len;
-		}
-
-		@Override
-		public int read(byte[] b, int off, int len) throws IOException {
-			if(closed) throw new IOException("Connection closed");
-			getBufferIfNecessary();
-			len = Math.min(len, buf.length - offset);
-			System.arraycopy(buf, offset, b, off, len);
-			offset += len;
-			return len;
-		}
-
-		private void getBufferIfNecessary() throws IOException {
-			if(buf == null || offset == buf.length) {
-				try {
-					buf = received.take();
-				} catch(InterruptedException e) {
-					if(LOG.isLoggable(WARNING))
-						LOG.warning("Interrupted while reading");
-					tryToClose(port);
-					Thread.currentThread().interrupt();
-					throw new IOException(e.toString());
-				}
-				if(buf.length == 0) throw new IOException("Connection closed");
-				offset = 0;
-			}
-		}
-	}
-
-	private class ModemOutputStream extends OutputStream {
-
-		private volatile boolean closed = false;
-
-		@Override
-		public void write(int b) throws IOException {
-			if(closed) throw new IOException("Connection closed");
-			try {
-				port.writeByte((byte) b);
-			} catch(SerialPortException e) {
-				tryToClose(port);
-				throw new IOException(e.toString());
-			}
-		}
-
-		@Override
-		public void write(byte[] b) throws IOException {
-			if(closed) throw new IOException("Connection closed");
-			try {
-				port.writeBytes(b);
-			} catch(SerialPortException e) {
-				tryToClose(port);
-				throw new IOException(e.toString());
-			}
-		}
-
-		@Override
-		public void write(byte[] b, int off, int len) throws IOException {
-			if(closed) throw new IOException("Connection closed");
-			if(len < b.length) {
-				byte[] copy = new byte[len];
-				System.arraycopy(b, off, copy, 0, len);
-				write(copy);
-			} else {
-				write(b);
-			}
-		}
-	}
 }

src/net/sf/briar/plugins/modem/ReadHandler.java

@@ -0,0 +1,8 @@
+package net.sf.briar.plugins.modem;
+
+import java.io.IOException;
+
+interface ReadHandler {
+
+	void handleRead(byte[] b, int length) throws IOException;
+}

src/net/sf/briar/plugins/modem/Receiver.java

@@ -0,0 +1,138 @@
+package net.sf.briar.plugins.modem;
+
+import static java.util.logging.Level.FINE;
+
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.logging.Logger;
+
+class Receiver implements ReadHandler {
+
+	private static final Logger LOG =
+			Logger.getLogger(Receiver.class.getName());
+
+	private static final int MAX_WINDOW_SIZE = 8 * Data.MAX_PAYLOAD_LENGTH;
+
+	private final Sender sender;
+	private final SortedSet<Data> dataFrames; // Locking: this
+
+	private int windowSize = MAX_WINDOW_SIZE; // Locking: this
+	private long finalSequenceNumber = Long.MAX_VALUE;
+	private long nextSequenceNumber = 1L;
+
+	Receiver(Sender sender) {
+		this.sender = sender;
+		dataFrames = new TreeSet<Data>(new SequenceNumberComparator());
+	}
+
+	synchronized Data read() throws IOException, InterruptedException {
+		while(true) {
+			if(dataFrames.isEmpty()) {
+				if(LOG.isLoggable(FINE)) LOG.fine("Waiting for a data frame");
+				wait();
+			} else {
+				Data d = dataFrames.first();
+				if(d.getSequenceNumber() == nextSequenceNumber) {
+					if(LOG.isLoggable(FINE))
+						LOG.fine("Reading #" + d.getSequenceNumber());
+					dataFrames.remove(d);
+					// Update the window
+					windowSize += d.getPayloadLength();
+					if(LOG.isLoggable(FINE))
+						LOG.fine("Window at receiver " + windowSize);
+					sender.sendAck(0L, windowSize);
+					nextSequenceNumber++;
+					return d;
+				} else {
+					if(LOG.isLoggable(FINE))
+						LOG.fine("Waiting for #" + nextSequenceNumber);
+					wait();
+				}
+			}
+		}
+	}
+
+	public void handleRead(byte[] b, int length) throws IOException {
+		if(length < Data.MIN_LENGTH || length > Data.MAX_LENGTH) {
+			if(LOG.isLoggable(FINE))
+				LOG.fine("Ignoring frame with invalid length");
+			return;
+		}
+		switch(b[0]) {
+		case 0:
+		case Frame.FIN_FLAG:
+			handleData(b, length);
+			break;
+		case Frame.ACK_FLAG:
+			sender.handleAck(b, length);
+			break;
+		default:
+			if(LOG.isLoggable(FINE)) LOG.fine("Ignoring unknown frame type");
+			return;
+		}
+	}
+
+	private synchronized void handleData(byte[] b, int length)
+			throws IOException {
+		Data d = new Data(b, length);
+		int payloadLength = d.getPayloadLength();
+		if(payloadLength > windowSize) {
+			if(LOG.isLoggable(FINE)) LOG.fine("No space in the window");
+			return;
+		}
+		if(d.getChecksum() != d.calculateChecksum()) {
+			if(LOG.isLoggable(FINE))
+				LOG.fine("Incorrect checksum on data frame");
+			return;
+		}
+		long sequenceNumber = d.getSequenceNumber();
+		if(sequenceNumber == 0L) {
+			if(LOG.isLoggable(FINE)) LOG.fine("Window probe");
+		} else if(sequenceNumber < nextSequenceNumber) {
+			if(LOG.isLoggable(FINE)) LOG.fine("Duplicate data frame");
+		} else if(d.isLastFrame()) {
+			finalSequenceNumber = sequenceNumber;
+			Iterator<Data> it = dataFrames.iterator();
+			while(it.hasNext()) {
+				Data d1 = it.next();
+				if(d1.getSequenceNumber() >= finalSequenceNumber) {
+					if(LOG.isLoggable(FINE))
+						LOG.fine("Received data frame after FIN");
+					it.remove();
+				}
+			}
+			if(LOG.isLoggable(FINE)) LOG.fine("Received #" + sequenceNumber);
+			if(dataFrames.add(d)) {
+				windowSize -= payloadLength;
+				if(LOG.isLoggable(FINE))
+					LOG.fine("Window at receiver " + windowSize);
+				notifyAll();
+			}
+		} else if(sequenceNumber < finalSequenceNumber) {
+			if(LOG.isLoggable(FINE)) LOG.fine("Received #" + sequenceNumber);
+			if(dataFrames.add(d)) {
+				windowSize -= payloadLength;
+				if(LOG.isLoggable(FINE))
+					LOG.fine("Window at receiver " + windowSize);
+				notifyAll();
+			}
+		} else {
+			if(LOG.isLoggable(FINE)) LOG.fine("Received data frame after FIN");
+		}
+		// Acknowledge the data frame even if it's a duplicate
+		sender.sendAck(sequenceNumber, windowSize);
+	}
+
+	private static class SequenceNumberComparator implements Comparator<Data> {
+
+		public int compare(Data d1, Data d2) {
+			long s1 = d1.getSequenceNumber(), s2 = d2.getSequenceNumber();
+			if(s1 < s2) return -1;
+			if(s1 > s2) return 1;
+			return 0;
+		}
+	}
+}

src/net/sf/briar/plugins/modem/ReceiverInputStream.java

@@ -0,0 +1,59 @@
+package net.sf.briar.plugins.modem;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+class ReceiverInputStream extends InputStream {
+
+	private final Receiver receiver;
+
+	private Data data = null;
+	private int offset = 0, length = 0;
+
+	ReceiverInputStream(Receiver receiver) {
+		this.receiver = receiver;
+	}
+
+	@Override
+	public int read() throws IOException {
+		if(length == -1) return -1;
+		while(length == 0) if(!receive()) return -1;
+		int b = data.getBuffer()[offset] & 0xff;
+		offset++;
+		length--;
+		return b;
+	}
+
+	@Override
+	public int read(byte[] b) throws IOException {
+		return read(b, 0, b.length);
+	}
+
+	@Override
+	public int read(byte[] b, int off, int len) throws IOException {
+		if(length == -1) return -1;
+		while(length == 0) if(!receive()) return -1;
+		len = Math.min(len, length);
+		System.arraycopy(data.getBuffer(), offset, b, off, len);
+		offset += len;
+		length -= len;
+		return len;
+	}
+
+	private boolean receive() throws IOException {
+		assert length == 0;
+		if(data != null && data.isLastFrame()) {
+			length = -1;
+			return false;
+		}
+		try {
+			data = receiver.read();
+		} catch(InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new IOException("Interrupted while reading");
+		}
+		offset = Data.HEADER_LENGTH;
+		length = data.getLength() - Data.MIN_LENGTH;
+		return true;
+	}
+}

src/net/sf/briar/plugins/modem/ReliabilityLayer.java

@@ -0,0 +1,52 @@
+package net.sf.briar.plugins.modem;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+class ReliabilityLayer implements ReadHandler, WriteHandler {
+
+	// Write side
+	private final WriteHandler writeHandler;
+	private final SlipEncoder encoder;
+	private final Sender sender;
+	private final SenderOutputStream outputStream;
+	// Read side
+	private final SlipDecoder decoder;
+	private final Receiver receiver;
+	private final ReceiverInputStream inputStream;
+
+	private volatile boolean valid = true;
+
+	ReliabilityLayer(WriteHandler writeHandler) {
+		this.writeHandler = writeHandler;
+		encoder = new SlipEncoder(this);
+		sender = new Sender(encoder);
+		outputStream = new SenderOutputStream(sender);
+		receiver = new Receiver(sender);
+		decoder = new SlipDecoder(receiver);
+		inputStream = new ReceiverInputStream(receiver);
+	}
+
+	InputStream getInputStream() {
+		return inputStream;
+	}
+
+	OutputStream getOutputStream() {
+		return outputStream;
+	}
+
+	void invalidate() {
+		valid = false;
+	}
+
+	public void handleRead(byte[] b, int length) throws IOException {
+		if(!valid) throw new IOException("Connection closed");
+		decoder.handleRead(b, length);
+	}
+
+	public void handleWrite(byte[] b, int length) throws IOException {
+		if(!valid) throw new IOException("Connection closed");
+		writeHandler.handleWrite(b, length);
+	}
+}

src/net/sf/briar/plugins/modem/Sender.java

@@ -0,0 +1,221 @@
+package net.sf.briar.plugins.modem;
+
+import static java.util.logging.Level.FINE;
+import static java.util.logging.Level.WARNING;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.logging.Logger;
+
+class Sender {
+
+	private static final Logger LOG =
+			Logger.getLogger(Sender.class.getName());
+
+	// All times are in milliseconds
+	private static final int MIN_TIMEOUT = 1000;
+	private static final int MAX_TIMEOUT = 60 * 1000;
+	private static final int INITIAL_RTT = 0;
+	private static final int INITIAL_RTT_VAR = 3 * 1000;
+
+	private final WriteHandler writeHandler;
+	private final LinkedList<Outstanding> outstanding; // Locking: this
+
+	private int outstandingBytes = 0; // Locking: this
+	private int windowSize = Data.MAX_PAYLOAD_LENGTH; // Locking: this
+	private int rtt = INITIAL_RTT, rttVar = INITIAL_RTT_VAR; // Locking: this
+	private int timeout = rtt + (rttVar << 2); // Locking: this
+	private long lastWindowUpdateOrProbe = Long.MAX_VALUE; // Locking: this
+	private boolean dataWaiting = false; // Locking: this
+
+	Sender(WriteHandler writeHandler) {
+		this.writeHandler = writeHandler;
+		outstanding = new LinkedList<Outstanding>();
+	}
+
+	void sendAck(long sequenceNumber, int windowSize) throws IOException {
+		Ack a = new Ack();
+		a.setSequenceNumber(sequenceNumber);
+		a.setWindowSize(windowSize);
+		a.setChecksum(a.calculateChecksum());
+		if(sequenceNumber == 0L) {
+			if(LOG.isLoggable(FINE)) LOG.fine("Sending window update");
+		} else {
+			if(LOG.isLoggable(FINE))
+				LOG.fine("Acknowledging #" + sequenceNumber);
+		}
+		writeHandler.handleWrite(a.getBuffer(), a.getLength());
+	}
+
+	void handleAck(byte[] b, int length) {
+		if(length != Ack.LENGTH) {
+			if(LOG.isLoggable(FINE))
+				LOG.fine("Ignoring ack frame with wrong length");
+			return;
+		}
+		Ack a = new Ack(b);
+		if(a.getChecksum() != a.calculateChecksum()) {
+			if(LOG.isLoggable(FINE))
+				LOG.fine("Incorrect checksum on ack frame");
+			return;
+		}
+		long sequenceNumber = a.getSequenceNumber();
+		long now = System.currentTimeMillis();
+		Outstanding fastRetransmit = null;
+		synchronized(this) {
+			// Remove the acked data frame if it's outstanding
+			int foundIndex = -1;
+			Iterator<Outstanding> it = outstanding.iterator();
+			for(int i = 0; it.hasNext(); i++) {
+				Outstanding o = it.next();
+				if(o.data.getSequenceNumber() == sequenceNumber) {
+					if(LOG.isLoggable(FINE))
+						LOG.fine("#" + sequenceNumber + " acknowledged");
+					it.remove();
+					outstandingBytes -= o.data.getPayloadLength();
+					foundIndex = i;
+					// Update the round-trip time and retransmission timer
+					if(!o.retransmitted) {
+						int sample = (int) (now - o.lastTransmitted);
+						int error = sample - rtt;
+						rtt += (error >> 3);
+						rttVar += (Math.abs(error) - rttVar) >> 2;
+						timeout = rtt + (rttVar << 2);
+						if(timeout < MIN_TIMEOUT) timeout = MIN_TIMEOUT;
+						else if(timeout > MAX_TIMEOUT) timeout = MAX_TIMEOUT;
+						if(LOG.isLoggable(FINE))
+							LOG.fine("RTT " + rtt + ", timeout " + timeout);
+					}
+					break;
+				}
+			}
+			// If any older data frames are outstanding, retransmit the oldest
+			if(foundIndex > 0) {
+				fastRetransmit = outstanding.poll();
+				if(LOG.isLoggable(FINE)) {
+					LOG.fine("Fast retransmitting #"
+							+ fastRetransmit.data.getSequenceNumber());
+				}
+				fastRetransmit.lastTransmitted = now;
+				fastRetransmit.retransmitted = true;
+				outstanding.add(fastRetransmit);
+			}
+			// Update the window
+			lastWindowUpdateOrProbe = now;
+			int oldWindowSize = windowSize;
+			windowSize = a.getWindowSize();
+			if(LOG.isLoggable(FINE)) LOG.fine("Window at sender " + windowSize);
+			// If space has become available, notify any waiting writers
+			if(windowSize > oldWindowSize || foundIndex != -1) notifyAll();
+		}
+		// Fast retransmission
+		if(fastRetransmit != null) {
+			Data d = fastRetransmit.data;
+			try {
+				writeHandler.handleWrite(d.getBuffer(), d.getLength());
+			} catch(IOException e) {
+				// FIXME: Do something more meaningful
+				if(LOG.isLoggable(WARNING)) LOG.warning(e.toString());
+			}
+		}
+	}
+
+	void tick() {
+		long now = System.currentTimeMillis();
+		List<Outstanding> retransmit = null;
+		boolean sendProbe = false;
+		synchronized(this) {
+			if(outstanding.isEmpty()) {
+				if(dataWaiting && now - lastWindowUpdateOrProbe > timeout) {
+					if(LOG.isLoggable(FINE)) LOG.fine("Sending window probe");
+					sendProbe = true;
+					timeout <<= 1;
+					if(timeout > MAX_TIMEOUT) timeout = MAX_TIMEOUT;
+					if(LOG.isLoggable(FINE))
+						LOG.fine("Increasing timeout to " + timeout);
+				}
+			} else {
+				Iterator<Outstanding> it = outstanding.iterator();
+				while(it.hasNext()) {
+					Outstanding o = it.next();
+					if(now - o.lastTransmitted > timeout) {
+						if(LOG.isLoggable(FINE)) {
+							LOG.fine("Retransmitting #"
+									+ o.data.getSequenceNumber());
+						}
+						it.remove();
+						if(retransmit == null)
+							retransmit = new ArrayList<Outstanding>();
+						retransmit.add(o);
+						timeout <<= 1;
+						if(timeout > MAX_TIMEOUT) timeout = MAX_TIMEOUT;
+						if(LOG.isLoggable(FINE))
+							LOG.fine("Increasing timeout to " + timeout);
+					}
+				}
+				if(retransmit != null) {
+					for(Outstanding o : retransmit) {
+						o.lastTransmitted = now;
+						o.retransmitted = true;
+						outstanding.add(o);
+					}
+				}
+			}
+		}
+		try {
+			// Send a window probe if necessary
+			if(sendProbe) {
+				byte[] buf = new byte[Data.MIN_LENGTH];
+				Data probe = new Data(buf, Data.MIN_LENGTH);
+				probe.setChecksum(probe.calculateChecksum());
+				writeHandler.handleWrite(buf, Data.MIN_LENGTH);
+			}
+			// Retransmit any lost data frames
+			if(retransmit != null) {
+				for(Outstanding o : retransmit) {
+					Data d = o.data;
+					writeHandler.handleWrite(d.getBuffer(), d.getLength());
+				}
+			}
+		} catch(IOException e) {
+			// FIXME: Do something more meaningful
+			if(LOG.isLoggable(WARNING)) LOG.warning(e.toString());
+			return;
+		}
+	}
+
+	void write(Data d) throws IOException, InterruptedException {
+		int payloadLength = d.getPayloadLength();
+		synchronized(this) {
+			while(outstandingBytes + payloadLength >= windowSize) {
+				if(LOG.isLoggable(FINE))
+					LOG.fine("Waiting for space in the window");
+				dataWaiting = true;
+				wait();
+			}
+			outstanding.add(new Outstanding(d));
+			outstandingBytes += payloadLength;
+			dataWaiting = false;
+		}
+		if(LOG.isLoggable(FINE))
+			LOG.fine("Transmitting #" + d.getSequenceNumber());
+		writeHandler.handleWrite(d.getBuffer(), d.getLength());
+	}
+
+	private static class Outstanding {
+
+		private final Data data;
+
+		private volatile long lastTransmitted;
+		private volatile boolean retransmitted;
+
+		private Outstanding(Data data) {
+			this.data = data;
+			lastTransmitted = System.currentTimeMillis();
+			retransmitted = false;
+		}
+	}
+}

src/net/sf/briar/plugins/modem/SenderOutputStream.java

@@ -0,0 +1,78 @@
+package net.sf.briar.plugins.modem;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+class SenderOutputStream extends OutputStream {
+
+	private final Sender sender;
+
+	private byte[] buf = null;
+	private int offset = 0;
+	private long sequenceNumber = 1L;
+
+	SenderOutputStream(Sender sender) {
+		this.sender = sender;
+	}
+
+	@Override
+	public void close() throws IOException {
+		if(buf == null) assignBuffer();
+		send(true);
+	}
+
+	@Override
+	public void flush() throws IOException {
+		if(buf != null) send(false);
+	}
+
+	@Override
+	public void write(int b) throws IOException {
+		if(buf == null) assignBuffer();
+		buf[offset] = (byte) b;
+		offset++;
+		if(offset == Data.HEADER_LENGTH + Data.MAX_PAYLOAD_LENGTH) send(false);
+	}
+
+	@Override
+	public void write(byte[] b) throws IOException {
+		write(b, 0, b.length);
+	}
+
+	@Override
+	public void write(byte[] b, int off, int len) throws IOException {
+		if(buf == null) assignBuffer();
+		int available = Data.MAX_LENGTH - offset - Data.FOOTER_LENGTH;
+		while(available <= len) {
+			System.arraycopy(b, off, buf, offset, available);
+			offset += available;
+			send(false);
+			assignBuffer();
+			off += available;
+			len -= available;
+			available = Data.MAX_LENGTH - offset - Data.FOOTER_LENGTH;
+		}
+		System.arraycopy(b, off, buf, offset, len);
+		offset += len;
+	}
+
+	private void assignBuffer() {
+		buf = new byte[Data.MAX_LENGTH];
+		offset = Data.HEADER_LENGTH;
+	}
+
+	private void send(boolean lastFrame) throws IOException {
+		Data d = new Data(buf, offset + Data.FOOTER_LENGTH);
+		d.setLastFrame(lastFrame);
+		d.setSequenceNumber(sequenceNumber++);
+		d.setChecksum(d.calculateChecksum());
+		try {
+			sender.write(d);
+		} catch(InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new IOException("Interrupted while writing");
+		}
+		buf = null;
+		offset = 0;
+	}
+}

src/net/sf/briar/plugins/modem/SlipDecoder.java

@@ -0,0 +1,81 @@
+package net.sf.briar.plugins.modem;
+
+import static java.util.logging.Level.FINE;
+
+import java.io.IOException;
+import java.util.logging.Logger;
+
+class SlipDecoder implements ReadHandler {
+
+	private static final Logger LOG =
+			Logger.getLogger(SlipDecoder.class.getName());
+
+	// https://tools.ietf.org/html/rfc1055
+	private static final byte END = (byte) 192, ESC = (byte) 219;
+	private static final byte TEND = (byte) 220, TESC = (byte) 221;
+
+	private final ReadHandler readHandler;
+
+	private byte[] buf = new byte[Data.MAX_LENGTH];
+	private int decodedLength = 0;
+	private boolean escape = false;
+
+	SlipDecoder(ReadHandler readHandler) {
+		this.readHandler = readHandler;
+	}
+
+	public void handleRead(byte[] b, int length) throws IOException {
+		for(int i = 0; i < length; i++) {
+			switch(b[i]) {
+			case END:
+				if(escape) {
+					reset(true);
+				} else {
+					if(decodedLength > 0) {
+						readHandler.handleRead(buf, decodedLength);
+						buf = new byte[Data.MAX_LENGTH];
+					}
+					reset(false);
+				}
+				break;
+			case ESC:
+				if(escape) reset(true);
+				else escape = true;
+				break;
+			case TEND:
+				if(escape) {
+					escape = false;
+					if(decodedLength == buf.length) reset(true);
+					else buf[decodedLength++] = END;
+				} else {
+					if(decodedLength == buf.length) reset(true);
+					else buf[decodedLength++] = TEND;
+				}
+				break;
+			case TESC:
+				if(escape) {
+					escape = false;
+					if(decodedLength == buf.length) reset(true);
+					else buf[decodedLength++] = ESC;
+				} else {
+					if(decodedLength == buf.length) reset(true);
+					else buf[decodedLength++] = TESC;
+				}
+				break;
+			default:
+				if(escape || decodedLength == buf.length) reset(true);
+				else buf[decodedLength++] = b[i];
+				break;
+			}
+		}
+	}
+
+	private void reset(boolean error) {
+		if(error) {
+			if(LOG.isLoggable(FINE))
+				LOG.fine("Decoding error after " + decodedLength + " bytes");
+		}
+		escape = false;
+		decodedLength = 0;
+	}
+}

src/net/sf/briar/plugins/modem/SlipEncoder.java

@@ -0,0 +1,39 @@
+package net.sf.briar.plugins.modem;
+
+import java.io.IOException;
+
+class SlipEncoder implements WriteHandler {
+
+	// https://tools.ietf.org/html/rfc1055
+	private static final byte END = (byte) 192, ESC = (byte) 219;
+	private static final byte TEND = (byte) 220, TESC = (byte) 221;
+
+	private final WriteHandler writeHandler;
+
+	SlipEncoder(WriteHandler writeHandler) {
+		this.writeHandler = writeHandler;
+	}
+
+	public void handleWrite(byte[] b, int length) throws IOException {
+		if(length > Data.MAX_LENGTH) throw new IllegalArgumentException();
+		int encodedLength = length + 2;
+		for(int i = 0; i < length; i++) {
+			if(b[i] == END || b[i] == ESC) encodedLength++;
+		}
+		byte[] buf = new byte[encodedLength];
+		buf[0] = END;
+		for(int i = 0, j = 1; i < length; i++) {
+			if(b[i] == END) {
+				buf[j++] = ESC;
+				buf[j++] = TEND;
+			} else if(b[i] == ESC) {
+				buf[j++] = ESC;
+				buf[j++] = TESC;
+			} else {
+				buf[j++] = b[i];
+			}
+		}
+		buf[encodedLength - 1] = END;
+		writeHandler.handleWrite(buf, encodedLength);
+	}
+}

src/net/sf/briar/plugins/modem/WriteHandler.java

@@ -0,0 +1,8 @@
+package net.sf.briar.plugins.modem;
+
+import java.io.IOException;
+
+interface WriteHandler {
+
+	void handleWrite(byte[] b, int length) throws IOException;
+}

src/net/sf/briar/util/ByteUtils.java

@@ -7,6 +7,11 @@ public class ByteUtils {
 	 */
 	public static final int MAX_16_BIT_UNSIGNED = 65535; // 2^16 - 1
 
+	/**
+	 * The maximum value that can be represented as an unsigned 24-bit integer.
+	 */
+	public static final int MAX_24_BIT_UNSIGNED = 16777215; // 2^24 - 1
+
 	/**
 	 * The maximum value that can be represented as an unsigned 32-bit integer.
 	 */
@@ -27,6 +32,15 @@ public class ByteUtils {
 		b[offset + 1] = (byte) (i & 0xFF);
 	}
 
+	public static void writeUint24(long i, byte[] b, int offset) {
+		if(i < 0L) throw new IllegalArgumentException();
+		if(i > MAX_24_BIT_UNSIGNED) throw new IllegalArgumentException();
+		if(b.length < offset + 3) throw new IllegalArgumentException();
+		b[offset] = (byte) (i >> 16);
+		b[offset + 1] = (byte) (i >> 8 & 0xFF);
+		b[offset + 2] = (byte) (i & 0xFF);
+	}
+
 	public static void writeUint32(long i, byte[] b, int offset) {
 		if(i < 0L) throw new IllegalArgumentException();
 		if(i > MAX_32_BIT_UNSIGNED) throw new IllegalArgumentException();
@@ -42,6 +56,12 @@ public class ByteUtils {
 		return ((b[offset] & 0xFF) << 8) | (b[offset + 1] & 0xFF);
 	}
 
+	public static int readUint24(byte[] b, int offset) {
+		if(b.length < offset + 3) throw new IllegalArgumentException();
+		return ((b[offset] & 0xFF) << 16) | ((b[offset + 1] & 0xFF) << 8)
+				| (b[offset + 2] & 0xFF);
+	}
+
 	public static long readUint32(byte[] b, int offset) {
 		if(b.length < offset + 4) throw new IllegalArgumentException();
 		return ((b[offset] & 0xFFL) << 24) | ((b[offset + 1] & 0xFFL) << 16)