code style cleanup

CMakeLists.txt

@@ -245,6 +245,7 @@ set(DROGON_SOURCES
     lib/src/HttpAppFrameworkImpl.cc
     lib/src/HttpBinder.cc
     lib/src/Http2Transport.cc
+    lib/src/Http1xTransport.cc
     lib/src/HttpClientImpl.cc
     lib/src/HttpControllersRouter.cc
     lib/src/HttpFileImpl.cc
@@ -291,6 +292,8 @@ set(private_headers
     lib/src/FiltersFunction.h
     lib/src/HttpAppFrameworkImpl.h
     lib/src/HttpClientImpl.h
+    lib/src/Http2Transport.h
+    lib/src/Http1xTransport.h
     lib/src/HttpControllersRouter.h
     lib/src/HttpFileImpl.h
     lib/src/HttpFileUploadRequest.h

examples/client_example/main.cc

@@ -16,10 +16,7 @@ int main()
 {
     trantor::Logger::setLogLevel(trantor::Logger::kTrace);
     {
-        auto client = HttpClient::newHttpClient("https://clehaxze.tw:8844",
-                                                nullptr,
-                                                false,
-                                                false);
+        auto client = HttpClient::newHttpClient("http://www.baidu.com");
         client->setSockOptCallback([](int fd) {
             std::cout << "setSockOptCallback:" << fd << std::endl;
 #ifdef __linux__
@@ -48,7 +45,7 @@ int main()
         req->setParameter("wd", "wx");
         req->setParameter("oq", "wx");
 
-        for (int i = 0; i < 1; ++i)
+        for (int i = 0; i < 10; ++i)
         {
             client->sendRequest(
                 req, [](ReqResult result, const HttpResponsePtr &response) {
@@ -74,8 +71,8 @@ int main()
                     }
                     std::cout << "count=" << nth_resp << std::endl;
                 });
-            LOG_INFO << "send request";
         }
     }
+
     app().run();
 }

lib/inc/drogon/HttpClient.h

@@ -184,6 +184,11 @@ class DROGON_EXPORT HttpClient : public trantor::NonCopyable
      * If this method is not called, the default depth value is 0 which means
      * the pipelining is disabled. For details about pipelining, see
      * rfc2616-8.1.2.2
+     *
+     * @param depth The depth value.
+     * @note This option is only valid for HTTP/1.x. If the client running in
+     * HTTP/2 mode, this settings have no effect. The maximum concurrent
+     * requests for HTTP/2 is 100 (unless the server has a different setting).
      */
     virtual void setPipeliningDepth(size_t depth) = 0;
 
@@ -238,15 +243,20 @@ class DROGON_EXPORT HttpClient : public trantor::NonCopyable
      * eanbled for HTTPS.
      * @param validateCert If the parameter is set to true, the client validates
      * the server certificate when SSL handshaking.
+     * @param targetVersion The target TLS version to use for HTTPS connections.
+     * This is a mere hint, and the actual version used will depend on the
+     * server and the client's capabilities.
      * @return HttpClientPtr The smart pointer to the new client object.
      * @note: The ip parameter support for both ipv4 and ipv6 address
      */
-    static HttpClientPtr newHttpClient(const std::string &ip,
-                                       uint16_t port,
-                                       bool useSSL = false,
-                                       trantor::EventLoop *loop = nullptr,
-                                       bool useOldTLS = false,
-                                       bool validateCert = true);
+    static HttpClientPtr newHttpClient(
+        const std::string &ip,
+        uint16_t port,
+        bool useSSL = false,
+        trantor::EventLoop *loop = nullptr,
+        bool useOldTLS = false,
+        bool validateCert = true,
+        std::optional<Version> targetVersion = std::nullopt);
 
     /// Get the event loop of the client;
     virtual trantor::EventLoop *getLoop() = 0;
@@ -303,6 +313,16 @@ class DROGON_EXPORT HttpClient : public trantor::NonCopyable
         const std::vector<std::pair<std::string, std::string>>
             &sslConfCmds) = 0;
 
+    /**
+     * @brief get the protocol version used by the HTTP connection
+     * @return std::optional<Version> the protocol version used by the HTTP
+     *
+     * NOTE: It could return std::nullopt if the connection is not established
+     * or is still negotiating the protocol. This is IMPORTANT as the client
+     * COULD be lazy and not connecting until the first request arrives.
+     */
+    virtual std::optional<Version> protocolVersion() const = 0;
+
     /// Create a Http client using the hostString to connect to server
     /**
      *
@@ -329,15 +349,25 @@ class DROGON_EXPORT HttpClient : public trantor::NonCopyable
      * @param validateCert If the parameter is set to true, the client validates
      * the server certificate when SSL handshaking.
      *
+     * @param targetVersion The target HTTP version that the client will attempt
+     * to use. **THIS IS ONLY A HINT**. The server may choose to ignore this as
+     * the server may not support the requested version. The preference is as
+     * follows:
+     *   HTTP/2 > HTTP/1.1 > HTTP/1.0
+     * Note that there is no way to auto-detect HTTP/1.0 servers, so it is only
+     * used if explicitly requested.
+     *
      * @note Don't add path and parameters in hostString, the request path and
      * parameters should be set in HttpRequestPtr when calling the sendRequest()
      * method.
      *
      */
-    static HttpClientPtr newHttpClient(const std::string &hostString,
-                                       trantor::EventLoop *loop = nullptr,
-                                       bool useOldTLS = false,
-                                       bool validateCert = true);
+    static HttpClientPtr newHttpClient(
+        const std::string &hostString,
+        trantor::EventLoop *loop = nullptr,
+        bool useOldTLS = false,
+        bool validateCert = true,
+        std::optional<Version> targetVersion = std::nullopt);
 
     virtual ~HttpClient()
     {

lib/src/Http1xTransport.cc

@@ -0,0 +1,104 @@
+#include "Http1xTransport.h"
+#include "HttpResponseParser.h"
+
+using namespace drogon;
+
+Http1xTransport::Http1xTransport(trantor::TcpConnectionPtr connPtr,
+                                 Version version,
+                                 size_t *bytesSent,
+                                 size_t *bytesReceived)
+    : connPtr(connPtr),
+      bytesSent_(bytesSent),
+      bytesReceived_(bytesReceived),
+      version_(version)
+{
+    connPtr->setContext(std::make_shared<HttpResponseParser>(connPtr));
+}
+
+void Http1xTransport::sendRequestInLoop(const HttpRequestPtr &req,
+                                        HttpReqCallback &&callback)
+{
+    sendReq(req);
+    pipeliningCallbacks_.emplace(std::move(req), std::move(callback));
+}
+
+void Http1xTransport::onRecvMessage(const trantor::TcpConnectionPtr &conn,
+                                    trantor::MsgBuffer *msg)
+{
+    auto responseParser = connPtr->getContext<HttpResponseParser>();
+    assert(responseParser != nullptr);
+    assert(connPtr.get() == conn.get());
+
+    // LOG_TRACE << "###:" << msg->readableBytes();
+    auto msgSize = msg->readableBytes();
+    while (msg->readableBytes() > 0)
+    {
+        if (pipeliningCallbacks_.empty())
+        {
+            LOG_ERROR << "More responses than expected!";
+            connPtr->shutdown();
+            return;
+        }
+        auto &firstReq = pipeliningCallbacks_.front();
+        if (firstReq.first->method() == Head)
+        {
+            responseParser->setForHeadMethod();
+        }
+        if (!responseParser->parseResponse(msg))
+        {
+            *bytesReceived_ += (msgSize - msg->readableBytes());
+            errorCallback(ReqResult::BadResponse);
+            return;
+        }
+        if (responseParser->gotAll())
+        {
+            auto resp = responseParser->responseImpl();
+            responseParser->reset();
+            *bytesReceived_ += (msgSize - msg->readableBytes());
+            msgSize = msg->readableBytes();
+            respCallback(resp, std::move(firstReq), conn);
+
+            pipeliningCallbacks_.pop();
+        }
+        else
+        {
+            break;
+        }
+    }
+}
+
+Http1xTransport::~Http1xTransport()
+{
+}
+
+bool Http1xTransport::handleConnectionClose()
+{
+    auto responseParser = connPtr->getContext<HttpResponseParser>();
+    if (responseParser && responseParser->parseResponseOnClose() &&
+        responseParser->gotAll())
+    {
+        auto &firstReq = pipeliningCallbacks_.front();
+        if (firstReq.first->method() == Head)
+        {
+            responseParser->setForHeadMethod();
+        }
+        auto resp = responseParser->responseImpl();
+        responseParser->reset();
+        respCallback(resp, std::move(firstReq), connPtr);
+        return false;
+    }
+    return true;
+}
+
+void Http1xTransport::sendReq(const HttpRequestPtr &req)
+{
+    trantor::MsgBuffer buffer;
+    assert(req);
+    auto implPtr = static_cast<HttpRequestImpl *>(req.get());
+    assert(version_ == Version::kHttp10 || version_ == Version::kHttp11);
+    implPtr->appendToBuffer(&buffer, version_);
+    LOG_TRACE << "Send request:"
+              << std::string_view(buffer.peek(), buffer.readableBytes());
+    *bytesSent_ += buffer.readableBytes();
+    connPtr->send(std::move(buffer));
+}

lib/src/Http1xTransport.h

@@ -0,0 +1,52 @@
+#pragma once
+
+#include <trantor/net/EventLoop.h>
+#include <trantor/net/TcpClient.h>
+#include <list>
+#include <queue>
+#include <vector>
+#include "HttpTransport.h"
+
+namespace drogon
+{
+class Http1xTransport : public HttpTransport
+{
+  private:
+    std::queue<std::pair<HttpRequestPtr, HttpReqCallback>> pipeliningCallbacks_;
+    trantor::TcpConnectionPtr connPtr;
+    size_t *bytesSent_;
+    size_t *bytesReceived_;
+    Version version_{Version::kHttp11};
+
+    void sendReq(const HttpRequestPtr &req);
+
+  public:
+    Http1xTransport(trantor::TcpConnectionPtr connPtr,
+                    Version version,
+                    size_t *bytesSent,
+                    size_t *bytesReceived);
+    virtual ~Http1xTransport();
+    void sendRequestInLoop(const HttpRequestPtr &req,
+                           HttpReqCallback &&callback) override;
+    void onRecvMessage(const trantor::TcpConnectionPtr &,
+                       trantor::MsgBuffer *) override;
+
+    size_t requestsInFlight() const override
+    {
+        return pipeliningCallbacks_.size();
+    }
+
+    bool handleConnectionClose() override;
+
+    void onError(ReqResult result) override
+    {
+        while (!pipeliningCallbacks_.empty())
+        {
+            auto &cb = pipeliningCallbacks_.front().second;
+            cb(result, nullptr);
+            pipeliningCallbacks_.pop();
+        }
+    }
+};
+
+}  // namespace drogon

lib/src/Http2Transport.cc

@@ -1,5 +1,7 @@
 #include "Http2Transport.h"
+#include "HttpFileUploadRequest.h"
 
+#include <fstream>
 #include <variant>
 
 using namespace drogon;
@@ -142,7 +144,7 @@ std::optional<WindowUpdateFrame> WindowUpdateFrame::parse(ByteStream &payload,
     }
     WindowUpdateFrame frame;
     // MSB is reserved for future use
-    auto [_, windowSizeIncrement] = payload.readBI32BE();
+    auto [_, windowSizeIncrement] = payload.readBI31BE();
     frame.windowSizeIncrement = windowSizeIncrement;
     return frame;
 }
@@ -188,7 +190,7 @@ std::optional<HeadersFrame> HeadersFrame::parse(ByteStream &payload,
 
     if (priority)
     {
-        auto [exclusive, streamDependency] = payload.readBI32BE();
+        auto [exclusive, streamDependency] = payload.readBI31BE();
         frame.exclusive = exclusive;
         frame.streamDependency = streamDependency;
         frame.weight = payload.readU8();
@@ -275,7 +277,7 @@ std::optional<DataFrame> DataFrame::parse(ByteStream &payload, uint8_t flags)
     }
 
     size_t minSize = frame.padLength;
-    if (payload.size() < minSize)
+    if (payload.remaining() < minSize)
     {
         LOG_TRACE << "Invalid data frame length";
         return std::nullopt;
@@ -385,7 +387,7 @@ std::optional<PushPromiseFrame> PushPromiseFrame::parse(ByteStream &payload,
         return std::nullopt;
     }
 
-    auto [_, promisedStreamId] = payload.readBI32BE();
+    auto [_, promisedStreamId] = payload.readBI31BE();
     frame.promisedStreamId = promisedStreamId;
     assert(payload.remaining() >= frame.padLength);
     frame.headerBlockFragment.resize(payload.remaining() - frame.padLength);
@@ -559,12 +561,11 @@ static void serializeFrame(OByteStream &buffer,
     buffer.overwriteU8(baseOffset + 4, flags);
 }
 
-// return streamId, frame, error and should continue parsing
+// return data {frame, streamId, h2flags, fatal-error}
 // Note that error can orrcur on a stream level or the entire connection
 // We need to handle both cases. Also it could happen that the TCP stream
 // just cuts off in the middle of a frame (or header). We need to handle that
 // too.
-// return data {frame, streamId, h2flags, fatal-error}
 static std::tuple<std::optional<H2Frame>, uint32_t, uint8_t, bool> parseH2Frame(
     trantor::MsgBuffer *msg)
 {
@@ -652,7 +653,15 @@ void Http2Transport::sendRequestInLoop(const HttpRequestPtr &req,
         return;
     }
 
-    const int32_t streamId = nextStreamId();
+    const auto sid = nextStreamId();
+    if (!sid.has_value())
+    {
+        // Upper HTTP client should see this and retry
+        // TODO: Need more elegant handling
+        connPtr->forceClose();
+        return;
+    }
+    const auto streamId = *sid;
     assert(streamId % 2 == 1);
     LOG_TRACE << "Sending HTTP/2 request: streamId=" << streamId;
     if (streams.find(streamId) != streams.end())
@@ -701,7 +710,10 @@ void Http2Transport::sendRequestInLoop(const HttpRequestPtr &req,
     encoded.resize(n);
     LOG_TRACE << "Encoded headers size: " << encoded.size();
 
-    bool haveBody = req->body().length() > 0;
+    bool haveBody =
+        req->body().length() > 0 ||
+        (req->contentType() != CT_MULTIPART_FORM_DATA &&
+         dynamic_cast<HttpFileUploadRequest *>(req.get()) != nullptr);
     auto &stream = createStream(streamId);
     stream.callback = std::move(callback);
     stream.request = req;
@@ -734,9 +746,8 @@ void Http2Transport::sendRequestInLoop(const HttpRequestPtr &req,
     if (!haveBody)
         return;
 
-    size_t sentOffset = sendBodyForStream(stream, 0);
-    assert(sentOffset <= req->body().length());
-    if (sentOffset != req->body().length())
+    auto [sentOffset, done] = sendBodyForStream(stream, 0);
+    if (!done)
     {
         auto it = pendingDataSend.find(streamId);
         if (it != pendingDataSend.end())
@@ -868,8 +879,8 @@ void Http2Transport::onRecvMessage(const trantor::TcpConnectionPtr &,
             do
             {
                 auto &stream = streams[it->first];
-                auto sentOffset = sendBodyForStream(stream, it->second);
-                if (sentOffset == stream.request->body().length())
+                auto [sentOffset, done] = sendBodyForStream(stream, it->second);
+                if (done)
                 {
                     pendingDataSend.erase(it);
                     it = pendingDataSend.begin();
@@ -976,6 +987,18 @@ void Http2Transport::onRecvMessage(const trantor::TcpConnectionPtr &,
             // Do nothing. RFC says to ignore unknown frames
         }
     }
+
+    if (!runningOutStreamId() || reconnectionIssued)
+        return;
+
+    // Only attempt to reconnect if there is no request in progress
+    if (requestsInFlight() == 0 && bufferedRequests.empty())
+    {
+        reconnectionIssued = true;
+        LOG_TRACE << "Reconnect due to running out of stream ids.";
+        connPtr->getLoop()->queueInLoop(
+            [connPtr = this->connPtr]() { connPtr->shutdown(); });
+    }
 }
 
 bool Http2Transport::parseAndApplyHeaders(internal::H2Stream &stream,
@@ -1212,9 +1235,8 @@ void Http2Transport::handleFrameForStream(const internal::H2Frame &frame,
         if (it == pendingDataSend.end())
             return;
 
-        auto sentOffset = sendBodyForStream(stream, it->second);
-        assert(sentOffset <= stream.request->body().length());
-        if (sentOffset == stream.request->body().length())
+        auto [sentOffset, done] = sendBodyForStream(stream, it->second);
+        if (done)
             pendingDataSend.erase(it);
         else
             it->second = sentOffset;
@@ -1254,6 +1276,9 @@ void Http2Transport::responseSuccess(internal::H2Stream &stream)
     auto it = streams.find(stream.streamId);
     assert(it != streams.end());
     auto streamId = stream.streamId;
+    // It is technically possible for a server to response before we fully
+    // send the body. So cehck and remove.
+    pendingDataSend.erase(streamId);
     // XXX: This callback seems to be able to cause the destruction of this
     // object
     respCallback(stream.response,
@@ -1261,12 +1286,9 @@ void Http2Transport::responseSuccess(internal::H2Stream &stream)
                  connPtr);
     streams.erase(it);  // NOTE: stream is now invalid
 
-    // It is technically possible for a server to response before we fully
-    // send the body. So cehck and remove.
-    pendingDataSend.erase(streamId);
-
     if (bufferedRequests.empty())
         return;
+    assert(streams.size() < maxConcurrentStreams);
     auto &[req, cb] = bufferedRequests.front();
     sendRequestInLoop(req, std::move(cb));
     bufferedRequests.pop();
@@ -1329,29 +1351,27 @@ bool Http2Transport::handleConnectionClose()
     return true;
 }
 
-size_t Http2Transport::sendBodyForStream(internal::H2Stream &stream,
-                                         size_t offset)
+std::pair<size_t, bool> Http2Transport::sendBodyForStream(
+    internal::H2Stream &stream,
+    const void *data,
+    size_t size)
 {
     auto streamId = stream.streamId;
     if (stream.avaliableTxWindow == 0 || avaliableTxWindow == 0)
-        return offset;
+        return {0, false};
 
-    assert(stream.request != nullptr);
-    assert(stream.request->body().length() >= offset);
-    size_t maxSendSize = stream.request->body().length() - offset;
+    size_t maxSendSize = size;
     maxSendSize = (std::min)(maxSendSize, stream.avaliableTxWindow);
     maxSendSize = (std::min)(maxSendSize, avaliableTxWindow);
-    size_t sendEndPos = offset + maxSendSize;
-    bool sendEverything =
-        maxSendSize == stream.request->body().length() - offset;
+    bool sendEverything = maxSendSize == size;
 
     size_t i = 0;
-    for (i = offset; i < sendEndPos; i += maxFrameSize)
+    for (i = 0; i < size; i += maxFrameSize)
     {
-        size_t remaining = stream.request->body().length() - i;
+        size_t remaining = size - i;
         size_t readSize = (std::min)(maxFrameSize, remaining);
-        bool endStream = sendEverything && i + maxFrameSize >= sendEndPos;
-        std::string_view sendData(stream.request->body().data() + i, readSize);
+        bool endStream = sendEverything && i + maxFrameSize >= size;
+        const std::string_view sendData((const char *)data + i, readSize);
         DataFrame dataFrame(sendData, endStream);
         LOG_TRACE << "Sending data frame: size=" << readSize
                   << " endStream=" << dataFrame.endStream;
@@ -1360,8 +1380,45 @@ size_t Http2Transport::sendBodyForStream(internal::H2Stream &stream,
         stream.avaliableTxWindow -= readSize;
         avaliableTxWindow -= readSize;
     }
-    assert(i >= offset && i <= sendEndPos);
-    return i;
+    i = (std::min)(i, size);
+    return {i, sendEverything};
+}
+
+std::pair<size_t, bool> Http2Transport::sendBodyForStream(
+    internal::H2Stream &stream,
+    size_t offset)
+{
+    auto streamId = stream.streamId;
+    if (stream.avaliableTxWindow == 0 || avaliableTxWindow == 0)
+        return {offset, false};
+
+    // Special handling for multipart because different underlying code
+    auto mPtr = dynamic_cast<HttpFileUploadRequest *>(stream.request.get());
+    if (mPtr)
+    {
+        // TODO: Don't put everything in memory. This causes a lot of memory
+        // when uploading large files. Howerver, we are not doing better in 1.x
+        // client either. So, meh.
+        if (stream.multipartData.readableBytes() == 0)
+            mPtr->renderMultipartFormData(stream.multipartData);
+        auto &content = stream.multipartData;
+        // CANNOT be empty. At least we are sending the boundary
+        assert(content.readableBytes() > 0);
+        auto [amount, done] =
+            sendBodyForStream(stream, content.peek(), content.readableBytes());
+
+        if (done)
+            // force release memory
+            stream.multipartData = trantor::MsgBuffer();
+        return {offset + amount, done};
+    }
+
+    auto [amount, done] =
+        sendBodyForStream(stream,
+                          stream.request->body().data() + offset,
+                          stream.request->body().length() - offset);
+    assert(offset + amount <= stream.request->body().length());
+    return {offset + amount, done};
 }
 
 void Http2Transport::sendFrame(const internal::H2Frame &frame, int32_t streamId)

lib/src/Http2Transport.h

@@ -6,6 +6,7 @@
 #include "hpack/HPacker.h"
 
 #include <variant>
+#include <climits>
 
 namespace drogon
 {
@@ -45,7 +46,7 @@ struct ByteStream
         return res;
     }
 
-    std::pair<bool, int32_t> readBI32BE()
+    std::pair<bool, int32_t> readBI31BE()
     {
         assert(length >= 4 && offset <= length - 4);
         int32_t res = ptr[offset] << 24 | ptr[offset + 1] << 16 |
@@ -426,6 +427,8 @@ struct H2Stream
     size_t avaliableTxWindow = 65535;
     size_t avaliableRxWindow = 65535;
     StreamState state = StreamState::ExpectingHeaders;
+
+    trantor::MsgBuffer multipartData;
 };
 }  // namespace internal
 
@@ -465,7 +468,7 @@ class Http2Transport : public HttpTransport
     int32_t expectngContinuationStreamId = 0;
 
     std::unordered_map<int32_t, size_t> pendingDataSend;
-    // TODO: Handle server-initiated stream creation
+    bool reconnectionIssued = false;
 
     // HTTP/2 client-wide settings (can be changed by server)
     size_t maxConcurrentStreams = 100;
@@ -475,8 +478,9 @@ class Http2Transport : public HttpTransport
 
     // Configuration settings
     const uint32_t windowIncreaseThreshold = 16384;
-    const uint32_t windowIncreaseSize = 10 * 1024 * 1024;  // 10 MiB
+    const uint32_t windowIncreaseSize = 128 * 1024;  // 128KB
     const uint32_t maxCompressiedHeaderSize = 2048;
+    const int32_t streamIdReconnectThreshold = INT_MAX - 8192;
 
     // HTTP/2 connection-wide state
     size_t avaliableTxWindow = 65535;
@@ -486,15 +490,25 @@ class Http2Transport : public HttpTransport
     void responseSuccess(internal::H2Stream &stream);
     void responseErrored(int32_t streamId, ReqResult result);
 
-    int32_t nextStreamId()
+    std::optional<int32_t> nextStreamId()
     {
-        // TODO: Handling stream ID requires to reconnect
-        // the entire connection. Handle this somehow
+        // XXX: Technically UB. But no one actually uses 1's complement
+        if (currentStreamId < 0)
+            return std::nullopt;
+
         int32_t streamId = currentStreamId;
         currentStreamId += 2;
         return streamId;
     }
 
+    // Returns true when we SHOULD reconnect due to exhausting stream IDs.
+    // Doesn't mean we will. We will force a reconnect when we actually
+    // run out.
+    inline bool runningOutStreamId()
+    {
+        return currentStreamId > streamIdReconnectThreshold;
+    }
+
     void handleFrameForStream(const internal::H2Frame &frame,
                               int32_t streamId,
                               uint8_t flags);
@@ -505,7 +519,11 @@ class Http2Transport : public HttpTransport
     bool parseAndApplyHeaders(internal::H2Stream &stream,
                               const void *data,
                               size_t len);
-    size_t sendBodyForStream(internal::H2Stream &stream, size_t offset);
+    std::pair<size_t, bool> sendBodyForStream(internal::H2Stream &stream,
+                                              const void *data,
+                                              size_t size);
+    std::pair<size_t, bool> sendBodyForStream(internal::H2Stream &stream,
+                                              size_t offset);
     void sendFrame(const internal::H2Frame &frame, int32_t streamId);
 
     void sendBufferedData();
@@ -522,7 +540,7 @@ class Http2Transport : public HttpTransport
 
     size_t requestsInFlight() const override
     {
-        return 0;
+        return streams.size();
     }
 
     bool handleConnectionClose() override;

lib/src/HttpClientImpl.cc

@@ -31,94 +31,12 @@ namespace trantor
 const static size_t kDefaultDNSTimeout{600};
 }
 
-Http1xTransport::Http1xTransport(trantor::TcpConnectionPtr connPtr,
-                                 size_t *bytesSent,
-                                 size_t *bytesReceived)
-    : connPtr(connPtr), bytesSent_(bytesSent), bytesReceived_(bytesReceived)
-{
-    connPtr->setContext(std::make_shared<HttpResponseParser>(connPtr));
-}
-
-void Http1xTransport::sendRequestInLoop(const HttpRequestPtr &req,
-                                        HttpReqCallback &&callback)
-{
-    sendReq(req);
-    pipeliningCallbacks_.emplace(std::move(req), std::move(callback));
-}
-
-void Http1xTransport::onRecvMessage(const trantor::TcpConnectionPtr &conn,
-                                    trantor::MsgBuffer *msg)
-{
-    auto responseParser = connPtr->getContext<HttpResponseParser>();
-    assert(responseParser != nullptr);
-    assert(connPtr.get() == conn.get());
-
-    // LOG_TRACE << "###:" << msg->readableBytes();
-    auto msgSize = msg->readableBytes();
-    while (msg->readableBytes() > 0)
-    {
-        if (pipeliningCallbacks_.empty())
-        {
-            LOG_ERROR << "More responses than expected!";
-            connPtr->shutdown();
-            return;
-        }
-        auto &firstReq = pipeliningCallbacks_.front();
-        if (firstReq.first->method() == Head)
-        {
-            responseParser->setForHeadMethod();
-        }
-        if (!responseParser->parseResponse(msg))
-        {
-            *bytesReceived_ += (msgSize - msg->readableBytes());
-            errorCallback(ReqResult::BadResponse);
-            return;
-        }
-        if (responseParser->gotAll())
-        {
-            auto resp = responseParser->responseImpl();
-            responseParser->reset();
-            *bytesReceived_ += (msgSize - msg->readableBytes());
-            msgSize = msg->readableBytes();
-            respCallback(resp, std::move(firstReq), conn);
-
-            pipeliningCallbacks_.pop();
-        }
-        else
-        {
-            break;
-        }
-    }
-}
-
-Http1xTransport::~Http1xTransport()
-{
-}
-
-bool Http1xTransport::handleConnectionClose()
-{
-    auto responseParser = connPtr->getContext<HttpResponseParser>();
-    if (responseParser && responseParser->parseResponseOnClose() &&
-        responseParser->gotAll())
-    {
-        auto &firstReq = pipeliningCallbacks_.front();
-        if (firstReq.first->method() == Head)
-        {
-            responseParser->setForHeadMethod();
-        }
-        auto resp = responseParser->responseImpl();
-        responseParser->reset();
-        respCallback(resp, std::move(firstReq), connPtr);
-        return false;
-    }
-    return true;
-}
-
 void HttpClientImpl::createTcpClient()
 {
     LOG_TRACE << "New TcpClient," << serverAddr_.toIpPort();
     tcpClientPtr_ =
         std::make_shared<trantor::TcpClient>(loop_, serverAddr_, "httpClient");
+    Version version = targetHttpVersion_.value_or(Version::kHttp2);
 
     if (useSSL_ && utils::supportsTls())
     {
@@ -130,8 +48,9 @@ void HttpClientImpl::createTcpClient()
             .setHostname(domain_)
             .setConfCmds(sslConfCmds_)
             .setCertPath(clientCertPath_)
-            .setKeyPath(clientKeyPath_)
-            .setAlpnProtocols({"h2", "http/1.1"});
+            .setKeyPath(clientKeyPath_);
+        if (version == Version::kHttp2)
+            policy->setAlpnProtocols({"h2", "http/1.1"});
         tcpClientPtr_->enableSSL(std::move(policy));
     }
 
@@ -156,13 +75,15 @@ void HttpClientImpl::createTcpClient()
             LOG_TRACE << "Connection established!";
 
             auto protocol = connPtr->applicationProtocol();
-            if (protocol.empty() || protocol == "http/1.1")
+            if (protocol == "http/1.1")
             {
                 LOG_TRACE << "Select http/1.1 protocol";
                 thisPtr->transport_ =
                     std::make_unique<Http1xTransport>(connPtr,
+                                                      Version::kHttp11,
                                                       &thisPtr->bytesSent_,
                                                       &thisPtr->bytesReceived_);
+                thisPtr->httpVersion_ = Version::kHttp11;
             }
             else if (protocol == "h2")
             {
@@ -171,6 +92,21 @@ void HttpClientImpl::createTcpClient()
                     std::make_unique<Http2Transport>(connPtr,
                                                      &thisPtr->bytesSent_,
                                                      &thisPtr->bytesReceived_);
+                thisPtr->httpVersion_ = Version::kHttp2;
+            }
+            else if (protocol.empty())
+            {
+                // Either we are not using TLS or the server does not support
+                // ALPN. Use HTTP/1.1 if not specified otherwise.
+                bool force1_0 = thisPtr->targetHttpVersion_.value_or(
+                                    Version::kUnknown) == Version::kHttp10;
+                auto version = force1_0 ? Version::kHttp10 : Version::kHttp11;
+                thisPtr->httpVersion_ = version;
+                thisPtr->transport_ =
+                    std::make_unique<Http1xTransport>(connPtr,
+                                                      version,
+                                                      &thisPtr->bytesSent_,
+                                                      &thisPtr->bytesReceived_);
             }
             else
             {
@@ -179,6 +115,9 @@ void HttpClientImpl::createTcpClient()
                 thisPtr->onError(ReqResult::BadResponse);
                 return;
             }
+
+            assert(thisPtr->httpVersion_.has_value());
+            assert(thisPtr->transport_);
             thisPtr->transport_->setRespCallback(
                 [weakPtr](const HttpResponseImplPtr &resp,
                           std::pair<HttpRequestPtr, HttpReqCallback> &&reqAndCb,
@@ -195,8 +134,13 @@ void HttpClientImpl::createTcpClient()
                 thisPtr->onError(result);
             });
 
-            // TODO: respect timeout and pipeliningDepth_
-            while (!thisPtr->requestsBuffer_.empty())
+            size_t maxSendReq = (*(thisPtr->httpVersion_) == Version::kHttp2)
+                                    ? size_t{0xfffffff}
+                                    : thisPtr->pipeliningDepth_;
+            if (maxSendReq == 0)
+                maxSendReq = 1;
+            while (!thisPtr->requestsBuffer_.empty() &&
+                   thisPtr->transport_->requestsInFlight() < maxSendReq)
             {
                 auto &reqAndCb = thisPtr->requestsBuffer_.front();
                 thisPtr->transport_->sendRequestInLoop(reqAndCb.first,
@@ -222,6 +166,8 @@ void HttpClientImpl::createTcpClient()
 
             // temporary fix of dead tcpClientPtr_
             // TODO: fix HttpResponseParser when content-length absence
+            // TODO: HTTP/2 transport also relies on this behavior to reconnect
+            //   when running out of stream IDs
             thisPtr->tcpClientPtr_.reset();
             if (!thisPtr->requestsBuffer_.empty())
             {
@@ -268,20 +214,26 @@ HttpClientImpl::HttpClientImpl(trantor::EventLoop *loop,
                                const trantor::InetAddress &addr,
                                bool useSSL,
                                bool useOldTLS,
-                               bool validateCert)
+                               bool validateCert,
+                               std::optional<Version> targetVersion)
     : loop_(loop),
       serverAddr_(addr),
       useSSL_(useSSL),
       validateCert_(validateCert),
-      useOldTLS_(useOldTLS)
+      useOldTLS_(useOldTLS),
+      targetHttpVersion_(targetVersion)
 {
 }
 
 HttpClientImpl::HttpClientImpl(trantor::EventLoop *loop,
                                const std::string &hostString,
                                bool useOldTLS,
-                               bool validateCert)
-    : loop_(loop), validateCert_(validateCert), useOldTLS_(useOldTLS)
+                               bool validateCert,
+                               std::optional<Version> targetVersion)
+    : loop_(loop),
+      validateCert_(validateCert),
+      useOldTLS_(useOldTLS),
+      targetHttpVersion_(targetVersion)
 {
     auto lowerHost = hostString;
     std::transform(lowerHost.begin(),
@@ -625,10 +577,15 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,
         return;
     }
     assert(transport_ != nullptr);
+    assert(httpVersion_.has_value());
+
+    size_t maxSendReq = (*httpVersion_ == Version::kHttp2) ? size_t{0xfffffffff}
+                                                           : pipeliningDepth_;
+    if (maxSendReq == 0)
+        maxSendReq = 1;
 
     // Connected, send request now
-    if (transport_->requestsInFlight() <= pipeliningDepth_ &&
-        requestsBuffer_.empty())
+    if (transport_->requestsInFlight() <= maxSendReq && requestsBuffer_.empty())
     {
         transport_->sendRequestInLoop(req, std::move(callback));
     }
@@ -644,18 +601,6 @@ void HttpClientImpl::sendRequestInLoop(const drogon::HttpRequestPtr &req,
     }
 }
 
-void Http1xTransport::sendReq(const HttpRequestPtr &req)
-{
-    trantor::MsgBuffer buffer;
-    assert(req);
-    auto implPtr = static_cast<HttpRequestImpl *>(req.get());
-    implPtr->appendToBuffer(&buffer);
-    LOG_TRACE << "Send request:"
-              << std::string(buffer.peek(), buffer.readableBytes());
-    *bytesSent_ += buffer.readableBytes();
-    connPtr->send(std::move(buffer));
-}
-
 void HttpClientImpl::handleResponse(
     const HttpResponseImplPtr &resp,
     std::pair<HttpRequestPtr, HttpReqCallback> &&reqAndCb,
@@ -722,7 +667,8 @@ HttpClientPtr HttpClient::newHttpClient(const std::string &ip,
                                         bool useSSL,
                                         trantor::EventLoop *loop,
                                         bool useOldTLS,
-                                        bool validateCert)
+                                        bool validateCert,
+                                        std::optional<Version> targetVersion)
 {
     bool isIpv6 = ip.find(':') == std::string::npos ? false : true;
     return std::make_shared<HttpClientImpl>(
@@ -730,19 +676,22 @@ HttpClientPtr HttpClient::newHttpClient(const std::string &ip,
         trantor::InetAddress(ip, port, isIpv6),
         useSSL,
         useOldTLS,
-        validateCert);
+        validateCert,
+        targetVersion);
 }
 
 HttpClientPtr HttpClient::newHttpClient(const std::string &hostString,
                                         trantor::EventLoop *loop,
                                         bool useOldTLS,
-                                        bool validateCert)
+                                        bool validateCert,
+                                        std::optional<Version> targetVersion)
 {
     return std::make_shared<HttpClientImpl>(
         loop == nullptr ? HttpAppFrameworkImpl::instance().getLoop() : loop,
         hostString,
         useOldTLS,
-        validateCert);
+        validateCert,
+        targetVersion);
 }
 
 void HttpClientImpl::onError(ReqResult result)

lib/src/HttpClientImpl.h

@@ -26,48 +26,11 @@
 #include "impl_forwards.h"
 #include "Http2Transport.h"
 #include "HttpTransport.h"
+#include "Http1xTransport.h"
 
 namespace drogon
 {
 
-class Http1xTransport : public HttpTransport
-{
-  private:
-    std::queue<std::pair<HttpRequestPtr, HttpReqCallback>> pipeliningCallbacks_;
-    trantor::TcpConnectionPtr connPtr;
-    size_t *bytesSent_;
-    size_t *bytesReceived_;
-
-    void sendReq(const HttpRequestPtr &req);
-
-  public:
-    Http1xTransport(trantor::TcpConnectionPtr connPtr,
-                    size_t *bytesSent,
-                    size_t *bytesReceived);
-    virtual ~Http1xTransport();
-    void sendRequestInLoop(const HttpRequestPtr &req,
-                           HttpReqCallback &&callback) override;
-    void onRecvMessage(const trantor::TcpConnectionPtr &,
-                       trantor::MsgBuffer *) override;
-
-    size_t requestsInFlight() const override
-    {
-        return pipeliningCallbacks_.size();
-    }
-
-    bool handleConnectionClose() override;
-
-    void onError(ReqResult result) override
-    {
-        while (!pipeliningCallbacks_.empty())
-        {
-            auto &cb = pipeliningCallbacks_.front().second;
-            cb(result, nullptr);
-            pipeliningCallbacks_.pop();
-        }
-    }
-};
-
 class HttpClientImpl final : public HttpClient,
                              public std::enable_shared_from_this<HttpClientImpl>
 {
@@ -76,11 +39,13 @@ class HttpClientImpl final : public HttpClient,
                    const trantor::InetAddress &addr,
                    bool useSSL = false,
                    bool useOldTLS = false,
-                   bool validateCert = true);
+                   bool validateCert = true,
+                   std::optional<Version> httpVersion = std::nullopt);
     HttpClientImpl(trantor::EventLoop *loop,
                    const std::string &hostString,
                    bool useOldTLS = false,
-                   bool validateCert = true);
+                   bool validateCert = true,
+                   std::optional<Version> httpVersion = std::nullopt);
     void sendRequest(const HttpRequestPtr &req,
                      const HttpReqCallback &callback,
                      double timeout = 0) override;
@@ -156,6 +121,11 @@ class HttpClientImpl final : public HttpClient,
         sockOptCallback_ = std::move(cb);
     }
 
+    std::optional<Version> protocolVersion() const override
+    {
+        return httpVersion_;
+    }
+
   private:
     std::shared_ptr<trantor::TcpClient> tcpClientPtr_;
     trantor::EventLoop *loop_;
@@ -191,6 +161,8 @@ class HttpClientImpl final : public HttpClient,
     std::string clientKeyPath_;
     std::function<void(int)> sockOptCallback_;
     std::unique_ptr<HttpTransport> transport_;
+    std::optional<Version> targetHttpVersion_;
+    std::optional<Version> httpVersion_;
 };
 
 using HttpClientImplPtr = std::shared_ptr<HttpClientImpl>;

lib/src/HttpFileUploadRequest.cc

@@ -16,6 +16,8 @@
 #include <drogon/UploadFile.h>
 #include <drogon/utils/Utilities.h>
 
+#include <fstream>
+
 using namespace drogon;
 
 HttpFileUploadRequest::HttpFileUploadRequest(
@@ -25,7 +27,74 @@ HttpFileUploadRequest::HttpFileUploadRequest(
       files_(files)
 {
     setMethod(drogon::Post);
-    setVersion(drogon::Version::kHttp11);
     setContentType("multipart/form-data; boundary=" + boundary_);
     contentType_ = CT_MULTIPART_FORM_DATA;
 }
+
+template <typename T>
+void renderMultipart(const HttpFileUploadRequest *mPtr, T &content)
+{
+    auto &boundary = mPtr->boundary();
+    for (auto &param : mPtr->getParameters())
+    {
+        content.append("--");
+        content.append(boundary);
+        content.append("\r\n");
+        content.append("content-disposition: form-data; name=\"");
+        content.append(param.first);
+        content.append("\"\r\n\r\n");
+        content.append(param.second);
+        content.append("\r\n");
+    }
+    for (auto &file : mPtr->files())
+    {
+        content.append("--");
+        content.append(boundary);
+        content.append("\r\n");
+        content.append("content-disposition: form-data; name=\"");
+        content.append(file.itemName());
+        content.append("\"; filename=\"");
+        content.append(file.fileName());
+        content.append("\"");
+        if (file.contentType() != CT_NONE)
+        {
+            content.append("\r\n");
+
+            auto &type = contentTypeToMime(file.contentType());
+            content.append("content-type: ");
+            content.append(type.data(), type.length());
+        }
+        content.append("\r\n\r\n");
+        std::ifstream infile(utils::toNativePath(file.path()),
+                             std::ifstream::binary);
+        if (!infile)
+        {
+            LOG_ERROR << file.path() << " not found";
+        }
+        else
+        {
+            std::streambuf *pbuf = infile.rdbuf();
+            std::streamsize filesize = pbuf->pubseekoff(0, infile.end);
+            pbuf->pubseekoff(0, infile.beg);  // rewind
+            std::string str;
+            str.resize(filesize);
+            pbuf->sgetn(&str[0], filesize);
+            content.append(std::move(str));
+        }
+        content.append("\r\n");
+    }
+    content.append("--");
+    content.append(boundary);
+    content.append("--");
+}
+
+void HttpFileUploadRequest::renderMultipartFormData(
+    trantor::MsgBuffer &buffer) const
+{
+    renderMultipart(this, buffer);
+}
+
+void HttpFileUploadRequest::renderMultipartFormData(std::string &buffer) const
+{
+    renderMultipart(this, buffer);
+}

lib/src/HttpFileUploadRequest.h

@@ -34,6 +34,9 @@ class HttpFileUploadRequest : public HttpRequestImpl
 
     explicit HttpFileUploadRequest(const std::vector<UploadFile> &files);
 
+    void renderMultipartFormData(trantor::MsgBuffer &buffer) const;
+    void renderMultipartFormData(std::string &buffer) const;
+
   private:
     std::string boundary_;
     std::vector<UploadFile> files_;

lib/src/HttpRequestImpl.cc

@@ -182,7 +182,8 @@ void HttpRequestImpl::parseParameters() const
     }
 }
 
-void HttpRequestImpl::appendToBuffer(trantor::MsgBuffer *output) const
+void HttpRequestImpl::appendToBuffer(trantor::MsgBuffer *output,
+                                     Version protoVer) const
 {
     switch (method_)
     {
@@ -266,11 +267,11 @@ void HttpRequestImpl::appendToBuffer(trantor::MsgBuffer *output) const
     }
 
     output->append(" ");
-    if (version_ == Version::kHttp11)
+    if (protoVer == Version::kHttp11)
     {
         output->append("HTTP/1.1");
     }
-    else if (version_ == Version::kHttp10)
+    else if (protoVer == Version::kHttp10)
     {
         output->append("HTTP/1.0");
     }
@@ -285,57 +286,7 @@ void HttpRequestImpl::appendToBuffer(trantor::MsgBuffer *output) const
         auto mReq = dynamic_cast<const HttpFileUploadRequest *>(this);
         if (mReq)
         {
-            for (auto &param : mReq->getParameters())
-            {
-                content.append("--");
-                content.append(mReq->boundary());
-                content.append("\r\n");
-                content.append("content-disposition: form-data; name=\"");
-                content.append(param.first);
-                content.append("\"\r\n\r\n");
-                content.append(param.second);
-                content.append("\r\n");
-            }
-            for (auto &file : mReq->files())
-            {
-                content.append("--");
-                content.append(mReq->boundary());
-                content.append("\r\n");
-                content.append("content-disposition: form-data; name=\"");
-                content.append(file.itemName());
-                content.append("\"; filename=\"");
-                content.append(file.fileName());
-                content.append("\"");
-                if (file.contentType() != CT_NONE)
-                {
-                    content.append("\r\n");
-
-                    auto &type = contentTypeToMime(file.contentType());
-                    content.append("content-type: ");
-                    content.append(type.data(), type.length());
-                }
-                content.append("\r\n\r\n");
-                std::ifstream infile(utils::toNativePath(file.path()),
-                                     std::ifstream::binary);
-                if (!infile)
-                {
-                    LOG_ERROR << file.path() << " not found";
-                }
-                else
-                {
-                    std::streambuf *pbuf = infile.rdbuf();
-                    std::streamsize filesize = pbuf->pubseekoff(0, infile.end);
-                    pbuf->pubseekoff(0, infile.beg);  // rewind
-                    std::string str;
-                    str.resize(filesize);
-                    pbuf->sgetn(&str[0], filesize);
-                    content.append(std::move(str));
-                }
-                content.append("\r\n");
-            }
-            content.append("--");
-            content.append(mReq->boundary());
-            content.append("--");
+            mReq->renderMultipartFormData(content);
         }
     }
     assert(!(!content.empty() && !content_.empty()));
@@ -506,7 +457,6 @@ HttpRequestPtr HttpRequest::newHttpRequest()
 {
     auto req = std::make_shared<HttpRequestImpl>(nullptr);
     req->setMethod(drogon::Get);
-    req->setVersion(drogon::Version::kHttp11);
     return req;
 }
 
@@ -514,7 +464,6 @@ HttpRequestPtr HttpRequest::newHttpFormPostRequest()
 {
     auto req = std::make_shared<HttpRequestImpl>(nullptr);
     req->setMethod(drogon::Post);
-    req->setVersion(drogon::Version::kHttp11);
     req->contentType_ = CT_APPLICATION_X_FORM;
     req->flagForParsingContentType_ = true;
     return req;
@@ -540,7 +489,6 @@ HttpRequestPtr HttpRequest::newHttpJsonRequest(const Json::Value &data)
     });
     auto req = std::make_shared<HttpRequestImpl>(nullptr);
     req->setMethod(drogon::Get);
-    req->setVersion(drogon::Version::kHttp11);
     req->contentType_ = CT_APPLICATION_JSON;
     req->setContent(writeString(builder, data));
     req->flagForParsingContentType_ = true;
@@ -599,6 +547,10 @@ const char *HttpRequestImpl::versionString() const
             result = "HTTP/1.1";
             break;
 
+        case Version::kHttp2:
+            result = "HTTP/2";
+            break;
+
         default:
             break;
     }

lib/src/HttpRequestImpl.h

@@ -87,8 +87,6 @@ class HttpRequestImpl : public HttpRequest
         return loop_;
     }
 
-    // [[deprcated("Now version is controlled by the HttpClient. Calling
-    // setVersion() will have no effect.")]]
     void setVersion(Version v)
     {
         version_ = v;
@@ -418,7 +416,7 @@ class HttpRequestImpl : public HttpRequest
         return passThrough_;
     }
 
-    void appendToBuffer(trantor::MsgBuffer *output) const;
+    void appendToBuffer(trantor::MsgBuffer *output, Version protoVer) const;
 
     const SessionPtr &session() const override
     {

lib/src/WebSocketClientImpl.cc

@@ -424,7 +424,7 @@ void WebSocketClientImpl::sendReq(const trantor::TcpConnectionPtr &connPtr)
     trantor::MsgBuffer buffer;
     assert(upgradeRequest_);
     auto implPtr = static_cast<HttpRequestImpl *>(upgradeRequest_.get());
-    implPtr->appendToBuffer(&buffer);
+    implPtr->appendToBuffer(&buffer, Version::kHttp11);
     LOG_TRACE << "Send request:"
               << std::string(buffer.peek(), buffer.readableBytes());
     connPtr->send(std::move(buffer));

lib/tests/integration_test/client/main.cc

@@ -1142,6 +1142,10 @@ void doTest(const HttpClientPtr &client, std::shared_ptr<test::Case> TEST_CTX)
         {
             FAIL("Unexpected exception, what(): " + std::string(e.what()));
         }
+
+        // Had to be put here else there's a chance that the client hasn't
+        // negotiated the protocol yet
+        CHECK(client->protocolVersion() == Version::kHttp11);
     });
 #endif
 }

patch.patch

@@ -0,0 +1,15 @@
+diff --git a/lib/src/Http2Transport.cc b/lib/src/Http2Transport.cc
+index f3f153e..11492b3 100644
+--- a/lib/src/Http2Transport.cc
++++ b/lib/src/Http2Transport.cc
+@@ -996,9 +996,8 @@ void Http2Transport::onRecvMessage(const trantor::TcpConnectionPtr &,
+     {
+         reconnectionIssued = true;
+         LOG_TRACE << "Reconnect due to running out of stream ids.";
+-        connPtr->getLoop()->queueInLoop([connPtr = this->connPtr]() {
+-            connPtr->shutdown();
+-        });
++        connPtr->getLoop()->queueInLoop(
++            [connPtr = this->connPtr]() { connPtr->shutdown(); });
+     }
+ }