Commit 64288d2c authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Add cursor field to protocol_t

parent 2fc0af54
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
__tmp->clen = 0; \ __tmp->clen = 0; \
__tmp->type = 0; \ __tmp->type = 0; \
__tmp->remaining = -1; \ __tmp->remaining = -1; \
__tmp->cursor = 0; \
__tmp->data = NULL; \ __tmp->data = NULL; \
} while(0) } while(0)
...@@ -344,6 +345,7 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons ...@@ -344,6 +345,7 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons
/* Store remaining objects for a complete multi bulk */ /* Store remaining objects for a complete multi bulk */
cur->remaining = (unsigned)i64.i64; cur->remaining = (unsigned)i64.i64;
cur->cursor = -1; /* Is incremented in "done" */
CALLBACK(array, cur, cur->remaining); CALLBACK(array, cur, cur->remaining);
goto done; goto done;
} }
...@@ -375,6 +377,9 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons ...@@ -375,6 +377,9 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons
cur->remaining -= available; cur->remaining -= available;
CALLBACK(string, cur, pos, available); CALLBACK(string, cur, pos, available);
pos += available; nread += available; pos += available; nread += available;
/* Add number of processed bytes to cursor */
cur->cursor += available;
goto finalize; goto finalize;
} }
...@@ -414,6 +419,9 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons ...@@ -414,6 +419,9 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons
/* No more data */ /* No more data */
CALLBACK(string, cur, mark, pos-mark); CALLBACK(string, cur, mark, pos-mark);
/* Add number of processed bytes to cursor */
cur->cursor += pos-mark;
} }
STATE(line_lf) { STATE(line_lf) {
...@@ -436,6 +444,7 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons ...@@ -436,6 +444,7 @@ size_t redis_parser_execute(redis_parser_t *parser, redis_protocol_t **dst, cons
/* Move to nested object when we see an incomplete array */ /* Move to nested object when we see an incomplete array */
if (cur->type == REDIS_ARRAY_T && cur->remaining) { if (cur->type == REDIS_ARRAY_T && cur->remaining) {
RESET_PROTOCOL_T(&stack[++stackidx]); RESET_PROTOCOL_T(&stack[++stackidx]);
cur->cursor++;
cur->remaining--; cur->remaining--;
break; break;
} }
......
...@@ -53,7 +53,8 @@ typedef enum redis_parser_errno_e { ...@@ -53,7 +53,8 @@ typedef enum redis_parser_errno_e {
struct redis_protocol_s { struct redis_protocol_s {
unsigned char type; /* payload type */ unsigned char type; /* payload type */
const redis_protocol_t* parent; /* when nested, parent object */ const redis_protocol_t* parent; /* when nested, parent object */
int64_t remaining; /* remaining bulk bytes/nested objects */ int remaining; /* remaining bulk bytes/nested objects */
int cursor; /* number of processed bulk bytes/nested objects */
void *data; /* user data */ void *data; /* user data */
size_t poff; /* protocol offset */ size_t poff; /* protocol offset */
size_t plen; /* protocol length */ size_t plen; /* protocol length */
......
...@@ -18,6 +18,7 @@ typedef struct log_entry_s log_entry_t; ...@@ -18,6 +18,7 @@ typedef struct log_entry_s log_entry_t;
struct log_entry_s { struct log_entry_s {
redis_protocol_t obj; redis_protocol_t obj;
int parent_cursor;
/* string_t specifics */ /* string_t specifics */
const char *string_buf; const char *string_buf;
...@@ -49,6 +50,10 @@ int on_string(redis_parser_t *parser, redis_protocol_t *obj, const char *buf, si ...@@ -49,6 +50,10 @@ int on_string(redis_parser_t *parser, redis_protocol_t *obj, const char *buf, si
.string_buf = buf, .string_buf = buf,
.string_len = len .string_len = len
}; };
if (obj->parent)
tmp.parent_cursor = obj->parent->cursor;
cb_log[cb_log_idx++] = tmp; cb_log[cb_log_idx++] = tmp;
return 0; return 0;
} }
...@@ -58,6 +63,10 @@ int on_array(redis_parser_t *parser, redis_protocol_t *obj, size_t len) { ...@@ -58,6 +63,10 @@ int on_array(redis_parser_t *parser, redis_protocol_t *obj, size_t len) {
.obj = *obj, .obj = *obj,
.array_len = len .array_len = len
}; };
if (obj->parent)
tmp.parent_cursor = obj->parent->cursor;
cb_log[cb_log_idx++] = tmp; cb_log[cb_log_idx++] = tmp;
return 0; return 0;
} }
...@@ -67,6 +76,10 @@ int on_integer(redis_parser_t *parser, redis_protocol_t *obj, int64_t value) { ...@@ -67,6 +76,10 @@ int on_integer(redis_parser_t *parser, redis_protocol_t *obj, int64_t value) {
.obj = *obj, .obj = *obj,
.integer_value = value .integer_value = value
}; };
if (obj->parent)
tmp.parent_cursor = obj->parent->cursor;
cb_log[cb_log_idx++] = tmp; cb_log[cb_log_idx++] = tmp;
return 0; return 0;
} }
...@@ -75,6 +88,10 @@ int on_nil(redis_parser_t *parser, redis_protocol_t *obj) { ...@@ -75,6 +88,10 @@ int on_nil(redis_parser_t *parser, redis_protocol_t *obj) {
log_entry_t tmp = { log_entry_t tmp = {
.obj = *obj .obj = *obj
}; };
if (obj->parent)
tmp.parent_cursor = obj->parent->cursor;
cb_log[cb_log_idx++] = tmp; cb_log[cb_log_idx++] = tmp;
return 0; return 0;
} }
...@@ -97,7 +114,6 @@ void test_char_by_char(redis_protocol_t *ref, const char *buf, size_t len) { ...@@ -97,7 +114,6 @@ void test_char_by_char(redis_protocol_t *ref, const char *buf, size_t len) {
size_t i; size_t i;
p = malloc(sizeof(redis_parser_t)); p = malloc(sizeof(redis_parser_t));
for (i = 1; i < (len-1); i++) { for (i = 1; i < (len-1); i++) {
RESET_PARSER_T(p); RESET_PARSER_T(p);
...@@ -110,6 +126,7 @@ void test_char_by_char(redis_protocol_t *ref, const char *buf, size_t len) { ...@@ -110,6 +126,7 @@ void test_char_by_char(redis_protocol_t *ref, const char *buf, size_t len) {
assert(NULL != res); assert(NULL != res);
/* Compare result with reference */ /* Compare result with reference */
ref->cursor = res->cursor = 0; /* ignore cursor */
assert(memcmp(ref, res, sizeof(redis_protocol_t)) == 0); assert(memcmp(ref, res, sizeof(redis_protocol_t)) == 0);
} }
...@@ -195,6 +212,7 @@ void test_array(redis_parser_t *p) { ...@@ -195,6 +212,7 @@ void test_array(redis_parser_t *p) {
assert(cb_log[1].obj.clen == 5); assert(cb_log[1].obj.clen == 5);
assert(cb_log[1].string_buf == buf+4+4); assert(cb_log[1].string_buf == buf+4+4);
assert(cb_log[1].string_len == 5); assert(cb_log[1].string_len == 5);
assert(cb_log[1].parent_cursor == 0);
assert(cb_log[2].obj.poff == 4+11); assert(cb_log[2].obj.poff == 4+11);
assert(cb_log[2].obj.plen == 4+5+2); assert(cb_log[2].obj.plen == 4+5+2);
...@@ -202,6 +220,7 @@ void test_array(redis_parser_t *p) { ...@@ -202,6 +220,7 @@ void test_array(redis_parser_t *p) {
assert(cb_log[2].obj.clen == 5); assert(cb_log[2].obj.clen == 5);
assert(cb_log[2].string_buf == buf+4+11+4); assert(cb_log[2].string_buf == buf+4+11+4);
assert(cb_log[2].string_len == 5); assert(cb_log[2].string_len == 5);
assert(cb_log[2].parent_cursor == 1);
/* Chunked check */ /* Chunked check */
test_char_by_char(res, buf, len); test_char_by_char(res, buf, len);
......
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