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

Merge branch 'unstable' of github.com:/antirez/redis into unstable

Browse Source
This commit is contained in:
antirez 2018-11-05 13:07:14 +01:00
commit 514bbdd670
1 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/t_stream.c
View File

@ -37,7 +37,7 @@
* mark the entry as deleted, or having the same field as the "master" * mark the entry as deleted, or having the same field as the "master"
* entry at the start of the listpack> */ * entry at the start of the listpack> */
#define STREAM_ITEM_FLAG_NONE 0 /* No special flags. */ #define STREAM_ITEM_FLAG_NONE 0 /* No special flags. */
#define STREAM_ITEM_FLAG_DELETED (1<<0) /* Entry is delted. Skip it. */ #define STREAM_ITEM_FLAG_DELETED (1<<0) /* Entry is deleted. Skip it. */
#define STREAM_ITEM_FLAG_SAMEFIELDS (1<<1) /* Same fields as master entry. */ #define STREAM_ITEM_FLAG_SAMEFIELDS (1<<1) /* Same fields as master entry. */
void streamFreeCG(streamCG *cg); void streamFreeCG(streamCG *cg);
@ -165,7 +165,7 @@ int streamCompareID(streamID *a, streamID *b) {
* Returns the new entry ID populating the 'added_id' structure. * Returns the new entry ID populating the 'added_id' structure.
* *
* If 'use_id' is not NULL, the ID is not auto-generated by the function, * If 'use_id' is not NULL, the ID is not auto-generated by the function,
* but instead the passed ID is uesd to add the new entry. In this case * but instead the passed ID is used to add the new entry. In this case
* adding the entry may fail as specified later in this comment. * adding the entry may fail as specified later in this comment.
* *
* The function returns C_OK if the item was added, this is always true * The function returns C_OK if the item was added, this is always true
@ -223,13 +223,13 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
* *
* count and deleted just represent respectively the total number of * count and deleted just represent respectively the total number of
* entries inside the listpack that are valid, and marked as deleted * entries inside the listpack that are valid, and marked as deleted
* (delted flag in the entry flags set). So the total number of items * (deleted flag in the entry flags set). So the total number of items
* actually inside the listpack (both deleted and not) is count+deleted. * actually inside the listpack (both deleted and not) is count+deleted.
* *
* The real entries will be encoded with an ID that is just the * The real entries will be encoded with an ID that is just the
* millisecond and sequence difference compared to the key stored at * millisecond and sequence difference compared to the key stored at
* the radix tree node containing the listpack (delta encoding), and * the radix tree node containing the listpack (delta encoding), and
* if the fields of the entry are the same as the master enty fields, the * if the fields of the entry are the same as the master entry fields, the
* entry flags will specify this fact and the entry fields and number * entry flags will specify this fact and the entry fields and number
* of fields will be omitted (see later in the code of this function). * of fields will be omitted (see later in the code of this function).
* *
@ -486,7 +486,7 @@ int64_t streamTrimByLength(stream *s, size_t maxlen, int approx) {
* } * }
* streamIteratorStop(&myiterator); */ * streamIteratorStop(&myiterator); */
void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev) { void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev) {
/* Intialize the iterator and translates the iteration start/stop /* Initialize the iterator and translates the iteration start/stop
* elements into a 128 big big-endian number. */ * elements into a 128 big big-endian number. */
if (start) { if (start) {
streamEncodeID(si->start_key,start); streamEncodeID(si->start_key,start);
@ -564,7 +564,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
si->lp_ele = lpLast(si->lp); si->lp_ele = lpLast(si->lp);
} }
} else if (si->rev) { } else if (si->rev) {
/* If we are itereating in the reverse order, and this is not /* If we are iterating in the reverse order, and this is not
* the first entry emitted for this listpack, then we already * the first entry emitted for this listpack, then we already
* emitted the current entry, and have to go back to the previous * emitted the current entry, and have to go back to the previous
* one. */ * one. */
@ -751,7 +751,7 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
} }
/* 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 * iterator, since the stream iterator itself is supposed to be stack
* allocated. */ * allocated. */
void streamIteratorStop(streamIterator *si) { void streamIteratorStop(streamIterator *si) {
raxStop(&si->ri); raxStop(&si->ri);
@ -867,15 +867,15 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
* *
* The behavior may be modified passing non-zero flags: * The behavior may be modified passing non-zero flags:
* *
* STREAM_RWR_NOACK: Do not craete PEL entries, that is, the point "3" above * STREAM_RWR_NOACK: Do not create PEL entries, that is, the point "3" above
* is not performed. * is not performed.
* STREAM_RWR_RAWENTRIES: Do not emit array boundaries, but just the entries, * STREAM_RWR_RAWENTRIES: Do not emit array boundaries, but just the entries,
* and return the number of entries emitted as usually. * and return the number of entries emitted as usually.
* This is used when the function is just used in order * This is used when the function is just used in order
* to emit data and there is some higher level logic. * to emit data and there is some higher level logic.
* *
* The final argument 'spi' (stream propagatino info pointer) is a structure * The final argument 'spi' (stream propagation info pointer) is a structure
* filled with information needed to propagte the command execution to AOF * filled with information needed to propagate the command execution to AOF
* and slaves, in the case a consumer group was passed: we need to generate * and slaves, in the case a consumer group was passed: we need to generate
* XCLAIM commands to create the pending list into AOF/slaves in that case. * XCLAIM commands to create the pending list into AOF/slaves in that case.
* *
@ -1140,7 +1140,7 @@ invalid:
} }
/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to /* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
* 0, to be used when - and + are accetable IDs. */ * 0, to be used when - and + are acceptable IDs. */
int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) { int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
return streamGenericParseIDOrReply(c,o,id,missing_seq,0); return streamGenericParseIDOrReply(c,o,id,missing_seq,0);
} }
@ -2159,7 +2159,7 @@ void xclaimCommand(client *c) {
/* If we stopped because some IDs cannot be parsed, perhaps they /* If we stopped because some IDs cannot be parsed, perhaps they
* are trailing options. */ * are trailing options. */
time_t now = mstime(); mstime_t now = mstime();
streamID last_id = {0,0}; streamID last_id = {0,0};
int propagate_last_id = 0; int propagate_last_id = 0;
for (; j < c->argc; j++) { for (; j < c->argc; j++) {