subscribe starts our subscriptions to peer and tunnnel updates in a new goroutine, and returns once we are subscribed or the querier context is canceled.
()
| 1187 | // subscribe starts our subscriptions to peer and tunnnel updates in a new goroutine, and returns once we are subscribed |
| 1188 | // or the querier context is canceled. |
| 1189 | func (q *querier) subscribe() { |
| 1190 | subscribed := make(chan struct{}) |
| 1191 | go func() { |
| 1192 | defer close(subscribed) |
| 1193 | eb := backoff.NewExponentialBackOff() |
| 1194 | eb.MaxElapsedTime = 0 // retry indefinitely |
| 1195 | eb.MaxInterval = dbMaxBackoff |
| 1196 | bkoff := backoff.WithContext(eb, q.ctx) |
| 1197 | var cancelPeer context.CancelFunc |
| 1198 | err := backoff.Retry(func() error { |
| 1199 | cancelFn, err := q.pubsub.SubscribeWithErr(eventPeerUpdate, q.listenPeer) |
| 1200 | if err != nil { |
| 1201 | q.logger.Warn(q.ctx, "failed to subscribe to peer updates", slog.Error(err)) |
| 1202 | return err |
| 1203 | } |
| 1204 | cancelPeer = cancelFn |
| 1205 | return nil |
| 1206 | }, bkoff) |
| 1207 | if err != nil { |
| 1208 | if q.ctx.Err() == nil { |
| 1209 | q.logger.Error(q.ctx, "code bug: retry failed before context canceled", slog.Error(err)) |
| 1210 | } |
| 1211 | return |
| 1212 | } |
| 1213 | defer func() { |
| 1214 | q.logger.Info(q.ctx, "canceling peer updates subscription") |
| 1215 | cancelPeer() |
| 1216 | }() |
| 1217 | bkoff.Reset() |
| 1218 | q.logger.Info(q.ctx, "subscribed to peer updates") |
| 1219 | |
| 1220 | var cancelTunnel context.CancelFunc |
| 1221 | err = backoff.Retry(func() error { |
| 1222 | cancelFn, err := q.pubsub.SubscribeWithErr(eventTunnelUpdate, q.listenTunnel) |
| 1223 | if err != nil { |
| 1224 | q.logger.Warn(q.ctx, "failed to subscribe to tunnel updates", slog.Error(err)) |
| 1225 | return err |
| 1226 | } |
| 1227 | cancelTunnel = cancelFn |
| 1228 | return nil |
| 1229 | }, bkoff) |
| 1230 | if err != nil { |
| 1231 | if q.ctx.Err() == nil { |
| 1232 | q.logger.Error(q.ctx, "code bug: retry failed before context canceled", slog.Error(err)) |
| 1233 | } |
| 1234 | return |
| 1235 | } |
| 1236 | defer func() { |
| 1237 | q.logger.Info(q.ctx, "canceling tunnel updates subscription") |
| 1238 | cancelTunnel() |
| 1239 | }() |
| 1240 | q.logger.Info(q.ctx, "subscribed to tunnel updates") |
| 1241 | |
| 1242 | var cancelRFH context.CancelFunc |
| 1243 | err = backoff.Retry(func() error { |
| 1244 | cancelFn, err := q.pubsub.SubscribeWithErr(eventReadyForHandshake, q.listenReadyForHandshake) |
| 1245 | if err != nil { |
| 1246 | q.logger.Warn(q.ctx, "failed to subscribe to ready for handshakes", slog.Error(err)) |
no test coverage detected