func (src *GCSSource) streamToPending(ctx context.Context, job tracker.Job) {
	defer close(src.pendingChan)
	files, _, err := src.fileLister(ctx)
	if err != nil {
		log.Println("Error streaming", err)
		metrics.ActiveErrors.WithLabelValues(src.Label(), "filelister").Inc()
		src.ctx.Fail(err)
		return
	}
	index := 0
	dataType := etl.DataType(job.Datatype)
	skipCount := dataType.SkipCount()
	for _, f := range files {
		debug.Println(f)
		if f == nil {
			log.Println("Nil file!!")
			metrics.ActiveErrors.WithLabelValues(src.Label(), "nil file").Inc()
			continue
		}
		if ctx.Err() != nil {
			metrics.ActiveErrors.WithLabelValues(src.Label(), "streamToPending").Inc()
			break
		}
		if index%(skipCount+1) == 0 {
			debug.Printf("Adding gs://%s/%s", f.Bucket, f.Name)
			src.pendingChan <- src.toRunnable(f)
		}
		index++
	}
}
streamToPending feeds tasks to the pending queue until all files have been submitted.
It fetches the list of files once, then converts files to Runnables until all files are
handled, or the context is canceled or expires.
The Runnables are pulled from the queue by Next().
package active
import (
	"context"
	"log"
	"regexp"
	"sync"
	"time"
	"google.golang.org/api/iterator"
	"cloud.google.com/go/storage"
	"github.com/m-lab/etl-gardener/tracker"
	"github.com/m-lab/etl/etl"
	"github.com/m-lab/etl/metrics"
	"github.com/m-lab/go/cloud/gcs"
	"github.com/m-lab/go/logx"
)
var debug = logx.Debug
type FileLister func(ctx context.Context) ([]*storage.ObjectAttrs, int64, error)
func FileListerFunc(bh *gcs.BucketHandle, prefix string, filter *regexp.Regexp) FileLister {
	return func(ctx context.Context) ([]*storage.ObjectAttrs, int64, error) {
		return bh.GetFilesSince(ctx, prefix, filter, time.Time{})
	}
}
type Context struct {
	context.Context
	cancel   func()
	lock     sync.Mutex
	otherErr error  
}
func (c *Context) Err() error {
	c.lock.Lock()
	defer c.lock.Unlock()
	select {
	case <-c.Done():
		if c.otherErr != nil {
			return c.otherErr
		}
		return c.Context.Err()
	default:
		return nil
	}
}
func (c *Context) Fail(err error) {
	debug.Output(2, "stopping")
	c.lock.Lock()
	defer c.lock.Unlock()
	c.otherErr = err
	c.cancel()
}
func WithFail(ctx context.Context) *Context {
	ctx, cancel := context.WithCancel(ctx)
	return &Context{Context: ctx, cancel: cancel}
}
type GCSSource struct {
	ctx *Context
	fileLister FileLister
	toRunnable func(*storage.ObjectAttrs) Runnable
	pendingChan chan Runnable
	label string
}
func NewGCSSource(ctx context.Context, job tracker.Job, fl FileLister, toRunnable func(*storage.ObjectAttrs) Runnable) (*GCSSource, error) {
	src := GCSSource{
		ctx:        WithFail(ctx),
		fileLister: fl,
		toRunnable: toRunnable,
		pendingChan: make(chan Runnable, 0),
		label:       job.Path(),
	}
	go src.streamToPending(ctx, job)
	return &src, nil
}
func (src *GCSSource) Label() string {
	return src.label
}
func (src *GCSSource) CancelStreaming() {
	src.ctx.cancel()
}
func (src *GCSSource) Next(ctx context.Context) (Runnable, error) {
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case next, ok := <-src.pendingChan:
		select {
		case <-ctx.Done():
			log.Println("Context expired in Next")
			metrics.ActiveErrors.WithLabelValues(src.Label(), "context expired in Next").Inc()
			return nil, ctx.Err()
		case <-src.ctx.Done():
			log.Println(src.ctx.Err())
			return nil, src.ctx.Err()
		default:
			if ok {
				return next, nil
			}
			debug.Println("iterator.Done")
			return nil, iterator.Done
		}
	}
}