dps-sdk/dpsv1/client.go

package dpsv1

import (
	"context"
	"encoding/json"
	"git.sxidc.com/service-supports/dps-sdk/pb/v1"
	"git.sxidc.com/service-supports/dps-sdk/pb/v1/request"
	"git.sxidc.com/service-supports/dps-sdk/ports"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

type Client struct {
	conn                 *grpc.ClientConn
	commandServiceClient v1.CommandServiceClient
	queryServiceClient   v1.QueryServiceClient
}

func NewClient(address string) (*Client, error) {
	conn, err := grpc.DialContext(context.Background(), address,
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		return nil, err
	}

	return &Client{
		conn:                 conn,
		commandServiceClient: v1.NewCommandServiceClient(conn),
		queryServiceClient:   v1.NewQueryServiceClient(conn),
	}, nil
}

func DestroyClient(client *Client) error {
	if client == nil {
		return nil
	}

	err := client.conn.Close()
	if err != nil {
		return err
	}

	client.conn = nil
	client.commandServiceClient = nil
	client.queryServiceClient = nil
	client = nil

	return nil
}

func (c *Client) AutoMigrate(req *ports.AutoMigrateRequest) error {
	tableModelDescribeJsonByte, err := json.Marshal(req.TableModelDescribe)
	if err != nil {
		return err
	}

	_, err = c.commandServiceClient.AutoMigrate(context.Background(), &request.AutoMigrateRequest{
		DatabaseID:            req.DatabaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		TableModelDescribe:    tableModelDescribeJsonByte,
	})
	if err != nil {
		return err
	}

	return nil
}

func (c *Client) Insert(req *ports.InsertRequest) (string, error) {
	keysJsonByte, err := json.Marshal(req.Keys)
	if err != nil {
		return "", err
	}

	tableRowJsonByte, err := json.Marshal(req.TableRow)
	if err != nil {
		return "", err
	}

	reply, err := c.commandServiceClient.Insert(context.Background(), &request.InsertRequest{
		DatabaseID:            req.DatabaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Keys:                  keysJsonByte,
		TableRow:              tableRowJsonByte,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

// TODO InsertBatch

func (c *Client) Delete(req *ports.DeleteRequest) (string, error) {
	keysJsonByte, err := json.Marshal(req.Keys)
	if err != nil {
		return "", err
	}

	reply, err := c.commandServiceClient.Delete(context.Background(), &request.DeleteRequest{
		DatabaseID:            req.DatabaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Keys:                  keysJsonByte,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

// TODO DeleteBatch

func (c *Client) Update(req *ports.UpdateRequest) (string, error) {
	keysJsonByte, err := json.Marshal(req.Keys)
	if err != nil {
		return "", err
	}

	newTableRowJsonByte, err := json.Marshal(req.NewTableRow)
	if err != nil {
		return "", err
	}

	reply, err := c.commandServiceClient.Update(context.Background(), &request.UpdateRequest{
		DatabaseID:            req.DatabaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Keys:                  keysJsonByte,
		NewTableRow:           newTableRowJsonByte,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}

func (c *Client) Replay(req *ports.ReplayRequest) (string, error) {
	keysJsonByte, err := json.Marshal(req.Keys)
	if err != nil {
		return "", err
	}

	reply, err := c.commandServiceClient.Replay(context.Background(), &request.ReplayRequest{
		DatabaseID:            req.DatabaseID,
		TablePrefixWithSchema: req.TablePrefixWithSchema,
		Version:               req.Version,
		Keys:                  keysJsonByte,
		UserID:                req.UserID,
	})
	if err != nil {
		return "", err
	}

	return reply.Statement, nil
}