diff --git a/lib/content_encoding.c b/lib/content_encoding.c index df62741932..a3724d319d 100644 --- a/lib/content_encoding.c +++ b/lib/content_encoding.c @@ -299,6 +299,7 @@ static const struct Curl_cwtype deflate_encoding = { NULL, deflate_do_init, deflate_do_write, + Curl_cwriter_def_flush, deflate_do_close, sizeof(struct zlib_writer) }; @@ -360,6 +361,7 @@ static const struct Curl_cwtype gzip_encoding = { "x-gzip", gzip_do_init, gzip_do_write, + Curl_cwriter_def_flush, gzip_do_close, sizeof(struct zlib_writer) }; @@ -493,6 +495,7 @@ static const struct Curl_cwtype brotli_encoding = { NULL, brotli_do_init, brotli_do_write, + Curl_cwriter_def_flush, brotli_do_close, sizeof(struct brotli_writer) }; @@ -606,6 +609,7 @@ static const struct Curl_cwtype zstd_encoding = { NULL, zstd_do_init, zstd_do_write, + Curl_cwriter_def_flush, zstd_do_close, sizeof(struct zstd_writer) }; @@ -617,6 +621,7 @@ static const struct Curl_cwtype identity_encoding = { "none", Curl_cwriter_def_init, Curl_cwriter_def_write, + Curl_cwriter_def_flush, Curl_cwriter_def_close, sizeof(struct Curl_cwriter) }; @@ -704,6 +709,7 @@ static const struct Curl_cwtype error_writer = { NULL, error_do_init, error_do_write, + Curl_cwriter_def_flush, error_do_close, sizeof(struct Curl_cwriter) }; diff --git a/lib/cw-out.c b/lib/cw-out.c index fad89b392c..b38fde455f 100644 --- a/lib/cw-out.c +++ b/lib/cw-out.c @@ -428,6 +428,8 @@ static CURLcode cw_out_write(struct Curl_easy *data, CURLcode result; bool flush_all = !!(type & CLIENTWRITE_EOS); + CURL_TRC_WRITE(data, "[OUT] write %zu bytes, type=%x", + blen, (unsigned int)type); if((type & CLIENTWRITE_BODY) || ((type & CLIENTWRITE_HEADER) && data->set.include_header)) { cw_out_type otype = (!blen && (type & CLIENTWRITE_0LEN)) ? @@ -446,31 +448,9 @@ static CURLcode cw_out_write(struct Curl_easy *data, return CURLE_OK; } -const struct Curl_cwtype Curl_cwt_out = { - "cw-out", - NULL, - cw_out_init, - cw_out_write, - cw_out_close, - sizeof(struct cw_out_ctx) -}; - -bool Curl_cw_out_is_paused(struct Curl_easy *data) -{ - struct Curl_cwriter *cw_out; - struct cw_out_ctx *ctx; - - cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out); - if(!cw_out) - return FALSE; - - ctx = (struct cw_out_ctx *)cw_out; - return (bool)ctx->paused; -} - -static CURLcode cw_out_flush(struct Curl_easy *data, - struct Curl_cwriter *cw_out, - bool flush_all) +static CURLcode cw_out_do_flush(struct Curl_easy *data, + struct Curl_cwriter *cw_out, + bool flush_all) { struct cw_out_ctx *ctx = (struct cw_out_ctx *)cw_out; CURLcode result = CURLE_OK; @@ -489,6 +469,36 @@ static CURLcode cw_out_flush(struct Curl_easy *data, return result; } +static CURLcode cw_out_flush(struct Curl_easy *data, + struct Curl_cwriter *writer) +{ + CURL_TRC_WRITE(data, "[OUT] flush"); + return cw_out_do_flush(data, writer, FALSE); +} + +const struct Curl_cwtype Curl_cwt_out = { + "cw-out", + NULL, + cw_out_init, + cw_out_write, + cw_out_flush, + cw_out_close, + sizeof(struct cw_out_ctx) +}; + +bool Curl_cw_out_is_paused(struct Curl_easy *data) +{ + struct Curl_cwriter *cw_out; + struct cw_out_ctx *ctx; + + cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out); + if(!cw_out) + return FALSE; + + ctx = (struct cw_out_ctx *)cw_out; + return (bool)ctx->paused; +} + CURLcode Curl_cw_out_unpause(struct Curl_easy *data) { struct Curl_cwriter *cw_out; @@ -499,9 +509,6 @@ CURLcode Curl_cw_out_unpause(struct Curl_easy *data) struct cw_out_ctx *ctx = (struct cw_out_ctx *)cw_out; CURL_TRC_WRITE(data, "[OUT] unpause"); ctx->paused = FALSE; - result = Curl_cw_pause_flush(data); - if(!result) - result = cw_out_flush(data, cw_out, FALSE); } return result; } @@ -514,9 +521,9 @@ CURLcode Curl_cw_out_done(struct Curl_easy *data) cw_out = Curl_cwriter_get_by_type(data, &Curl_cwt_out); if(cw_out) { CURL_TRC_WRITE(data, "[OUT] done"); - result = Curl_cw_pause_flush(data); + result = Curl_client_flush(data); if(!result) - result = cw_out_flush(data, cw_out, TRUE); + result = cw_out_do_flush(data, cw_out, TRUE); } return result; } diff --git a/lib/cw-pause.c b/lib/cw-pause.c index 5561a8d7b7..a42f61f4ba 100644 --- a/lib/cw-pause.c +++ b/lib/cw-pause.c @@ -138,6 +138,9 @@ static CURLcode cw_pause_flush(struct Curl_easy *data, *plast = NULL; } } + + if(!result) + result = Curl_cwriter_flush(data, cw_pause->next); return result; } @@ -166,8 +169,6 @@ static CURLcode cw_pause_write(struct Curl_easy *data, if(wlen < blen) wtype &= ~CLIENTWRITE_EOS; result = Curl_cwriter_write(data, writer->next, wtype, buf, wlen); - CURL_TRC_WRITE(data, "[PAUSE] writing %zu/%zu bytes of type %x -> %d", - wlen, blen, (unsigned int)wtype, (int)result); if(result) return result; buf += wlen; @@ -210,18 +211,7 @@ const struct Curl_cwtype Curl_cwt_pause = { NULL, cw_pause_init, cw_pause_write, + cw_pause_flush, cw_pause_close, sizeof(struct cw_pause_ctx) }; - -CURLcode Curl_cw_pause_flush(struct Curl_easy *data) -{ - struct Curl_cwriter *cw_pause; - CURLcode result = CURLE_OK; - - cw_pause = Curl_cwriter_get_by_type(data, &Curl_cwt_pause); - if(cw_pause) - result = cw_pause_flush(data, cw_pause); - - return result; -} diff --git a/lib/cw-pause.h b/lib/cw-pause.h index 544cbfa577..9a41a1c9b1 100644 --- a/lib/cw-pause.h +++ b/lib/cw-pause.h @@ -33,6 +33,4 @@ struct Curl_easy; */ extern const struct Curl_cwtype Curl_cwt_pause; -CURLcode Curl_cw_pause_flush(struct Curl_easy *data); - #endif /* HEADER_CURL_CW_PAUSE_H */ diff --git a/lib/ftp.c b/lib/ftp.c index c4b1b5fcfd..b3da888b44 100644 --- a/lib/ftp.c +++ b/lib/ftp.c @@ -445,6 +445,7 @@ static const struct Curl_cwtype ftp_cw_lc = { NULL, Curl_cwriter_def_init, ftp_cw_lc_write, + Curl_cwriter_def_flush, Curl_cwriter_def_close, sizeof(struct ftp_cw_lc_ctx) }; diff --git a/lib/headers.c b/lib/headers.c index b290a9e5b3..5d2101ee88 100644 --- a/lib/headers.c +++ b/lib/headers.c @@ -317,6 +317,7 @@ static const struct Curl_cwtype hds_cw_collect = { NULL, Curl_cwriter_def_init, hds_cw_collect_write, + Curl_cwriter_def_flush, Curl_cwriter_def_close, sizeof(struct hds_cw_collect_ctx) }; diff --git a/lib/http_chunks.c b/lib/http_chunks.c index 9596fc6693..93f5bb6c42 100644 --- a/lib/http_chunks.c +++ b/lib/http_chunks.c @@ -468,6 +468,7 @@ const struct Curl_cwtype Curl_httpchunk_unencoder = { NULL, cw_chunked_init, cw_chunked_write, + Curl_cwriter_def_flush, cw_chunked_close, sizeof(struct chunked_writer) }; diff --git a/lib/sendf.c b/lib/sendf.c index 7559f64f84..b830995345 100644 --- a/lib/sendf.c +++ b/lib/sendf.c @@ -134,6 +134,15 @@ CURLcode Curl_cwriter_write(struct Curl_easy *data, return writer->cwt->do_write(data, writer, type, buf, nbytes); } +CURLcode Curl_cwriter_flush(struct Curl_easy *data, + struct Curl_cwriter *writer) +{ + if(!writer) { + return CURLE_WRITE_ERROR; + } + return writer->cwt->do_flush(data, writer); +} + CURLcode Curl_cwriter_def_init(struct Curl_easy *data, struct Curl_cwriter *writer) { @@ -149,6 +158,12 @@ CURLcode Curl_cwriter_def_write(struct Curl_easy *data, return Curl_cwriter_write(data, writer->next, type, buf, nbytes); } +CURLcode Curl_cwriter_def_flush(struct Curl_easy *data, + struct Curl_cwriter *writer) +{ + return Curl_cwriter_flush(data, writer->next); +} + void Curl_cwriter_def_close(struct Curl_easy *data, struct Curl_cwriter *writer) { @@ -298,6 +313,7 @@ static const struct Curl_cwtype cw_download = { NULL, Curl_cwriter_def_init, cw_download_write, + Curl_cwriter_def_flush, Curl_cwriter_def_close, sizeof(struct cw_download_ctx) }; @@ -319,6 +335,7 @@ static const struct Curl_cwtype cw_raw = { NULL, Curl_cwriter_def_init, cw_raw_write, + Curl_cwriter_def_flush, Curl_cwriter_def_close, sizeof(struct Curl_cwriter) }; @@ -401,6 +418,22 @@ CURLcode Curl_client_write(struct Curl_easy *data, int type, const char *buf, return result; } +CURLcode Curl_client_flush(struct Curl_easy *data) +{ + CURLcode result; + + if(!data->req.writer_stack) { + result = do_init_writer_stack(data); + if(result) + return result; + DEBUGASSERT(data->req.writer_stack); + } + + result = Curl_cwriter_flush(data, data->req.writer_stack); + CURL_TRC_WRITE(data, "client_flush() -> %d", (int)result); + return result; +} + /* Create an unencoding writer stage using the given handler. */ CURLcode Curl_cwriter_create(struct Curl_cwriter **pwriter, struct Curl_easy *data, @@ -510,7 +543,10 @@ bool Curl_cwriter_is_paused(struct Curl_easy *data) CURLcode Curl_cwriter_unpause(struct Curl_easy *data) { - return Curl_cw_out_unpause(data); + CURLcode result = Curl_cw_out_unpause(data); + if(!result) + result = Curl_cwriter_flush(data, data->req.writer_stack); + return result; } CURLcode Curl_creader_read(struct Curl_easy *data, diff --git a/lib/sendf.h b/lib/sendf.h index 787fe7ff69..0cf6245ff6 100644 --- a/lib/sendf.h +++ b/lib/sendf.h @@ -61,6 +61,8 @@ struct Curl_easy; CURLcode Curl_client_write(struct Curl_easy *data, int type, const char *buf, size_t len) WARN_UNUSED_RESULT; +CURLcode Curl_client_flush(struct Curl_easy *data); + /** * Free all resources related to client writing. */ @@ -115,6 +117,8 @@ struct Curl_cwtype { CURLcode (*do_write)(struct Curl_easy *data, struct Curl_cwriter *writer, int type, const char *buf, size_t nbytes); + CURLcode (*do_flush)(struct Curl_easy *data, + struct Curl_cwriter *writer); void (*do_close)(struct Curl_easy *data, struct Curl_cwriter *writer); size_t cwriter_size; /* sizeof() allocated struct Curl_cwriter */ @@ -179,6 +183,8 @@ struct Curl_cwriter *Curl_cwriter_get_by_name(struct Curl_easy *data, CURLcode Curl_cwriter_write(struct Curl_easy *data, struct Curl_cwriter *writer, int type, const char *buf, size_t nbytes); +CURLcode Curl_cwriter_flush(struct Curl_easy *data, + struct Curl_cwriter *writer); /** * Return TRUE iff client writer is paused. @@ -201,6 +207,8 @@ CURLcode Curl_cwriter_def_init(struct Curl_easy *data, CURLcode Curl_cwriter_def_write(struct Curl_easy *data, struct Curl_cwriter *writer, int type, const char *buf, size_t nbytes); +CURLcode Curl_cwriter_def_flush(struct Curl_easy *data, + struct Curl_cwriter *writer); void Curl_cwriter_def_close(struct Curl_easy *data, struct Curl_cwriter *writer); diff --git a/lib/ws.c b/lib/ws.c index 9820c3e4bd..3082860389 100644 --- a/lib/ws.c +++ b/lib/ws.c @@ -477,7 +477,7 @@ static CURLcode ws_dec_read_head(struct ws_decoder *dec, dec->frame_age = 0; dec->payload_offset = 0; - ws_dec_info(dec, data, "decoded"); + ws_dec_info(dec, data, "head"); return CURLE_OK; } return CURLE_AGAIN; @@ -496,7 +496,8 @@ static CURLcode ws_dec_pass_payload(struct ws_decoder *dec, size_t remain = curlx_sotouz_range(dec->payload_len - dec->payload_offset, 0, SIZE_MAX); - while(remain && Curl_bufq_peek(inraw, &inbuf, &inlen)) { + while(remain && Curl_bufq_peek(inraw, &inbuf, &inlen) && + !Curl_cwriter_is_paused(data)) { if(inlen > remain) inlen = remain; result = write_cb(inbuf, inlen, dec->frame_age, dec->frame_flags, @@ -733,7 +734,7 @@ static CURLcode ws_cw_write(struct Curl_easy *data, } } - while(!Curl_bufq_is_empty(&ctx->buf)) { + while(!Curl_bufq_is_empty(&ctx->buf) && !Curl_cwriter_is_paused(data)) { struct ws_cw_dec_ctx pass_ctx; pass_ctx.data = data; pass_ctx.ws = ws; @@ -768,12 +769,63 @@ out: return result; } +static CURLcode ws_cw_flush(struct Curl_easy *data, + struct Curl_cwriter *writer) +{ + CURLcode result = CURLE_OK; + + CURL_TRC_WRITE(data, "[ws] flush"); + if(!data->set.ws_raw_mode) { + struct ws_cw_ctx *ctx = writer->ctx; + struct websocket *ws; + + /* Frames should be written one by one, else the meta data does + * not fit. Flush the next writer first, so it does not aggregate + * our flushed data with anything it might have buffered. */ + result = Curl_cwriter_flush(data, writer->next); + if(result) + goto out; + + ws = Curl_conn_meta_get(data->conn, CURL_META_PROTO_WS_CONN); + if(!ws) { + failf(data, "[WS] not a websocket transfer"); + return CURLE_FAILED_INIT; + } + + while(!Curl_bufq_is_empty(&ctx->buf) && !Curl_cwriter_is_paused(data)) { + struct ws_cw_dec_ctx pass_ctx; + pass_ctx.data = data; + pass_ctx.ws = ws; + pass_ctx.next_writer = writer->next; + pass_ctx.cw_type = CLIENTWRITE_BODY; + result = ws_dec_pass(&ws->dec, data, &ctx->buf, + ws_cw_dec_next, &pass_ctx); + if(result == CURLE_AGAIN) { + /* insufficient amount of data, keep it for later. + * we pretend to have written all since we have a copy */ + result = CURLE_OK; + goto out; + } + else if(result) { + failf(data, "[WS] decode payload error %d", (int)result); + return result; + } + } + } + +out: + if(!result) + result = Curl_cwriter_flush(data, writer->next); + return result; +} + /* WebSocket payload decoding client writer. */ static const struct Curl_cwtype ws_cw_decode = { "ws-decode", NULL, ws_cw_init, ws_cw_write, + ws_cw_flush, ws_cw_close, sizeof(struct ws_cw_ctx) }; diff --git a/tests/http/Makefile.am b/tests/http/Makefile.am index 6b4b6054eb..4f2fad7d02 100644 --- a/tests/http/Makefile.am +++ b/tests/http/Makefile.am @@ -38,7 +38,8 @@ TESTENV = \ testenv/ports.py \ testenv/sshd.py \ testenv/vsftpd.py \ - testenv/ws_echo_server.py + testenv/ws_echo_server.py \ + testenv/ws_4frames_server.py EXTRA_DIST = \ CMakeLists.txt \ diff --git a/tests/http/test_20_websockets.py b/tests/http/test_20_websockets.py index a3f5e3a3c0..5ee8b5e6a9 100644 --- a/tests/http/test_20_websockets.py +++ b/tests/http/test_20_websockets.py @@ -44,13 +44,19 @@ from testenv.ports import alloc_ports_and_do log = logging.getLogger(__name__) -@pytest.mark.skipif(condition=not Env.curl_has_protocol('ws'), - reason='curl lacks ws protocol support') -class TestWebsockets: +class WsServer: - PORT_SPECS = { - 'ws': socket.SOCK_STREAM, - } + def __init__(self, name, env, cmd): + self.name = name + self.env = env + self.run_dir = os.path.join(env.gen_dir, self.name) + self.err_file = os.path.join(self.run_dir, 'stderr') + self._rmrf(self.run_dir) + self._mkpath(self.run_dir) + self.cmd = cmd + self.wsproc = None + self.cerr = None + self.port = 0 def check_alive(self, env, port, timeout=Env.SERVER_TIMEOUT): curl = CurlClient(env=env) @@ -71,42 +77,62 @@ class TestWebsockets: if os.path.exists(path): shutil.rmtree(path) - @pytest.fixture(autouse=True, scope='class') - def ws_echo(self, env): - self.run_dir = os.path.join(env.gen_dir, 'ws_echo_server') - err_file = os.path.join(self.run_dir, 'stderr') - self._rmrf(self.run_dir) - self._mkpath(self.run_dir) - self.cmd = os.path.join(env.project_dir, - 'tests/http/testenv/ws_echo_server.py') - self.wsproc = None - self.cerr = None + def startup(self): def startup(ports: Dict[str, int]) -> bool: - wargs = [self.cmd, '--port', str(ports['ws'])] + self.port = ports[self.name] + wargs = [self.cmd, '--port', str(self.port)] log.info(f'start_ {wargs}') self.wsproc = subprocess.Popen(args=wargs, cwd=self.run_dir, stderr=self.cerr, stdout=self.cerr) - if self.check_alive(env, ports['ws']): - env.update_ports(ports) + if self.check_alive(self.env, self.port): + self.env.update_ports(ports) return True log.error(f'not alive {wargs}') self.wsproc.terminate() self.wsproc = None return False - with open(err_file, 'w') as self.cerr: - assert alloc_ports_and_do(TestWebsockets.PORT_SPECS, startup, - env.gen_root, max_tries=3) - assert self.wsproc - yield - self.wsproc.terminate() + self.cerr = open(self.err_file, 'w') + port_spec = { + self.name: socket.SOCK_STREAM + } + assert alloc_ports_and_do(port_spec, startup, + self.env.gen_root, max_tries=3) + assert self.wsproc + + def shutdown(self): + self.wsproc.terminate() + self.cerr.close() + + +@pytest.mark.skipif(condition=not Env.curl_has_protocol('ws'), + reason='curl lacks ws protocol support') +class TestWebsockets: + + @pytest.fixture(autouse=True, scope='class') + def ws_echo(self, env): + cmd = os.path.join(env.project_dir, + 'tests/http/testenv/ws_echo_server.py') + server = WsServer('ws_echo', env, cmd) + server.startup() + yield server + server.shutdown() + + @pytest.fixture(autouse=True, scope='class') + def ws_4frames(self, env): + cmd = os.path.join(env.project_dir, + 'tests/http/testenv/ws_4frames_server.py') + server = WsServer('ws_4frames', env, cmd) + server.startup() + yield server + server.shutdown() def test_20_01_basic(self, env: Env, ws_echo): curl = CurlClient(env=env) - url = f'http://localhost:{env.ws_port}/' + url = f'http://localhost:{ws_echo.port}/' r = curl.http_download(urls=[url]) r.check_response(http_status=426) @@ -115,7 +141,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_pingpong') if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' r = client.run(args=[url, payload]) r.check_exit_code(0) @@ -125,7 +151,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_pingpong') if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' r = client.run(args=[url, payload]) r.check_exit_code(100) # CURLE_TOO_LARGE @@ -137,7 +163,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_data') if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' r = client.run(args=[f'-{model}', '-m', str(1), '-M', str(10), url]) r.check_exit_code(0) @@ -149,7 +175,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_data') if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' r = client.run(args=[f'-{model}', '-m', str(120), '-M', str(130), url]) r.check_exit_code(0) @@ -161,7 +187,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_data') if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' r = client.run(args=[f'-{model}', '-m', str(65535 - 5), '-M', str(65535 + 5), url]) r.check_exit_code(0) @@ -175,7 +201,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_data', run_env=run_env) if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' r = client.run(args=[f'-{model}', '-m', str(65535 - 5), '-M', str(65535 + 5), url]) r.check_exit_code(0) @@ -191,7 +217,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_data', run_env=run_env) if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' count = 10 large = 20000 r = client.run(args=[f'-{model}', '-c', str(count), '-m', str(large), url]) @@ -205,7 +231,7 @@ class TestWebsockets: client = LocalClient(env=env, name='cli_ws_data') if not client.exists(): pytest.skip(f'example client not built: {client.name}') - url = f'ws://localhost:{env.ws_port}/' + url = f'ws://localhost:{ws_echo.port}/' count = 10 large = 0 r = client.run(args=[f'-{model}', '-c', str(count), '-m', str(large), url]) @@ -214,7 +240,7 @@ class TestWebsockets: # use ws:// URL with HTTP proxy, check that it tunnels automatically def test_20_10_proxy_http(self, env: Env, httpd, ws_echo): curl = CurlClient(env=env) - url = f'ws://127.0.0.1:{env.ws_port}/' + url = f'ws://127.0.0.1:{ws_echo.port}/' xargs = curl.get_proxy_args(proxys=False) xargs.extend([ '--max-time', '2' @@ -300,3 +326,13 @@ class TestWebsockets: assert r.profile, f'{r}' rss2 = r.profile.stats['rss'] / (1024 * 1024) assert (rss1 * 1.1) >= rss2, 'bad memory increase' + + # test frame delivery when pausing + def test_20_12_pause_frames(self, env: Env, ws_4frames): + payload = 127 * "x" + client = LocalClient(env=env, name='cli_ws_pause') + if not client.exists(): + pytest.skip(f'example client not built: {client.name}') + url = f'ws://localhost:{ws_4frames.port}/' + r = client.run(args=[url, payload]) + r.check_exit_code(0) diff --git a/tests/http/testenv/env.py b/tests/http/testenv/env.py index 78ab3abec9..8f619271a2 100644 --- a/tests/http/testenv/env.py +++ b/tests/http/testenv/env.py @@ -862,10 +862,6 @@ class Env: def vsftpd(self) -> str: return self.CONFIG.vsftpd - @property - def ws_port(self) -> int: - return self.CONFIG.ports["ws"] - @property def curl(self) -> str: return self.CONFIG.curl diff --git a/tests/http/testenv/ws_4frames_server.py b/tests/http/testenv/ws_4frames_server.py new file mode 100755 index 0000000000..8d5655aa5c --- /dev/null +++ b/tests/http/testenv/ws_4frames_server.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +#*************************************************************************** +# _ _ ____ _ +# Project ___| | | | _ \| | +# / __| | | | |_) | | +# | (__| |_| | _ <| |___ +# \___|\___/|_| \_\_____| +# +# Copyright (C) Daniel Stenberg, , et al. +# +# This software is licensed as described in the file COPYING, which +# you should have received as part of this distribution. The terms +# are also available at https://curl.se/docs/copyright.html. +# +# You may opt to use, copy, modify, merge, publish, distribute and/or sell +# copies of the Software, and permit persons to whom the Software is +# furnished to do so, under the terms of the COPYING file. +# +# This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY +# KIND, either express or implied. +# +# SPDX-License-Identifier: curl +# +########################################################################### +# +import argparse +import asyncio +import logging + +import websockets + +MESSAGES = [ + "Hello 1", + "Hello 2", + "Hello 3", + "Hello 4", +] + + +async def handler(websocket): + peer = websocket.remote_address + print(f"client from {peer[0]}:{peer[1]}", flush=True) + print("handshake complete", flush=True) + + await asyncio.sleep(0.1) + for index, payload in enumerate(MESSAGES, start=1): + await websocket.send(payload) + print(f"sent frame {index}: {payload!r}", flush=True) + # await asyncio.sleep(0.2) + + # await asyncio.sleep(2.0) + print("server done", flush=True) + + +async def main(): + parser = argparse.ArgumentParser(prog='scorecard', description=""" + Run a websocket 4frames server. + """) + parser.add_argument("--port", type=int, + default=9876, help="port to listen on") + args = parser.parse_args() + + logging.basicConfig( + format="%(asctime)s %(message)s", + level=logging.DEBUG, + ) + + print(f"listening on ws://localhost:{args.port}", flush=True) + async with websockets.serve(handler, 'localhost', args.port): + await asyncio.Future() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/libtest/Makefile.inc b/tests/libtest/Makefile.inc index 64508d1285..140ec0bd58 100644 --- a/tests/libtest/Makefile.inc +++ b/tests/libtest/Makefile.inc @@ -63,6 +63,7 @@ TESTS_C = \ cli_tls_session_reuse.c \ cli_upload_pausing.c \ cli_ws_data.c \ + cli_ws_pause.c \ cli_ws_pingpong.c \ \ lib500.c lib501.c lib502.c lib503.c lib504.c lib505.c lib506.c lib507.c \ diff --git a/tests/libtest/cli_ws_pause.c b/tests/libtest/cli_ws_pause.c new file mode 100644 index 0000000000..b423314003 --- /dev/null +++ b/tests/libtest/cli_ws_pause.c @@ -0,0 +1,171 @@ +/*************************************************************************** + * _ _ ____ _ + * Project ___| | | | _ \| | + * / __| | | | |_) | | + * | (__| |_| | _ <| |___ + * \___|\___/|_| \_\_____| + * + * Copyright (C) Daniel Stenberg, , et al. + * + * This software is licensed as described in the file COPYING, which + * you should have received as part of this distribution. The terms + * are also available at https://curl.se/docs/copyright.html. + * + * You may opt to use, copy, modify, merge, publish, distribute and/or sell + * copies of the Software, and permit persons to whom the Software is + * furnished to do so, under the terms of the COPYING file. + * + * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY + * KIND, either express or implied. + * + * SPDX-License-Identifier: curl + * + ***************************************************************************/ +#include "first.h" + +#include "testtrace.h" + +#ifndef CURL_DISABLE_WEBSOCKETS + +struct test_ws_pause_ctx { + CURL *easy; + int callback_count; + int paused; + int frames; + int errors; + int closed; +}; + +static size_t test_ws_pause_write_cb(char *ptr, size_t size, size_t nmemb, + void *userdata) +{ + struct test_ws_pause_ctx *ctx = userdata; + size_t nbytes = size * nmemb; + const struct curl_ws_frame *meta = curl_ws_meta(ctx->easy); + + ctx->callback_count++; + if(!meta) { + curl_mfprintf(stderr, "write_cb: ERROR call #%d with meta=NULL\n", + ctx->callback_count); + ++ctx->errors; + return CURL_WRITEFUNC_ERROR; + } + + ++ctx->frames; + if(meta->len < nbytes) { + curl_mfprintf(stderr, "write_cb: ERROR call #%d more data than current " + "frame, FRAME[flags=0x%x age=%d offset=%" FMT_OFF_T + " bytesleft=%" FMT_OFF_T " len=%zu], bytes=%zu\n", + ctx->callback_count, + (unsigned int)meta->flags, meta->age, + meta->offset, meta->bytesleft, meta->len, nbytes); + ++ctx->errors; + return CURL_WRITEFUNC_ERROR; + } + + if(meta->flags == 0x1) { /* TEXT frame */ + curl_mfprintf(stderr, "write_cb: call #%d FRAME[TEXT age=%d offset=%" + FMT_OFF_T " bytesleft=%" FMT_OFF_T " len=%zu] '%.*s'\n", + ctx->callback_count, meta->age, + meta->offset, meta->bytesleft, meta->len, + (int)nbytes, ptr); + } + else if(meta->flags == 0x8) { + curl_mfprintf(stderr, "write_cb: call #%d FRAME[CLOSE age=%d offset=%" + FMT_OFF_T " bytesleft=%" FMT_OFF_T " len=%zu] bytes=%zu\n", + ctx->callback_count, meta->age, + meta->offset, meta->bytesleft, meta->len, nbytes); + ctx->closed = TRUE; + } + else { + curl_mfprintf(stderr, "write_cb: call #%d FRAME[flags=0x%x age=%d offset=%" + FMT_OFF_T " bytesleft=%" FMT_OFF_T " len=%zu\n", + ctx->callback_count, + (unsigned int)meta->flags, meta->age, + meta->offset, meta->bytesleft, meta->len); + } + + if(ctx->callback_count == 1 || ctx->callback_count == 3) { + ctx->paused = 1; + curl_mfprintf(stderr, "write_cb: call #%d PAUSING\n", ctx->callback_count); + return CURL_WRITEFUNC_PAUSE; + } + return nbytes; +} +#endif /* CURL_DISABLE_WEBSOCKETS */ + +static CURLcode test_cli_ws_pause(const char *URL) +{ +#ifndef CURL_DISABLE_WEBSOCKETS + struct test_ws_pause_ctx ctx; + CURLM *multi; + int still_running = 0; + int msgs_left = 0; + int done = 0; + + memset(&ctx, 0, sizeof(ctx)); + setbuf(stdout, NULL); + + curl_global_init(CURL_GLOBAL_ALL); + + ctx.easy = curl_easy_init(); + multi = curl_multi_init(); + if(!ctx.easy || !multi) { + curl_mfprintf(stderr, "main: ERROR creating easy/multi\n"); + ctx.errors = 1; + goto out; + } + + curl_easy_setopt(ctx.easy, CURLOPT_URL, URL); + curl_easy_setopt(ctx.easy, CURLOPT_WRITEFUNCTION, test_ws_pause_write_cb); + curl_easy_setopt(ctx.easy, CURLOPT_WRITEDATA, &ctx); + curl_easy_setopt(ctx.easy, CURLOPT_VERBOSE, 1L); + + curl_multi_add_handle(multi, ctx.easy); + curl_multi_perform(multi, &still_running); + + while(still_running && !ctx.closed && !ctx.errors) { + + if(ctx.paused) { + curl_mfprintf(stderr, "main: wait and UNPAUSE\n"); + curlx_wait_ms(500); + ctx.paused = 0; + curl_easy_pause(ctx.easy, CURLPAUSE_CONT); + } + + curl_mfprintf(stderr, "main: poll\n"); + curl_multi_poll(multi, NULL, 0, 100, NULL); + curl_mfprintf(stderr, "main: perform\n"); + curl_multi_perform(multi, &still_running); + + while(!done) { + CURLMsg *msg = curl_multi_info_read(multi, &msgs_left); + if(!msg) + break; + if(msg->msg == CURLMSG_DONE) { + curl_mfprintf(stderr, "main: done result=%d (%s)\n", + (int)msg->data.result, + curl_easy_strerror(msg->data.result)); + done = 1; + } + } + } + +out: + if(ctx.easy) { + if(multi) + curl_multi_remove_handle(multi, ctx.easy); + curl_easy_cleanup(ctx.easy); + } + if(multi) + curl_multi_cleanup(multi); + curl_global_cleanup(); + + return ctx.errors ? CURLE_WRITE_ERROR : CURLE_OK; + +#else /* !CURL_DISABLE_WEBSOCKETS */ + (void)URL; + curl_mfprintf(stderr, "WebSockets not enabled in libcurl\n"); + return (CURLcode)1; +#endif /* CURL_DISABLE_WEBSOCKETS */ +}