Ssl refactoring

GitOrigin-RevId: f5916787608227b6914c10520dfe7a7039522ef9

1 files changed, 43 insertions, 22 deletions

diff --git a/tdnet/td/net/HttpConnectionBase.cpp b/tdnet/td/net/HttpConnectionBase.cpp
index c093efc75..3d05a6fd4 100644
--- a/tdnet/td/net/HttpConnectionBase.cpp
+++ b/tdnet/td/net/HttpConnectionBase.cpp
@@ -14,14 +14,23 @@
 namespace td {
 namespace detail {
 
-HttpConnectionBase::HttpConnectionBase(State state, FdProxy fd, size_t max_post_size, size_t max_files,
-                                       int32 idle_timeout)
+HttpConnectionBase::HttpConnectionBase(State state, SocketFd fd, SslStream ssl_stream, size_t max_post_size,
+                                       size_t max_files, int32 idle_timeout)
     : state_(state)
-    , stream_connection_(std::move(fd))
+    , fd_(std::move(fd))
+    , ssl_stream_(std::move(ssl_stream))
     , max_post_size_(max_post_size)
     , max_files_(max_files)
     , idle_timeout_(idle_timeout) {
   CHECK(state_ != State::Close);
+
+  if (ssl_stream_) {
+    read_source_ >> ssl_stream_.read_byte_flow() >> read_sink_;
+    write_source_ >> ssl_stream_.write_byte_flow() >> write_sink_;
+  } else {
+    read_source_ >> read_sink_;
+    write_source_ >> write_sink_;
+  }
 }
 
 void HttpConnectionBase::live_event() {
@@ -31,9 +40,9 @@ void HttpConnectionBase::live_event() {
 }
 
 void HttpConnectionBase::start_up() {
-  stream_connection_.get_fd().set_observer(this);
-  subscribe(stream_connection_.get_fd());
-  reader_.init(&stream_connection_.input_buffer(), max_post_size_, max_files_);
+  fd_.get_fd().set_observer(this);
+  subscribe(fd_.get_fd());
+  reader_.init(read_sink_.get_output(), max_post_size_, max_files_);
   if (state_ == State::Read) {
     current_query_ = make_unique<HttpQuery>();
   }
@@ -41,13 +50,13 @@ void HttpConnectionBase::start_up() {
   yield();
 }
 void HttpConnectionBase::tear_down() {
-  unsubscribe_before_close(stream_connection_.get_fd());
-  stream_connection_.close();
+  unsubscribe_before_close(fd_.get_fd());
+  fd_.close();
 }
 
 void HttpConnectionBase::write_next(BufferSlice buffer) {
   CHECK(state_ == State::Write);
-  stream_connection_.output_buffer().append(std::move(buffer));
+  write_buffer_.append(std::move(buffer));
   loop();
 }
 
@@ -69,7 +78,7 @@ void HttpConnectionBase::write_error(Status error) {
 void HttpConnectionBase::timeout_expired() {
   LOG(INFO) << "Idle timeout expired";
 
-  if (stream_connection_.need_flush_write()) {
+  if (fd_.need_flush_write()) {
     on_error(Status::Error("Write timeout expired"));
   } else if (state_ == State::Read) {
     on_error(Status::Error("Read timeout expired"));
@@ -78,9 +87,9 @@ void HttpConnectionBase::timeout_expired() {
   stop();
 }
 void HttpConnectionBase::loop() {
-  if (can_read(stream_connection_)) {
+  if (can_read(fd_)) {
     LOG(DEBUG) << "Can read from the connection";
-    auto r = stream_connection_.flush_read();
+    auto r = fd_.flush_read();
     if (r.is_error()) {
       if (!begins_with(r.error().message(), "SSL error {336134278")) {  // if error is not yet outputed
         LOG(INFO) << "flush_read error: " << r.error();
@@ -89,6 +98,7 @@ void HttpConnectionBase::loop() {
       return stop();
     }
   }
+  read_source_.wakeup();
 
   // TODO: read_next even when state_ == State::Write
 
@@ -102,7 +112,7 @@ void HttpConnectionBase::loop() {
       HttpHeaderCreator hc;
       hc.init_status_line(res.error().code());
       hc.set_content_size(0);
-      stream_connection_.output_buffer().append(hc.finish().ok());
+      write_buffer_.append(hc.finish().ok());
       close_after_write_ = true;
       on_error(Status::Error(res.error().public_message()));
     } else if (res.ok() == 0) {
@@ -115,34 +125,45 @@ void HttpConnectionBase::loop() {
     }
   }
 
-  if (can_write(stream_connection_)) {
+  write_source_.wakeup();
+
+  if (can_write(fd_)) {
     LOG(DEBUG) << "Can write to the connection";
-    auto r = stream_connection_.flush_write();
+    auto r = fd_.flush_write();
     if (r.is_error()) {
       LOG(INFO) << "flush_write error: " << r.error();
       on_error(Status::Error(r.error().public_message()));
     }
-    if (close_after_write_ && !stream_connection_.need_flush_write()) {
+    if (close_after_write_ && !fd_.need_flush_write()) {
       return stop();
     }
   }
 
-  if (stream_connection_.get_fd().has_pending_error()) {
-    auto pending_error = stream_connection_.get_pending_error();
+  Status pending_error;
+  if (fd_.get_fd().has_pending_error()) {
+    pending_error = fd_.get_pending_error();
+  }
+  if (pending_error.is_ok() && write_sink_.status().is_error()) {
+    pending_error = std::move(write_sink_.status());
+  }
+  if (pending_error.is_ok() && read_sink_.status().is_error()) {
+    pending_error = std::move(read_sink_.status());
+  }
+  if (pending_error.is_error()) {
     LOG(INFO) << pending_error;
     if (!close_after_write_) {
       on_error(Status::Error(pending_error.public_message()));
     }
     state_ = State::Close;
   }
-  if (can_close(stream_connection_)) {
+
+  if (can_close(fd_)) {
     LOG(DEBUG) << "Can close the connection";
     state_ = State::Close;
   }
   if (state_ == State::Close) {
-    LOG_IF(INFO, stream_connection_.need_flush_write()) << "Close nonempty connection";
-    LOG_IF(INFO, want_read &&
-                     (stream_connection_.input_buffer().size() > 0 || current_query_->type_ != HttpQuery::Type::EMPTY))
+    LOG_IF(INFO, fd_.need_flush_write()) << "Close nonempty connection";
+    LOG_IF(INFO, want_read && (fd_.input_buffer().size() > 0 || current_query_->type_ != HttpQuery::Type::EMPTY))
         << "Close connection while reading request/response";
     return stop();
   }