| 155 | } |
| 156 | |
| 157 | func (u *upstream) startEventsLoop(doneChan chan struct{}) { |
| 158 | go func() { |
| 159 | for { |
| 160 | select { |
| 161 | case <-doneChan: |
| 162 | return |
| 163 | case event, ok := <-u.events: |
| 164 | if !ok { |
| 165 | return |
| 166 | } |
| 167 | |
| 168 | // We need this loop to catch up if we got a lot of change events |
| 169 | u.eventBufferMutex.Lock() |
| 170 | u.eventBuffer = append(u.eventBuffer, event) |
| 171 | for eventsLeft := true; eventsLeft; { |
| 172 | select { |
| 173 | case event := <-u.events: |
| 174 | u.eventBuffer = append(u.eventBuffer, event) |
| 175 | break |
| 176 | default: |
| 177 | eventsLeft = false |
| 178 | break |
| 179 | } |
| 180 | } |
| 181 | u.eventBufferMutex.Unlock() |
| 182 | } |
| 183 | } |
| 184 | }() |
| 185 | } |
| 186 | |
| 187 | func (u *upstream) getEvents() []notify.EventInfo { |
| 188 | var eventsRef []notify.EventInfo |