func (o *UnaryOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
	log := ctx.GetLogger()
	log.Debugf("Unary operator %s is started", o.name)

	if len(o.outputs) <= 0 {
		go func() { errCh <- fmt.Errorf("no output channel found") }()
		return
	}

	// validate p
	if o.concurrency < 1 {
		o.concurrency = 1
	}
	//reset status
	o.statManagers = nil

	for i := 0; i < o.concurrency; i++ { // workers
		instance := i
		go o.doOp(ctx.WithInstance(instance), errCh)
	}
}
// Exec is the entry point for the executor
Exec is the entry point for the executor
// Exec is the entry point for the executor
func (o *UnaryOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
	log := ctx.GetLogger()
	log.Debugf("Unary operator %s is started", o.name)
	if len(o.outputs) <= 0 {
		go func() { errCh <- fmt.Errorf("no output channel found") }()
		return
	}
	if o.concurrency < 1 {
		o.concurrency = 1
	}
	o.statManagers = nil
	for i := 0; i < o.concurrency; i++ { 
		instance := i
		go o.doOp(ctx.WithInstance(instance), errCh)
	}
}
func ( o * UnaryOperator ) Exec ( ctx api . StreamContext , errCh chan <- error ) { log := ctx . GetLogger ( ) log . Debugf ( " " , o . name ) if len ( o . outputs ) <= 0 { go func ( ) { errCh <- fmt . Errorf ( " " ) } ( ) return } if o . concurrency < 1 { o . concurrency = 1 } o . statManagers = nil for i := 0 ; i < o . concurrency ; i ++ { instance := i go o . doOp ( ctx . WithInstance ( instance ) , errCh ) } }
Exec is the entry point for the executor
Exec is the entry point for the executor
// validate p //reset status // workers