diff --git a/src/blocked.c b/src/blocked.c index d8ae596d..4a667501 100644 --- a/src/blocked.c +++ b/src/blocked.c @@ -397,10 +397,7 @@ void handleClientsBlockedOnKeys(void) { } } - if (s->last_id.ms > gt->ms || - (s->last_id.ms == gt->ms && - s->last_id.seq > gt->seq)) - { + if (streamCompareID(&s->last_id, gt) > 0) { streamID start = *gt; start.seq++; /* Can't overflow, it's an uint64_t */ diff --git a/src/stream.h b/src/stream.h index 61210f95..ef08753b 100644 --- a/src/stream.h +++ b/src/stream.h @@ -108,5 +108,6 @@ streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int create); streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id); streamNACK *streamCreateNACK(streamConsumer *consumer); void streamDecodeID(void *buf, streamID *id); +int streamCompareID(streamID *a, streamID *b); #endif diff --git a/src/t_stream.c b/src/t_stream.c index 3b4ebea7..7b1076b1 100644 --- a/src/t_stream.c +++ b/src/t_stream.c @@ -1401,9 +1401,7 @@ void xreadCommand(client *c) { * synchronously in case the group top item delivered is smaller * than what the stream has inside. */ streamID *last = &groups[i]->last_id; - if (s->last_id.ms > last->ms || - (s->last_id.ms == last->ms && s->last_id.seq > last->seq)) - { + if (streamCompareID(&s->last_id, last) > 0) { serve_synchronously = 1; *gt = *last; } @@ -1411,9 +1409,7 @@ void xreadCommand(client *c) { } else { /* For consumers without a group, we serve synchronously if we can * actually provide at least one item from the stream. */ - if (s->last_id.ms > gt->ms || - (s->last_id.ms == gt->ms && s->last_id.seq > gt->seq)) - { + if (streamCompareID(&s->last_id, gt) > 0) { serve_synchronously = 1; } }