Adds a new item into the stream 's' having the specified number of * field-value pairs as specified in 'numfields' and stored into 'argv'. * 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, * but instead the passed ID is used to add the new entry. In this case * adding the entry may fail as specified late
| 445 | * 2. If a size of a single element or the sum of the elements is too big to |
| 446 | * be stored into the stream. errno will be set to ERANGE. */ |
| 447 | int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id) { |
| 448 | |
| 449 | /* Generate the new entry ID. */ |
| 450 | streamID id; |
| 451 | if (use_id) |
| 452 | id = *use_id; |
| 453 | else |
| 454 | streamNextID(&s->last_id,&id); |
| 455 | |
| 456 | /* Check that the new ID is greater than the last entry ID |
| 457 | * or return an error. Automatically generated IDs might |
| 458 | * overflow (and wrap-around) when incrementing the sequence |
| 459 | part. */ |
| 460 | if (streamCompareID(&id,&s->last_id) <= 0) { |
| 461 | errno = EDOM; |
| 462 | return C_ERR; |
| 463 | } |
| 464 | |
| 465 | /* Avoid overflow when trying to add an element to the stream (listpack |
| 466 | * can only host up to 32bit length sttrings, and also a total listpack size |
| 467 | * can't be bigger than 32bit length. */ |
| 468 | size_t totelelen = 0; |
| 469 | for (int64_t i = 0; i < numfields*2; i++) { |
| 470 | sds ele = szFromObj(argv[i]); |
| 471 | totelelen += sdslen(ele); |
| 472 | } |
| 473 | if (totelelen > STREAM_LISTPACK_MAX_SIZE) { |
| 474 | errno = ERANGE; |
| 475 | return C_ERR; |
| 476 | } |
| 477 | |
| 478 | /* Add the new entry. */ |
| 479 | raxIterator ri; |
| 480 | raxStart(&ri,s->rax); |
| 481 | raxSeek(&ri,"$",NULL,0); |
| 482 | |
| 483 | size_t lp_bytes = 0; /* Total bytes in the tail listpack. */ |
| 484 | unsigned char *lp = NULL; /* Tail listpack pointer. */ |
| 485 | |
| 486 | /* Get a reference to the tail node listpack. */ |
| 487 | if (raxNext(&ri)) { |
| 488 | lp = (unsigned char*)ri.data; |
| 489 | lp_bytes = lpBytes(lp); |
| 490 | } |
| 491 | raxStop(&ri); |
| 492 | |
| 493 | /* We have to add the key into the radix tree in lexicographic order, |
| 494 | * to do so we consider the ID as a single 128 bit number written in |
| 495 | * big endian, so that the most significant bytes are the first ones. */ |
| 496 | uint64_t rax_key[2]; /* Key in the radix tree containing the listpack.*/ |
| 497 | streamID master_id; /* ID of the master entry in the listpack. */ |
| 498 | |
| 499 | /* Create a new listpack and radix tree node if needed. Note that when |
| 500 | * a new listpack is created, we populate it with a "master entry". This |
| 501 | * is just a set of fields that is taken as references in order to compress |
| 502 | * the stream entries that we'll add inside the listpack. |
| 503 | * |
| 504 | * Note that while we use the first added entry fields to create |
no test coverage detected