webserver: comparison src/server/proxy/httpclient.c

-:3ddfd45d4e47
+:fea7c3d74cc6
 static int client_connected(EventHandler *ev, Event *event);
 static int client_io(EventHandler *ev, Event *event);
 static int client_finished(EventHandler *ev, Event *event);
-static int client_send_request(HttpClient *client);
+static int client_send_buf(HttpClient *client);
 static int client_send_request_body(HttpClient *client);
 static int client_read_response_header(HttpClient *client);
 static int client_read_response_body(HttpClient *client);
 HttpClient* http_client_new(EventHandler *ev) {
 http_parser_free(client->parser);
 if(client->stream) {
 client->stream->st.free(&client->stream->st);
 }
 free(client->buffer.inbuf);
+free(client->transfer_buffer);
+free(client->transfer2_buffer);
 free(client->addr);
 free(client->method);
 free(client->uri);
 free(client);
 }
 close(socketfd);
 }
 return ret;
 }
+int http_client_process(HttpClient *client) {
+return client_io(client->ev, &client->event);
+}
+size_t http_client_message_buf_size_available(HttpClient *client) {
+return client->transfer_buffer_alloc - client->transfer_buffer_len;
+}
+int http_client_add_message(HttpClient *client, const void *buf, size_t size) {
+size_t available = http_client_message_buf_size_available(client);
+if(available == 0) {
+return HTTP_CLIENT_CALLBACK_WOULD_BLOCK;
+}
+if(size > available) {
+size = available;
+}
+memcpy(client->transfer_buffer + client->transfer_buffer_len, buf, size);
+return size;
+}
 static int create_req_buffer(HttpClient *client) {
 CxBuffer buf;
 if(cxBufferInit(&buf, cxDefaultAllocator, NULL, HTTP_CLIENT_BUFFER_SIZE, CX_BUFFER_AUTO_EXTEND)) {
 return 1;
 }
 static int client_io(EventHandler *ev, Event *event) {
 HttpClient *client = event->cookie;
 if(client->stage == 0) {
 if(client->transfer_buffer_pos < client->transfer_buffer_len) {
-if(client_send_request(client)) {
+if(client_send_buf(client)) {
 return client->error == 0;
 }
 }
 // do we need to send a request body?
 if(client->req_content_length != 0) {
 if(client_send_request_body(client)) {
 return client->error == 0;
 }
 }
-client->transfer_buffer_pos = 0;
-client->transfer_buffer_len = 0;
 }
 // writing complete, switch to read events
 event->events = EVENT_POLLIN;
 client->stage = 1;
 if(client_read_response_header(client)) {
 return client->error == 0;
 }
+int ret = 0;
+if(client->stage == 2) {
+// websocket: write message buffer
+ret = client_send_buf(client);
+}
 if(client_read_response_body(client)) {
+ret = 1;
+}
+if(ret) {
 return client->error == 0;
 }
 return 0;
 }
 }
 return 0;
 }
