159
}
160
161
>
func (r *ReaderImpl) Stop() {
reader.go
162
>
if !atomic.CompareAndSwapInt32(
163
>
&r.status,
164
>
common.DaemonStatusStarted,
165
>
common.DaemonStatusStopped,
166
>
) {
167
return
168
}
169
170
>
r.monitor.RemoveReader(r.readerID)
reader.go
171
>
172
>
close(r.shutdownCh)
173
>
r.rateLimitContextCancel()
174
>
if success := common.AwaitWaitGroup(&r.shutdownWG, time.Minute); !success {
175
r.logger.Warn("queue reader shutdown timed out waiting for event loop", tag.LifeCycleStopTimedout)
176
}
177
>
r.logger.Info("queue reader stopped", tag.LifeCycleStopped)
reader.go
178
}
179