dapr_client/client.go

package dapr_client

import (
	"context"
	dapr "github.com/dapr/go-sdk/client"
	"os"
	"time"
)

const (
	daprPortDefault    = "50001"
	daprPortEnvVarName = "DAPR_GRPC_PORT"
)

type Client struct {
	client     dapr.Client
	timeoutSec int64
}

func newClient(timeoutSec int64) (*Client, error) {
	port := os.Getenv(daprPortEnvVarName)
	if port == "" {
		port = daprPortDefault
	}

	client, err := dapr.NewClientWithPort(port)
	if err != nil {
		return nil, err
	}

	c := new(Client)
	c.client = client
	c.timeoutSec = timeoutSec

	return c, nil
}

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

	client.client.Close()
	client.client = nil
	client = nil
}

func (client *Client) InvokeGetMethod(appID string, methodName string) ([]byte, error) {
	return client.invokeMethod(appID, methodName, "get")
}

func (client *Client) InvokePostMethodJson(appID string, methodName string, jsonData []byte) ([]byte, error) {
	return client.invokeMethodJson(appID, methodName, "post", jsonData)
}

func (client *Client) InvokePutMethodJson(appID string, methodName string, jsonData []byte) ([]byte, error) {
	return client.invokeMethodJson(appID, methodName, "put", jsonData)
}

func (client *Client) InvokeDeleteMethod(appID string, methodName string) ([]byte, error) {
	return client.invokeMethod(appID, methodName, "delete")
}

func (client *Client) Publish(pubsubName string, topic string, content []byte) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(client.timeoutSec)*time.Second)
	defer cancel()

	return client.client.PublishEvent(ctx, pubsubName, topic, content)
}

func (client *Client) invokeMethod(appID string, methodName string, verb string) ([]byte, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(client.timeoutSec)*time.Second)
	defer cancel()

	return client.client.InvokeMethod(ctx, appID, methodName, verb)
}

func (client *Client) invokeMethodJson(appID string, methodName string, verb string, jsonData []byte) ([]byte, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(client.timeoutSec)*time.Second)
	defer cancel()

	return client.client.InvokeMethodWithContent(ctx, appID, methodName, verb,
		&dapr.DataContent{
			Data:        jsonData,
			ContentType: "application/json",
		})
}

func (client *Client) SaveState(stateStoreName string, key string, value []byte, meta map[string]string, so ...dapr.StateOption) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(client.timeoutSec)*time.Second)
	defer cancel()

	return client.client.SaveState(ctx, stateStoreName, key, value, meta, so...)
}

func (client *Client) DeleteState(stateStoreName string, key string, meta map[string]string) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(client.timeoutSec)*time.Second)
	defer cancel()

	return client.client.DeleteState(ctx, stateStoreName, key, meta)
}

func (client *Client) GetState(stateStoreName string, key string, meta map[string]string) (*dapr.StateItem, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(client.timeoutSec)*time.Second)
	defer cancel()

	return client.client.GetState(ctx, stateStoreName, key, meta)
}

func (client *Client) ExecuteStateTransaction(stateStoreName string, meta map[string]string, ops []*dapr.StateOperation) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(client.timeoutSec)*time.Second)
	defer cancel()

	return client.client.ExecuteStateTransaction(ctx, stateStoreName, meta, ops)
}