func (a *vAdapter) readEvents(ctx context.Context, c *event.HistoryCollector) error {
	logger := logging.FromContext(ctx)

	var (
		lastEvent              types.BaseEvent
		lastCheckpointEventKey int32
	)

	bOff := backoff.Backoff{
		Factor: 2,
		Jitter: false,
		Min:    time.Second,
		Max:    5 * time.Second,
	}

	cpTicker := time.NewTicker(a.CpConfig.Period)
	defer cpTicker.Stop()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()

		// checkpoints
		case <-cpTicker.C:
			// avoid unnecessary K8s API calls
			skip := lastEvent == nil || lastCheckpointEventKey == lastEvent.GetEvent().Key
			if !skip {
				logger.Debug("creating checkpoint")
				if err := a.KVStore.Save(ctx); err != nil {
					return fmt.Errorf("save checkpoint: %w", err)
				}
				lastCheckpointEventKey = lastEvent.GetEvent().Key
			} else {
				logger.Debug("skipping checkpoint: no new events since last checkpoint")
			}

		// poll vCenter events
		default:
			events, err := c.ReadNextEvents(ctx, maxEventsBatch)
			if err != nil {
				return fmt.Errorf("read events from vcenter: %w", err)
			}

			if len(events) == 0 {
				delay := bOff.Duration()
				logger.Debugw("no new events, backing off", zap.String("delaySeconds", delay.String()))
				time.Sleep(delay)
				continue
			}

			logger.Debugf("got %d events", len(events))

			n, err := a.sendEvents(ctx, events)
			if err != nil {
				// TODO: return and fail instead?
				logger.Errorf("send events: success %d (total %d): %v", n, len(events), err)

				// 	special case: all events failed so skipping checkpoint
				if n == 0 {
					continue
				}
			}

			if n == 0 && err == nil {
				panic("we should never get here")
			}

			// last successfully sent event from batch
			lastEvent = events[n-1]
			cp := checkpoint{
				VCenter:               a.Source,
				LastEventKey:          lastEvent.GetEvent().Key,
				LastEventType:         getEventDetails(lastEvent).Type,
				LastEventKeyTimestamp: lastEvent.GetEvent().CreatedTime,
				CreatedTimestamp:      time.Now().UTC(),
			}
			if err = a.KVStore.Set(ctx, checkpointKey, cp); err != nil {
				return fmt.Errorf("set checkpoint: %w", err)
			}

			bOff.Reset()
		}
	}
}
// readEvents polls vCenter for new events starting at the configured begin time
// in the provided event history collector. A checkpoint will be periodically
// created and stored in Kubernetes to track successfully processed events
// (ACK-ed by sink).
readEvents polls vCenter for new events starting at the configured begin time
in the provided event history collector. A checkpoint will be periodically
created and stored in Kubernetes to track successfully processed events
(ACK-ed by sink).
// readEvents polls vCenter for new events starting at the configured begin time // in the provided event history collector . A checkpoint will be periodically // created and stored in Kubernetes to track successfully processed events // ( ACK - ed by sink ) .
func (a *vAdapter) readEvents(ctx context.Context, c *event.HistoryCollector) error {
	logger := logging.FromContext(ctx)
	var (
		lastEvent              types.BaseEvent
		lastCheckpointEventKey int32
	)
	bOff := backoff.Backoff{
		Factor: 2,
		Jitter: false,
		Min:    time.Second,
		Max:    5 * time.Second,
	}
	cpTicker := time.NewTicker(a.CpConfig.Period)
	defer cpTicker.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-cpTicker.C:
			skip := lastEvent == nil || lastCheckpointEventKey == lastEvent.GetEvent().Key
			if !skip {
				logger.Debug("creating checkpoint")
				if err := a.KVStore.Save(ctx); err != nil {
					return fmt.Errorf("save checkpoint: %w", err)
				}
				lastCheckpointEventKey = lastEvent.GetEvent().Key
			} else {
				logger.Debug("skipping checkpoint: no new events since last checkpoint")
			}
		default:
			events, err := c.ReadNextEvents(ctx, maxEventsBatch)
			if err != nil {
				return fmt.Errorf("read events from vcenter: %w", err)
			}
			if len(events) == 0 {
				delay := bOff.Duration()
				logger.Debugw("no new events, backing off", zap.String("delaySeconds", delay.String()))
				time.Sleep(delay)
				continue
			}
			logger.Debugf("got %d events", len(events))
			n, err := a.sendEvents(ctx, events)
			if err != nil {
				logger.Errorf("send events: success %d (total %d): %v", n, len(events), err)
				if n == 0 {
					continue
				}
			}
			if n == 0 && err == nil {
				panic("we should never get here")
			}
			lastEvent = events[n-1]
			cp := checkpoint{
				VCenter:               a.Source,
				LastEventKey:          lastEvent.GetEvent().Key,
				LastEventType:         getEventDetails(lastEvent).Type,
				LastEventKeyTimestamp: lastEvent.GetEvent().CreatedTime,
				CreatedTimestamp:      time.Now().UTC(),
			}
			if err = a.KVStore.Set(ctx, checkpointKey, cp); err != nil {
				return fmt.Errorf("set checkpoint: %w", err)
			}
			bOff.Reset()
		}
	}
}
func ( a * vAdapter ) readEvents ( ctx context . Context , c * event . HistoryCollector ) error { logger := logging . FromContext ( ctx ) var ( lastEvent types . BaseEvent lastCheckpointEventKey int32 ) bOff := backoff . Backoff { Factor : 2 , Jitter : false , Min : time . Second , Max : 5 * time . Second , } cpTicker := time . NewTicker ( a . CpConfig . Period ) defer cpTicker . Stop ( ) for { select { case <- ctx . Done ( ) : return ctx . Err ( ) case <- cpTicker . C : skip := lastEvent == nil || lastCheckpointEventKey == lastEvent . GetEvent ( ) . Key if ! skip { logger . Debug ( " " ) if err := a . KVStore . Save ( ctx ) ; err != nil { return fmt . Errorf ( " " , err ) } lastCheckpointEventKey = lastEvent . GetEvent ( ) . Key } else { logger . Debug ( " " ) } default : events , err := c . ReadNextEvents ( ctx , maxEventsBatch ) if err != nil { return fmt . Errorf ( " " , err ) } if len ( events ) == 0 { delay := bOff . Duration ( ) logger . Debugw ( " " , zap . String ( " " , delay . String ( ) ) ) time . Sleep ( delay ) continue } logger . Debugf ( " " , len ( events ) ) n , err := a . sendEvents ( ctx , events ) if err != nil { logger . Errorf ( " " , n , len ( events ) , err ) if n == 0 { continue } } if n == 0 && err == nil { panic ( " " ) } lastEvent = events [ n - 1 ] cp := checkpoint { VCenter : a . Source , LastEventKey : lastEvent . GetEvent ( ) . Key , LastEventType : getEventDetails ( lastEvent ) . Type , LastEventKeyTimestamp : lastEvent . GetEvent ( ) . CreatedTime , CreatedTimestamp : time . Now ( ) . UTC ( ) , } if err = a . KVStore . Set ( ctx , checkpointKey , cp ) ; err != nil { return fmt . Errorf ( " " , err ) } bOff . Reset ( ) } } }
readEvents polls vCenter for new events starting at the configured begin time
in the provided event history collector.
readEvents polls vCenter for new events starting at the configured begin time in the provided event history collector .
// checkpoints // avoid unnecessary K8s API calls // poll vCenter events // TODO: return and fail instead? // 	special case: all events failed so skipping checkpoint // last successfully sent event from batch