mirror of https://github.com/dapr/go-sdk.git
WIP: state methods no longer leak proto types
This commit is contained in:
Mark Chmarny
parent
14f04f2e95
commit
5b5c0f23d8
|
|
@ -1,4 +1,4 @@
|
|||
# dapr SDK for Go
|
||||
# Dapr SDK for Go
|
||||
|
||||
This is the dapr SDK (client) for Go.
|
||||
|
||||
|
|
@ -10,7 +10,7 @@ go get github.com/dapr/go-sdk
|
|||
|
||||
## Usage
|
||||
|
||||
The `example` folder contains a Dapr enabled `serving` app a `client` app that uses this SDK to invoke dapr API for state and events, `serving` app for service to service invocation, and a simple HTTP binding to illustrate output binding. To run the example:
|
||||
The `example` folder contains a Dapr enabled `serving` app a `client` app that uses this SDK to invoke Dapr API for state and events, `serving` app for service to service invocation, and a simple HTTP binding to illustrate output binding. To run the example:
|
||||
|
||||
1. Start the `serving` app in the `example/serving` directory
|
||||
|
||||
|
|
|
|||
263
client/state.go
263
client/state.go
|
|
@ -3,50 +3,199 @@ package client
|
|||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
v1 "github.com/dapr/go-sdk/dapr/proto/common/v1"
|
||||
pb "github.com/dapr/go-sdk/dapr/proto/runtime/v1"
|
||||
duration "github.com/golang/protobuf/ptypes/duration"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
var (
|
||||
// StateOptionConsistencyDefault is strong
|
||||
StateOptionConsistencyDefault = v1.StateOptions_CONSISTENCY_STRONG
|
||||
const (
|
||||
// StateConsistencyUndefined is the undefined value for state consistency
|
||||
StateConsistencyUndefined StateConsistency = 0
|
||||
// StateConsistencyEventual represents eventual state consistency value
|
||||
StateConsistencyEventual StateConsistency = 1
|
||||
// StateConsistencyStrong represents strong state consistency value
|
||||
StateConsistencyStrong StateConsistency = 2
|
||||
|
||||
// StateOptionConcurrencyDefault is last write
|
||||
StateOptionConcurrencyDefault = v1.StateOptions_CONCURRENCY_LAST_WRITE
|
||||
// StateConcurrencyUndefined is the undefined value for state concurrency
|
||||
StateConcurrencyUndefined StateConcurrency = 0
|
||||
// StateConcurrencyFirstWrite represents first write concurrency value
|
||||
StateConcurrencyFirstWrite StateConcurrency = 1
|
||||
// StateConcurrencyLastWrite represents last write concurrency value
|
||||
StateConcurrencyLastWrite StateConcurrency = 2
|
||||
|
||||
// StateOptionRetryPolicyDefault is threshold 3
|
||||
StateOptionRetryPolicyDefault = &v1.StateRetryPolicy{
|
||||
Threshold: 3,
|
||||
// RetryPatternUndefined is the undefined value for retry pattern
|
||||
RetryPatternUndefined RetryPattern = 0
|
||||
// RetryPatternLinear represents the linear retry pattern value
|
||||
RetryPatternLinear RetryPattern = 1
|
||||
// RetryPatternExponential represents the exponential retry pattern value
|
||||
RetryPatternExponential RetryPattern = 2
|
||||
)
|
||||
|
||||
type (
|
||||
// StateConsistency is the consistency enum type
|
||||
StateConsistency int
|
||||
// StateConcurrency is the concurrency enum type
|
||||
StateConcurrency int
|
||||
// RetryPattern is the retry pattern enum type
|
||||
RetryPattern int
|
||||
)
|
||||
|
||||
func (c StateConsistency) String() string {
|
||||
names := [...]string{
|
||||
"Undefined",
|
||||
"Strong",
|
||||
"Eventual",
|
||||
}
|
||||
if c < StateConsistencyStrong || c > StateConsistencyEventual {
|
||||
return "Undefined"
|
||||
}
|
||||
|
||||
// StateOptionDefault is the optimistic state option (last write concurency and strong consistency)
|
||||
StateOptionDefault = &v1.StateOptions{
|
||||
Concurrency: StateOptionConcurrencyDefault,
|
||||
Consistency: StateOptionConsistencyDefault,
|
||||
RetryPolicy: StateOptionRetryPolicyDefault,
|
||||
return names[c]
|
||||
}
|
||||
|
||||
// End Consistency
|
||||
|
||||
func (c StateConcurrency) String() string {
|
||||
names := [...]string{
|
||||
"Undefined",
|
||||
"FirstWrite",
|
||||
"LastWrite",
|
||||
}
|
||||
if c < StateConcurrencyFirstWrite || c > StateConcurrencyLastWrite {
|
||||
return "Undefined"
|
||||
}
|
||||
|
||||
return names[c]
|
||||
}
|
||||
|
||||
// END Concurrency
|
||||
func (c RetryPattern) String() string {
|
||||
names := [...]string{
|
||||
"Undefined",
|
||||
"Linear",
|
||||
"Exponential",
|
||||
}
|
||||
if c < RetryPatternLinear || c > RetryPatternExponential {
|
||||
return "Undefined"
|
||||
}
|
||||
|
||||
return names[c]
|
||||
}
|
||||
|
||||
// END Retry Pattern
|
||||
|
||||
var (
|
||||
stateOptionRetryPolicyDefault = &v1.StateRetryPolicy{
|
||||
Threshold: 3,
|
||||
Pattern: v1.StateRetryPolicy_RETRY_EXPONENTIAL,
|
||||
}
|
||||
|
||||
stateOptionDefault = &v1.StateOptions{
|
||||
Concurrency: v1.StateOptions_CONCURRENCY_LAST_WRITE,
|
||||
Consistency: v1.StateOptions_CONSISTENCY_STRONG,
|
||||
RetryPolicy: stateOptionRetryPolicyDefault,
|
||||
}
|
||||
)
|
||||
|
||||
// State is a collection of StateItems with a store name
|
||||
type State struct {
|
||||
StoreName string
|
||||
States []*StateItem
|
||||
}
|
||||
|
||||
// StateItem represents the state to be persisted
|
||||
type StateItem struct {
|
||||
Key string
|
||||
Value []byte
|
||||
Etag string
|
||||
Metadata map[string]string
|
||||
Options *StateOptions
|
||||
}
|
||||
|
||||
// StateOptions represents the state store persistence policy
|
||||
type StateOptions struct {
|
||||
Concurrency StateConcurrency
|
||||
Consistency StateConsistency
|
||||
RetryPolicy *StateRetryPolicy
|
||||
}
|
||||
|
||||
// StateRetryPolicy represents the state store invocation retry policy
|
||||
type StateRetryPolicy struct {
|
||||
Threshold int32
|
||||
Pattern RetryPattern
|
||||
Interval time.Duration
|
||||
}
|
||||
|
||||
// *** Converters
|
||||
|
||||
func toProtoSaveStateRequest(s *State) (req *pb.SaveStateRequest) {
|
||||
r := &pb.SaveStateRequest{
|
||||
StoreName: s.StoreName,
|
||||
States: make([]*v1.StateItem, 0),
|
||||
}
|
||||
|
||||
for _, si := range s.States {
|
||||
item := toProtoSaveStateItem(si)
|
||||
r.States = append(r.States, item)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func toProtoSaveStateItem(si *StateItem) (item *v1.StateItem) {
|
||||
return &v1.StateItem{
|
||||
Etag: si.Etag,
|
||||
Key: si.Key,
|
||||
Metadata: si.Metadata,
|
||||
Value: si.Value,
|
||||
Options: toProtoStateOptions(si.Options),
|
||||
}
|
||||
}
|
||||
|
||||
func toProtoStateOptions(so *StateOptions) (opts *v1.StateOptions) {
|
||||
if so == nil {
|
||||
return stateOptionDefault
|
||||
}
|
||||
return &v1.StateOptions{
|
||||
Concurrency: (v1.StateOptions_StateConcurrency(so.Concurrency)),
|
||||
Consistency: (v1.StateOptions_StateConsistency(so.Consistency)),
|
||||
RetryPolicy: &v1.StateRetryPolicy{
|
||||
Interval: toProtoDuration(so.RetryPolicy.Interval),
|
||||
Pattern: (v1.StateRetryPolicy_RetryPattern(so.RetryPolicy.Pattern)),
|
||||
Threshold: so.RetryPolicy.Threshold,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func toProtoDuration(d time.Duration) *duration.Duration {
|
||||
nanos := d.Nanoseconds()
|
||||
secs := nanos / 1e9
|
||||
nanos -= secs * 1e9
|
||||
return &duration.Duration{
|
||||
Seconds: int64(secs),
|
||||
Nanos: int32(nanos),
|
||||
}
|
||||
}
|
||||
|
||||
// *** Save State ***
|
||||
|
||||
// SaveState saves the fully loaded save state request
|
||||
func (c *Client) SaveState(ctx context.Context, req *pb.SaveStateRequest) error {
|
||||
if req == nil {
|
||||
return errors.New("nil request")
|
||||
func (c *Client) SaveState(ctx context.Context, s *State) error {
|
||||
if s == nil || s.StoreName == "" || s.States == nil || len(s.States) < 1 {
|
||||
return errors.New("nil or invalid state")
|
||||
}
|
||||
|
||||
req := toProtoSaveStateRequest(s)
|
||||
_, err := c.protoClient.SaveState(authContext(ctx), req)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error saving state")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// SaveStateItem saves a single state item
|
||||
func (c *Client) SaveStateItem(ctx context.Context, store string, item *v1.StateItem) error {
|
||||
func (c *Client) SaveStateItem(ctx context.Context, store string, item *StateItem) error {
|
||||
if store == "" {
|
||||
return errors.New("nil store")
|
||||
}
|
||||
|
|
@ -54,9 +203,9 @@ func (c *Client) SaveStateItem(ctx context.Context, store string, item *v1.State
|
|||
return errors.New("nil item")
|
||||
}
|
||||
|
||||
req := &pb.SaveStateRequest{
|
||||
req := &State{
|
||||
StoreName: store,
|
||||
States: []*v1.StateItem{item},
|
||||
States: []*StateItem{item},
|
||||
}
|
||||
|
||||
return c.SaveState(ctx, req)
|
||||
|
|
@ -71,13 +220,12 @@ func (c *Client) SaveStateWithData(ctx context.Context, store, key string, data
|
|||
return errors.New("nil key")
|
||||
}
|
||||
|
||||
req := &pb.SaveStateRequest{
|
||||
req := &State{
|
||||
StoreName: store,
|
||||
States: []*v1.StateItem{
|
||||
States: []*StateItem{
|
||||
{
|
||||
Key: key,
|
||||
Value: data,
|
||||
Options: StateOptionDefault,
|
||||
Key: key,
|
||||
Value: data,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
|
@ -99,10 +247,19 @@ func (c *Client) SaveStateJSON(ctx context.Context, store, key string, in interf
|
|||
|
||||
// *** Get State ***
|
||||
|
||||
// GetStateWithRequest retreaves state from specific store using provided request
|
||||
func (c *Client) GetStateWithRequest(ctx context.Context, req *pb.GetStateRequest) (out []byte, err error) {
|
||||
if req == nil {
|
||||
return nil, errors.New("nil request")
|
||||
// GetStateWithConsistency retreaves state from specific store using provided request
|
||||
func (c *Client) GetStateWithConsistency(ctx context.Context, store, key string, sc StateConsistency) (out []byte, err error) {
|
||||
if store == "" {
|
||||
return nil, errors.New("nil store")
|
||||
}
|
||||
if key == "" {
|
||||
return nil, errors.New("nil key")
|
||||
}
|
||||
|
||||
req := &pb.GetStateRequest{
|
||||
StoreName: store,
|
||||
Key: key,
|
||||
Consistency: (v1.StateOptions_StateConsistency(sc)),
|
||||
}
|
||||
|
||||
result, err := c.protoClient.GetState(authContext(ctx), req)
|
||||
|
|
@ -115,27 +272,25 @@ func (c *Client) GetStateWithRequest(ctx context.Context, req *pb.GetStateReques
|
|||
|
||||
// GetState retreaves state from specific store using default consistency option
|
||||
func (c *Client) GetState(ctx context.Context, store, key string) (out []byte, err error) {
|
||||
if store == "" {
|
||||
return nil, errors.New("nil store")
|
||||
}
|
||||
if key == "" {
|
||||
return nil, errors.New("nil key")
|
||||
}
|
||||
req := &pb.GetStateRequest{
|
||||
StoreName: store,
|
||||
Key: key,
|
||||
Consistency: StateOptionConsistencyDefault,
|
||||
}
|
||||
|
||||
return c.GetStateWithRequest(ctx, req)
|
||||
return c.GetStateWithConsistency(ctx, store, key, StateConsistencyStrong)
|
||||
}
|
||||
|
||||
// *** Delete State ***
|
||||
|
||||
// DeleteStateWithRequest deletes content from store using provided request
|
||||
func (c *Client) DeleteStateWithRequest(ctx context.Context, req *pb.DeleteStateRequest) error {
|
||||
if req == nil {
|
||||
return errors.New("nil request")
|
||||
// DeleteStateWithOptions deletes content from store using provided state options and etag
|
||||
func (c *Client) DeleteStateWithOptions(ctx context.Context, store, key, etag string, opts *StateOptions) error {
|
||||
if store == "" {
|
||||
return errors.New("nil store")
|
||||
}
|
||||
if key == "" {
|
||||
return errors.New("nil key")
|
||||
}
|
||||
|
||||
req := &pb.DeleteStateRequest{
|
||||
StoreName: store,
|
||||
Key: key,
|
||||
Etag: etag,
|
||||
Options: toProtoStateOptions(opts),
|
||||
}
|
||||
|
||||
_, err := c.protoClient.DeleteState(authContext(ctx), req)
|
||||
|
|
@ -148,17 +303,5 @@ func (c *Client) DeleteStateWithRequest(ctx context.Context, req *pb.DeleteState
|
|||
|
||||
// DeleteState deletes content from store using default state options
|
||||
func (c *Client) DeleteState(ctx context.Context, store, key string) error {
|
||||
if store == "" {
|
||||
return errors.New("nil store")
|
||||
}
|
||||
if key == "" {
|
||||
return errors.New("nil key")
|
||||
}
|
||||
req := &pb.DeleteStateRequest{
|
||||
StoreName: store,
|
||||
Key: key,
|
||||
Options: StateOptionDefault,
|
||||
}
|
||||
|
||||
return c.DeleteStateWithRequest(ctx, req)
|
||||
return c.DeleteStateWithOptions(ctx, store, key, "", nil)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,36 @@
|
|||
package client
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
v1 "github.com/dapr/go-sdk/dapr/proto/common/v1"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestDurationConverter(t *testing.T) {
|
||||
d := time.Duration(10 * time.Second)
|
||||
pd := toProtoDuration(d)
|
||||
assert.NotNil(t, pd)
|
||||
assert.Equal(t, pd.Seconds, int64(10))
|
||||
}
|
||||
|
||||
func TestStateOptionsConverter(t *testing.T) {
|
||||
s := &StateOptions{
|
||||
Concurrency: StateConcurrencyLastWrite,
|
||||
Consistency: StateConsistencyStrong,
|
||||
RetryPolicy: &StateRetryPolicy{
|
||||
Threshold: 3,
|
||||
Interval: time.Duration(10 * time.Second),
|
||||
Pattern: RetryPatternExponential,
|
||||
},
|
||||
}
|
||||
p := toProtoStateOptions(s)
|
||||
assert.NotNil(t, p)
|
||||
assert.Equal(t, p.Concurrency, v1.StateOptions_CONCURRENCY_LAST_WRITE)
|
||||
assert.Equal(t, p.Consistency, v1.StateOptions_CONSISTENCY_STRONG)
|
||||
assert.NotNil(t, p.RetryPolicy)
|
||||
assert.Equal(t, p.RetryPolicy.Threshold, int32(3))
|
||||
assert.Equal(t, p.RetryPolicy.Interval.Seconds, int64(10))
|
||||
assert.Equal(t, p.RetryPolicy.Pattern, v1.StateRetryPolicy_RETRY_EXPONENTIAL)
|
||||
}
|
||||
Loading…
Reference in New Issue