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). */
| 376 | * when there may be clients blocked on a stream key, and there may be new |
| 377 | * data to fetch (the key is ready). */ |
| 378 | void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { |
| 379 | dictEntry *de = dictFind(rl->db->blocking_keys,rl->key); |
| 380 | stream *s = o->ptr; |
| 381 | |
| 382 | /* We need to provide the new data arrived on the stream |
| 383 | * to all the clients that are waiting for an offset smaller |
| 384 | * than the current top item. */ |
| 385 | if (de) { |
| 386 | list *clients = dictGetVal(de); |
| 387 | listNode *ln; |
| 388 | listIter li; |
| 389 | listRewind(clients,&li); |
| 390 | |
| 391 | while((ln = listNext(&li))) { |
| 392 | client *receiver = listNodeValue(ln); |
| 393 | if (receiver->btype != BLOCKED_STREAM) continue; |
| 394 | bkinfo *bki = dictFetchValue(receiver->bpop.keys,rl->key); |
| 395 | streamID *gt = &bki->stream_id; |
| 396 | |
| 397 | /* If we blocked in the context of a consumer |
| 398 | * group, we need to resolve the group and update the |
| 399 | * last ID the client is blocked for: this is needed |
| 400 | * because serving other clients in the same consumer |
| 401 | * group will alter the "last ID" of the consumer |
| 402 | * group, and clients blocked in a consumer group are |
| 403 | * always blocked for the ">" ID: we need to deliver |
| 404 | * only new messages and avoid unblocking the client |
| 405 | * otherwise. */ |
| 406 | streamCG *group = NULL; |
| 407 | if (receiver->bpop.xread_group) { |
| 408 | group = streamLookupCG(s, |
| 409 | receiver->bpop.xread_group->ptr); |
| 410 | /* If the group was not found, send an error |
| 411 | * to the consumer. */ |
| 412 | if (!group) { |
| 413 | addReplyError(receiver, |
| 414 | "-NOGROUP the consumer group this client " |
| 415 | "was blocked on no longer exists"); |
| 416 | unblockClient(receiver); |
| 417 | continue; |
| 418 | } else { |
| 419 | *gt = group->last_id; |
| 420 | } |
| 421 | } |
| 422 | |
| 423 | if (streamCompareID(&s->last_id, gt) > 0) { |
| 424 | streamID start = *gt; |
| 425 | streamIncrID(&start); |
| 426 | |
| 427 | /* Lookup the consumer for the group, if any. */ |
| 428 | streamConsumer *consumer = NULL; |
| 429 | int noack = 0; |
| 430 | |
| 431 | if (group) { |
| 432 | int created = 0; |
| 433 | consumer = |
| 434 | streamLookupConsumer(group, |
| 435 | receiver->bpop.xread_consumer->ptr, |
no test coverage detected