run is the main loop of the notifier.
(success chan<- dispatchResult, failure chan<- dispatchResult)
| 87 | |
| 88 | // run is the main loop of the notifier. |
| 89 | func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchResult) error { |
| 90 | n.log.Info(n.outerCtx, "started") |
| 91 | |
| 92 | defer func() { |
| 93 | close(n.done) |
| 94 | n.log.Info(context.Background(), "gracefully stopped") |
| 95 | }() |
| 96 | |
| 97 | // TODO: idea from Cian: instead of querying the database on a short interval, we could wait for pubsub notifications. |
| 98 | // if 100 notifications are enqueued, we shouldn't activate this routine for each one; so how to debounce these? |
| 99 | // PLUS we should also have an interval (but a longer one, maybe 1m) to account for retries (those will not get |
| 100 | // triggered by a code path, but rather by a timeout expiring which makes the message retryable) |
| 101 | |
| 102 | // run the ticker with the graceful context, so we stop fetching after stop() is called |
| 103 | tick := n.clock.TickerFunc(n.gracefulCtx, n.cfg.FetchInterval.Value(), func() error { |
| 104 | // Check if notifier is not paused. |
| 105 | ok, err := n.ensureRunning(n.outerCtx) |
| 106 | if err != nil { |
| 107 | n.log.Warn(n.outerCtx, "failed to check notifier state", slog.Error(err)) |
| 108 | } |
| 109 | |
| 110 | if ok { |
| 111 | err = n.process(n.outerCtx, success, failure) |
| 112 | if err != nil { |
| 113 | n.log.Error(n.outerCtx, "failed to process messages", slog.Error(err)) |
| 114 | } |
| 115 | } |
| 116 | // we don't return any errors because we don't want to kill the loop because of them. |
| 117 | return nil |
| 118 | }, "notifier", "fetchInterval") |
| 119 | |
| 120 | _ = tick.Wait() |
| 121 | // only errors we can return are context errors. Only return an error if the outer context |
| 122 | // was canceled, not if we were gracefully stopped. |
| 123 | if n.outerCtx.Err() != nil { |
| 124 | return xerrors.Errorf("notifier %q context canceled: %w", n.id, n.outerCtx.Err()) |
| 125 | } |
| 126 | return nil |
| 127 | } |
| 128 | |
| 129 | // ensureRunning checks if notifier is not paused. |
| 130 | func (n *notifier) ensureRunning(ctx context.Context) (bool, error) { |