fluencelabs/redis
1
0
Fork 0
mirror of https://github.com/fluencelabs/redis synced 2025-03-19 17:10:50 +00:00
Code Issues Projects Releases Wiki Activity

Streams: stream iteration refactoring, WIP 2.

Browse Source
This commit is contained in:
antirez 2017-09-14 14:23:27 +02:00
parent b1ec333633
commit a58733cacf
1 changed files with 19 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/t_stream.c
View File

@ -208,6 +208,11 @@ typedef struct streamIterator {
raxIterator ri; /* Rax iterator. */ raxIterator ri; /* Rax iterator. */
unsigned char *lp; /* Current listpack. */ unsigned char *lp; /* Current listpack. */
unsigned char *lp_ele; /* Current listpack cursor. */ unsigned char *lp_ele; /* Current listpack cursor. */
/* Buffers used to hold the string of lpGet() when the element is
* integer encoded, so that there is no string representation of the
* element inside the listpack itself. */
unsigned char field_buf[LP_INTBUF_SIZE];
unsigned char value_buf[LP_INTBUF_SIZE];
} streamIterator; } streamIterator;
/* Initialize the stream iterator, so that we can call iterating functions /* Initialize the stream iterator, so that we can call iterating functions
@ -289,6 +294,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, size_t *numfields) {
return 0; /* We are already out of range. */ return 0; /* We are already out of range. */
streamDecodeID(e,id); streamDecodeID(e,id);
*numfields = lpGetInteger(si->lp_ele); *numfields = lpGetInteger(si->lp_ele);
si->lp_ele = lpNext(si->lp,si->lp_ele);
return 1; /* Valid item returned. */ return 1; /* Valid item returned. */
} else { } else {
/* If we do not emit, we have to discard. */ /* If we do not emit, we have to discard. */
@ -303,6 +309,19 @@ int streamIteratorGetID(streamIterator *si, streamID *id, size_t *numfields) {
} }
} }
/* Get the field and value of the current item we are iterating. This should
* be called immediately after streamIteratorGetID(), and for each field
* according to the number of fields returned by streamIteratorGetID().
* The function populates the field and value pointers and the corresponding
* lengths by reference, that are valid until the next iterator call, assuming
* no one touches the stream meanwhile. */
void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsigned char **valueptr, int64_t *fieldlen, int64_t *valuelen) {
*fieldptr = lpGet(si->lp_ele,fieldlen,si->field_buf);
si->lp_ele = lpNext(si->lp,si->lp_ele);
*valueptr = lpGet(si->lp_ele,valuelen,si->value_buf);
si->lp_ele = lpNext(si->lp,si->lp_ele);
}
/* Stop the stream iterator. The only cleanup we need is to free the rax /* Stop the stream iterator. The only cleanup we need is to free the rax
* itereator, since the stream iterator itself is supposed to be stack * itereator, since the stream iterator itself is supposed to be stack
* allocated. */ * allocated. */