Initial try at compression

examples/FormData.vala

@@ -489,6 +489,7 @@ void main() {
         .add_endpoint(new FormDebugEndpoint());
     
     var pipeline = new Pipeline()
+        .add_component(new Compression())
         .add_component(router);
 
     var server = new Server(8084, pipeline);

src/Components/Compression.vala

@@ -0,0 +1,38 @@
+
+namespace Astralis {
+
+    public class Compression : Object, PipelineComponent {
+
+        public async HttpResult process_request (HttpContext http_context, PipelineContext pipeline_context) throws Error {
+            var result  = yield pipeline_context.next();
+            var accept_encoding = http_context.request.headers.get_any_or_default("Accept-Encoding");
+            if(accept_encoding != null && accept_encoding.contains("gzip")) {
+                return new CompressionResult (result, ZlibCompressorFormat.GZIP)
+                    .set_header("Content-Encoding", "gzip");
+            }
+            return result;
+        }
+
+        private class CompressionResult : HttpResult {
+            private HttpResult inner_result;
+            private ZlibCompressorFormat format;
+
+            public CompressionResult(HttpResult result, ZlibCompressorFormat format) {
+                base(result.status, result.content_length);
+                replace_headers(result.headers);
+
+                this.inner_result = result;
+                this.format = format;
+            }
+
+            public async override void send_body (AsyncOutput output) throws Error {
+                var converter = new ZlibCompressor(format);
+                var converter_output = new ConverterAsyncOutput(output, converter);
+                yield inner_result.send_body(converter_output);
+            }
+            
+        }
+
+    }
+
+}

src/Core/AsyncInput.vala

@@ -5,8 +5,8 @@ namespace Astralis {
     public interface AsyncInput : Object {
 
         public abstract BinaryData? peek();
-        public abstract async BinaryData? read();
-        public abstract async BinaryData read_all();
+        public abstract async BinaryData? read() throws Error;
+        public abstract async BinaryData read_all() throws Error;
 
     }
 

src/Core/AsyncOutput.vala

@@ -1,11 +1,79 @@
 using Invercargill;
+using Invercargill.DataStructures;
 
 namespace Astralis {
 
     public interface AsyncOutput : Object {
 
-        public abstract async void write(BinaryData data);
-        public abstract async void write_stream(OutputStream stream);
+        public abstract async void write(BinaryData data) throws Error;
+        public abstract async void write_stream(InputStream stream) throws Error;
+
+    }
+
+    public class ConverterAsyncOutput : Object, AsyncOutput {
+
+        private AsyncOutput inner;
+        private Converter converter;
+
+        public ConverterAsyncOutput(AsyncOutput output, Converter converter) {
+            this.inner = output;
+            this.converter = converter;
+        }
+
+        public async void write(Invercargill.BinaryData data) throws Error {
+            var in_buffer = data.to_array();
+            size_t in_offset = 0;
+            const size_t BUFFER_SIZE = 4096;
+            
+            while (in_offset < in_buffer.length) {
+                uint8[] out_buffer = new uint8[BUFFER_SIZE];
+                size_t bytes_read = 0;
+                size_t bytes_written = 0;
+                
+                ConverterFlags flags = ConverterFlags.NONE;
+                if (in_offset + bytes_read >= in_buffer.length) {
+                    flags |= ConverterFlags.INPUT_AT_END;
+                }
+                
+                ConverterResult result;
+                try {
+                    result = converter.convert(
+                        in_buffer[in_offset:in_buffer.length],
+                        out_buffer,
+                        flags,
+                        out bytes_read,
+                        out bytes_written
+                    );
+                } catch (IOError.NO_SPACE e) {
+                    // Need larger output buffer, try again with bigger buffer
+                    out_buffer = new uint8[BUFFER_SIZE * 2];
+                    result = converter.convert(
+                        in_buffer[in_offset:in_buffer.length],
+                        out_buffer,
+                        flags,
+                        out bytes_read,
+                        out bytes_written
+                    );
+                }
+                
+                in_offset += bytes_read;
+                
+                if (bytes_written > 0) {
+                    yield inner.write(new ByteBuffer.from_byte_array(out_buffer[0:bytes_written]));
+                }
+                
+                if (result == ConverterResult.FINISHED || result == ConverterResult.FLUSHED) {
+                    break;
+                }
+            }
+        }
+
+        public async void write_stream(GLib.InputStream stream) throws Error {
+            var converter_stream = new ConverterInputStream(stream, converter);
+            yield inner.write_stream(converter_stream);
+        }
+
+        
 
     }
 

src/Core/HttpResult.vala

@@ -22,6 +22,18 @@ namespace Astralis {
             return this;
         }
 
+        public HttpResult set_all_headers(Enumerable<KeyValuePair<string, string>> headers) {
+            foreach(var header in headers) {
+                this.headers[header.key] = header.value;
+            }
+            return this;
+        }
+
+        protected void replace_headers(Enumerable<KeyValuePair<string, string>> headers) {
+            this.headers.clear();
+            set_all_headers(headers);
+        }
+
         public abstract async void send_body(AsyncOutput output) throws Error;
     }
 
@@ -35,7 +47,7 @@ namespace Astralis {
             bytes = buffer;
         }
 
-        public async override void send_body(AsyncOutput output) {
+        public async override void send_body(AsyncOutput output) throws Error {
             yield output.write(bytes);
         }
     }
@@ -49,14 +61,14 @@ namespace Astralis {
 
     public class HttpStreamResult : HttpResult {
 
-        private OutputStream stream;
+        private InputStream stream;
 
-        public HttpStreamResult(OutputStream stream, uint64? content_length = null, StatusCode status = StatusCode.OK) {
+        public HttpStreamResult(InputStream stream, uint64? content_length = null, StatusCode status = StatusCode.OK) {
             this.stream = stream;
             base(status, content_length);
         }
 
-        public async override void send_body(AsyncOutput output) {
+        public async override void send_body(AsyncOutput output) throws Error {
             yield output.write_stream(stream);
         }
 

src/Server/ServerInput.vala

@@ -12,7 +12,7 @@ namespace Astralis {
             return chunks.first_or_default();
         }
 
-        public async BinaryData? read() {
+        public async BinaryData? read() throws Error {
             while (chunks.length == 0 && !writes_complete) {
                 Idle.add(read.callback);
                 yield;

src/Server/ServerOutput.vala

@@ -21,7 +21,7 @@ namespace Astralis {
             }
         }
 
-        public async void write_stream (GLib.OutputStream stream) {
+        public async void write_stream (GLib.InputStream stream) {
             assert_not_reached ();
         }
 

src/meson.build

@@ -6,7 +6,8 @@ sources = files(
     'Core/AsyncOutput.vala',
     'Core/Pipeline.vala',
     'Data/FormDataParser.vala',
-    'Handlers/EndpointRouter.vala',
+    'Components/EndpointRouter.vala',
+    'Components/Compression.vala',
     'Server/Server.vala',
     'Server/RequestContext.vala',
     'Server/ResponseContext.vala',