Implement publisher subscriber library using redis streams [NIT-2319] #2196
Conversation
cla-bot
bot
added
the
s
anodar
changed the title
pubsub/consumer.go
Outdated
| var DefaultConsumerConfig = &ConsumerConfig{ | ||
| ResponseEntryTimeout: time.Hour, | ||
| KeepAliveTimeout: 5 * time.Minute, | ||
| RedisStream: "default", |
There was a problem hiding this comment.
for both producer and consumer:
A single redis server should have multiple streams. To avoid confusion/misconfiguration - default should be empty, and producer/consumer should fail to init with empty string (like redisURL)
There was a problem hiding this comment.
Done.
Did the same for RedisGroups as well, I assume same applies there.
pubsub/producer.go
Outdated
| if err != nil { | ||
| return false | ||
| } | ||
| return time.Now().UnixMilli()-val < int64(p.cfg.KeepAliveTimeout.Milliseconds()) |
There was a problem hiding this comment.
redis timeout will delete the value and that is good enough for us.
In the race condition that we detect consumer death only on the next loop - it's o.k. to be a little late
on the other hand, we don't want to mistakingly consider a server dead because there's some offset between it's local time and our local time.
pubsub/producer.go
Outdated
| } | ||
|
|
||
| // Unmarshal converts a JSON byte slice back to the generic type object. | ||
| func (j jsonMarshaller[T]) Unmarshal(val []byte) (T, error) { |
There was a problem hiding this comment.
you don't need Marshaller at all. You can just call json.Marshal/Unmarshal directly from producer/consumer
|
approved with comments |
No description provided.