github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-14 06:15:45 +00:00
Code Issues Projects Releases Wiki Activity

Expose HistoricalFIFO's pop with cancel channel

Browse Source
This commit is contained in:
Dr. Stefan Schimanski 2015-09-17 11:05:10 +02:00
parent d49ca164ef
commit a496e8dd20
1 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
contrib/mesos/pkg/queue/historical.go
View File

@ -222,7 +222,7 @@ func (f *HistoricalFIFO) Poll(id string, t EventType) bool {
func (q *HistoricalFIFO) Await(timeout time.Duration) interface{} {
cancel := make(chan struct{})
ch := make(chan interface{}, 1)
go func() { ch <- q.pop(cancel) }()
go func() { ch <- q.CancelablePop(cancel) }()
select {
case <-time.After(timeout):
close(cancel)
@ -232,10 +232,10 @@ func (q *HistoricalFIFO) Await(timeout time.Duration) interface{} {
}
}
func (f *HistoricalFIFO) Pop() interface{} {
return f.pop(nil)
return f.CancelablePop(nil)
}
func (f *HistoricalFIFO) pop(cancel chan struct{}) interface{} {
func (f *HistoricalFIFO) CancelablePop(cancel <-chan struct{}) interface{} {
popEvent := (Entry)(nil)
defer func() {
f.carrier(popEvent)
@ -383,7 +383,7 @@ func (f *HistoricalFIFO) merge(id string, obj UniqueCopyable) (notifications []E
// NewHistorical returns a Store which can be used to queue up items to
// process. If a non-nil Mux is provided, then modifications to the
// the FIFO are delivered on a channel specific to this fifo.
func NewHistorical(ch chan<- Entry) FIFO {
func NewHistorical(ch chan<- Entry) *HistoricalFIFO {
carrier := dead
if ch != nil {
carrier = func(msg Entry) {