1 #ifndef STREAM_H 2 #define STREAM_H 3 4 #include "rax.h" 5 #include "listpack.h" 6 7 /* Stream item ID: a 128 bit number composed of a milliseconds time and 8 * a sequence counter. IDs generated in the same millisecond (or in a past 9 * millisecond if the clock jumped backward) will use the millisecond time 10 * of the latest generated ID and an incremented sequence. */ 11 typedef struct streamID { 12 uint64_t ms; /* Unix time in milliseconds. */ 13 uint64_t seq; /* Sequence number. */ 14 } streamID; 15 16 typedef struct stream { 17 rax *rax; /* The radix tree holding the stream. */ 18 uint64_t length; /* Number of elements inside this stream. */ 19 streamID last_id; /* Zero if there are yet no items. */ 20 rax *cgroups; /* Consumer groups dictionary: name -> streamCG */ 21 } stream; 22 23 /* We define an iterator to iterate stream items in an abstract way, without 24 * caring about the radix tree + listpack representation. Technically speaking 25 * the iterator is only used inside streamReplyWithRange(), so could just 26 * be implemented inside the function, but practically there is the AOF 27 * rewriting code that also needs to iterate the stream to emit the XADD 28 * commands. */ 29 typedef struct streamIterator { 30 stream *stream; /* The stream we are iterating. */ 31 streamID master_id; /* ID of the master entry at listpack head. */ 32 uint64_t master_fields_count; /* Master entries # of fields. */ 33 unsigned char *master_fields_start; /* Master entries start in listpack. */ 34 unsigned char *master_fields_ptr; /* Master field to emit next. */ 35 int entry_flags; /* Flags of entry we are emitting. */ 36 int rev; /* True if iterating end to start (reverse). */ 37 uint64_t start_key[2]; /* Start key as 128 bit big endian. */ 38 uint64_t end_key[2]; /* End key as 128 bit big endian. */ 39 raxIterator ri; /* Rax iterator. */ 40 unsigned char *lp; /* Current listpack. */ 41 unsigned char *lp_ele; /* Current listpack cursor. */ 42 unsigned char *lp_flags; /* Current entry flags pointer. */ 43 /* Buffers used to hold the string of lpGet() when the element is 44 * integer encoded, so that there is no string representation of the 45 * element inside the listpack itself. */ 46 unsigned char field_buf[LP_INTBUF_SIZE]; 47 unsigned char value_buf[LP_INTBUF_SIZE]; 48 } streamIterator; 49 50 /* Consumer group. */ 51 typedef struct streamCG { 52 streamID last_id; /* Last delivered (not acknowledged) ID for this 53 group. Consumers that will just ask for more 54 messages will served with IDs > than this. */ 55 rax *pel; /* Pending entries list. This is a radix tree that 56 has every message delivered to consumers (without 57 the NOACK option) that was yet not acknowledged 58 as processed. The key of the radix tree is the 59 ID as a 64 bit big endian number, while the 60 associated value is a streamNACK structure.*/ 61 rax *consumers; /* A radix tree representing the consumers by name 62 and their associated representation in the form 63 of streamConsumer structures. */ 64 } streamCG; 65 66 /* A specific consumer in a consumer group. */ 67 typedef struct streamConsumer { 68 mstime_t seen_time; /* Last time this consumer was active. */ 69 sds name; /* Consumer name. This is how the consumer 70 will be identified in the consumer group 71 protocol. Case sensitive. */ 72 rax *pel; /* Consumer specific pending entries list: all 73 the pending messages delivered to this 74 consumer not yet acknowledged. Keys are 75 big endian message IDs, while values are 76 the same streamNACK structure referenced 77 in the "pel" of the conumser group structure 78 itself, so the value is shared. */ 79 } streamConsumer; 80 81 /* Pending (yet not acknowledged) message in a consumer group. */ 82 typedef struct streamNACK { 83 mstime_t delivery_time; /* Last time this message was delivered. */ 84 uint64_t delivery_count; /* Number of times this message was delivered.*/ 85 streamConsumer *consumer; /* The consumer this message was delivered to 86 in the last delivery. */ 87 } streamNACK; 88 89 /* Stream propagation informations, passed to functions in order to propagate 90 * XCLAIM commands to AOF and slaves. */ 91 typedef struct sreamPropInfo { 92 robj *keyname; 93 robj *groupname; 94 } streamPropInfo; 95 96 /* Prototypes of exported APIs. */ 97 struct client; 98 99 stream *streamNew(void); 100 void freeStream(stream *s); 101 size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count, int rev, streamCG *group, streamConsumer *consumer, int flags, streamPropInfo *spi); 102 void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev); 103 int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields); 104 void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsigned char **valueptr, int64_t *fieldlen, int64_t *valuelen); 105 void streamIteratorStop(streamIterator *si); 106 streamCG *streamLookupCG(stream *s, sds groupname); 107 streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int create); 108 streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id); 109 streamNACK *streamCreateNACK(streamConsumer *consumer); 110 void streamDecodeID(void *buf, streamID *id); 111 int streamCompareID(streamID *a, streamID *b); 112 113 #endif 114