b940cb1566

lib/content_encoding.c+6 -0

@@ -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)
	};

lib/cw-out.c+33 -26

@@ -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,11 +448,40 @@ static CURLcode cw_out_write(struct Curl_easy *data,
	return CURLE_OK;
	}

+	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;
+
+	if(ctx->errored)
+	return CURLE_WRITE_ERROR;
+	if(ctx->paused)
+	return CURLE_OK; /* not doing it */
+
+	result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all);
+	if(result) {
+	ctx->errored = TRUE;
+	cw_out_bufs_free(ctx);
+	return result;
+	}
+	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)
	};
@@ -468,27 +499,6 @@ bool Curl_cw_out_is_paused(struct Curl_easy *data)
	return (bool)ctx->paused;
	}

-	static CURLcode cw_out_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;
-
-	if(ctx->errored)
-	return CURLE_WRITE_ERROR;
-	if(ctx->paused)
-	return CURLE_OK; /* not doing it */
-
-	result = cw_out_flush_chain(ctx, data, &ctx->buf, flush_all);
-	if(result) {
-	ctx->errored = TRUE;
-	cw_out_bufs_free(ctx);
-	return result;
-	}
-	return result;
-	}
-
	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;
	}

lib/cw-pause.c+4 -14

@@ -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;
-	}

lib/cw-pause.h+0 -2

@@ -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 */

lib/ftp.c+1 -0

@@ -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)
	};

lib/headers.c+1 -0

@@ -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)
	};

lib/http_chunks.c+1 -0

@@ -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)
	};

lib/sendf.c+37 -1

@@ -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,

lib/sendf.h+8 -0

@@ -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);

lib/ws.c+55 -3

@@ -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)
	};

tests/http/Makefile.am+2 -1

@@ -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 \

tests/http/test_20_websockets.py+71 -35

@@ -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)

tests/http/testenv/env.py+0 -4

@@ -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

tests/http/testenv/ws_4frames_server.py+75 -0

@@ -0,0 +1,75 @@
+	#!/usr/bin/env python3
+	# -- coding: utf-8 --
+	#***************************************************************************
+	# _ _ ____ _
+	# Project ___\| \| \| \| _ \\| \|
+	# / __\| \| \| \| \|_) \| \|
+	# \| (__\| \|_\| \| _ <\| \|___
+	# \___\|\___/\|_\| \_\_____\|
+	#
+	# Copyright (C) Daniel Stenberg, <[email protected]>, 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())

tests/libtest/Makefile.inc+1 -0

@@ -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 \

tests/libtest/cli_ws_pause.c+171 -0

@@ -0,0 +1,171 @@
+	/***************************************************************************
+	* _ _ ____ _
+	* Project ___\| \| \| \| _ \\| \|
+	* / __\| \| \| \| \|_) \| \|
+	* \| (__\| \|_\| \| _ <\| \|___
+	* \___\|\___/\|_\| \_\_____\|
+	*
+	* Copyright (C) Daniel Stenberg, <[email protected]>, 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 */
+	}

Subject

Binary

Architecture & Compiler Flags

Build Log

Commit Diff GitHub →