code: func (b *bufferedBackend) processIncomingEvents(stopCh <-chan struct{}) {
  	defer close(b.buffer)
  
  	var (
  		maxWaitChan  <-chan time.Time
  		maxWaitTimer *time.Timer
  	)
  	// Only use max wait batching if batching is enabled.
  	if b.maxBatchSize > 1 {
  		maxWaitTimer = time.NewTimer(b.maxBatchWait)
  		maxWaitChan = maxWaitTimer.C
  		defer maxWaitTimer.Stop()
  	}
  
  	for {
  		func() {
  			// Recover from any panics caused by this function so a panic in the
  			// goroutine can't bring down the main routine.
  			defer runtime.HandleCrash()
  
  			if b.maxBatchSize > 1 {
  				maxWaitTimer.Reset(b.maxBatchWait)
  			}
  			b.processEvents(b.collectEvents(maxWaitChan, stopCh))
  		}()
  
  		select {
  		case <-stopCh:
  			return
  		default:
  		}
  	}
  }
docs: // processIncomingEvents runs a loop that collects events from the buffer. When
  // b.stopCh is closed, processIncomingEvents stops and closes the buffer.