finish stream, echo_client

Author: netcan

CMakeLists.txt

@@ -27,13 +27,16 @@ set(ASYNCIO_INC
         include/asyncio/concept/awaitable.h
         include/asyncio/gather.h
         include/asyncio/result.h
-        include/asyncio/callstack.h)
+        include/asyncio/callstack.h
+        include/asyncio/open_connection.h
+        include/asyncio/stream.h
+        )
 
 include_directories(${CMAKE_SOURCE_DIR}/include)
 
 add_library(asyncio STATIC
         ${ASYNC_INC}
-        src/event_loop.cpp include/asyncio/open_connection.h)
+        src/event_loop.cpp)
 target_link_libraries(asyncio PUBLIC fmt::fmt)
 
 add_subdirectory(test)

include/asyncio/open_connection.h

@@ -5,25 +5,17 @@
 #ifndef ASYNCIO_OPEN_CONNECTION_H
 #define ASYNCIO_OPEN_CONNECTION_H
 #include <asyncio/asyncio_ns.h>
+#include <asyncio/stream.h>
 #include <asyncio/selector/event.h>
 #include <exception>
 #include <asyncio/task.h>
 #include <fcntl.h>
-#include <unistd.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <system_error>
 #include <netdb.h>
 
 ASYNCIO_NS_BEGIN
-struct Stream: NonCopyable {
-    Stream(int fd): fd_(fd) {}
-    Stream(Stream&& other): fd_{std::exchange(other.fd_, -1) } {}
-    ~Stream() { if (fd_ > 0) { close(fd_); } }
-private:
-    int fd_{-1};
-};
-
 namespace detail {
 Task<bool> connect(int fd, const sockaddr *addr, socklen_t len) noexcept {
     int rc = ::connect(fd, addr, len);
@@ -73,8 +65,7 @@ Task<Stream> open_connection(std::string_view ip, uint16_t port) {
             continue;
         }
         if (co_await detail::connect(sockfd, p->ai_addr, p->ai_addrlen) ) {
-            close(sockfd);
-            continue;
+            break;
         }
     }
     if (sockfd == -1) {

include/asyncio/selector/epoll_selector.h

@@ -32,9 +32,7 @@ struct EpollSelector {
         return result;
     }
     ~EpollSelector() {
-        if (epfd_ > 0) {
-            close(epfd_);
-        }
+        if (epfd_ > 0) { close(epfd_); }
     }
     bool is_stop() { return register_event_count_ == 1; }
     void register_event(const Event& event) {

include/asyncio/stream.h

@@ -0,0 +1,80 @@
+//
+// Created by netcan on 2021/11/30.
+//
+
+#ifndef ASYNCIO_STREAM_H
+#define ASYNCIO_STREAM_H
+#include <asyncio/asyncio_ns.h>
+#include <asyncio/noncopyable.h>
+#include <asyncio/task.h>
+#include <utility>
+#include <vector>
+#include <unistd.h>
+
+ASYNCIO_NS_BEGIN
+struct Stream: NonCopyable {
+    using Buffer = std::vector<char>;
+    Stream(int fd): fd_(fd) {}
+    Stream(Stream&& other): fd_{std::exchange(other.fd_, -1) } {}
+    ~Stream() { close(); }
+
+    void close() {
+        if (fd_ > 0) { ::close(fd_); }
+        fd_ = -1;
+    }
+
+    Task<Buffer> read(ssize_t sz = -1) {
+        if (sz < 0) { co_return co_await read_until_eof(); }
+
+        Buffer result(sz, 0);
+        Event ev { .fd = fd_, .events = EPOLLIN };
+        auto& loop = get_event_loop();
+        co_await loop.wait_event(ev);
+        sz = ::read(fd_, result.data(), result.size());
+        if (sz == -1) {
+            throw std::system_error(std::make_error_code(static_cast<std::errc>(errno)));
+        }
+        result.resize(sz);
+        co_return result;
+    }
+
+    Task<> write(const Buffer& buf) {
+        Event ev { .fd = fd_, .events = EPOLLOUT };
+        auto& loop = get_event_loop();
+        ssize_t total_write = 0;
+        while (total_write < buf.size()) {
+            co_await loop.wait_event(ev);
+            ssize_t sz = ::write(fd_, buf.data() + total_write, buf.size() - total_write);
+            if (sz == -1) {
+                throw std::system_error(std::make_error_code(static_cast<std::errc>(errno)));
+            }
+            total_write += sz;
+        }
+    }
+
+private:
+    Task<Buffer> read_until_eof() {
+        auto& loop = get_event_loop();
+
+        Buffer result(chunk_size, 0);
+        Event ev { .fd = fd_, .events = EPOLLIN };
+        int current_read = 0;
+        int total_read = 0;
+        do {
+            co_await loop.wait_event(ev);
+            current_read = ::read(fd_, result.data() + total_read, chunk_size);
+            if (current_read == -1) {
+                throw std::system_error(std::make_error_code(static_cast<std::errc>(errno)));
+            }
+            if (current_read < chunk_size) { result.resize(total_read + current_read); }
+            total_read += current_read;
+            result.resize(total_read + chunk_size);
+        } while (current_read > 0);
+        co_return result;
+    }
+private:
+    int fd_{-1};
+    constexpr static size_t chunk_size = 4096;
+};
+ASYNCIO_NS_END
+#endif // ASYNCIO_STREAM_H

test/st/echo_client.cpp

@@ -4,10 +4,19 @@
 #include <asyncio/open_connection.h>
 #include <asyncio/runner.h>
 using asyncio::Task;
+using asyncio::Stream;
 
 Task<void> tcp_echo_client(std::string_view message) {
     auto stream = co_await asyncio::open_connection("127.0.0.1", 8888);
-    co_return;
+
+    fmt::print("Send: {}!\n", message);
+    co_await stream.write(Stream::Buffer(message.begin(), message.end()));
+
+    auto data = co_await stream.read(100);
+    fmt::print("Received: {}!\n", data.data());
+
+    fmt::print("Close the connection\n");
+    stream.close();
 }
 
 int main(int argc, char** argv) {