Allow custom socket streams in library/cpp/http/server

Prepare to use external coroutine/fiber/etc pool with non-blocking read/writes instead of plain blocking system threads. We already can replace system thread pool with something else, allow to inject different streams around connection socket:

- move common part of http connection (http streams and output buffer) directly into it;
- replace connection Impl with socket streams provider;
- add TClientRequest::CreateHttpConnection factory;
- check that we are able to override socket streams in unittest.
commit_hash:afe39ce57ee1d10673f4c36a12e01b467d9f77b0

1 files changed, 36 insertions, 45 deletions

diff --git a/library/cpp/http/server/conn.cpp b/library/cpp/http/server/conn.cpp
index 38a76c4c309..4b72c3f77f0 100644
--- a/library/cpp/http/server/conn.cpp
+++ b/library/cpp/http/server/conn.cpp
@@ -1,47 +1,38 @@
 #include "conn.h"
 
 #include <util/network/socket.h>
-#include <util/stream/buffered.h>
 
-class THttpServerConn::TImpl {
-public:
-    inline TImpl(const TSocket& s, size_t outputBufferSize)
-        : S_(s)
-        , SI_(S_)
-        , SO_(S_)
-        , BO_(&SO_, outputBufferSize)
-        , HI_(&SI_)
-        , HO_(&BO_, &HI_)
-    {
-    }
-
-    inline ~TImpl() {
-    }
-
-    inline THttpInput* Input() noexcept {
-        return &HI_;
-    }
+namespace {
+    class TBlockingSocketStreams: public THttpServerConn::ISocketStreams {
+    public:
+        explicit TBlockingSocketStreams(const TSocket& s)
+            : Socket_(s)
+            , Input_(Socket_)
+            , Output_(Socket_)
+        {
+        }
 
-    inline THttpOutput* Output() noexcept {
-        return &HO_;
-    }
+        IInputStream* Input() override {
+            return &Input_;
+        }
 
-    inline void Reset() {
-        if (S_ != INVALID_SOCKET) {
-            // send RST packet to client
-            S_.SetLinger(true, 0);
-            S_.Close();
+        IOutputStream* Output() override {
+            return &Output_;
         }
-    }
 
-private:
-    TSocket S_;
-    TSocketInput SI_;
-    TSocketOutput SO_;
-    TBufferedOutput BO_;
-    THttpInput HI_;
-    THttpOutput HO_;
-};
+        void Reset() override {
+            if (Socket_ != INVALID_SOCKET) {
+                // send RST packet to client
+                Socket_.SetLinger(true, 0);
+                Socket_.Close();
+           }
+        }
+    private:
+        TSocket Socket_;
+        TSocketInput Input_;
+        TSocketOutput Output_;
+    };
+}
 
 THttpServerConn::THttpServerConn(const TSocket& s)
     : THttpServerConn(s, s.MaximumTransferUnit())
@@ -49,21 +40,21 @@ THttpServerConn::THttpServerConn(const TSocket& s)
 }
 
 THttpServerConn::THttpServerConn(const TSocket& s, size_t outputBufferSize)
-    : Impl_(new TImpl(s, outputBufferSize))
+    : THttpServerConn(MakeHolder<TBlockingSocketStreams>(s), outputBufferSize)
 {
 }
 
-THttpServerConn::~THttpServerConn() {
-}
-
-THttpInput* THttpServerConn::Input() noexcept {
-    return Impl_->Input();
+THttpServerConn::THttpServerConn(THolder<ISocketStreams> socketStreams, size_t outputBufferSize)
+    : SocketStreams_(std::move(socketStreams))
+    , BufferedOutput_(SocketStreams_->Output(), outputBufferSize)
+    , HttpInput_(SocketStreams_->Input())
+    , HttpOutput_(&BufferedOutput_, &HttpInput_)
+{
 }
 
-THttpOutput* THttpServerConn::Output() noexcept {
-    return Impl_->Output();
+THttpServerConn::~THttpServerConn() {
 }
 
 void THttpServerConn::Reset() {
-    return Impl_->Reset();
+    return SocketStreams_->Reset();
 }