func (sw *StreamWatcher) Stop() {
	sw.Lock()
	defer sw.Unlock()
	select {
	case <-sw.done:
	default:
		close(sw.done)
		sw.source.Close()
	}
}
Stop implements Interface.
package watch
import (
	"fmt"
	"io"
	"sync"
	"k8s.io/klog/v2"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/net"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
)
type Decoder interface {
	Decode() (action EventType, object runtime.Object, err error)
	Close()
}
type Reporter interface {
	AsObject(err error) runtime.Object
}
type StreamWatcher struct {
	sync.Mutex
	source   Decoder
	reporter Reporter
	result   chan Event
	done     chan struct{}
}
func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {
	sw := &StreamWatcher{
		source:   d,
		reporter: r,
		result: make(chan Event),
		done: make(chan struct{}),
	}
	go sw.receive()
	return sw
}
func (sw *StreamWatcher) ResultChan() <-chan Event {
	return sw.result
}