Reporting using kafka #18
Conversation
… feature/reporting-kafka
…tible with gobbler's module system; configuration for Kafka (brokers)
…to Kafka; config, treating errors
…gging errors of asyncproducer, flush periodically
cosminrentea
requested review from
zeisss,
stefanscheidt,
fabzo,
bogh,
marian-craciunescu,
everesio and
robinnepomukmai
| @@ -188,6 +196,9 @@ func assertArguments(a *assert.Assertions) { | |||
| a.Equal("dev", *Config.EnvName) | |||
| a.Equal("mem", *Config.Profile) | |||
|
|
|||
| a.Equal("[ 127.0.0.1:9092 127.0.0.1:9091]", (*Config.KafkaProducer.Brokers).String()) | |||
There was a problem hiding this comment.
There is a space at the beginning in this string. Is this correct?
There was a problem hiding this comment.
took the implementation from another project; will fix it in the next PR.
| type Producer interface { | ||
| service.Startable | ||
| service.Stopable | ||
| Report(topic string, bytes []byte, key string) |
There was a problem hiding this comment.
As a suggestion here. Wouldn't it be nice if we use a io.ReadCloser as the body? If possible 💃
There was a problem hiding this comment.
interesting, I will take a look and try to change the func signature in the next PR, which will contain also more tests
There was a problem hiding this comment.
This would allow a streaming mechanism any time 🗡️ For example you can pass a json.Encoder which accepts an io.writer. And this way you can connect the writer to a reader and encode the data and save memory. All this is theory btw :D not completely sure
| @@ -150,39 +202,52 @@ func (ns *NexmoSender) Send(msg *protocol.Message) error { | |||
| Jitter: true, | |||
| }, | |||
| } | |||
|
|
|||
| err = withRetry.executeAndCheck(sendSms) | |||
| if err != nil && err == ErrRetryFailed { | |||
There was a problem hiding this comment.
I guess it's fine that I removed the first condition err != nil, it was redundant
More Tests will follow in another PR
@bogh @marian-craciunescu