Alternatively you can stop by context. This is better because with channels for stopping it could block if for some reason the other side is already stopped or blocked by something else, etc.
type Watcher struct {
events chan Event
cancel context.CancelFunc
}
func (w *Watcher) Watch() <-chan Event {
ctx, cancel := context.WithCancel(context.Background())
w.cancel = cancel
go w.fetch(ctx)
return w.events
}
func (w *Watcher) Stop() {
w.cancel()
}
func (w *Watcher) fetch(ctx context.Context) {
for {
e := fetchEvent()
select {
case <-ctx.Done():
return
case w.events <- e:
}
}
}
This example creates a Context in the Watch function (before fetching gets started) and registers the cancel function in the Watcher. The ctx is then passed on to fetch which checks for ctx.Done on every iteration. If the context is Done, it will stop fetching.
You might want to initialize the cancel function to be an empty function so it is never nil and Stop can be called without a panic even if the Watcher was never started.
func NewWatcher() *Watcher {
return &Watcher{
// ...
cancel: func() {},
}
}
If you additionally want to pass in a context from the outside, you can rewrite your Watch function to:
func (w *Watcher) Watch(ctx context.Context) <-chan Event {
ctx, cancel := context.WithCancel(ctx)
w.cancel = cancel
w.fetch(ctx)
return w.events
}
Now the Watcher will stop to fetch if the Stop function is called or the outside ctx is cancelled (e.g. if your application shuts down).
If the Watcher.events channel is not used anywhere else, you can create it in the Watch function and pass it to fetch as well as return it. That way the Watch function can also be called multiple times and they will have separate channels, and not multiple readers reading from the same channel.
If you do this, you probably also want to have a way to cancel them individually, so you could return the cancel function from the Watch function and remove it from the Watcher struct. Also Stop is now obsolete.
func (w *Watcher) Watch() (<-chan Event, context.CancelFunc) {
ctx, cancel := context.WithCancel(context.Background())
// with small buffer for channel to smooth sending/receiving
ch := make(chan Event, 5)
go w.fetch(ctx, cancel, ch)
return ch, cancel
}
func (w *Watcher) fetch(ctx context.Context, cancel context.CancelFunc, ch chan<- Event) {
// make sure the context is cancelled always if `fetch` stops to avoid goroutine leak.
defer cancel()
for {
select {
case <-ctx.Done():
// now you can close it without problems
close(ch)
return
case ch <- fetchEvent():
}
}
}