Added Socket::write_sync()

Author: matyhtf

core-tests/src/network/socket.cpp

@@ -333,3 +333,46 @@ TEST(socket, check_liveness) {
     t1.join();
     t2.join();
 }
+
+#define CRLF "\r\n"
+
+TEST(socket, sync) {
+	auto sock = make_socket(SW_SOCK_TCP, SW_FD_STREAM, 0);
+
+	swoole::network::Address addr;
+	ASSERT_TRUE(addr.assign("tcp://httpbin.org:80"));
+
+	ASSERT_EQ(sock->connect(addr), 0);
+
+	const char *req = "GET / HTTP/1.1" CRLF \
+			"Host: httpbin.org" CRLF \
+			"User-Agent: curl/7.81.0" CRLF \
+			"Accept: */*" CRLF \
+			"Connection: close" CRLF \
+			CRLF CRLF;
+	ssize_t n = strlen(req);
+	ASSERT_EQ(sock->write_sync(req, n), n);
+
+	string resp;
+	SW_LOOP {
+		char buf[1024];
+		n = sock->read_sync(buf, sizeof(buf));
+		if (n == 0) {
+			break;
+		}
+		ASSERT_GT(n, 0);
+		resp.append(buf, n);
+	}
+
+	ASSERT_GT(resp.length(), 4096);
+	sock->free();
+}
+
+TEST(socket, ipv6_addr) {
+	auto sock = make_socket(SW_SOCK_TCP6, SW_FD_STREAM, 0);
+	swoole::network::Address addr;
+	ASSERT_TRUE(addr.assign("tcp://[::1]:12345"));
+	ASSERT_EQ(sock->connect(addr), SW_ERR);
+	ASSERT_EQ(errno, ECONNREFUSED);
+	sock->free();
+}

include/swoole_error.h

@@ -54,6 +54,7 @@ enum swErrorCode {
 
     SW_ERROR_BAD_IPV6_ADDRESS = 720,
     SW_ERROR_UNREGISTERED_SIGNAL,
+	SW_ERROR_BAD_HOST_ADDR,
 
     // EventLoop
     SW_ERROR_EVENT_SOCKET_REMOVED = 800,

include/swoole_socket.h

@@ -108,9 +108,12 @@ struct Address {
     SocketType type;
 
     bool assign(SocketType _type, const std::string &_host, int _port);
+    bool assign(const std::string &url);
+
     const char *get_ip() {
         return get_addr();
     }
+
     int get_port();
     const char *get_addr();
 
@@ -498,6 +501,12 @@ struct Socket {
      */
     ssize_t read_sync(void *__buf, size_t __len, int timeout_ms = -1);
 
+    /**
+     * Write data to the socket synchronously without setting non-blocking or blocking IO,
+     * and allow interruptions by signals.
+     */
+    ssize_t write_sync(const void *__buf, size_t __len, int timeout_ms = -1);
+
     int shutdown(int __how) {
         return ::shutdown(fd, __how);
     }

src/network/address.cc

@@ -16,6 +16,8 @@
 
 #include "swoole_socket.h"
 
+#include <regex>
+
 namespace swoole {
 namespace network {
 
@@ -74,5 +76,40 @@ bool Address::assign(SocketType _type, const std::string &_host, int _port) {
     return false;
 }
 
+bool Address::assign(const std::string &url) {
+    std::regex pattern(R"((tcp|udp)://([\[\]a-zA-Z0-9.-:]+):(\d+))");
+    std::smatch match;
+
+    if (std::regex_match(url, match, pattern)) {
+        std::string host = match[2];
+        auto port = std::stoi(match[3]);
+
+        if (host[0] == '[') {
+        	type = SW_SOCK_TCP6;
+        	addr.inet_v6.sin6_family = AF_INET6;
+            addr.inet_v6.sin6_port = htons(port);
+            len = sizeof(addr.inet_v6);
+            if (inet_pton(AF_INET6, host.substr(1, host.size() - 2).c_str(), addr.inet_v6.sin6_addr.s6_addr)) {
+                return true;
+            }
+        } else {
+        	type = SW_SOCK_TCP;
+        	addr.inet_v4.sin_family = AF_INET;
+            addr.inet_v4.sin_port = htons(port);
+    		len = sizeof(addr.inet_v4);
+			if (!inet_pton(AF_INET, host.c_str(), &addr.inet_v4.sin_addr.s_addr)) {
+		        if (gethostbyname(AF_INET, host.c_str(), (char *) &addr.inet_v4.sin_addr.s_addr) < 0) {
+		            swoole_set_last_error(SW_ERROR_DNSLOOKUP_RESOLVE_FAILED);
+		            return false;
+		        }
+			}
+			return true;
+        }
+    }
+
+	swoole_error_log(SW_LOG_NOTICE, SW_ERROR_BAD_HOST_ADDR, "Invalid address['%s']", url.c_str());
+	return false;
+}
+
 }  // namespace network
 }  // namespace swoole

src/network/socket.cc

@@ -781,6 +781,17 @@ ssize_t Socket::read_sync(void *__buf, size_t __len, int timeout_ms) {
     }
 }
 
+ssize_t Socket::write_sync(const void *__buf, size_t __len, int timeout_ms) {
+    struct pollfd event;
+    event.fd = fd;
+    event.events = POLLOUT;
+    if (poll(&event, 1, timeout_ms) == 1) {
+        return write(__buf, __len);
+    } else {
+        return -1;
+    }
+}
+
 ssize_t Socket::readv(IOVector *io_vector) {
     ssize_t retval;