issue 1840/S1: let a process put a key whole and list a state's keys, as a tools bundle can
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -128,3 +129,41 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) {
|
||||
}
|
||||
return rev, nil
|
||||
}
|
||||
|
||||
// Put writes a key whole, whatever it held, and returns the revision written.
|
||||
func (s State) Put(key string, value any) (uint64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
data, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Put(ctx, key, data)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
||||
}
|
||||
return rev, nil
|
||||
}
|
||||
|
||||
// Keys is every key with a current value, sorted; none is an empty list.
|
||||
func (s State) Keys() ([]string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
keys, err := kv.Keys(ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return []string{}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
@@ -94,3 +94,38 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
|
||||
t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// 1840/S1: given a process connection granted its bucket, when it puts keys whole and lists them, then Put
|
||||
// returns a new revision each time and Keys lists them sorted (none for a bucket with none).
|
||||
func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) {
|
||||
url, account := grantedSendProcess(t)
|
||||
_, adminJetStream := adminJS(t, url)
|
||||
if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
|
||||
t.Fatalf("the admin could not make the bucket: %v", err)
|
||||
}
|
||||
conn, err := Connect(account)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
s := conn.State("sends")
|
||||
|
||||
if keys, err := s.Keys(); err != nil || len(keys) != 0 {
|
||||
t.Fatalf("Keys of an empty bucket: %v, %v", keys, err)
|
||||
}
|
||||
rev, err := s.Put("b", map[string]any{"v": 1})
|
||||
if err != nil || rev == 0 {
|
||||
t.Fatalf("Put on a granted bucket: %d, %v", rev, err)
|
||||
}
|
||||
rev2, err := s.Put("b", map[string]any{"v": 2})
|
||||
if err != nil || rev2 <= rev {
|
||||
t.Fatalf("Put over a key: %d, %v", rev2, err)
|
||||
}
|
||||
if _, err := s.Put("a", 1); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keys, err := s.Keys()
|
||||
if err != nil || strings.Join(keys, ",") != "a,b" {
|
||||
t.Fatalf("Keys: %v, %v, want a,b", keys, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,8 @@ package process
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
@@ -68,6 +70,28 @@ func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision ui
|
||||
return f.next, nil
|
||||
}
|
||||
|
||||
func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.next++
|
||||
f.values[key], f.revs[key] = value, f.next
|
||||
return f.next, nil
|
||||
}
|
||||
|
||||
func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if len(f.values) == 0 {
|
||||
return nil, jetstream.ErrNoKeysFound
|
||||
}
|
||||
keys := make([]string, 0, len(f.values))
|
||||
for k := range f.values {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Sort(sort.Reverse(sort.StringSlice(keys))) // as a bucket lists them: in no order the caller may rely on
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
func stateOn(kv jetstream.KeyValue) State {
|
||||
return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }}
|
||||
}
|
||||
@@ -103,3 +127,38 @@ func TestStateSatisfiesAskerStore(t *testing.T) {
|
||||
t.Fatal("Update did not advance the revision")
|
||||
}
|
||||
}
|
||||
|
||||
// 1840/S1: given a state, when a key is put twice, then each Put returns a new revision and the key reads
|
||||
// back the last value, whether or not it had one.
|
||||
func TestStatePutWritesAKeyWhole(t *testing.T) {
|
||||
s := stateOn(newFakeKV())
|
||||
rev, err := s.Put("k", map[string]any{"a": 1})
|
||||
if err != nil || rev == 0 {
|
||||
t.Fatalf("Put of a new key: %d, %v", rev, err)
|
||||
}
|
||||
rev2, err := s.Put("k", map[string]any{"a": 2})
|
||||
if err != nil || rev2 <= rev {
|
||||
t.Fatalf("Put over a key: %d, %v, want a revision after %d", rev2, err, rev)
|
||||
}
|
||||
e, err := s.Get("k")
|
||||
if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` {
|
||||
t.Fatalf("Get after Put: %+v, %v", e, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error.
|
||||
func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) {
|
||||
s := stateOn(newFakeKV())
|
||||
if keys, err := s.Keys(); err != nil || len(keys) != 0 {
|
||||
t.Fatalf("Keys of an empty state: %v, %v", keys, err)
|
||||
}
|
||||
for _, k := range []string{"b", "c", "a"} {
|
||||
if _, err := s.Put(k, 1); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
keys, err := s.Keys()
|
||||
if err != nil || strings.Join(keys, ",") != "a,b,c" {
|
||||
t.Fatalf("Keys: %v, %v, want a,b,c", keys, err)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user