MCPcopy Create free account
hub / github.com/Snapchat/KeyDB / streamIteratorGetID

Function streamIteratorGetID

src/t_stream.cpp:1090–1228  ·  view source on GitHub ↗

Return 1 and store the current item ID at 'id' if there are still * elements within the iteration range, otherwise return 0 in order to * signal the iteration terminated. */

Source from the content-addressed store, hash-verified

1088 * elements within the iteration range, otherwise return 0 in order to
1089 * signal the iteration terminated. */
1090int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
1091 while(1) { /* Will stop when element > stop_key or end of radix tree. */
1092 /* If the current listpack is set to NULL, this is the start of the
1093 * iteration or the previous listpack was completely iterated.
1094 * Go to the next node. */
1095 if (si->lp == NULL || si->lp_ele == NULL) {
1096 if (!si->rev && !raxNext(&si->ri)) return 0;
1097 else if (si->rev && !raxPrev(&si->ri)) return 0;
1098 serverAssert(si->ri.key_len == sizeof(streamID));
1099 /* Get the master ID. */
1100 streamDecodeID(si->ri.key,&si->master_id);
1101 /* Get the master fields count. */
1102 si->lp = (unsigned char*)si->ri.data;
1103 si->lp_ele = lpFirst(si->lp); /* Seek items count */
1104 si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek deleted count. */
1105 si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek num fields. */
1106 si->master_fields_count = lpGetInteger(si->lp_ele);
1107 si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek first field. */
1108 si->master_fields_start = si->lp_ele;
1109 /* We are now pointing to the first field of the master entry.
1110 * We need to seek either the first or the last entry depending
1111 * on the direction of the iteration. */
1112 if (!si->rev) {
1113 /* If we are iterating in normal order, skip the master fields
1114 * to seek the first actual entry. */
1115 for (uint64_t i = 0; i < si->master_fields_count; i++)
1116 si->lp_ele = lpNext(si->lp,si->lp_ele);
1117 } else {
1118 /* If we are iterating in reverse direction, just seek the
1119 * last part of the last entry in the listpack (that is, the
1120 * fields count). */
1121 si->lp_ele = lpLast(si->lp);
1122 }
1123 } else if (si->rev) {
1124 /* If we are iterating in the reverse order, and this is not
1125 * the first entry emitted for this listpack, then we already
1126 * emitted the current entry, and have to go back to the previous
1127 * one. */
1128 int64_t lp_count = lpGetInteger(si->lp_ele);
1129 while(lp_count--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
1130 /* Seek lp-count of prev entry. */
1131 si->lp_ele = lpPrev(si->lp,si->lp_ele);
1132 }
1133
1134 /* For every radix tree node, iterate the corresponding listpack,
1135 * returning elements when they are within range. */
1136 while(1) {
1137 if (!si->rev) {
1138 /* If we are going forward, skip the previous entry
1139 * lp-count field (or in case of the master entry, the zero
1140 * term field) */
1141 si->lp_ele = lpNext(si->lp,si->lp_ele);
1142 if (si->lp_ele == NULL) break;
1143 } else {
1144 /* If we are going backward, read the number of elements this
1145 * entry is composed of, and jump backward N times to seek
1146 * its start. */
1147 int64_t lp_count = lpGetInteger(si->lp_ele);

Callers 7

streamDeleteItemFunction · 0.85
streamLastValidIDFunction · 0.85
streamReplyWithRangeFunction · 0.85
xclaimCommandFunction · 0.85
xorObjectDigestFunction · 0.85
RM_StreamIteratorNextIDFunction · 0.85
rewriteStreamObjectFunction · 0.85

Calls 8

raxNextFunction · 0.85
raxPrevFunction · 0.85
streamDecodeIDFunction · 0.85
lpFirstFunction · 0.85
lpNextFunction · 0.85
lpLastFunction · 0.85
lpPrevFunction · 0.85
streamEncodeIDFunction · 0.85

Tested by

no test coverage detected