Commit b7188721 authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Request queue is responsible for parsing replies

The change in d0eb3f65 moved responsibility to the request itself, but
kept ownership of the parser struct with the request queue. In
retrospect, it is cleaner to keep protocol parsing in the queue and have
it pass the parsed protocol via the read callback.
parent 796b2f1f
......@@ -173,7 +173,9 @@ int redis_request_queue_write_cb(redis_request_queue *self, size_t len) {
int redis_request_queue_read_cb(redis_request_queue *self, const char *buf, size_t len) {
ngx_queue_t *q = NULL;
redis_request *req = NULL;
int nread, done;
redis_protocol *p = NULL;
size_t nparsed = 0;
int done;
while (len) {
if (ngx_queue_empty(&self->request_wait_read)) {
......@@ -184,17 +186,21 @@ int redis_request_queue_read_cb(redis_request_queue *self, const char *buf, size
req = ngx_queue_data(q, redis_request, rq);
done = 0;
nparsed = redis_parser_execute(&self->parser, &p, buf, len);
if (nparsed < len && p == NULL) {
errno = redis_parser_err(&self->parser);
return REDIS_EPARSER;
}
assert(req->read_cb);
nread = req->read_cb(req, buf, len, &done);
req->read_cb(req, p, buf, nparsed, &done);
/* Abort on error */
if (nread < 0) {
return nread;
}
/* Request cannot be done on partial reply */
assert(p || !done);
assert((unsigned)nread <= len);
buf += nread;
len -= nread;
/* Update buffer */
buf += nparsed;
len -= nparsed;
/* Remove this request from `wait_read` when done reading */
if (done) {
......
......@@ -48,14 +48,12 @@ typedef int (redis_request_write_cb)(redis_request *self,
* buf buffer with reply data
* len length of buffer with reply data
* done set to non-zero by the request when it has been read in full
*
* Return:
* the number of bytes (<= n) actually used
*/
typedef int (redis_request_read_cb)(redis_request *self,
const char *buf,
size_t len,
int *done);
typedef void (redis_request_read_cb)(redis_request *self,
redis_protocol *p,
const char *buf,
size_t len,
int *done);
/*
* Free the request. This function is called after the last reply has been
......@@ -126,7 +124,6 @@ struct redis_request_queue_s {
ngx_queue_t request_to_write;
ngx_queue_t request_wait_write;
ngx_queue_t request_wait_read;
redis_parser parser;
redis_request_queue_to_write_cb *request_to_write_cb;
redis_request_queue_wait_write_cb *request_wait_write_cb;
......@@ -136,6 +133,8 @@ struct redis_request_queue_s {
redis_request_queue_io_fn *start_read_fn;
redis_request_queue_io_fn *stop_read_fn;
redis_parser parser;
void *data;
};
......
......@@ -206,18 +206,8 @@ int t1_write_cb(redis_request *_self, int n, int *done) {
return to_write;
}
int t1_read_cb(redis_request *_self, const char *buf, size_t len, int *done) {
void t1_read_cb(redis_request *_self, redis_protocol *p, const char *buf, size_t len, int *done) {
t1_redis_request *self = (t1_redis_request*)_self;
redis_protocol *p = NULL;
size_t nparsed;
nparsed = redis_parser_execute(&self->request_queue->parser, &p, buf, len);
/* Test for parse error */
if (nparsed < len && p == NULL) {
errno = redis_parser_err(&self->request_queue->parser);
return REDIS_EPARSER;
}
if (p != NULL) {
self->reply = p;
......@@ -225,9 +215,7 @@ int t1_read_cb(redis_request *_self, const char *buf, size_t len, int *done) {
}
self->read_raw_buf = buf;
self->read_raw_len = nparsed;
return nparsed;
self->read_raw_len = len;
}
void t1_free(redis_request *_self) {
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment