mirror of
https://github.com/curl/curl.git
synced 2026-08-26 12:33:38 +03:00
websocket: support CURLOPT_READFUNCTION
Add support for CURLOPT_READFUNCTION with WebSocket urls when *not* in connect-only mode, e.g. when using curl_multi_perform. Install the callback function and set CURLOPT_UPLOAD. Return CURL_READFUNC_PAUSE when having nothing more to send and unpause the transfer when more data is ready. This will send the read bytes in a WebSocket BINARY frame. Add support for this mode in the pytest "ws_data" client and have all tests run in 'curl_ws_send/recv' and 'peform' mode as well. Add `curl_ws_start_frame()`. Document, cover in libcurl-ws.md and explain the READFUNCTION mode for websockets. Add example `websocket-updown` for this. Closes #17683
This commit is contained in:
Stefan Eissing
Daniel Stenberg
parent
756c0718c2
commit
37cecfc7b9
17 changed files with 878 additions and 156 deletions
|
|
@ -1822,7 +1822,9 @@ void Curl_http_method(struct Curl_easy *data, struct connectdata *conn,
|
|||
{
|
||||
Curl_HttpReq httpreq = (Curl_HttpReq)data->state.httpreq;
|
||||
const char *request;
|
||||
if((conn->handler->protocol&(PROTO_FAMILY_HTTP|CURLPROTO_FTP)) &&
|
||||
if(conn->handler->protocol&(CURLPROTO_WS|CURLPROTO_WSS))
|
||||
httpreq = HTTPREQ_GET;
|
||||
else if((conn->handler->protocol&(PROTO_FAMILY_HTTP|CURLPROTO_FTP)) &&
|
||||
data->state.upload)
|
||||
httpreq = HTTPREQ_PUT;
|
||||
|
||||
|
|
@ -3771,9 +3773,6 @@ static CURLcode http_on_response(struct Curl_easy *data,
|
|||
if(result)
|
||||
goto out;
|
||||
*pconsumed += blen; /* ws accept handled the data */
|
||||
k->header = FALSE; /* we will not get more responses */
|
||||
if(data->set.connect_only)
|
||||
k->keepon &= ~KEEP_RECV; /* read no more content */
|
||||
}
|
||||
#endif
|
||||
else {
|
||||
|
|
|
|||
|
|
@ -96,3 +96,4 @@ curl_version_info
|
|||
curl_ws_meta
|
||||
curl_ws_recv
|
||||
curl_ws_send
|
||||
curl_ws_start_frame
|
||||
|
|
|
|||
448
lib/ws.c
448
lib/ws.c
|
|
@ -116,7 +116,7 @@ struct websocket {
|
|||
struct ws_encoder enc; /* decode of we frames */
|
||||
struct bufq recvbuf; /* raw data from the server */
|
||||
struct bufq sendbuf; /* raw data to be sent to the server */
|
||||
struct curl_ws_frame frame; /* the current WS FRAME received */
|
||||
struct curl_ws_frame recvframe; /* the current WS FRAME received */
|
||||
size_t sendbuf_payload; /* number of payload bytes in sendbuf */
|
||||
};
|
||||
|
||||
|
|
@ -220,62 +220,66 @@ static int ws_frame_firstbyte2flags(struct Curl_easy *data,
|
|||
}
|
||||
}
|
||||
|
||||
static unsigned char ws_frame_flags2firstbyte(struct Curl_easy *data,
|
||||
unsigned int flags,
|
||||
bool contfragment,
|
||||
CURLcode *err)
|
||||
static CURLcode ws_frame_flags2firstbyte(struct Curl_easy *data,
|
||||
unsigned int flags,
|
||||
bool contfragment,
|
||||
unsigned char *pfirstbyte)
|
||||
{
|
||||
*pfirstbyte = 0;
|
||||
switch(flags & ~CURLWS_OFFSET) {
|
||||
case 0:
|
||||
if(contfragment) {
|
||||
infof(data, "[WS] no flags given; interpreting as continuation "
|
||||
"fragment for compatibility");
|
||||
return (WSBIT_OPCODE_CONT | WSBIT_FIN);
|
||||
*pfirstbyte = (WSBIT_OPCODE_CONT | WSBIT_FIN);
|
||||
return CURLE_OK;
|
||||
}
|
||||
failf(data, "[WS] no flags given");
|
||||
*err = CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
return 0xff;
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
case CURLWS_CONT:
|
||||
if(contfragment) {
|
||||
infof(data, "[WS] setting CURLWS_CONT flag without message type is "
|
||||
"supported for compatibility but highly discouraged");
|
||||
return WSBIT_OPCODE_CONT;
|
||||
*pfirstbyte = WSBIT_OPCODE_CONT;
|
||||
return CURLE_OK;
|
||||
}
|
||||
failf(data, "[WS] No ongoing fragmented message to continue");
|
||||
*err = CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
return 0xff;
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
case CURLWS_TEXT:
|
||||
return contfragment ? (WSBIT_OPCODE_CONT | WSBIT_FIN)
|
||||
: (WSBIT_OPCODE_TEXT | WSBIT_FIN);
|
||||
*pfirstbyte = contfragment ? (WSBIT_OPCODE_CONT | WSBIT_FIN)
|
||||
: (WSBIT_OPCODE_TEXT | WSBIT_FIN);
|
||||
return CURLE_OK;
|
||||
case (CURLWS_TEXT | CURLWS_CONT):
|
||||
return contfragment ? WSBIT_OPCODE_CONT : WSBIT_OPCODE_TEXT;
|
||||
*pfirstbyte = contfragment ? WSBIT_OPCODE_CONT : WSBIT_OPCODE_TEXT;
|
||||
return CURLE_OK;
|
||||
case CURLWS_BINARY:
|
||||
return contfragment ? (WSBIT_OPCODE_CONT | WSBIT_FIN)
|
||||
: (WSBIT_OPCODE_BIN | WSBIT_FIN);
|
||||
*pfirstbyte = contfragment ? (WSBIT_OPCODE_CONT | WSBIT_FIN)
|
||||
: (WSBIT_OPCODE_BIN | WSBIT_FIN);
|
||||
return CURLE_OK;
|
||||
case (CURLWS_BINARY | CURLWS_CONT):
|
||||
return contfragment ? WSBIT_OPCODE_CONT : WSBIT_OPCODE_BIN;
|
||||
*pfirstbyte = contfragment ? WSBIT_OPCODE_CONT : WSBIT_OPCODE_BIN;
|
||||
return CURLE_OK;
|
||||
case CURLWS_CLOSE:
|
||||
return WSBIT_OPCODE_CLOSE | WSBIT_FIN;
|
||||
*pfirstbyte = WSBIT_OPCODE_CLOSE | WSBIT_FIN;
|
||||
return CURLE_OK;
|
||||
case (CURLWS_CLOSE | CURLWS_CONT):
|
||||
failf(data, "[WS] CLOSE frame must not be fragmented");
|
||||
*err = CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
return 0xff;
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
case CURLWS_PING:
|
||||
return WSBIT_OPCODE_PING | WSBIT_FIN;
|
||||
*pfirstbyte = WSBIT_OPCODE_PING | WSBIT_FIN;
|
||||
return CURLE_OK;
|
||||
case (CURLWS_PING | CURLWS_CONT):
|
||||
failf(data, "[WS] PING frame must not be fragmented");
|
||||
*err = CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
return 0xff;
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
case CURLWS_PONG:
|
||||
return WSBIT_OPCODE_PONG | WSBIT_FIN;
|
||||
*pfirstbyte = WSBIT_OPCODE_PONG | WSBIT_FIN;
|
||||
return CURLE_OK;
|
||||
case (CURLWS_PONG | CURLWS_CONT):
|
||||
failf(data, "[WS] PONG frame must not be fragmented");
|
||||
*err = CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
return 0xff;
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
default:
|
||||
failf(data, "[WS] unknown flags: %x", flags);
|
||||
*err = CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
return 0xff;
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -518,7 +522,7 @@ static CURLcode ws_dec_pass(struct ws_decoder *dec,
|
|||
result = ws_dec_read_head(dec, data, inraw);
|
||||
if(result) {
|
||||
if(result != CURLE_AGAIN) {
|
||||
infof(data, "[WS] decode error %d", (int)result);
|
||||
failf(data, "[WS] decode frame error %d", (int)result);
|
||||
break; /* real error */
|
||||
}
|
||||
/* incomplete ws frame head */
|
||||
|
|
@ -562,11 +566,11 @@ static void update_meta(struct websocket *ws,
|
|||
{
|
||||
curl_off_t bytesleft = (payload_len - payload_offset - cur_len);
|
||||
|
||||
ws->frame.age = frame_age;
|
||||
ws->frame.flags = frame_flags;
|
||||
ws->frame.offset = payload_offset;
|
||||
ws->frame.len = cur_len;
|
||||
ws->frame.bytesleft = bytesleft;
|
||||
ws->recvframe.age = frame_age;
|
||||
ws->recvframe.flags = frame_flags;
|
||||
ws->recvframe.offset = payload_offset;
|
||||
ws->recvframe.len = cur_len;
|
||||
ws->recvframe.bytesleft = bytesleft;
|
||||
}
|
||||
|
||||
/* WebSockets decoding client writer */
|
||||
|
|
@ -644,6 +648,7 @@ static CURLcode ws_cw_write(struct Curl_easy *data,
|
|||
struct websocket *ws;
|
||||
CURLcode result;
|
||||
|
||||
CURL_TRC_WRITE(data, "ws_cw_write(len=%zu, type=%d)", nbytes, type);
|
||||
if(!(type & CLIENTWRITE_BODY) || data->set.ws_raw_mode)
|
||||
return Curl_cwriter_write(data, writer->next, type, buf, nbytes);
|
||||
|
||||
|
|
@ -674,11 +679,10 @@ static CURLcode ws_cw_write(struct Curl_easy *data,
|
|||
if(result == CURLE_AGAIN) {
|
||||
/* insufficient amount of data, keep it for later.
|
||||
* we pretend to have written all since we have a copy */
|
||||
CURL_TRC_WS(data, "buffered incomplete frame head");
|
||||
return CURLE_OK;
|
||||
}
|
||||
else if(result) {
|
||||
infof(data, "[WS] decode error %d", (int)result);
|
||||
failf(data, "[WS] decode payload error %d", (int)result);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
|
@ -748,36 +752,33 @@ static void ws_enc_init(struct ws_encoder *enc)
|
|||
+---------------------------------------------------------------+
|
||||
*/
|
||||
|
||||
static ssize_t ws_enc_write_head(struct Curl_easy *data,
|
||||
static CURLcode ws_enc_write_head(struct Curl_easy *data,
|
||||
struct ws_encoder *enc,
|
||||
unsigned int flags,
|
||||
curl_off_t payload_len,
|
||||
struct bufq *out,
|
||||
CURLcode *err)
|
||||
struct bufq *out)
|
||||
{
|
||||
unsigned char firstbyte = 0;
|
||||
unsigned char firstb = 0;
|
||||
unsigned char head[14];
|
||||
size_t hlen, n;
|
||||
CURLcode result;
|
||||
size_t hlen, nwritten;
|
||||
|
||||
if(payload_len < 0) {
|
||||
failf(data, "[WS] starting new frame with negative payload length %"
|
||||
FMT_OFF_T, payload_len);
|
||||
*err = CURLE_SEND_ERROR;
|
||||
return -1;
|
||||
return CURLE_SEND_ERROR;
|
||||
}
|
||||
|
||||
if(enc->payload_remain > 0) {
|
||||
/* trying to write a new frame before the previous one is finished */
|
||||
failf(data, "[WS] starting new frame with %zd bytes from last one "
|
||||
"remaining to be sent", (ssize_t)enc->payload_remain);
|
||||
*err = CURLE_SEND_ERROR;
|
||||
return -1;
|
||||
return CURLE_SEND_ERROR;
|
||||
}
|
||||
|
||||
firstbyte = ws_frame_flags2firstbyte(data, flags, enc->contfragment, err);
|
||||
if(*err) {
|
||||
return -1;
|
||||
}
|
||||
result = ws_frame_flags2firstbyte(data, flags, enc->contfragment, &firstb);
|
||||
if(result)
|
||||
return result;
|
||||
|
||||
/* fragmentation only applies to data frames (text/binary);
|
||||
* control frames (close/ping/pong) do not affect the CONT status */
|
||||
|
|
@ -788,23 +789,20 @@ static ssize_t ws_enc_write_head(struct Curl_easy *data,
|
|||
if(flags & CURLWS_PING && payload_len > 125) {
|
||||
/* The maximum valid size of PING frames is 125 bytes. */
|
||||
failf(data, "[WS] given PING frame is too big");
|
||||
*err = CURLE_TOO_LARGE;
|
||||
return -1;
|
||||
return CURLE_TOO_LARGE;
|
||||
}
|
||||
if(flags & CURLWS_PONG && payload_len > 125) {
|
||||
/* The maximum valid size of PONG frames is 125 bytes. */
|
||||
failf(data, "[WS] given PONG frame is too big");
|
||||
*err = CURLE_TOO_LARGE;
|
||||
return -1;
|
||||
return CURLE_TOO_LARGE;
|
||||
}
|
||||
if(flags & CURLWS_CLOSE && payload_len > 125) {
|
||||
/* The maximum valid size of CLOSE frames is 125 bytes. */
|
||||
failf(data, "[WS] given CLOSE frame is too big");
|
||||
*err = CURLE_TOO_LARGE;
|
||||
return -1;
|
||||
return CURLE_TOO_LARGE;
|
||||
}
|
||||
|
||||
head[0] = enc->firstbyte = firstbyte;
|
||||
head[0] = enc->firstbyte = firstb;
|
||||
if(payload_len > 65535) {
|
||||
head[1] = 127 | WSBIT_MASK;
|
||||
head[2] = (unsigned char)((payload_len >> 56) & 0xff);
|
||||
|
|
@ -837,29 +835,28 @@ static ssize_t ws_enc_write_head(struct Curl_easy *data,
|
|||
/* reset for payload to come */
|
||||
enc->xori = 0;
|
||||
|
||||
*err = Curl_bufq_write(out, head, hlen, &n);
|
||||
if(*err)
|
||||
return -1;
|
||||
if(n != hlen) {
|
||||
result = Curl_bufq_write(out, head, hlen, &nwritten);
|
||||
if(result)
|
||||
return result;
|
||||
if(nwritten != hlen) {
|
||||
/* We use a bufq with SOFT_LIMIT, writing should always succeed */
|
||||
DEBUGASSERT(0);
|
||||
*err = CURLE_SEND_ERROR;
|
||||
return -1;
|
||||
return CURLE_SEND_ERROR;
|
||||
}
|
||||
return (ssize_t)n;
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
static ssize_t ws_enc_write_payload(struct ws_encoder *enc,
|
||||
struct Curl_easy *data,
|
||||
const unsigned char *buf, size_t buflen,
|
||||
struct bufq *out, CURLcode *err)
|
||||
static CURLcode ws_enc_write_payload(struct ws_encoder *enc,
|
||||
struct Curl_easy *data,
|
||||
const unsigned char *buf, size_t buflen,
|
||||
struct bufq *out, size_t *pnwritten)
|
||||
{
|
||||
CURLcode result;
|
||||
size_t i, len, n;
|
||||
|
||||
if(Curl_bufq_is_full(out)) {
|
||||
*err = CURLE_AGAIN;
|
||||
return -1;
|
||||
}
|
||||
*pnwritten = 0;
|
||||
if(Curl_bufq_is_full(out))
|
||||
return CURLE_AGAIN;
|
||||
|
||||
/* not the most performant way to do this */
|
||||
len = buflen;
|
||||
|
|
@ -868,21 +865,139 @@ static ssize_t ws_enc_write_payload(struct ws_encoder *enc,
|
|||
|
||||
for(i = 0; i < len; ++i) {
|
||||
unsigned char c = buf[i] ^ enc->mask[enc->xori];
|
||||
*err = Curl_bufq_write(out, &c, 1, &n);
|
||||
if(*err) {
|
||||
if((*err != CURLE_AGAIN) || !i)
|
||||
return -1;
|
||||
result = Curl_bufq_write(out, &c, 1, &n);
|
||||
if(result) {
|
||||
if((result != CURLE_AGAIN) || !i)
|
||||
return result;
|
||||
break;
|
||||
}
|
||||
enc->xori++;
|
||||
enc->xori &= 3;
|
||||
}
|
||||
*pnwritten = i;
|
||||
enc->payload_remain -= (curl_off_t)i;
|
||||
ws_enc_info(enc, data, "buffered");
|
||||
return (ssize_t)i;
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
|
||||
|
||||
struct cr_ws_ctx {
|
||||
struct Curl_creader super;
|
||||
BIT(read_eos); /* we read an EOS from the next reader */
|
||||
BIT(eos); /* we have returned an EOS */
|
||||
};
|
||||
|
||||
static CURLcode cr_ws_init(struct Curl_easy *data, struct Curl_creader *reader)
|
||||
{
|
||||
(void)data;
|
||||
(void)reader;
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
static void cr_ws_close(struct Curl_easy *data, struct Curl_creader *reader)
|
||||
{
|
||||
(void)data;
|
||||
(void)reader;
|
||||
}
|
||||
|
||||
static CURLcode cr_ws_read(struct Curl_easy *data,
|
||||
struct Curl_creader *reader,
|
||||
char *buf, size_t blen,
|
||||
size_t *pnread, bool *peos)
|
||||
{
|
||||
struct cr_ws_ctx *ctx = reader->ctx;
|
||||
CURLcode result = CURLE_OK;
|
||||
size_t nread, n;
|
||||
struct websocket *ws;
|
||||
bool eos;
|
||||
|
||||
*pnread = 0;
|
||||
if(ctx->eos) {
|
||||
*peos = TRUE;
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
if(Curl_bufq_is_empty(&ws->sendbuf)) {
|
||||
if(ctx->read_eos) {
|
||||
ctx->eos = TRUE;
|
||||
*peos = TRUE;
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
if(ws->enc.payload_remain) {
|
||||
CURL_TRC_WS(data, "current frame, %" FMT_OFF_T " remaining",
|
||||
ws->enc.payload_remain);
|
||||
if(ws->enc.payload_remain < (curl_off_t)blen)
|
||||
blen = (size_t)ws->enc.payload_remain;
|
||||
}
|
||||
|
||||
result = Curl_creader_read(data, reader->next, buf, blen, &nread, &eos);
|
||||
if(result)
|
||||
return result;
|
||||
ctx->read_eos = eos;
|
||||
|
||||
if(!nread) {
|
||||
/* nothing to convert, return this right away */
|
||||
if(ctx->read_eos)
|
||||
ctx->eos = TRUE;
|
||||
*pnread = nread;
|
||||
*peos = ctx->eos;
|
||||
goto out;
|
||||
}
|
||||
|
||||
if(!ws->enc.payload_remain) {
|
||||
/* encode the data as a new BINARY frame */
|
||||
result = ws_enc_write_head(data, &ws->enc, CURLWS_BINARY, nread,
|
||||
&ws->sendbuf);
|
||||
if(result)
|
||||
goto out;
|
||||
}
|
||||
|
||||
result = ws_enc_write_payload(&ws->enc, data, (unsigned char *)buf,
|
||||
nread, &ws->sendbuf, &n);
|
||||
if(result)
|
||||
goto out;
|
||||
CURL_TRC_READ(data, "cr_ws_read, added %zu payload, len=%zu", nread, n);
|
||||
}
|
||||
|
||||
DEBUGASSERT(!Curl_bufq_is_empty(&ws->sendbuf));
|
||||
*peos = FALSE;
|
||||
result = Curl_bufq_cread(&ws->sendbuf, buf, blen, pnread);
|
||||
if(!result && ctx->read_eos && Curl_bufq_is_empty(&ws->sendbuf)) {
|
||||
/* no more data, read all, done. */
|
||||
ctx->eos = TRUE;
|
||||
*peos = TRUE;
|
||||
}
|
||||
|
||||
out:
|
||||
CURL_TRC_READ(data, "cr_ws_read(len=%zu) -> %d, nread=%zu, eos=%d",
|
||||
blen, result, *pnread, *peos);
|
||||
return result;
|
||||
}
|
||||
|
||||
static const struct Curl_crtype ws_cr_encode = {
|
||||
"ws-encode",
|
||||
cr_ws_init,
|
||||
cr_ws_read,
|
||||
cr_ws_close,
|
||||
Curl_creader_def_needs_rewind,
|
||||
Curl_creader_def_total_length,
|
||||
Curl_creader_def_resume_from,
|
||||
Curl_creader_def_rewind,
|
||||
Curl_creader_def_unpause,
|
||||
Curl_creader_def_is_paused,
|
||||
Curl_creader_def_done,
|
||||
sizeof(struct cr_ws_ctx)
|
||||
};
|
||||
|
||||
|
||||
struct wsfield {
|
||||
const char *name;
|
||||
const char *val;
|
||||
|
|
@ -968,7 +1083,8 @@ CURLcode Curl_ws_accept(struct Curl_easy *data,
|
|||
{
|
||||
struct SingleRequest *k = &data->req;
|
||||
struct websocket *ws;
|
||||
struct Curl_cwriter *ws_dec_writer;
|
||||
struct Curl_cwriter *ws_dec_writer = NULL;
|
||||
struct Curl_creader *ws_enc_reader = NULL;
|
||||
CURLcode result;
|
||||
|
||||
DEBUGASSERT(data->conn);
|
||||
|
|
@ -1044,13 +1160,11 @@ CURLcode Curl_ws_accept(struct Curl_easy *data,
|
|||
result = Curl_cwriter_create(&ws_dec_writer, data, &ws_cw_decode,
|
||||
CURL_CW_CONTENT_DECODE);
|
||||
if(result)
|
||||
return result;
|
||||
|
||||
goto out;
|
||||
result = Curl_cwriter_add(data, ws_dec_writer);
|
||||
if(result) {
|
||||
Curl_cwriter_free(data, ws_dec_writer);
|
||||
return result;
|
||||
}
|
||||
if(result)
|
||||
goto out;
|
||||
ws_dec_writer = NULL; /* owned by transfer now */
|
||||
|
||||
if(data->set.connect_only) {
|
||||
size_t nwritten;
|
||||
|
|
@ -1060,18 +1174,55 @@ CURLcode Curl_ws_accept(struct Curl_easy *data,
|
|||
result = Curl_bufq_write(&ws->recvbuf, (const unsigned char *)mem,
|
||||
nread, &nwritten);
|
||||
if(result)
|
||||
return result;
|
||||
goto out;
|
||||
DEBUGASSERT(nread == nwritten);
|
||||
infof(data, "%zu bytes websocket payload", nread);
|
||||
k->keepon &= ~KEEP_RECV; /* read no more content */
|
||||
}
|
||||
else { /* !connect_only */
|
||||
if(data->set.method == HTTPREQ_PUT) {
|
||||
CURL_TRC_WS(data, "UPLOAD set, add ws-encode reader");
|
||||
result = Curl_creader_set_fread(data, -1);
|
||||
if(result)
|
||||
goto out;
|
||||
|
||||
if(!data->set.ws_raw_mode) {
|
||||
/* Add our client readerr encoding WS BINARY frames */
|
||||
result = Curl_creader_create(&ws_enc_reader, data, &ws_cr_encode,
|
||||
CURL_CR_CONTENT_ENCODE);
|
||||
if(result)
|
||||
goto out;
|
||||
result = Curl_creader_add(data, ws_enc_reader);
|
||||
if(result)
|
||||
goto out;
|
||||
ws_enc_reader = NULL; /* owned by transfer now */
|
||||
}
|
||||
|
||||
/* start over with sending */
|
||||
data->req.eos_read = FALSE;
|
||||
k->keepon |= KEEP_SEND;
|
||||
}
|
||||
|
||||
/* And pass any additional data to the writers */
|
||||
if(nread) {
|
||||
result = Curl_client_write(data, CLIENTWRITE_BODY, mem, nread);
|
||||
if(result)
|
||||
goto out;
|
||||
}
|
||||
}
|
||||
k->upgr101 = UPGR101_RECEIVED;
|
||||
|
||||
k->upgr101 = UPGR101_RECEIVED;
|
||||
k->header = FALSE; /* we will not get more responses */
|
||||
|
||||
out:
|
||||
if(ws_dec_writer)
|
||||
Curl_cwriter_free(data, ws_dec_writer);
|
||||
if(ws_enc_reader)
|
||||
Curl_creader_free(data, ws_enc_reader);
|
||||
if(result)
|
||||
CURL_TRC_WS(data, "Curl_ws_accept() failed -> %d", result);
|
||||
else
|
||||
CURL_TRC_WS(data, "websocket established, %s mode",
|
||||
data->set.connect_only ? "connect-only" : "callback");
|
||||
return result;
|
||||
}
|
||||
|
||||
|
|
@ -1226,11 +1377,12 @@ CURLcode curl_ws_recv(CURL *d, void *buffer,
|
|||
/* update frame information to be passed back */
|
||||
update_meta(ws, ctx.frame_age, ctx.frame_flags, ctx.payload_offset,
|
||||
ctx.payload_len, ctx.bufidx);
|
||||
*metap = &ws->frame;
|
||||
*nread = ws->frame.len;
|
||||
*metap = &ws->recvframe;
|
||||
*nread = ws->recvframe.len;
|
||||
CURL_TRC_WS(data, "curl_ws_recv(len=%zu) -> %zu bytes (frame at %"
|
||||
FMT_OFF_T ", %" FMT_OFF_T " left)",
|
||||
buflen, *nread, ws->frame.offset, ws->frame.bytesleft);
|
||||
buflen, *nread, ws->recvframe.offset,
|
||||
ws->recvframe.bytesleft);
|
||||
return CURLE_OK;
|
||||
}
|
||||
|
||||
|
|
@ -1378,7 +1530,7 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
{
|
||||
struct websocket *ws;
|
||||
const unsigned char *buffer = buffer_arg;
|
||||
ssize_t n;
|
||||
size_t n;
|
||||
CURLcode result = CURLE_OK;
|
||||
struct Curl_easy *data = d;
|
||||
|
||||
|
|
@ -1387,7 +1539,16 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
CURL_TRC_WS(data, "curl_ws_send(len=%zu, fragsize=%" FMT_OFF_T
|
||||
", flags=%x), raw=%d",
|
||||
buflen, fragsize, flags, data->set.ws_raw_mode);
|
||||
*sent = 0;
|
||||
|
||||
if(sent)
|
||||
*sent = 0;
|
||||
|
||||
if(!buffer && buflen) {
|
||||
failf(data, "[WS] buffer is NULL when buflen is not");
|
||||
result = CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
goto out;
|
||||
}
|
||||
|
||||
if(!data->conn && data->set.connect_only) {
|
||||
result = Curl_connect_only_attach(data);
|
||||
if(result)
|
||||
|
|
@ -1412,6 +1573,14 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
if(result)
|
||||
goto out;
|
||||
|
||||
if(!buffer) {
|
||||
failf(data, "[WS] buffer is NULL in raw mode");
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
}
|
||||
if(!sent) {
|
||||
failf(data, "[WS] sent is NULL in raw mode");
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
}
|
||||
if(fragsize || flags) {
|
||||
failf(data, "[WS] fragsize and flags must be zero in raw mode");
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
|
|
@ -1446,16 +1615,18 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
}
|
||||
else {
|
||||
/* starting a new frame, we want a clean sendbuf */
|
||||
curl_off_t payload_len = (flags & CURLWS_OFFSET) ?
|
||||
fragsize : (curl_off_t)buflen;
|
||||
result = ws_flush(data, ws, Curl_is_in_callback(data));
|
||||
if(result)
|
||||
goto out;
|
||||
|
||||
n = ws_enc_write_head(data, &ws->enc, flags, payload_len,
|
||||
&ws->sendbuf, &result);
|
||||
if(n < 0)
|
||||
result = ws_enc_write_head(data, &ws->enc, flags,
|
||||
(flags & CURLWS_OFFSET) ?
|
||||
fragsize : (curl_off_t)buflen,
|
||||
&ws->sendbuf);
|
||||
if(result) {
|
||||
CURL_TRC_WS(data, "curl_ws_send(), error writing frame head %d", result);
|
||||
goto out;
|
||||
}
|
||||
}
|
||||
|
||||
/* While there is either sendbuf to flush OR more payload to encode... */
|
||||
|
|
@ -1463,11 +1634,11 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
/* Try to add more payload to sendbuf */
|
||||
if(buflen > ws->sendbuf_payload) {
|
||||
size_t prev_len = Curl_bufq_len(&ws->sendbuf);
|
||||
n = ws_enc_write_payload(&ws->enc, data,
|
||||
buffer + ws->sendbuf_payload,
|
||||
buflen - ws->sendbuf_payload,
|
||||
&ws->sendbuf, &result);
|
||||
if(n < 0 && (result != CURLE_AGAIN))
|
||||
result = ws_enc_write_payload(&ws->enc, data,
|
||||
buffer + ws->sendbuf_payload,
|
||||
buflen - ws->sendbuf_payload,
|
||||
&ws->sendbuf, &n);
|
||||
if(result && (result != CURLE_AGAIN))
|
||||
goto out;
|
||||
ws->sendbuf_payload += Curl_bufq_len(&ws->sendbuf) - prev_len;
|
||||
if(!ws->sendbuf_payload) {
|
||||
|
|
@ -1479,7 +1650,8 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
/* flush, blocking when in callback */
|
||||
result = ws_flush(data, ws, Curl_is_in_callback(data));
|
||||
if(!result && ws->sendbuf_payload > 0) {
|
||||
*sent += ws->sendbuf_payload;
|
||||
if(sent)
|
||||
*sent += ws->sendbuf_payload;
|
||||
buffer += ws->sendbuf_payload;
|
||||
buflen -= ws->sendbuf_payload;
|
||||
ws->sendbuf_payload = 0;
|
||||
|
|
@ -1489,7 +1661,8 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
/* blocked, part of payload bytes remain, report length
|
||||
* that we managed to send. */
|
||||
size_t flushed = (ws->sendbuf_payload - Curl_bufq_len(&ws->sendbuf));
|
||||
*sent += flushed;
|
||||
if(sent)
|
||||
*sent += flushed;
|
||||
ws->sendbuf_payload -= flushed;
|
||||
result = CURLE_OK;
|
||||
goto out;
|
||||
|
|
@ -1499,7 +1672,7 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
* OK on 0-length send (caller counts only payload) and EAGAIN */
|
||||
CURL_TRC_WS(data, "EAGAIN flushing sendbuf, payload_encoded: %zu/%zu",
|
||||
ws->sendbuf_payload, buflen);
|
||||
DEBUGASSERT(*sent == 0);
|
||||
DEBUGASSERT(!sent || *sent == 0);
|
||||
result = CURLE_AGAIN;
|
||||
goto out;
|
||||
}
|
||||
|
|
@ -1511,7 +1684,8 @@ CURLcode curl_ws_send(CURL *d, const void *buffer_arg,
|
|||
out:
|
||||
CURL_TRC_WS(data, "curl_ws_send(len=%zu, fragsize=%" FMT_OFF_T
|
||||
", flags=%x, raw=%d) -> %d, %zu",
|
||||
buflen, fragsize, flags, data->set.ws_raw_mode, result, *sent);
|
||||
buflen, fragsize, flags, data->set.ws_raw_mode, result,
|
||||
sent ? *sent : 0);
|
||||
return result;
|
||||
}
|
||||
|
||||
|
|
@ -1537,12 +1711,63 @@ const struct curl_ws_frame *curl_ws_meta(CURL *d)
|
|||
struct websocket *ws;
|
||||
ws = Curl_conn_meta_get(data->conn, CURL_META_PROTO_WS_CONN);
|
||||
if(ws)
|
||||
return &ws->frame;
|
||||
return &ws->recvframe;
|
||||
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
CURL_EXTERN CURLcode curl_ws_start_frame(CURL *d,
|
||||
unsigned int flags,
|
||||
curl_off_t frame_len)
|
||||
{
|
||||
struct websocket *ws;
|
||||
CURLcode result = CURLE_OK;
|
||||
struct Curl_easy *data = d;
|
||||
|
||||
if(!GOOD_EASY_HANDLE(data))
|
||||
return CURLE_BAD_FUNCTION_ARGUMENT;
|
||||
if(data->set.ws_raw_mode) {
|
||||
failf(data, "cannot curl_ws_start_frame() with CURLWS_RAW_MODE enabled");
|
||||
return CURLE_FAILED_INIT;
|
||||
}
|
||||
|
||||
CURL_TRC_WS(data, "curl_start_frame(flags=%x, frame_len=%" FMT_OFF_T,
|
||||
flags, frame_len);
|
||||
|
||||
if(!data->conn) {
|
||||
failf(data, "[WS] No associated connection");
|
||||
result = CURLE_SEND_ERROR;
|
||||
goto out;
|
||||
}
|
||||
ws = Curl_conn_meta_get(data->conn, CURL_META_PROTO_WS_CONN);
|
||||
if(!ws) {
|
||||
failf(data, "[WS] Not a websocket transfer");
|
||||
result = CURLE_SEND_ERROR;
|
||||
goto out;
|
||||
}
|
||||
|
||||
if(data->set.ws_raw_mode) {
|
||||
failf(data, "[WS] cannot start frame in raw mode");
|
||||
result = CURLE_SEND_ERROR;
|
||||
goto out;
|
||||
}
|
||||
|
||||
if(ws->enc.payload_remain) {
|
||||
failf(data, "[WS] previous frame not finished");
|
||||
result = CURLE_SEND_ERROR;
|
||||
goto out;
|
||||
}
|
||||
|
||||
result = ws_enc_write_head(data, &ws->enc, flags, frame_len, &ws->sendbuf);
|
||||
if(result)
|
||||
CURL_TRC_WS(data, "curl_start_frame(), error adding frame head %d",
|
||||
result);
|
||||
|
||||
out:
|
||||
return result;
|
||||
}
|
||||
|
||||
const struct Curl_handler Curl_handler_ws = {
|
||||
"WS", /* scheme */
|
||||
ws_setup_conn, /* setup_connection */
|
||||
|
|
@ -1631,4 +1856,15 @@ const struct curl_ws_frame *curl_ws_meta(CURL *data)
|
|||
(void)data;
|
||||
return NULL;
|
||||
}
|
||||
|
||||
CURL_EXTERN CURLcode curl_ws_start_frame(CURL *curl,
|
||||
unsigned int flags,
|
||||
curl_off_t frame_len)
|
||||
{
|
||||
(void)curl;
|
||||
(void)flags;
|
||||
(void)frame_len;
|
||||
return CURLE_NOT_BUILT_IN;
|
||||
}
|
||||
|
||||
#endif /* !CURL_DISABLE_WEBSOCKETS */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue