This helper function serializes a consumer group Pending Entries List (PEL) * into the RDB file. The 'nacks' argument tells the function if also persist * the informations about the not acknowledged message, or if to persist * just the IDs: this is useful because for the global consumer group PEL * we serialized the NACKs as well, but when serializing the local consumer * PELs we just add the
| 716 | * PELs we just add the ID, that will be resolved inside the global PEL to |
| 717 | * put a reference to the same structure. */ |
| 718 | ssize_t rdbSaveStreamPEL(rio *rdb, rax *pel, int nacks) { |
| 719 | ssize_t n, nwritten = 0; |
| 720 | |
| 721 | /* Number of entries in the PEL. */ |
| 722 | if ((n = rdbSaveLen(rdb,raxSize(pel))) == -1) return -1; |
| 723 | nwritten += n; |
| 724 | |
| 725 | /* Save each entry. */ |
| 726 | raxIterator ri; |
| 727 | raxStart(&ri,pel); |
| 728 | raxSeek(&ri,"^",NULL,0); |
| 729 | while(raxNext(&ri)) { |
| 730 | /* We store IDs in raw form as 128 big big endian numbers, like |
| 731 | * they are inside the radix tree key. */ |
| 732 | if ((n = rdbWriteRaw(rdb,ri.key,sizeof(streamID))) == -1) { |
| 733 | raxStop(&ri); |
| 734 | return -1; |
| 735 | } |
| 736 | nwritten += n; |
| 737 | |
| 738 | if (nacks) { |
| 739 | streamNACK *nack = ri.data; |
| 740 | if ((n = rdbSaveMillisecondTime(rdb,nack->delivery_time)) == -1) { |
| 741 | raxStop(&ri); |
| 742 | return -1; |
| 743 | } |
| 744 | nwritten += n; |
| 745 | if ((n = rdbSaveLen(rdb,nack->delivery_count)) == -1) { |
| 746 | raxStop(&ri); |
| 747 | return -1; |
| 748 | } |
| 749 | nwritten += n; |
| 750 | /* We don't save the consumer name: we'll save the pending IDs |
| 751 | * for each consumer in the consumer PEL, and resolve the consumer |
| 752 | * at loading time. */ |
| 753 | } |
| 754 | } |
| 755 | raxStop(&ri); |
| 756 | return nwritten; |
| 757 | } |
| 758 | |
| 759 | /* Serialize the consumers of a stream consumer group into the RDB. Helper |
| 760 | * function for the stream data type serialization. What we do here is to |
no test coverage detected