-static int client_send_request(HttpClient *client) {
+// sends the content of the transfer buffer to client->socketfd
+static int client_send_buf(HttpClient *client) {
 size_t nbytes = client->transfer_buffer_len - client->transfer_buffer_pos;
 ssize_t w;
 while((w = write(client->socketfd, client->transfer_buffer + client->transfer_buffer_pos, nbytes)) > 0) {
 client->transfer_buffer_pos += w;
 nbytes = client->transfer_buffer_len - client->transfer_buffer_pos;
 }
 client->req_contentlength_pos += r;
 client->transfer_buffer_pos = startpos;
 client->transfer_buffer_len = rbody_buf_offset + r;
-if(client_send_request(client)) {
+if(client_send_buf(client)) {
 return 1;
 }
 }
 // chunked transfer encoding: terminate
 if(client->req_content_length == -1 && !client->request_body_terminated) {
 memcpy(client->transfer_buffer, "0\r\n\r\n", 5);
 client->transfer_buffer_pos = 0;
 client->transfer_buffer_len = 5;
 client->request_body_terminated = 1;
-if(client_send_request(client)) {
+if(client_send_buf(client)) {
 return 1;
 }
 } else if(client->req_content_length != client->req_contentlength_pos) {
 // incomplete request body
 // initialize httpstream
 HeaderArray *headers = client->parser->headers;
 long long contentlength = 0;
 int chunkedtransferenc = 0;
+cxmutstr hdr_connection = CX_NULLSTR;
+cxmutstr hdr_upgrade = CX_NULLSTR;
 while(headers) {
 for(int i=0;i<headers->len;i++) {
-if(!cx_strcasecmp(headers->headers[i].name, "content-length")) {
+cxmutstr header = headers->headers[i].name;
-if(!cx_strtoll(headers->headers[i].value, &contentlength, 10)) {
+cxmutstr hvalue = headers->headers[i].value;
+if(!cx_strcasecmp(header, "content-length")) {
+if(!cx_strtoll(hvalue, &contentlength, 10)) {
 headers = NULL;
 break;
 }
-} else if(!cx_strcasecmp(headers->headers[i].name, "transfer-encoding")) {
+} else if(!cx_strcasecmp(header, "transfer-encoding")) {
-if(!cx_strcmp(headers->headers[i].value, "chunked")) {
+if(!cx_strcmp(hvalue, "chunked")) {
 chunkedtransferenc = 1;
 headers = NULL;
 break;
 }
+} else if(!cx_strcasecmp(header, "connection")) {
+hdr_connection = hvalue;
+}
+if(client->statuscode == 101 && !cx_strcasecmp(header, "upgrade")) {
+hdr_upgrade = hvalue;
 }
 }
 if(headers) {
 headers = headers->next;
 }
 }
-if(contentlength > 0 || chunkedtransferenc) {
+if(client->statuscode == 101) {
+if(!cx_strcasecmp(hdr_upgrade, "websocket") && !cx_strcasecmp(hdr_connection, "upgrade")) {
+client->stage = 2;
+client->event.events = EVENT_POLLIN|EVENT_POLLOUT;
+// prepare IO buffers for websockets
+client->transfer_buffer_len = 0;
+client->transfer_buffer_pos = 0;
+client->transfer2_buffer = malloc(HTTP_CLIENT_BUFFER_SIZE);
+if(!client->transfer2_buffer) {
+client->error = 1;
+return 1;
+}
+client->transfer2_buffer_alloc = HTTP_CLIENT_BUFFER_SIZE;
+client->transfer2_buffer_len = 0;
+client->transfer2_buffer_pos = 0;
+} else {
+// error: unknown protocol
+log_ereport(LOG_FAILURE, "http-client: unknown protocol upgrade: %.*s", (int)hdr_upgrade.length, hdr_upgrade.ptr);
+client->error = 1;
+return 1;
+}
+} else if(contentlength > 0 || chunkedtransferenc) {
 IOStream *fd = Sysstream_new(NULL, client->socketfd);
 if(!fd) {
 client->error = 1;
 return 1;
 }
 HttpStream *http = (HttpStream*)httpstream_new(NULL, fd);
 if(!http) {
 fd->free(fd);
 }
+// we can reuse the already allocated transfer_bufer for transfer2
+client->transfer2_buffer = client->transfer_buffer;
+client->transfer2_buffer_alloc = client->transfer_buffer_alloc;
+client->transfer2_buffer_len = 0;
+client->transfer_buffer_pos = 0;
+client->transfer_buffer = NULL;
+client->transfer_buffer_alloc = 0;
 if(contentlength > 0) {
 http->max_read = contentlength;
 httpstream_enable_buffered_read(&http->st, (char*)client->buffer.inbuf, client->buffer.maxsize, &client->buffer.cursize, &client->buffer.pos);
 } else if(chunkedtransferenc) {
 httpstream_enable_chunked_read(&http->st, (char*)client->buffer.inbuf, client->buffer.maxsize, &client->buffer.cursize, &client->buffer.pos);
 // uses the response_body_write callback to write the content of the
 // transfer buffer
 // returns 0 success
 //         1 would block or error
 static int client_write_response(HttpClient *client) {
-while(client->transfer_buffer_pos < client->transfer_buffer_len) {
+while(client->transfer2_buffer_pos < client->transfer2_buffer_len) {
-char *buf = client->transfer_buffer + client->transfer_buffer_pos;
+char *buf = client->transfer2_buffer + client->transfer2_buffer_pos;
-size_t len = client->transfer_buffer_len - client->transfer_buffer_pos;
+size_t len = client->transfer2_buffer_len - client->transfer2_buffer_pos;
 int ret = client->response_body_write(client, buf, len, client->response_body_write_userdata);
 if(ret > 0) {
-client->transfer_buffer_pos += ret;
+client->transfer2_buffer_pos += ret;
 } else if(ret == 0) {
 // EOF?
 // check if the write is incomplete, which would be an error
-client->error == client->transfer_buffer_pos < client->transfer_buffer_len;
+client->error == client->transfer2_buffer_pos < client->transfer2_buffer_len;
 return client->error;
 } else {
 if(ret != HTTP_CLIENT_CALLBACK_WOULD_BLOCK) {
 client->error = 1;
 }
 // does the transfer buffer still contains bytes, that should be written?
 if(client_write_response(client)) {
 return 1;
 }
-char *buf = client->transfer_buffer;
+char *buf = client->transfer2_buffer;
-size_t nbytes = client->transfer_buffer_alloc;
+size_t nbytes = client->transfer2_buffer_alloc;
 ssize_t r;
 while((r = net_read(&client->stream->st, buf, nbytes)) > 0) {
-client->transfer_buffer_len = r;
+client->transfer2_buffer_len = r;
-client->transfer_buffer_pos = 0;
+client->transfer2_buffer_pos = 0;
 if(client->response_body_write) {
 if(client_write_response(client)) {
 return 1;
 }
 }
 client->transfer_buffer = str;
 client->transfer_buffer_len = len;
 // test client_send_request
-int ret = client_send_request(client);
+int ret = client_send_buf(client);
 // It is very likely that the first client_send_request call doesn't
 // fully write the request buffer to the socket
 // In that case it returns 1 but without the error flag
 CX_TEST_ASSERT(ret == 1 && !client->error);
 CX_TEST_ASSERT(client->transfer_buffer_pos > 0);
 int writes = 1;
 while(client->transfer_buffer_pos < client->transfer_buffer_len && writes < 2000000) {
 ssize_t r = read(sock, tmpbuf, 1024);
 CX_TEST_ASSERT(r >= 0);
 cxBufferWrite(tmpbuf, 1, r, &buf);
-ret = client_send_request(client);
+ret = client_send_buf(client);
 CX_TEST_ASSERT(ret == 0 || (ret == 1 && !client->error));
 writes++;
 }
 CX_TEST_ASSERT(client->transfer_buffer_pos == client->transfer_buffer_len);

Mercurial > hg > webserver / file comparison

comparison: src/server/proxy/httpclient.c

src/server/proxy/httpclient.c