Method Run

input/kcl.go:198–229

Run implements baker.Input.

(inch chan<- *baker.Data)

// Run implements baker.Input.
func (k *KCL) Run(inch chan<- *baker.Data) error {
	k.inch = inch

	wk := worker.NewWorker(k, k.workerCfg)
	if err := wk.Start(); err != nil {
		return fmt.Errorf("input: kcl: can't start the worker with: %v", err)
	}
	defer wk.Shutdown()

	ticker := time.NewTicker(60 * time.Second)
	defer ticker.Stop()
	// Run a loop that periodically checks the number of shards available in the stream,
	// quitting baker as soon as the number changes
	for {
		select {
		case <-k.done:
			return nil
		case <-ticker.C:
			log.Debug("Refreshing shards number")
			n, err := k.totalShards()
			if err != nil {
				log.Errorf("Error refreshing the shards number: %v", err)
				continue
			}
			log.Debugf("Total shards: %d, new shards count: %d", k.streamShards, n)
			if k.streamShards != n {
				log.Info("Shard number has changed, shutting down")
				return nil
			}
		}
	}
}

// Stats implements baker.Input
func (k *KCL) Stats() baker.InputStats {