G A-40: the held writer seeds what it holds from its state at Start
Builds: design 47 G A-40, A-29
This commit is contained in:
+66
-20
@@ -14,8 +14,12 @@ import (
|
||||
// It is a Writer[F, C, S, A] with that Observe, so no part writes the adapter.
|
||||
type HeldWriter[F, C, S, A any] struct {
|
||||
*Writer[F, C, S, A]
|
||||
mu sync.Mutex
|
||||
held map[string]F
|
||||
mu sync.Mutex
|
||||
held map[string]F
|
||||
state FindingState
|
||||
started bool
|
||||
pending map[string]F // written before Start, written through the writer at Start
|
||||
cleared map[string][4]string // cleared before Start, cleared through the writer at Start
|
||||
}
|
||||
|
||||
// NewHeldWriter makes the writer for cfg, whose Observe is none (a given Observe is refused): the findings that stand in
|
||||
@@ -29,23 +33,7 @@ func NewHeldWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A]) (*HeldWriter[F,
|
||||
return nil, errors.New("a finding writer needs the state `findings`")
|
||||
}
|
||||
h := &HeldWriter[F, C, S, A]{held: map[string]F{}}
|
||||
keys, err := cfg.State.Keys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, k := range keys {
|
||||
e, err := cfg.State.Get(k)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var f F
|
||||
if e == nil || json.Unmarshal(e.Value, &f) != nil {
|
||||
continue
|
||||
}
|
||||
if m, err := wireOf(&f); err == nil && keyOf(m) == k {
|
||||
h.held[k] = f
|
||||
}
|
||||
}
|
||||
h.state, h.pending, h.cleared = cfg.State, map[string]F{}, map[string][4]string{}
|
||||
cfg.Observe = func(context.Context) (Observed[F], error) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
@@ -77,6 +65,12 @@ func (h *HeldWriter[F, C, S, A]) Write(ctx context.Context, finding F) error {
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.held[keyOf(m)] = finding
|
||||
if !h.started {
|
||||
h.pending[keyOf(m)] = finding
|
||||
delete(h.cleared, keyOf(m))
|
||||
h.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
h.mu.Unlock()
|
||||
return h.Writer.Write(ctx, finding)
|
||||
}
|
||||
@@ -84,7 +78,14 @@ func (h *HeldWriter[F, C, S, A]) Write(ctx context.Context, finding F) error {
|
||||
// Clear stops holding the finding of scope, id and kind and clears it through the writer.
|
||||
func (h *HeldWriter[F, C, S, A]) Clear(ctx context.Context, scope, id, kind, why string) error {
|
||||
h.mu.Lock()
|
||||
delete(h.held, FindingKey(scope, id, kind))
|
||||
k := FindingKey(scope, id, kind)
|
||||
delete(h.held, k)
|
||||
if !h.started {
|
||||
delete(h.pending, k)
|
||||
h.cleared[k] = [4]string{scope, id, kind, why}
|
||||
h.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
h.mu.Unlock()
|
||||
return h.Writer.Clear(ctx, scope, id, kind, why)
|
||||
}
|
||||
@@ -110,3 +111,48 @@ func (h *HeldWriter[F, C, S, A]) Held() []F {
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Start seeds what the writer holds from the keys of its state `findings` (design 47 G A-40), what was written before it
|
||||
// standing over them, writes what was written and clears what was cleared before it, and starts the writer. A part makes the
|
||||
// writer before its module serves, so nothing is read from the state until now.
|
||||
func (h *HeldWriter[F, C, S, A]) Start(ctx context.Context) error {
|
||||
keys, err := h.state.Keys()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
h.mu.Lock()
|
||||
for _, k := range keys {
|
||||
e, err := h.state.Get(k)
|
||||
if err != nil {
|
||||
h.mu.Unlock()
|
||||
return err
|
||||
}
|
||||
if _, ok := h.held[k]; ok {
|
||||
continue
|
||||
}
|
||||
if _, ok := h.cleared[k]; ok {
|
||||
continue
|
||||
}
|
||||
var f F
|
||||
if e == nil || json.Unmarshal(e.Value, &f) != nil {
|
||||
continue
|
||||
}
|
||||
if m, err := wireOf(&f); err == nil && keyOf(m) == k {
|
||||
h.held[k] = f
|
||||
}
|
||||
}
|
||||
pending, cleared := h.pending, h.cleared
|
||||
h.started, h.pending, h.cleared = true, nil, nil
|
||||
h.mu.Unlock()
|
||||
for _, f := range pending {
|
||||
if err := h.Writer.Write(ctx, f); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, c := range cleared {
|
||||
if err := h.Writer.Clear(ctx, c[0], c[1], c[2], c[3]); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return h.Writer.Start(ctx)
|
||||
}
|
||||
|
||||
@@ -526,6 +526,42 @@ func subHeldWriterStatesWhatItHoldsFromItsOwnState(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A held writer reads its state at Start, not when made: what was written before Start is held and written at Start, over
|
||||
// what stood in the state (design 47 G A-40).
|
||||
func subHeldWriterSeedsAtStart(t *testing.T) {
|
||||
state := &fakeState{m: map[string][]byte{}}
|
||||
raw, _ := json.Marshal(finding("seeded", "stood"))
|
||||
state.m["module.seeded.k"] = raw
|
||||
var mu sync.Mutex
|
||||
cfg := testConfig{Module: "m", State: state, Node: "n1", Asked: asked,
|
||||
Publish: func(context.Context, string, []byte, string) error { mu.Lock(); mu.Unlock(); return nil },
|
||||
Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }}
|
||||
h, err := views.NewHeldWriter(cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
if err := h.Write(ctx, finding("early", "before")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if held := h.Held(); len(held) != 1 || held[0].ID != "early" {
|
||||
t.Fatalf("before Start Held answers what was written so far: %v", held)
|
||||
}
|
||||
if len(state.m) != 1 {
|
||||
t.Fatalf("a write before Start touched the state: %v", state.m)
|
||||
}
|
||||
if err := h.Start(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer h.Close()
|
||||
if held := h.Held(); len(held) != 2 {
|
||||
t.Fatalf("after Start Held answers the seeded and the early: %v", held)
|
||||
}
|
||||
if _, ok := state.m["module.early.k"]; !ok {
|
||||
t.Fatalf("the early finding was written at Start: %v", state.m)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFindingSubjectsAreTheEventsStreamForm runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
|
||||
func TestFindingSubjectsAreTheEventsStreamForm(t *testing.T) {
|
||||
t.Run("FindingSubjectsAreTheEventsStreamForm", subFindingSubjectsAreTheEventsStreamForm)
|
||||
@@ -537,4 +573,5 @@ func TestFindingSubjectsAreTheEventsStreamForm(t *testing.T) {
|
||||
t.Run("ReaderFindingsAreLaidOutAndCheckFindingRefuses", subReaderFindingsAreLaidOutAndCheckFindingRefuses)
|
||||
t.Run("WriterRefusesABodyWithoutAContractField", subWriterRefusesABodyWithoutAContractField)
|
||||
t.Run("HeldWriterStatesWhatItHoldsFromItsOwnState", subHeldWriterStatesWhatItHoldsFromItsOwnState)
|
||||
t.Run("HeldWriterSeedsAtStart", subHeldWriterSeedsAtStart)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user