Method Run

fs/instance_impl.go:71–130

(runtime api.FunctionRuntime, sources []<-chan contube.Record,
	sink chan<- contube.Record)

}

func (instance *FunctionInstanceImpl) Run(runtime api.FunctionRuntime, sources []<-chan contube.Record,
	sink chan<- contube.Record) {
	logger := instance.logger
	defer close(sink)

	defer logger.Info("function instance has been stopped")

	logger.Info("function instance is running")

	logCounter := common.LogCounter()
	channels := make([]reflect.SelectCase, len(sources)+1)
	for i, s := range sources {
		channels[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(s)}
	}
	channels[len(sources)] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(instance.ctx.Done())}

	for len(channels) > 0 {
		// Use reflect.Select to select a channel from the slice
		chosen, value, ok := reflect.Select(channels)
		if !ok {
			// The selected channel has been closed, remove it from the slice
			channels = append(channels[:chosen], channels[chosen+1:]...)
			continue
		}

		// Convert the selected value to the type Record
		record := value.Interface().(contube.Record)
		if logger.DebugEnabled() {
			logger.Debug("Calling process function", "count", logCounter)
		}

		// Call the processing function
		output, err := runtime.Call(record)
		if err != nil {
			if errors.Is(err, context.Canceled) {
				return
			}
			// Log the error if there's an issue with the processing function
			logger.Error(err, "failed to process record")
			return
		}

		// If the output is nil, continue with the next iteration
		if output == nil {
			continue
		}

		// Try to send the output to the sink, but also listen to the context's Done channel
		select {
		case sink <- output:
		case <-instance.ctx.Done():
			return
		}

		// If the selected channel is the context's Done channel, exit the loop
		if chosen == len(channels)-1 {
			return
		}