Finish file transfer after receiving enough data

This means that we no longer rely on the remote end to close the
connection after sending the file, but additionally use the `<size>`
element from the initial file transfer `<description>` to check whether
the file transfer has been completed.

This was motivated by Conversations not closing the connection for
SOCKS5 file transfers.

1 files changed, 53 insertions, 8 deletions

diff --git a/xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala b/xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala
index cce7b967..dc9851d4 100644
--- a/xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala
+++ b/xmpp-vala/src/module/xep/0234_jingle_file_transfer.vala
@@ -64,7 +64,7 @@ public class Parameters : Jingle.ContentParameters, Object {
     public int64 size { get; private set; }
     public StanzaNode original_description { get; private set; }
 
-    public Parameters(Module parent, StanzaNode original_description, string? media_type, string? name, int64? size) {
+    public Parameters(Module parent, StanzaNode original_description, string? media_type, string? name, int64 size) {
         this.parent = parent;
         this.original_description = original_description;
         this.media_type = media_type;
@@ -86,12 +86,15 @@ public class Parameters : Jingle.ContentParameters, Object {
         string? size_raw = size_node != null ? size_node.get_string_content() : null;
         // TODO(hrxi): For some reason, the ?:-expression does not work due to a type error.
         //int64? size = size_raw != null ? int64.parse(size_raw) : null; // TODO(hrxi): this has no error handling
-        int64 size = -1;
-        if (size_raw != null) {
-            size = int64.parse(size_raw);
-            if (size < 0) {
-                throw new Jingle.IqError.BAD_REQUEST("negative file size is invalid");
-            }
+        if (size_raw == null) {
+            // Jingle file transfers (XEP-0234) theoretically SHOULD send a
+            // file size, however, we do require it in order to reliably find
+            // the end of the file transfer.
+            throw new Jingle.IqError.BAD_REQUEST("file offer without file size");
+        }
+        int64 size = int64.parse(size_raw);
+        if (size < 0) {
+            throw new Jingle.IqError.BAD_REQUEST("negative file size is invalid");
         }
 
         return new Parameters(parent, description, media_type, name, size);
@@ -102,6 +105,47 @@ public class Parameters : Jingle.ContentParameters, Object {
     }
 }
 
+// Does nothing except wrapping an input stream to signal EOF after reading
+// `max_size` bytes.
+private class FileTransferInputStream : InputStream {
+    InputStream inner;
+    int64 remaining_size;
+    public FileTransferInputStream(InputStream inner, int64 max_size) {
+        this.inner = inner;
+        this.remaining_size = max_size;
+    }
+    private ssize_t update_remaining(ssize_t read) {
+        this.remaining_size -= read;
+        return read;
+    }
+    public override ssize_t read(uint8[] buffer_, Cancellable? cancellable = null) throws IOError {
+        unowned uint8[] buffer = buffer_;
+        if (remaining_size <= 0) {
+            return 0;
+        }
+        if (buffer.length > remaining_size) {
+            buffer = buffer[0:remaining_size];
+        }
+        return update_remaining(inner.read(buffer, cancellable));
+    }
+    public override async ssize_t read_async(uint8[]? buffer_, int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
+        unowned uint8[] buffer = buffer_;
+        if (remaining_size <= 0) {
+            return 0;
+        }
+        if (buffer.length > remaining_size) {
+            buffer = buffer[0:remaining_size];
+        }
+        return update_remaining(yield inner.read_async(buffer, io_priority, cancellable));
+    }
+    public override bool close(Cancellable? cancellable = null) throws IOError {
+        return inner.close(cancellable);
+    }
+    public override async bool close_async(int io_priority = GLib.Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
+        return yield inner.close_async(io_priority, cancellable);
+    }
+}
+
 public class FileTransfer : Object {
     Jingle.Session session;
     Parameters parameters;
@@ -110,11 +154,12 @@ public class FileTransfer : Object {
     public string? file_name { get { return parameters.name; } }
     public int64 size { get { return parameters.size; } }
 
-    public InputStream? stream { get { return session.conn != null ? session.conn.input_stream : null; } }
+    public InputStream? stream { get; private set; }
 
     public FileTransfer(Jingle.Session session, Parameters parameters) {
         this.session = session;
         this.parameters = parameters;
+        this.stream = new FileTransferInputStream(session.conn.input_stream, parameters.size);
     }
 
     public void accept(XmppStream stream) {