forked from
tangled.org/core
Monorepo for Tangled
knotmirror/knotstream: use long-live context for subscription
Signed-off-by: Seongmin Lee <git@boltless.me>
authored by
Seongmin Lee
and committed by
tangled.org
1164e7a3
ffd2e000
+18
-3
+12
-2
knotmirror/knotstream/knotstream.go
+12
-2
knotmirror/knotstream/knotstream.go
···
66
66
if host.Status == models.HostStatusBanned {
67
67
return fmt.Errorf("cannot subscribe to banned knot")
68
68
}
69
-
return s.slurper.Subscribe(ctx, *host)
69
+
// `Subscribe` expects long-living context
70
+
if err := s.slurper.Subscribe(*host); err != nil {
71
+
return fmt.Errorf("slurper: %w", err)
72
+
}
73
+
74
+
host.Status = models.HostStatusActive
75
+
if err := db.UpsertHost(ctx, s.db, host); err != nil {
76
+
return fmt.Errorf("upserting host status to db: %w", err)
77
+
}
78
+
79
+
return nil
70
80
}
71
81
72
82
func (s *KnotStream) ResubscribeAllHosts(ctx context.Context) error {
···
78
88
for _, host := range hosts {
79
89
l := s.logger.With("hostname", host.Hostname)
80
90
l.Info("re-subscribing to active host")
81
-
if err := s.slurper.Subscribe(ctx, host); err != nil {
91
+
if err := s.slurper.Subscribe(host); err != nil {
82
92
l.Warn("failed to re-subscribe to host", "err", err)
83
93
}
84
94
// sleep for a very short period, so we don't open tons of sockets at the same time
+6
-1
knotmirror/knotstream/slurper.go
+6
-1
knotmirror/knotstream/slurper.go
···
89
89
return err
90
90
}
91
91
92
-
func (s *KnotSlurper) Subscribe(ctx context.Context, host models.Host) error {
92
+
func (s *KnotSlurper) Subscribe(host models.Host) error {
93
93
s.subsLk.Lock()
94
94
defer s.subsLk.Unlock()
95
95
···
109
109
}
110
110
s.subs[host.Hostname] = sub
111
111
112
+
// TODO: use service level context, not the top-most one.
113
+
// Using top-most context should be avoided to do graceful shutdown.
114
+
ctx := context.TODO()
115
+
112
116
sub.scheduler.Start(ctx)
113
117
go s.subscribeWithRedialer(ctx, host, sub)
114
118
return nil
···
120
124
s.subsLk.Lock()
121
125
defer s.subsLk.Unlock()
122
126
127
+
l.Info("unsubscribing knot")
123
128
delete(s.subs, host.Hostname)
124
129
}()
125
130