WIP: Reading brotli encoded HTTP messages. First-draft-ugly implementation, but its works

dbms/CMakeLists.txt

@@ -89,6 +89,9 @@ set(dbms_sources)
 
 include(../cmake/dbms_glob_sources.cmake)
 
+# temp ugly hack
+include_directories(${BROTLI_INCLUDE_DIR})
+
 add_headers_and_sources(clickhouse_common_io src/Common)
 add_headers_and_sources(clickhouse_common_io src/Common/HashTable)
 add_headers_and_sources(clickhouse_common_io src/IO)

dbms/programs/server/HTTPHandler.cpp

@@ -17,6 +17,7 @@
 #include <Common/setThreadName.h>
 #include <IO/ReadBufferFromIStream.h>
 #include <IO/ZlibInflatingReadBuffer.h>
+#include <IO/BrotliReadBuffer.h>
 #include <IO/ReadBufferFromString.h>
 #include <IO/ConcatReadBuffer.h>
 #include <Compression/CompressedReadBuffer.h>
@@ -396,19 +397,25 @@ void HTTPHandler::processQuery(
     String http_request_compression_method_str = request.get("Content-Encoding", "");
     if (!http_request_compression_method_str.empty())
     {
-        ZlibCompressionMethod method;
-        if (http_request_compression_method_str == "gzip")
-        {
-            method = ZlibCompressionMethod::Gzip;
-        }
-        else if (http_request_compression_method_str == "deflate")
-        {
-            method = ZlibCompressionMethod::Zlib;
-        }
-        else
+        do {
+            if (http_request_compression_method_str == "gzip")
+            {
+                in_post = std::make_unique<ZlibInflatingReadBuffer>(*in_post_raw, ZlibCompressionMethod::Gzip);
+                break;
+            }
+            if (http_request_compression_method_str == "deflate")
+            {
+                in_post = std::make_unique<ZlibInflatingReadBuffer>(*in_post_raw, ZlibCompressionMethod::Zlib);
+                break;
+            }
+            if (http_request_compression_method_str == "br")
+            {
+                in_post = std::make_unique<BrotliReadBuffer>(*in_post_raw);
+                break;
+            }
             throw Exception("Unknown Content-Encoding of HTTP request: " + http_request_compression_method_str,
-                ErrorCodes::UNKNOWN_COMPRESSION_METHOD);
-        in_post = std::make_unique<ZlibInflatingReadBuffer>(*in_post_raw, method);
+                            ErrorCodes::UNKNOWN_COMPRESSION_METHOD);
+        } while(0);
     }
     else
         in_post = std::move(in_post_raw);

dbms/src/IO/BrotliReadBuffer.cpp

@@ -0,0 +1,41 @@
+#include "BrotliReadBuffer.h"
+
+namespace DB
+{
+BrotliReadBuffer::BrotliReadBuffer(ReadBuffer &in_, size_t buf_size, char *existing_memory, size_t alignment)
+        : BufferWithOwnMemory<ReadBuffer>(buf_size, existing_memory, alignment)
+        , in(in_)
+        , eof(false)
+{
+    bstate_ = BrotliDecoderCreateInstance(NULL,NULL,NULL);
+}
+
+BrotliReadBuffer::~BrotliReadBuffer()
+{
+    BrotliDecoderDestroyInstance(bstate_);
+}
+
+bool BrotliReadBuffer::nextImpl()
+{
+    if (eof)
+        return false;
+
+    auto ptr_in = reinterpret_cast<const uint8_t *>(in.position());
+    size_t size_in = in.buffer().end() - in.position();
+
+    auto ptr_out = reinterpret_cast<uint8_t *>(internal_buffer.begin());
+    size_t size_out = internal_buffer.size();
+
+    BrotliDecoderDecompressStream(bstate_, &size_in, &ptr_in, &size_out, &ptr_out, nullptr);
+
+    in.position() = in.buffer().end() - size_in;
+    working_buffer.resize(internal_buffer.size() - size_out);
+
+    if (in.eof()) {
+        eof = true;
+        return working_buffer.size() != 0;
+    }
+
+    return true;
+}
+}

dbms/src/IO/BrotliReadBuffer.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <IO/ReadBuffer.h>
+#include <IO/BufferWithOwnMemory.h>
+
+#include <brotli/decode.h>
+
+namespace DB {
+
+class BrotliReadBuffer : public BufferWithOwnMemory<ReadBuffer> {
+public:
+    BrotliReadBuffer(
+            ReadBuffer &in_,
+            size_t buf_size = DBMS_DEFAULT_BUFFER_SIZE,
+            char *existing_memory = nullptr,
+            size_t alignment = 0);
+
+    ~BrotliReadBuffer() override;
+
+private:
+    bool nextImpl() override;
+
+    ReadBuffer &in;
+    BrotliDecoderState * bstate_;
+    bool eof;
+};
+}