Copilot commented on code in PR #868: URL: https://github.com/apache/incubator-seata-go/pull/868#discussion_r2261992054
########## pkg/saga/statemachine/engine/invoker/local_invoker.go: ########## @@ -0,0 +1,178 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package invoker + +import ( + "context" + "fmt" + "github.com/seata/seata-go/pkg/saga/statemachine/statelang/state" + "reflect" + "sync" +) + +type LocalServiceInvoker struct { + serviceRegistry map[string]interface{} + methodCache map[string]*reflect.Method + jsonParser JsonParser + mutex sync.RWMutex +} + +func NewLocalServiceInvoker() *LocalServiceInvoker { + return &LocalServiceInvoker{ + serviceRegistry: make(map[string]interface{}), + methodCache: make(map[string]*reflect.Method), + jsonParser: &DefaultJsonParser{}, + } +} + +func (l *LocalServiceInvoker) RegisterService(serviceName string, instance interface{}) { + l.mutex.Lock() + defer l.mutex.Unlock() + l.serviceRegistry[serviceName] = instance +} + +func (l *LocalServiceInvoker) Invoke(ctx context.Context, input []any, service state.ServiceTaskState) ([]reflect.Value, error) { + serviceName := service.ServiceName() + instance, exists := l.serviceRegistry[serviceName] + if !exists { + return nil, fmt.Errorf("service %s not registered", serviceName) + } + + methodName := service.ServiceMethod() + method, err := l.getMethod(serviceName, methodName, service.ParameterTypes()) + if err != nil { + return nil, err + } + + params, err := l.resolveParameters(input, method.Type) + if err != nil { + return nil, err + } + + return l.invokeMethod(instance, method, params), nil +} + +func (l *LocalServiceInvoker) resolveMethod(key, serviceName, methodName string) (*reflect.Method, error) { + l.mutex.Lock() + defer l.mutex.Unlock() + + if cachedMethod, ok := l.methodCache[key]; ok { + return cachedMethod, nil + } + + instance, exists := l.serviceRegistry[serviceName] + if !exists { + return nil, fmt.Errorf("service %s not found", serviceName) + } + + objType := reflect.TypeOf(instance) + method, ok := objType.MethodByName(methodName) + if !ok { + return nil, fmt.Errorf("method %s not found in service %s", methodName, serviceName) + } + + l.methodCache[key] = &method + return &method, nil +} + +func (l *LocalServiceInvoker) getMethod(serviceName, methodName string, paramTypes []string) (*reflect.Method, error) { + key := fmt.Sprintf("%s.%s", serviceName, methodName) + + l.mutex.RLock() + if method, ok := l.methodCache[key]; ok { + l.mutex.RUnlock() + return method, nil + } + l.mutex.RUnlock() + + return l.resolveMethod(key, serviceName, methodName) +} + +func (l *LocalServiceInvoker) resolveParameters(input []any, methodType reflect.Type) ([]reflect.Value, error) { + paramCount := methodType.NumIn() - 1 + params := make([]reflect.Value, paramCount) + + for i := 0; i < paramCount; i++ { + methodParamIndex := i + 1 + paramType := methodType.In(methodParamIndex) + + if i >= len(input) { + params[i] = reflect.Zero(paramType) + continue + } + Review Comment: When input length is less than required parameters, the function sets missing parameters to zero values, but this may not be the intended behavior. Consider returning an error for insufficient parameters instead of silently using zero values. ```suggestion if len(input) < paramCount { return nil, fmt.Errorf("insufficient parameters: expected %d, got %d", paramCount, len(input)) } params := make([]reflect.Value, paramCount) for i := 0; i < paramCount; i++ { methodParamIndex := i + 1 paramType := methodType.In(methodParamIndex) ``` ########## pkg/saga/statemachine/engine/invoker/local_invoker.go: ########## @@ -0,0 +1,178 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package invoker + +import ( + "context" + "fmt" + "github.com/seata/seata-go/pkg/saga/statemachine/statelang/state" + "reflect" + "sync" +) + +type LocalServiceInvoker struct { + serviceRegistry map[string]interface{} + methodCache map[string]*reflect.Method + jsonParser JsonParser + mutex sync.RWMutex +} + +func NewLocalServiceInvoker() *LocalServiceInvoker { + return &LocalServiceInvoker{ + serviceRegistry: make(map[string]interface{}), + methodCache: make(map[string]*reflect.Method), + jsonParser: &DefaultJsonParser{}, + } +} + +func (l *LocalServiceInvoker) RegisterService(serviceName string, instance interface{}) { + l.mutex.Lock() + defer l.mutex.Unlock() + l.serviceRegistry[serviceName] = instance +} + +func (l *LocalServiceInvoker) Invoke(ctx context.Context, input []any, service state.ServiceTaskState) ([]reflect.Value, error) { + serviceName := service.ServiceName() + instance, exists := l.serviceRegistry[serviceName] + if !exists { + return nil, fmt.Errorf("service %s not registered", serviceName) + } + + methodName := service.ServiceMethod() + method, err := l.getMethod(serviceName, methodName, service.ParameterTypes()) + if err != nil { + return nil, err + } + + params, err := l.resolveParameters(input, method.Type) + if err != nil { + return nil, err + } + + return l.invokeMethod(instance, method, params), nil +} + +func (l *LocalServiceInvoker) resolveMethod(key, serviceName, methodName string) (*reflect.Method, error) { + l.mutex.Lock() + defer l.mutex.Unlock() + + if cachedMethod, ok := l.methodCache[key]; ok { + return cachedMethod, nil + } + + instance, exists := l.serviceRegistry[serviceName] + if !exists { + return nil, fmt.Errorf("service %s not found", serviceName) + } + + objType := reflect.TypeOf(instance) + method, ok := objType.MethodByName(methodName) + if !ok { + return nil, fmt.Errorf("method %s not found in service %s", methodName, serviceName) + } + + l.methodCache[key] = &method + return &method, nil +} + +func (l *LocalServiceInvoker) getMethod(serviceName, methodName string, paramTypes []string) (*reflect.Method, error) { + key := fmt.Sprintf("%s.%s", serviceName, methodName) + + l.mutex.RLock() + if method, ok := l.methodCache[key]; ok { + l.mutex.RUnlock() + return method, nil + } + l.mutex.RUnlock() + + return l.resolveMethod(key, serviceName, methodName) +} + +func (l *LocalServiceInvoker) resolveParameters(input []any, methodType reflect.Type) ([]reflect.Value, error) { + paramCount := methodType.NumIn() - 1 + params := make([]reflect.Value, paramCount) + + for i := 0; i < paramCount; i++ { + methodParamIndex := i + 1 Review Comment: The parameter count calculation assumes the first parameter is always the receiver, but this may not be correct for all method types. This could cause index out of bounds errors when methodType.NumIn() returns 0. ```suggestion numIn := methodType.NumIn() var paramStart, paramCount int // If methodType represents a method (not a function), the first parameter is the receiver. // For functions, there is no receiver. if numIn == 0 { return []reflect.Value{}, nil } // Assume receiver is present (as in original code), but guard against negative values. paramStart = 1 paramCount = numIn - paramStart if paramCount < 0 { paramCount = 0 } params := make([]reflect.Value, paramCount) for i := 0; i < paramCount; i++ { methodParamIndex := i + paramStart ``` ########## pkg/saga/statemachine/engine/invoker/javascript_script_invoker.go: ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package invoker + +import ( + "context" + "fmt" + "sync" + + "github.com/robertkrimen/otto" +) + +type JavaScriptScriptInvoker struct { + mutex sync.Mutex + jsonParser JsonParser + closed bool +} + +func NewJavaScriptScriptInvoker() *JavaScriptScriptInvoker { + return &JavaScriptScriptInvoker{ + jsonParser: &DefaultJsonParser{}, + closed: false, + } +} + +func (j *JavaScriptScriptInvoker) Type() string { + return "javascript" +} + +func (j *JavaScriptScriptInvoker) Invoke(ctx context.Context, script string, params map[string]interface{}) (interface{}, error) { + j.mutex.Lock() + defer j.mutex.Unlock() + + if j.closed { + return nil, fmt.Errorf("javascript invoker has been closed") + } + + vm := otto.New() Review Comment: Creating a new Otto VM instance for each invocation is inefficient. Consider implementing VM pooling or reusing VM instances to improve performance, especially under high concurrency. ```suggestion if j.closed { return nil, fmt.Errorf("javascript invoker has been closed") } var vm *otto.Otto select { case vm = <-j.vmPool: // got a VM from the pool default: // pool empty, create a new one vm = otto.New() } // Clean up VM state before use (optional, depending on use case) defer func() { // Return VM to pool if not closed j.mutex.Lock() defer j.mutex.Unlock() if !j.closed { select { case j.vmPool <- vm: default: // pool full, discard VM } } }() ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@seata.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: notifications-unsubscr...@seata.apache.org For additional commands, e-mail: notifications-h...@seata.apache.org
