Implement Message Handler #20
Conversation
Codecov Report❌ Patch coverage is ❌ Your project check has failed because the head coverage (67.66%) is below the target coverage (72.00%). You can increase the head coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## main #20 +/- ##
===========================================
- Coverage 84.64% 67.66% -16.99%
===========================================
Files 20 24 +4
Lines 1029 1373 +344
===========================================
+ Hits 871 929 +58
- Misses 141 427 +286
Partials 17 17
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
piccione99
left a comment
piccione99
left a comment
There was a problem hiding this comment.
Looks good! Just a few minor comments.
| topic: "device-lifecycle" | ||
| - pattern: "device-status" | ||
| topic: "device-status" | ||
| - pattern: "*" |
There was a problem hiding this comment.
for ease of use, we might want to just keep just the last pattern and topic (* and default-events) as the default route so we only need one route in our default kafka setup
internal/consumer/consumer.go
Outdated
| } | ||
|
|
||
| consumer.client = client | ||
| consumer.handler = consumer.config.handler |
There was a problem hiding this comment.
can we change the WithMessageHandler option in options.go to directly set consumer.handler so we can remove the above line? (I think the consumerConfig object is superfluous and awkward, my bad. I can clean that up.)
internal/consumer/handler.go
Outdated
| h.emitLog(log.LevelError, "failed to produce WRP message", map[string]any{ | ||
| "error": err.Error(), | ||
| }) | ||
| return fmt.Errorf("production failed: %w", err) |
There was a problem hiding this comment.
can we say "produce failed" since "production failed" sounds more alarming
internal/consumer/handler.go
Outdated
| return fmt.Errorf("production failed: %w", err) | ||
| } | ||
|
|
||
| h.emitLog(log.LevelInfo, "successfully routed WRP message", map[string]any{ |
There was a problem hiding this comment.
can we make this debug since this will be extremely high volume
| return outcome, fmt.Errorf("failed to produce message: %w", err) | ||
| } | ||
|
|
||
| p.metricEmitter.Notify(metrics.Event{ |
There was a problem hiding this comment.
can we make this "messsages_produced" and use "outcome" as a label? Then we can remove the metric on line 160 (and return outside the if statement) since that should cover errors and successes. Once we set up the metrics and inject them (that can be a separate PR), we probably want to define constants in metrics/fx.go and use them instead of hardcoding them. Given how particular prometheus is about labels. But the labels are fine for this PR.
There was a problem hiding this comment.
we can also generate metrics using a PublishEvent listener for the async messages. but that's fine to leave until the prometheus metrics are set up in the metrics package.
| } | ||
|
|
||
| // Emit metrics about the message being produced | ||
| p.metricEmitter.Notify(metrics.Event{ |
There was a problem hiding this comment.
does this metric add something different than the metric on line 169 or can we remove it?
| opts := []publisher.Option{ | ||
| // Observability | ||
| publisher.WithLogEmitter(in.LogEmitter), | ||
| publisher.WithMetricsEmitter(in.MetricEmitter), |
There was a problem hiding this comment.
Just FYI - since we are injecting the metric emitter instead of using the noop, we will get errors when we try to emit the metrics for the publisher, since they are missing. that's probably fine because the call to emit metrics is async so just that go routine will panic.
2 participants
Closes #7