Method Process

fs/runtime/grpc/grpc_func.go:250–288

(req *proto.FunctionProcessRequest, stream proto.Function_ProcessServer)

}

func (f *FunctionServerImpl) Process(req *proto.FunctionProcessRequest, stream proto.Function_ProcessServer) error {
	runtime, err := f.reconcileSvr.getFunc(req.Name)
	if err != nil {
		return err
	}
	log := runtime.log
	log.Info("Start processing events using GRPC")
	runtime.readyOnce.Do(func() {
		runtime.readyCh <- err
	})
	errCh := make(chan error)

	defer func() {
		if runtime.processing.Load() {
			runtime.output <- nil
			runtime.processing.Store(false)
		}
	}()

	logCounter := common.LogCounter()
	for {
		select {
		case event := <-runtime.input:
			log.Debug("sending event", "count", logCounter)
			runtime.processing.Store(true)
			err := stream.Send(&proto.Event{Payload: string(event.GetPayload())}) // TODO: Change payload type to bytes
			if err != nil {
				log.Error(err, "failed to send event")
				return err
			}
		case <-stream.Context().Done():
			return nil
		case <-runtime.ctx.Done():
			return nil
		case e := <-errCh:
			return e
		}
	}
}

func (f *FunctionServerImpl) getFunctionRuntime(ctx context.Context) (*FuncRuntime, error) {
	md, ok := metadata.FromIncomingContext(ctx)