Modify client/client-handler to support HttpObjectAggregator (#393)

HttpObjectAggregator modifies the response to aggregate all HttpMessage(s) up to a maximum length (it throws a TooLongFrameException if the length exceed the limit).

When this handler is part of the pipeline, the channel receives an instance of FullHttpResponse with the whole aggregated response.

Author: rborer

src/aleph/http/client.clj

@@ -33,8 +33,12 @@
      Channel
      ChannelHandler ChannelHandlerContext
      ChannelPipeline]
-    [io.netty.handler.stream ChunkedWriteHandler]
-    [io.netty.handler.codec.http FullHttpRequest]
+    [io.netty.handler.codec
+     TooLongFrameException]
+    [io.netty.handler.stream 
+     ChunkedWriteHandler]
+    [io.netty.handler.codec.http 
+     FullHttpRequest]
     [io.netty.handler.codec.http.websocketx
      CloseWebSocketFrame
      PingWebSocketFrame
@@ -159,7 +163,13 @@
 
       :exception-caught
       ([_ ctx ex]
-        (when-not (instance? IOException ex)
+        (cond
+          ; could happens when io.netty.handler.codec.http.HttpObjectAggregator
+          ; is part of the pipeline
+          (instance? TooLongFrameException ex)
+          (s/put! response-stream ex)
+
+          (not (instance? IOException ex))
           (log/warn ex "error in HTTP client")))
 
       :channel-inactive
@@ -176,6 +186,18 @@
 
         (cond
 
+          ; happens when io.netty.handler.codec.http.HttpObjectAggregator is part of the pipeline
+          (instance? FullHttpResponse msg)
+          (let [^FullHttpResponse rsp msg
+                content (.content rsp)
+                c (d/deferred)
+                s (netty/buffered-source (netty/channel ctx) #(alength ^bytes %) buffer-capacity)]
+            (s/on-closed s #(d/success! c true))
+            (s/put! s (netty/buf->array content))
+            (netty/release content)
+            (handle-response rsp c s)
+            (s/close! s))
+
           (instance? HttpResponse msg)
           (let [rsp msg]
             (if (HttpUtil/isTransferEncodingChunked rsp)