ds-sdk/sdk/sdk.go

package sdk

import (
	"encoding/json"
	"errors"
	"git.sxidc.com/go-tools/utils/strutils"
	"git.sxidc.com/service-supports/ds-sdk/client"
	"git.sxidc.com/service-supports/ds-sdk/grpc_client"
	v1 "git.sxidc.com/service-supports/ds-sdk/grpc_client/v1"
	"git.sxidc.com/service-supports/ds-sdk/grpc_client/v1/request"
	"io"
)

type SDK struct {
	options    *Options
	client     *client.Client
	grpcClient *grpc_client.Client
}

func (s *SDK) CreateDataContainer(name string, spec map[string]any) error {
	if strutils.IsStringEmpty(name) {
		return errors.New("没有传递数据容器名称")
	}

	options := s.options

	err := s.client.CreateDataContainer(options.token, options.baseUrl,
		options.namespace, options.dataSource, name, spec)
	if err != nil {
		return err
	}

	return nil
}

func (s *SDK) DeleteDataContainer(name string) error {
	if strutils.IsStringEmpty(name) {
		return errors.New("没有传递数据容器名称")
	}

	options := s.options

	err := s.client.DeleteDataContainer(options.token, options.baseUrl,
		options.namespace, options.dataSource, name)
	if err != nil {
		return err
	}

	return nil
}

func (s *SDK) GetDataContainers(name string, pageNo int, pageSize int) ([]client.DataContainerInfo, error) {
	options := s.options

	return s.client.GetDataContainers(options.token, options.baseUrl,
		options.namespace, options.dataSource, name, pageNo, pageSize)
}

// TODO 增加Event版本的Container创建

func (s *SDK) ExecuteRawSql(sql string, executeParams map[string]any) ([]map[string]any, error) {
	if strutils.IsStringEmpty(sql) {
		return make([]map[string]any, 0), nil
	}

	options := s.options

	results, err := s.client.ExecuteRawSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, sql, executeParams)
	if err != nil {
		return nil, err
	}

	return results, nil
}

func (s *SDK) CreateSQL(name string, spec map[string]any) error {
	if strutils.IsStringEmpty(name) {
		return errors.New("没有传递SQL资源名称")
	}

	options := s.options

	err := s.client.CreateSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, name, spec)
	if err != nil {
		return err
	}

	return nil
}

func (s *SDK) DeleteSQL(name string) error {
	if strutils.IsStringEmpty(name) {
		return errors.New("没有传递SQL资源名称")
	}

	options := s.options

	err := s.client.DeleteSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, name)
	if err != nil {
		return err
	}

	return nil
}

func (s *SDK) ExecuteSql(name string, executeParams map[string]any) ([]map[string]any, error) {
	if strutils.IsStringEmpty(name) {
		return nil, errors.New("没有传递SQL资源名称")
	}

	options := s.options

	results, err := s.client.ExecuteSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, name, executeParams)
	if err != nil {
		return nil, err
	}

	return results, nil
}

type TxFunc func(tx *Transaction) error

type Transaction struct {
	stream v1.SqlService_TransactionClient
}

func (tx *Transaction) ExecuteRawSql(sql string, executeParams map[string]any) error {
	var retErr error

	defer func() {
		if retErr != nil {
			innerErr := tx.stream.CloseSend()
			if innerErr != nil {
				panic(innerErr)
			}
		}
	}()

	executeParamsJsonBytes, err := json.Marshal(executeParams)
	if err != nil {
		retErr = err
		return retErr
	}

	err = tx.stream.SendMsg(&request.TransactionOperation{
		Request: &request.TransactionOperation_ExecuteRawSqlRequest{
			ExecuteRawSqlRequest: &request.ExecuteRawSqlRequest{
				SQL:           sql,
				ExecuteParams: string(executeParamsJsonBytes),
			},
		},
	})
	if err != nil {
		retErr = err
		return retErr
	}

	_, err = tx.stream.Recv()
	if err != nil {
		retErr = err
		return retErr
	}

	return nil
}

func (tx *Transaction) ExecuteSql(name string, executeParams map[string]any) error {
	var retErr error

	defer func() {
		if retErr != nil {
			innerErr := tx.stream.CloseSend()
			if innerErr != nil {
				panic(innerErr)
			}
		}
	}()

	executeParamsJsonBytes, err := json.Marshal(executeParams)
	if err != nil {
		retErr = err
		return retErr
	}

	err = tx.stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_ExecuteSqlRequest{
			ExecuteSqlRequest: &request.ExecuteSqlRequest{
				Name:          name,
				ExecuteParams: string(executeParamsJsonBytes),
			},
		},
	})
	if err != nil {
		retErr = err
		return retErr
	}

	_, err = tx.stream.Recv()
	if err != nil {
		retErr = err
		return retErr
	}

	return nil
}

func (s *SDK) Transaction(txFunc TxFunc) error {
	stream, err := s.grpcClient.Transaction()
	if err != nil {
		return err
	}

	defer func() {
		innerErr := stream.CloseSend()
		if innerErr != nil {
			panic(innerErr)
		}
	}()

	err = stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_TransactionBeginRequest{
			TransactionBeginRequest: &request.TransactionBeginRequest{
				Token:      s.options.token,
				Namespace:  s.options.namespace,
				DataSource: s.options.dataSource,
			},
		}})
	if err != nil {
		return err
	}

	err = txFunc(&Transaction{
		stream: stream,
	})
	if err != nil {
		return err
	}

	err = stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_TransactionEndRequest{
			TransactionEndRequest: &request.TransactionEndRequest{},
		}})
	if err != nil {
		return err
	}

	_, err = stream.Recv()
	if err != nil && err != io.EOF {
		return err
	}

	return nil
}