Compare commits

...

4 Commits

Author SHA1 Message Date
Ken Schalk 7c60e69c33 Remove redunant call to close_socket (#911) 2021-04-23 17:07:19 -04:00
yhirose 33e94891ee Updated test.cc 2021-04-22 08:04:46 -04:00
yhirose 73e0729f63 Change sink.write() to return boolean 2021-04-22 07:14:08 -04:00
yhirose 21c529229c Fixed timeout issues 2021-04-22 07:14:08 -04:00
2 changed files with 81 additions and 39 deletions
+70 -35
View File
@@ -308,7 +308,7 @@ public:
DataSink(DataSink &&) = delete;
DataSink &operator=(DataSink &&) = delete;
std::function<void(const char *data, size_t data_len)> write;
std::function<bool(const char *data, size_t data_len)> write;
std::function<void()> done;
std::function<bool()> is_writable;
std::ostream os;
@@ -2091,8 +2091,9 @@ socket_t create_socket(const char *host, int port, int address_family,
for (auto rp = result; rp; rp = rp->ai_next) {
// Create a socket
#ifdef _WIN32
auto sock = WSASocketW(rp->ai_family, rp->ai_socktype, rp->ai_protocol,
nullptr, 0, WSA_FLAG_NO_HANDLE_INHERIT);
auto sock =
WSASocketW(rp->ai_family, rp->ai_socktype, rp->ai_protocol, nullptr, 0,
WSA_FLAG_NO_HANDLE_INHERIT | WSA_FLAG_OVERLAPPED);
/**
* Since the WSA_FLAG_NO_HANDLE_INHERIT is only supported on Windows 7 SP1
* and above the socket creation fails on older Windows Systems.
@@ -2214,11 +2215,12 @@ inline std::string if2ip(const std::string &ifn) {
}
#endif
inline socket_t create_client_socket(const char *host, int port,
int address_family, bool tcp_nodelay,
SocketOptions socket_options,
time_t timeout_sec, time_t timeout_usec,
const std::string &intf, Error &error) {
inline socket_t create_client_socket(
const char *host, int port, int address_family, bool tcp_nodelay,
SocketOptions socket_options, time_t connection_timeout_sec,
time_t connection_timeout_usec, time_t read_timeout_sec,
time_t read_timeout_usec, time_t write_timeout_sec,
time_t write_timeout_usec, const std::string &intf, Error &error) {
auto sock = create_socket(
host, port, address_family, 0, tcp_nodelay, std::move(socket_options),
[&](socket_t sock, struct addrinfo &ai) -> bool {
@@ -2240,14 +2242,28 @@ inline socket_t create_client_socket(const char *host, int port,
if (ret < 0) {
if (is_connection_error() ||
!wait_until_socket_is_ready(sock, timeout_sec, timeout_usec)) {
close_socket(sock);
!wait_until_socket_is_ready(sock, connection_timeout_sec,
connection_timeout_usec)) {
error = Error::Connection;
return false;
}
}
set_nonblocking(sock, false);
{
timeval tv;
tv.tv_sec = static_cast<long>(read_timeout_sec);
tv.tv_usec = static_cast<decltype(tv.tv_usec)>(read_timeout_usec);
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv, sizeof(tv));
}
{
timeval tv;
tv.tv_sec = static_cast<long>(write_timeout_sec);
tv.tv_usec = static_cast<decltype(tv.tv_usec)>(write_timeout_usec);
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv, sizeof(tv));
}
error = Error::Success;
return true;
});
@@ -3004,7 +3020,7 @@ inline bool write_content(Stream &strm, const ContentProvider &content_provider,
auto ok = true;
DataSink data_sink;
data_sink.write = [&](const char *d, size_t l) {
data_sink.write = [&](const char *d, size_t l) -> bool {
if (ok) {
if (write_data(strm, d, l)) {
offset += l;
@@ -3012,6 +3028,7 @@ inline bool write_content(Stream &strm, const ContentProvider &content_provider,
ok = false;
}
}
return ok;
};
data_sink.is_writable = [&](void) { return ok && strm.is_writable(); };
@@ -3050,11 +3067,12 @@ write_content_without_length(Stream &strm,
auto ok = true;
DataSink data_sink;
data_sink.write = [&](const char *d, size_t l) {
data_sink.write = [&](const char *d, size_t l) -> bool {
if (ok) {
offset += l;
if (!write_data(strm, d, l)) { ok = false; }
}
return ok;
};
data_sink.done = [&](void) { data_available = false; };
@@ -3077,30 +3095,30 @@ write_content_chunked(Stream &strm, const ContentProvider &content_provider,
auto ok = true;
DataSink data_sink;
data_sink.write = [&](const char *d, size_t l) {
if (!ok) { return; }
data_sink.write = [&](const char *d, size_t l) -> bool {
if (ok) {
data_available = l > 0;
offset += l;
data_available = l > 0;
offset += l;
std::string payload;
if (!compressor.compress(d, l, false,
[&](const char *data, size_t data_len) {
payload.append(data, data_len);
return true;
})) {
ok = false;
return;
}
if (!payload.empty()) {
// Emit chunked response header and footer for each chunk
auto chunk = from_i_to_hex(payload.size()) + "\r\n" + payload + "\r\n";
if (!write_data(strm, chunk.data(), chunk.size())) {
std::string payload;
if (compressor.compress(d, l, false,
[&](const char *data, size_t data_len) {
payload.append(data, data_len);
return true;
})) {
if (!payload.empty()) {
// Emit chunked response header and footer for each chunk
auto chunk =
from_i_to_hex(payload.size()) + "\r\n" + payload + "\r\n";
if (!write_data(strm, chunk.data(), chunk.size())) {
ok = false;
}
}
} else {
ok = false;
return;
}
}
return ok;
};
data_sink.done = [&](void) {
@@ -4847,6 +4865,19 @@ inline bool Server::listen_internal() {
break;
}
{
timeval tv;
tv.tv_sec = static_cast<long>(read_timeout_sec_);
tv.tv_usec = static_cast<decltype(tv.tv_usec)>(read_timeout_usec_);
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv, sizeof(tv));
}
{
timeval tv;
tv.tv_sec = static_cast<long>(write_timeout_sec_);
tv.tv_usec = static_cast<decltype(tv.tv_usec)>(write_timeout_usec_);
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv, sizeof(tv));
}
#if __cplusplus > 201703L
task_queue->enqueue([=, this]() { process_and_close_socket(sock); });
#else
@@ -5263,11 +5294,14 @@ inline socket_t ClientImpl::create_client_socket(Error &error) const {
return detail::create_client_socket(
proxy_host_.c_str(), proxy_port_, address_family_, tcp_nodelay_,
socket_options_, connection_timeout_sec_, connection_timeout_usec_,
interface_, error);
read_timeout_sec_, read_timeout_usec_, write_timeout_sec_,
write_timeout_usec_, interface_, error);
}
return detail::create_client_socket(
host_.c_str(), port_, address_family_, tcp_nodelay_, socket_options_,
connection_timeout_sec_, connection_timeout_usec_, interface_, error);
connection_timeout_sec_, connection_timeout_usec_, read_timeout_sec_,
read_timeout_usec_, write_timeout_sec_, write_timeout_usec_, interface_,
error);
}
inline bool ClientImpl::create_and_connect_socket(Socket &socket,
@@ -5713,7 +5747,7 @@ inline std::unique_ptr<Response> ClientImpl::send_with_content_provider(
size_t offset = 0;
DataSink data_sink;
data_sink.write = [&](const char *data, size_t data_len) {
data_sink.write = [&](const char *data, size_t data_len) -> bool {
if (ok) {
auto last = offset + data_len == content_length;
@@ -5729,6 +5763,7 @@ inline std::unique_ptr<Response> ClientImpl::send_with_content_provider(
ok = false;
}
}
return ok;
};
data_sink.is_writable = [&](void) { return ok && true; };
+11 -4
View File
@@ -1366,7 +1366,8 @@ protected:
const auto &d = *data;
auto out_len =
std::min(static_cast<size_t>(length), DATA_CHUNK_SIZE);
sink.write(&d[static_cast<size_t>(offset)], out_len);
auto ret = sink.write(&d[static_cast<size_t>(offset)], out_len);
EXPECT_TRUE(ret);
return true;
},
[data] { delete data; });
@@ -2521,7 +2522,8 @@ TEST_F(ServerTest, SlowPost) {
auto res = cli_.Post(
"/slowpost", 64 * 1024 * 1024,
[&](size_t /*offset*/, size_t /*length*/, DataSink &sink) {
sink.write(buffer, sizeof(buffer));
auto ret = sink.write(buffer, sizeof(buffer));
EXPECT_TRUE(ret);
return true;
},
"text/plain");
@@ -3145,7 +3147,10 @@ static bool send_request(time_t read_timeout_sec, const std::string &req,
auto client_sock =
detail::create_client_socket(HOST, PORT, AF_UNSPEC, false, nullptr,
/*timeout_sec=*/5, 0, std::string(), error);
/*connection_timeout_sec=*/5, 0,
/*read_timeout_sec=*/5, 0,
/*write_timeout_sec=*/5, 0,
std::string(), error);
if (client_sock == INVALID_SOCKET) { return false; }
@@ -3346,7 +3351,8 @@ TEST(ServerStopTest, StopServerWithChunkedTransmission) {
DataSink &sink) {
char buffer[27];
auto size = static_cast<size_t>(sprintf(buffer, "data:%ld\n\n", offset));
sink.write(buffer, size);
auto ret = sink.write(buffer, size);
EXPECT_TRUE(ret);
std::this_thread::sleep_for(std::chrono::seconds(1));
return true;
});
@@ -3874,6 +3880,7 @@ TEST(SSLClientTest, WildcardHostNameMatch) {
cli.set_ca_cert_path(CA_CERT_FILE);
cli.enable_server_certificate_verification(true);
cli.set_follow_location(true);
auto res = cli.Get("/");
ASSERT_TRUE(res);