diff --git a/go/process/chunks.go b/go/process/chunks.go new file mode 100644 index 00000000..ee5c14e9 --- /dev/null +++ b/go/process/chunks.go @@ -0,0 +1,110 @@ +package process + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "time" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/calladdress" +) + +// CallChunks calls a seat verb whose answer file is sent in chunks (a verb its seat definition marks `chunks`, design 47 G +// ยง681, A-139), the one way a process receives such an answer (A-291): it subscribes to the chunks of call (the 128 random +// bits the verb's input carries as its call) before it asks, so none is missed, asks key ("seat:.[@]") +// with body as Call does, hands the verb's answer to accept (nil accepts every answer; an error ends the call before a chunk +// is taken), then receives the number of chunks the answer's member `chunks` names, in order, acknowledging each, into w. +// It answers the verb's answer. A chunk that does not come within buscalls.SeatAckWithin, or ctx ending, ends the receive. +func (c *Conn) CallChunks(ctx context.Context, key, call string, body any, accept func(answer json.RawMessage) error, w io.Writer) (json.RawMessage, error) { + a, err := calladdress.Parse(key) + if err != nil { + return nil, err + } + if a.Seat == "" { + return nil, fmt.Errorf("%q names no seat verb, so it sends no chunks", key) + } + if call == "" { + return nil, fmt.Errorf("the call of %s names no call token for its chunks", key) + } + send := buscalls.SeatChunks{Seat: a.Seat, Verb: a.Verb, Call: call} + chunks := make(chan *nats.Msg, 4) + sub, err := c.nc.ChanSubscribe(send.Chunks(), chunks) + if err != nil { + return nil, fmt.Errorf("the chunks of %s cannot be heard: %w", key, err) + } + defer func() { _ = sub.Unsubscribe() }() + answer, err := c.Call(key, body) + if err != nil { + return nil, err + } + if accept != nil { + if err := accept(answer); err != nil { + return answer, err + } + } + var count struct { + Chunks int64 `json:"chunks"` + } + if err := json.Unmarshal(answer, &count); err != nil { + return answer, fmt.Errorf("%s answered no count of its chunks: %w", key, err) + } + next := func() (string, []byte, error) { + bound := time.NewTimer(buscalls.SeatAckWithin) + defer bound.Stop() + select { + case m := <-chunks: + return m.Subject, m.Data, nil + case <-bound.C: + return "", nil, buscalls.ErrSeatChunkMissed + case <-ctx.Done(): + return "", nil, ctx.Err() + } + } + ack := func(subject string) error { + _, err := c.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin) + return err + } + if _, err := send.Receive(count.Chunks, next, ack, w); err != nil { + return answer, fmt.Errorf("the answer of %s: %w", key, err) + } + return answer, nil +} + +// CallFile is CallChunks for an answer read whole, in the form a bundle's ordinary seat call answers a chunked verb (design 47 +// G A-249, A-253): the verb's answer with the whole file added as its member `file` (base64), none when no chunk was sent. The +// call token is body's member `call`. +func (c *Conn) CallFile(key string, body any) (json.RawMessage, error) { + call := callOf(body) + var file bytes.Buffer + answer, err := c.CallChunks(context.Background(), key, call, body, nil, &file) + if err != nil { + return nil, err + } + var whole map[string]any + if err := json.Unmarshal(answer, &whole); err != nil { + return nil, fmt.Errorf("%s answered what is no object: %w", key, err) + } + if file.Len() > 0 { + whole["file"] = base64.StdEncoding.EncodeToString(file.Bytes()) + } + return json.Marshal(whole) +} + +// callOf is the member call of a verb's input. +func callOf(body any) string { + raw, err := json.Marshal(body) + if err != nil { + return "" + } + var in struct { + Call string `json:"call"` + } + _ = json.Unmarshal(raw, &in) + return in.Call +} diff --git a/go/process/chunks_integration_test.go b/go/process/chunks_integration_test.go new file mode 100644 index 00000000..60d83d67 --- /dev/null +++ b/go/process/chunks_integration_test.go @@ -0,0 +1,71 @@ +//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()) + } +}