func (ap *AbstractPoller) waitForEventBatch(eventsChan chan nuclio.Event,
	maxBatchSize int,
	maxBatchDuration time.Duration) ([]nuclio.Event, bool, error) {
	done := false
	eventCycleCompleted := false
	events := make([]nuclio.Event, 0, maxBatchSize)
	deadline := time.Now().Add(maxBatchDuration)
	for !done {
		timeLeft := time.Until(deadline)
		select {
		case receivedEvent := <-eventsChan:
			if receivedEvent == nil {
				eventCycleCompleted = true
				done = true
			} else {
				events = append(events, receivedEvent)
				if len(events) > maxBatchSize {
					done = true
				}
			}
		case <-time.After(timeLeft):
			done = true
		}
	}
	return events, eventCycleCompleted, nil
}
gets a batch of events from the channel. will return when either the max number of events per batch is read, if a
timeout expires or if we get a nil event from the channel indicating the reader completed a cycle
package poller
import (
	"time"
	"github.com/nuclio/nuclio/pkg/functionconfig"
	"github.com/nuclio/nuclio/pkg/processor/trigger"
	"github.com/nuclio/nuclio/pkg/processor/worker"
	"github.com/nuclio/errors"
	"github.com/nuclio/logger"
	"github.com/nuclio/nuclio-sdk-go"
)
type AbstractPoller struct {
	trigger.AbstractTrigger
	configuration *Configuration
	poller        Poller
}
func NewAbstractPoller(logger logger.Logger,
	workerAllocator worker.Allocator,
	configuration *Configuration) (*AbstractPoller, error) {
	abstractTrigger, err := trigger.NewAbstractTrigger(logger,
		workerAllocator,
		&configuration.Configuration,
		"batch",
		"poller",
		configuration.Name)
	if err != nil {
		return nil, errors.New("Failed to create abstract trigger")
	}
	newTrigger := &AbstractPoller{
		AbstractTrigger: abstractTrigger,
		configuration:   configuration,
	}
	return newTrigger, nil
}
func (ap *AbstractPoller) SetPoller(poller Poller) {
	ap.poller = poller
}
func (ap *AbstractPoller) Start(checkpoint functionconfig.Checkpoint) error {
	go ap.getEventsSingleCycle()
	return nil
}
func (ap *AbstractPoller) Stop(force bool) (functionconfig.Checkpoint, error) {
	return nil, nil
}
func (ap *AbstractPoller) getEventsSingleCycle() {
	var eventBatch []nuclio.Event
	var err error
	eventsChan := make(chan nuclio.Event)
	for {
		eventCycleCompleted := false
		go ap.poller.GetNewEvents(eventsChan)  
		for !eventCycleCompleted {
			eventBatch, eventCycleCompleted, err = ap.waitForEventBatch(eventsChan,
				ap.configuration.MaxBatchSize,
				time.Duration(ap.configuration.MaxBatchWaitMs)*time.Millisecond)
			if err != nil {
				continue
			}
			ap.Logger.DebugWith("Got events", "num", len(eventBatch))
			eventResponses, submitError, eventErrors := ap.AllocateWorkerAndSubmitEvents(eventBatch, nil, 10*time.Second)
			if submitError != nil {
				continue
			}
			ap.poller.PostProcessEvents(eventBatch, eventResponses, eventErrors)
		}
		time.Sleep(time.Duration(ap.configuration.IntervalMs) * time.Millisecond)
	}
}