* Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
*
* Licensed 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 yr
import (
"fmt"
"yuanrong.org/kernel/runtime/libruntime"
"yuanrong.org/kernel/runtime/libruntime/api"
"yuanrong.org/kernel/runtime/libruntime/clibruntime"
)
type ClusterModeRuntime struct {
}
func (r *ClusterModeRuntime) ReleaseInstance(allocation api.InstanceAllocation, stateID string,
abnormal bool, option api.InvokeOptions) {
return
}
func (r *ClusterModeRuntime) AcquireInstance(state string, funcMeta api.FunctionMeta,
invokeOpt api.InvokeOptions) (api.InstanceAllocation, error) {
return api.InstanceAllocation{}, fmt.Errorf("not implement")
}
func (r *ClusterModeRuntime) Init() error {
return libruntime.Init(GetConfigManager().Config)
}
func (r *ClusterModeRuntime) InvokeByFunctionName(
funcMeta api.FunctionMeta, args []api.Arg, invokeOpt api.InvokeOptions,
) (string, error) {
return clibruntime.InvokeByFunctionName(funcMeta, args, invokeOpt)
}
func (r *ClusterModeRuntime) CreateInstance(
funcMeta api.FunctionMeta, args []api.Arg, invokeOpt api.InvokeOptions,
) (string, error) {
return clibruntime.CreateInstance(funcMeta, args, invokeOpt)
}
func (r *ClusterModeRuntime) InvokeByInstanceId(
funcMeta api.FunctionMeta, instanceID string, args []api.Arg, invokeOpt api.InvokeOptions,
) (string, error) {
return clibruntime.InvokeByInstanceId(funcMeta, instanceID, args, invokeOpt)
}
func (r *ClusterModeRuntime) Kill(instanceID string, signal int, payload []byte, invokeOpt api.InvokeOptions) error {
var routeAddress, proxyID string
if invokeOpt.CreateOpt != nil {
if r, ok := invokeOpt.CreateOpt["YR_ROUTE"]; ok && r != "" {
routeAddress = r
}
if p, ok := invokeOpt.CreateOpt["YR_PROXY_ID"]; ok && p != "" {
proxyID = p
}
}
return clibruntime.Kill(instanceID, signal, payload, routeAddress, proxyID)
}
func (r *ClusterModeRuntime) CreateInstanceRaw(createReqRaw []byte, option api.RawRequestOption) ([]byte, error) {
return nil, nil
}
func (r *ClusterModeRuntime) InvokeByInstanceIdRaw(invokeReqRaw []byte, option api.RawRequestOption) ([]byte, error) {
return nil, nil
}
func (r *ClusterModeRuntime) KillRaw(killReqRaw []byte, option api.RawRequestOption) ([]byte, error) {
return nil, nil
}
func (r *ClusterModeRuntime) SaveState(state []byte) (string, error) { return "", nil }
func (r *ClusterModeRuntime) LoadState(checkpointID string) ([]byte, error) { return nil, nil }
func (r *ClusterModeRuntime) Exit(code int, message string) {}
func (r *ClusterModeRuntime) Finalize() {
clibruntime.Finalize()
}
func (r *ClusterModeRuntime) KVSet(key string, value []byte, param api.SetParam) error {
return clibruntime.KVSet(key, value, param)
}
func (r *ClusterModeRuntime) KVSetWithoutKey(value []byte, param api.SetParam) (string, error) {
return "", nil
}
func (r *ClusterModeRuntime) KVMSetTx(keys []string, values [][]byte, param api.MSetParam) error {
return clibruntime.KVMSetTx(keys, values, param)
}
func (r *ClusterModeRuntime) KVGet(key string, timeoutms uint) ([]byte, error) {
return clibruntime.KVGet(key, timeoutms)
}
func (r *ClusterModeRuntime) KVGetMulti(keys []string, timeoutms uint) ([][]byte, error) {
return clibruntime.KVGetMulti(keys, timeoutms)
}
func (r *ClusterModeRuntime) KVDel(key string) error {
return clibruntime.KVDel(key)
}
func (r *ClusterModeRuntime) KVDelMulti(keys []string) ([]string, error) {
return clibruntime.KVDelMulti(keys)
}
func (r *ClusterModeRuntime) CreateProducer(
streamName string, producerConf api.ProducerConf,
) (api.StreamProducer, error) {
return clibruntime.CreateStreamProducer(streamName, producerConf)
}
func (r *ClusterModeRuntime) Subscribe(streamName string, config api.SubscriptionConfig) (api.StreamConsumer, error) {
return clibruntime.CreateStreamConsumer(streamName, config)
}
func (r *ClusterModeRuntime) DeleteStream(streamName string) error {
return clibruntime.DeleteStream(streamName)
}
func (r *ClusterModeRuntime) QueryGlobalProducersNum(streamName string) (uint64, error) {
return clibruntime.QueryGlobalProducersNum(streamName)
}
func (r *ClusterModeRuntime) QueryGlobalConsumersNum(streamName string) (uint64, error) {
return clibruntime.QueryGlobalConsumersNum(streamName)
}
func (r *ClusterModeRuntime) SetTraceID(traceID string) {}
func (r *ClusterModeRuntime) SetTenantID(tenantID string) error {
return clibruntime.SetTenantID(tenantID)
}
func (r *ClusterModeRuntime) Put(objectID string, value []byte, param api.PutParam, nestedObjectIDs ...string) error {
return clibruntime.Put(objectID, value, param, nestedObjectIDs...)
}
func (r *ClusterModeRuntime) PutRaw(
objectID string, value []byte, param api.PutParam, nestedObjectIDs ...string,
) error {
return clibruntime.PutRaw(objectID, value, param, nestedObjectIDs...)
}
func (r *ClusterModeRuntime) Get(objectIDs []string, timeoutMs int) ([][]byte, error) {
return clibruntime.Get(objectIDs, timeoutMs)
}
func (r *ClusterModeRuntime) GetRaw(objectIDs []string, timeoutMs int) ([][]byte, error) {
return clibruntime.GetRaw(objectIDs, timeoutMs)
}
func (r *ClusterModeRuntime) Wait(
objectIDs []string, waitNum uint64, timeoutMs int,
) ([]string, []string, map[string]error) {
return clibruntime.Wait(objectIDs, waitNum, timeoutMs)
}
func (r *ClusterModeRuntime) GIncreaseRef(objectIDs []string, remoteClientID ...string) ([]string, error) {
return clibruntime.GIncreaseRef(objectIDs, remoteClientID...)
}
func (r *ClusterModeRuntime) GIncreaseRefRaw(objectIDs []string, remoteClientID ...string) ([]string, error) {
return clibruntime.GIncreaseRefRaw(objectIDs, remoteClientID...)
}
func (r *ClusterModeRuntime) GDecreaseRef(objectIDs []string, remoteClientID ...string) ([]string, error) {
return nil, nil
}
func (r *ClusterModeRuntime) GDecreaseRefRaw(objectIDs []string, remoteClientID ...string) ([]string, error) {
return clibruntime.GDecreaseRefRaw(objectIDs, remoteClientID...)
}
func (r *ClusterModeRuntime) GetAsync(objectID string, cb api.GetAsyncCallback) {}
func (r *ClusterModeRuntime) GetEvent(objectID string, cb api.GetEventCallback) {}
func (r *ClusterModeRuntime) DeleteGetEventCallback(objectID string) {}
func (r *ClusterModeRuntime) GetFormatLogger() api.FormatLogger { return nil }
func (r *ClusterModeRuntime) ReleaseGRefs(remoteClientID string) error {
return clibruntime.ReleaseGRefs(remoteClientID)
}
func (r *ClusterModeRuntime) CreateClient(config api.ConnectArguments) (api.KvClient, error) {
return nil, nil
}
func (r *ClusterModeRuntime) GetCredential() api.Credential {
return api.Credential{}
}
func (r *ClusterModeRuntime) UpdateSchdulerInfo(schedulerName string, schedulerId string, option string) {
}
func (r *ClusterModeRuntime) SetGauge(data api.GaugeData) error {
return clibruntime.SetGauge(data)
}
func (r *ClusterModeRuntime) IncreaseGauge(data api.GaugeData) error {
return clibruntime.IncreaseGauge(data)
}
func (r *ClusterModeRuntime) DecreaseGauge(data api.GaugeData) error {
return clibruntime.DecreaseGauge(data)
}
func (r *ClusterModeRuntime) IncreaseUInt64Counter(data api.UInt64CounterData) error {
return clibruntime.IncreaseUInt64Counter(data)
}
func (r *ClusterModeRuntime) IsHealth() bool {
return true
}
func (r *ClusterModeRuntime) IsDsHealth() bool {
return true
}
func (r *ClusterModeRuntime) GetActiveMasterAddr() string {
return clibruntime.GetActiveMasterAddr()
}