jschoubben
842a901229
Conn.CallChunks subscribes to the call's chunks before it asks the verb, hands the verb's answer to an accept hook, then receives the chunks the answer counts, in order, acknowledging each, into a writer; CallFile answers the whole body as the member file, the form a bundle's seat call answers. The checker, the indexer, the assigner and the recovery keeper read a step's answer kept as an object through it. Builds: design 47 G A-291, A-290, A-254 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
72 lines
2.7 KiB
Go
72 lines
2.7 KiB
Go
//go:build integration
|
|
|
|
package process
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
)
|
|
|
|
// Design 47 G A-291: a process receives a seat verb's chunked answer through one helper: it subscribes to the call's chunks
|
|
// before it asks, acknowledges each, and answers the whole body (CallChunks into a writer, CallFile as the member file);
|
|
// an answer accept refuses takes no chunk.
|
|
func TestAProcessReceivesAChunkedSeatAnswerWhole(t *testing.T) {
|
|
url := testbus.URL(t)
|
|
holder, err := nats.Connect(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer holder.Close()
|
|
body := []byte(strings.Repeat("0123456789", 2*buscalls.SeatChunkBytes/10+7)) // three chunks
|
|
of := int64((len(body) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes)
|
|
sub, err := holder.Subscribe("mesh.seat.node-build-go.call.output.ace.>", func(m *nats.Msg) {
|
|
var in struct{ Ask, Call string }
|
|
_ = json.Unmarshal(m.Data, &in)
|
|
answer, _ := json.Marshal(map[string]any{"result": map[string]any{"ask": in.Ask, "chunks": of}})
|
|
_ = m.Respond(answer)
|
|
go func() {
|
|
_ = buscalls.SeatChunks{Seat: "node-build-go", Verb: "output", Call: in.Call}.Send(holder, body, buscalls.SeatAckWithin, t.Logf)
|
|
}()
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = sub.Unsubscribe() }()
|
|
_ = holder.Flush()
|
|
nc, err := nats.Connect(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c := &Conn{nc: nc, account: Account{User: "novox.mesh-delivery-checker"}}
|
|
defer c.Close()
|
|
var got bytes.Buffer
|
|
answer, err := c.CallChunks(context.Background(), "seat:node-build-go.output@ace", "c1", map[string]any{"ask": "a1", "call": "c1"}, nil, &got)
|
|
if err != nil || !bytes.Equal(got.Bytes(), body) || !strings.Contains(string(answer), `"ask":"a1"`) {
|
|
t.Fatalf("answered %s and %d bytes (%v), want the verb's answer and the whole %d bytes", answer, got.Len(), err, len(body))
|
|
}
|
|
raw, err := c.CallFile("seat:node-build-go.output@ace", map[string]any{"ask": "a2", "call": "c2"})
|
|
var file struct{ Ask, File string }
|
|
if err != nil || json.Unmarshal(raw, &file) != nil || file.Ask != "a2" {
|
|
t.Fatalf("CallFile answered %.80s, %v", raw, err)
|
|
}
|
|
if whole, _ := base64.StdEncoding.DecodeString(file.File); !bytes.Equal(whole, body) {
|
|
t.Fatalf("CallFile's file holds %d bytes, want %d", len(whole), len(body))
|
|
}
|
|
refused := errors.New("too large for this disk")
|
|
got.Reset()
|
|
if _, err := c.CallChunks(context.Background(), "seat:node-build-go.output@ace", "c3", map[string]any{"ask": "a3", "call": "c3"},
|
|
func(json.RawMessage) error { return refused }, &got); !errors.Is(err, refused) || got.Len() != 0 {
|
|
t.Fatalf("an answer accept refused: %v, %d bytes taken", err, got.Len())
|
|
}
|
|
}
|