Serialize the consumers of a stream consumer group into the RDB. Helper * function for the stream data type serialization. What we do here is to * persist the consumer metadata, and it's PEL, for each consumer. */
| 776 | * function for the stream data type serialization. What we do here is to |
| 777 | * persist the consumer metadata, and it's PEL, for each consumer. */ |
| 778 | size_t rdbSaveStreamConsumers(rio *rdb, streamCG *cg) { |
| 779 | ssize_t n, nwritten = 0; |
| 780 | |
| 781 | /* Number of consumers in this consumer group. */ |
| 782 | if ((n = rdbSaveLen(rdb,raxSize(cg->consumers))) == -1) return -1; |
| 783 | nwritten += n; |
| 784 | |
| 785 | /* Save each consumer. */ |
| 786 | raxIterator ri; |
| 787 | raxStart(&ri,cg->consumers); |
| 788 | raxSeek(&ri,"^",NULL,0); |
| 789 | while(raxNext(&ri)) { |
| 790 | streamConsumer *consumer = (streamConsumer*)ri.data; |
| 791 | |
| 792 | /* Consumer name. */ |
| 793 | if ((n = rdbSaveRawString(rdb,ri.key,ri.key_len)) == -1) { |
| 794 | raxStop(&ri); |
| 795 | return -1; |
| 796 | } |
| 797 | nwritten += n; |
| 798 | |
| 799 | /* Last seen time. */ |
| 800 | if ((n = rdbSaveMillisecondTime(rdb,consumer->seen_time)) == -1) { |
| 801 | raxStop(&ri); |
| 802 | return -1; |
| 803 | } |
| 804 | nwritten += n; |
| 805 | |
| 806 | /* Consumer PEL, without the ACKs (see last parameter of the function |
| 807 | * passed with value of 0), at loading time we'll lookup the ID |
| 808 | * in the consumer group global PEL and will put a reference in the |
| 809 | * consumer local PEL. */ |
| 810 | if ((n = rdbSaveStreamPEL(rdb,consumer->pel,0)) == -1) { |
| 811 | raxStop(&ri); |
| 812 | return -1; |
| 813 | } |
| 814 | nwritten += n; |
| 815 | } |
| 816 | raxStop(&ri); |
| 817 | return nwritten; |
| 818 | } |
| 819 | |
| 820 | /* Save a Redis object. |
| 821 | * Returns -1 on error, number of bytes written on success. */ |
no test coverage detected