this repo has no description
new scheduler when reconnecting firehose
+6
-6
+6
-6
nexus/firehose.go
+6
-6
nexus/firehose.go
···
22
22
}
23
23
24
24
func (fc *FirehoseConsumer) Run(ctx context.Context) error {
25
-
scheduler := parallel.NewScheduler(
26
-
fc.Parallelism,
27
-
100,
28
-
fc.RelayHost,
29
-
fc.Callbacks.EventHandler,
30
-
)
31
25
32
26
u, err := url.Parse(fc.RelayHost)
33
27
if err != nil {
···
73
67
fc.Logger.Info("connected to firehose")
74
68
retries = 0
75
69
70
+
scheduler := parallel.NewScheduler(
71
+
fc.Parallelism,
72
+
100,
73
+
fc.RelayHost,
74
+
fc.Callbacks.EventHandler,
75
+
)
76
76
if err := events.HandleRepoStream(ctx, con, scheduler, nil); err != nil {
77
77
fc.Logger.Warn("firehose connection failed", "err", err)
78
78
}