code: func (w *bulkWorker) work(ctx context.Context) {
  	defer func() {
  		w.p.workerWg.Done()
  		close(w.flushAckC)
  		close(w.flushC)
  	}()
  
  	var stop bool
  	for !stop {
  		var err error
  		select {
  		case req, open := <-w.p.requestsC:
  			if open {
  				// Received a new request
  				if _, err = req.Source(); err == nil {
  					w.service.Add(req)
  					if w.commitRequired() {
  						err = w.commit(ctx)
  					}
  				}
  			} else {
  				// Channel closed: Stop.
  				stop = true
  				if w.service.NumberOfActions() > 0 {
  					err = w.commit(ctx)
  				}
  			}
  
  		case <-w.flushC:
  			// Commit outstanding requests
  			if w.service.NumberOfActions() > 0 {
  				err = w.commit(ctx)
  			}
  			w.flushAckC <- struct{}{}
  		}
  		if err != nil {
  			w.p.c.errorf("elastic: bulk processor %q was unable to perform work: %v", w.p.name, err)
  			if !stop {
  				waitForActive := func() {
  					// Add back pressure to prevent Add calls from filling up the request queue
  					ready := make(chan struct{})
  					go w.waitForActiveConnection(ready)
  					<-ready
  				}
  				if _, ok := err.(net.Error); ok {
  					waitForActive()
  				} else if IsConnErr(err) {
  					waitForActive()
  				}
  			}
  		}
  	}
  }
docs: // work waits for bulk requests and manual flush calls on the respective
  // channels and is invoked as a goroutine when the bulk processor is started.