Helper function for handleClientsBlockedOnKeys(). This function is called * when there may be clients blocked on a stream key, and there may be new * data to fetch (the key is ready). */
| 398 | * when there may be clients blocked on a stream key, and there may be new |
| 399 | * data to fetch (the key is ready). */ |
| 400 | void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { |
| 401 | dictEntry *de = dictFind(rl->db->blocking_keys,rl->key); |
| 402 | stream *s = (stream*)ptrFromObj(o); |
| 403 | |
| 404 | /* We need to provide the new data arrived on the stream |
| 405 | * to all the clients that are waiting for an offset smaller |
| 406 | * than the current top item. */ |
| 407 | if (de) { |
| 408 | list *clients = (list*)dictGetVal(de); |
| 409 | listNode *ln; |
| 410 | listIter li; |
| 411 | listRewind(clients,&li); |
| 412 | |
| 413 | while((ln = listNext(&li))) { |
| 414 | client *receiver = (client*)listNodeValue(ln); |
| 415 | if (receiver->btype != BLOCKED_STREAM) continue; |
| 416 | std::unique_lock<decltype(receiver->lock)> lock(receiver->lock); |
| 417 | bkinfo *bki = (bkinfo*)dictFetchValue(receiver->bpop.keys,rl->key); |
| 418 | streamID *gt = &bki->stream_id; |
| 419 | |
| 420 | /* If we blocked in the context of a consumer |
| 421 | * group, we need to resolve the group and update the |
| 422 | * last ID the client is blocked for: this is needed |
| 423 | * because serving other clients in the same consumer |
| 424 | * group will alter the "last ID" of the consumer |
| 425 | * group, and clients blocked in a consumer group are |
| 426 | * always blocked for the ">" ID: we need to deliver |
| 427 | * only new messages and avoid unblocking the client |
| 428 | * otherwise. */ |
| 429 | streamCG *group = NULL; |
| 430 | if (receiver->bpop.xread_group) { |
| 431 | group = streamLookupCG(s, |
| 432 | szFromObj(receiver->bpop.xread_group)); |
| 433 | /* If the group was not found, send an error |
| 434 | * to the consumer. */ |
| 435 | if (!group) { |
| 436 | addReplyError(receiver, |
| 437 | "-NOGROUP the consumer group this client " |
| 438 | "was blocked on no longer exists"); |
| 439 | unblockClient(receiver); |
| 440 | continue; |
| 441 | } else { |
| 442 | *gt = group->last_id; |
| 443 | } |
| 444 | } |
| 445 | |
| 446 | if (streamCompareID(&s->last_id, gt) > 0) { |
| 447 | streamID start = *gt; |
| 448 | streamIncrID(&start); |
| 449 | |
| 450 | /* Lookup the consumer for the group, if any. */ |
| 451 | streamConsumer *consumer = NULL; |
| 452 | int noack = 0; |
| 453 | |
| 454 | if (group) { |
| 455 | int created = 0; |
| 456 | consumer = |
| 457 | streamLookupConsumer(group, |
no test coverage detected