http client support uds (#461)

Author: beef9999

fs/test/CMakeLists.txt

@@ -12,6 +12,6 @@ add_executable(test-filecopy test_filecopy.cpp)
 target_link_libraries(test-filecopy PRIVATE photon_shared)
 add_test(NAME test-filecopy COMMAND $<TARGET_FILE:test-filecopy>)
 
-add_executable(test-throttled test_throttledfile.cpp)
-target_link_libraries(test-throttled PRIVATE photon_shared)
-add_test(NAME test-throttled COMMAND $<TARGET_FILE:test-throttled>)
+add_executable(test-throttle-file test_throttledfile.cpp)
+target_link_libraries(test-throttle-file PRIVATE photon_shared)
+add_test(NAME test-throttle-file COMMAND $<TARGET_FILE:test-throttle-file>)

net/http/client.cpp

@@ -38,6 +38,7 @@ class PooledDialer {
     bool tls_ctx_ownership;
     std::unique_ptr<ISocketClient> tcpsock;
     std::unique_ptr<ISocketClient> tlssock;
+    std::unique_ptr<ISocketClient> udssock;
     std::unique_ptr<Resolver> resolver;
 
     //etsocket seems not support multi thread very well, use tcp_socket now. need to find out why
@@ -49,6 +50,7 @@ class PooledDialer {
         auto tls_cli = new_tls_client(tls_ctx, new_tcp_socket_client(), true);
         tcpsock.reset(new_tcp_socket_pool(tcp_cli, -1, true));
         tlssock.reset(new_tcp_socket_pool(tls_cli, -1, true));
+        udssock.reset(new_uds_client());
     }
 
     ~PooledDialer() {
@@ -63,6 +65,8 @@ class PooledDialer {
     ISocketStream* dial(const T& x, uint64_t timeout = -1UL) {
         return dial(x.host_no_port(), x.port(), x.secure(), timeout);
     }
+
+    ISocketStream* dial(std::string_view uds_path, uint64_t timeout = -1UL);
 };
 
 ISocketStream* PooledDialer::dial(std::string_view host, uint16_t port, bool secure, uint64_t timeout) {
@@ -96,6 +100,14 @@ ISocketStream* PooledDialer::dial(std::string_view host, uint16_t port, bool sec
     return nullptr;
 }
 
+ISocketStream* PooledDialer::dial(std::string_view uds_path, uint64_t timeout) {
+    udssock->timeout(timeout);
+    auto stream = udssock->connect(uds_path.data());
+    if (!stream)
+        LOG_ERRNO_RETURN(0, nullptr, "failed to dial to unix socket `", uds_path);
+    return stream;
+}
+
 constexpr uint64_t code3xx() { return 0; }
 template<typename...Ts>
 constexpr uint64_t code3xx(uint64_t x, Ts...xs)
@@ -164,9 +176,13 @@ class ClientImpl : public Client {
         if (tmo.timeout() == 0)
             LOG_ERROR_RETURN(ETIMEDOUT, ROUNDTRIP_FAILED, "connection timedout");
         auto &req = op->req;
-        auto s = (m_proxy && !m_proxy_url.empty())
-                     ? m_dialer.dial(m_proxy_url, tmo.timeout())
-                     : m_dialer.dial(req, tmo.timeout());
+        ISocketStream* s;
+        if (m_proxy && !m_proxy_url.empty())
+            s = m_dialer.dial(m_proxy_url, tmo.timeout());
+        else if (!op->uds_path.empty())
+            s = m_dialer.dial(op->uds_path, tmo.timeout());
+        else
+            s = m_dialer.dial(req, tmo.timeout());
         if (!s) {
             if (errno == ECONNREFUSED || errno == ENOENT) {
                 LOG_ERROR_RETURN(0, ROUNDTRIP_FAST_RETRY, "connection refused")

net/http/client.h

@@ -48,7 +48,9 @@ class Client : public Object {
         uint16_t retry = 5;                       // default retry: 5 at most
         Response resp;                            // response
         int status_code = -1;                     // status code in response
-        bool enable_proxy;
+        bool enable_proxy = false;
+        std::string_view uds_path;                // If set, Unix Domain Socket will be used instead of TCP.
+                                                  // URL should still be the format of http://localhost/xxx
         IStream* body_stream = nullptr;                 // use body_stream as body
         using BodyWriter = Delegate<ssize_t, Request*>; // or call body_writer if body_stream
         BodyWriter body_writer = {};                    // is not set
@@ -68,6 +70,11 @@ class Client : public Object {
             if (!_client) return -1;
             return _client->call(this);
         }
+        int call(std::string_view unix_socket_path) {
+            if (!_client) return -1;
+            uds_path = unix_socket_path;
+            return _client->call(this);
+        }
 
     protected:
         Client* _client;
@@ -83,15 +90,15 @@ class Client : public Object {
         Operation(uint16_t buf_size) : req(_buf, buf_size) {}
     };
 
-    Operation* new_operation(Verb v, std::string_view url, uint16_t buf_size = 64 * 1024 - 1) {
+    Operation* new_operation(Verb v, std::string_view url, uint16_t buf_size = UINT16_MAX) {
         return Operation::create(this, v, url, buf_size);
     }
 
-    Operation* new_operation(uint16_t buf_size = 64 * 1024 - 1) {
+    Operation* new_operation(uint16_t buf_size = UINT16_MAX) {
         return Operation::create(this, buf_size);
     }
 
-    template<uint16_t BufferSize>
+    template<uint16_t BufferSize = UINT16_MAX>
     class OperationOnStack : public Operation {
         char _buf[BufferSize];
     public:

net/http/test/client_function_test.cpp

@@ -61,6 +61,21 @@ int timeout_writer(void *self, IStream* stream) {
     return 0;
 }
 
+class SimpleHandler : public http::HTTPHandler {
+public:
+    int handle_request(http::Request& req, http::Response& resp, std::string_view) {
+        std::string url_path(req.target());
+        resp.set_result(200);
+        resp.headers.content_length(url_path.size());
+        resp.headers.insert("Content-Type", "application/octet-stream");
+        auto n = resp.write(url_path.data(), url_path.size());
+        if (n != (ssize_t) url_path.size()) {
+            LOG_ERRNO_RETURN(0, -1, "send body failed");
+        }
+        return 0;
+    }
+};
+
 TEST(http_client, get) {
     system("mkdir -p /tmp/ease_ut/http_test/");
     system("echo \"this is a http_client request body text for socket stream\" > /tmp/ease_ut/http_test/ease-httpclient-gettestfile");
@@ -523,6 +538,41 @@ TEST(DISABLED_http_client, ipv6) {  // make sure runing in a ipv6-ready environm
     EXPECT_EQ(200, op->resp.status_code());
 }
 
+TEST(http_client, unix_socket) {
+    const char* uds_path = "test-http-client.sock";
+
+    auto http_server = new_http_server();
+    DEFER(delete http_server);
+    http_server->add_handler(new SimpleHandler, true, "/simple-api");
+
+    auto socket_server = new_uds_server(true);
+    DEFER(delete socket_server);
+
+    socket_server->set_handler(http_server->get_connection_handler());
+    ASSERT_EQ(0, socket_server->bind(uds_path));
+    ASSERT_EQ(0, socket_server->listen());
+    ASSERT_EQ(0, socket_server->start_loop(false));
+
+    auto client = new_http_client();
+    DEFER(delete client);
+
+    Client::OperationOnStack<> op(client, Verb::GET, "http://localhost/simple-api");
+    int ret = op.call(uds_path);
+    ASSERT_EQ(0, ret);
+    ASSERT_EQ(200, op.resp.status_code());
+
+    char buf[UINT16_MAX];
+    ssize_t n = op.resp.read(buf, op.resp.body_size());
+    ASSERT_EQ(n, (ssize_t) op.resp.body_size());
+    LOG_INFO(buf);
+
+    // A wrong hostname or HTTPS doesn't have effect on unix socket
+    Client::OperationOnStack<> op2(client, Verb::GET, "https://www.wrong.hostname/simple-api");
+    ret = op2.call(uds_path);
+    ASSERT_EQ(0, ret);
+    ASSERT_EQ(200, op2.resp.status_code());
+}
+
 TEST(url, url_escape_unescape) {
     EXPECT_EQ(
         url_escape("?a=x:b&b=cd&c= feg&d=2/1[+]@alibaba.com&e='!bad';"),
@@ -582,5 +632,5 @@ int main(int argc, char** arg) {
 #endif
     set_log_output_level(ALOG_INFO);
     ::testing::InitGoogleTest(&argc, arg);
-    LOG_DEBUG("test result:`", RUN_ALL_TESTS());
+    return RUN_ALL_TESTS();
 }