jschoubben
a8be5acd80
The bus module sends snapshot-get's file in chunks and the recovery module receives them; both now use buscalls.SeatChunks (subjects, body, receive), so the form lives once in mesh-sdk. Builds: design 47 G §681, A-64, A-139 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
73 lines
2.7 KiB
Go
73 lines
2.7 KiB
Go
package buscalls
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
// TestSeatChunksReceive: the chunks of a send are written in order and each acknowledged before the next; a missed,
|
|
// misplaced or miscounted chunk ends the receive unacknowledged (design 47 G §681, A-139).
|
|
func TestSeatChunksReceive(t *testing.T) {
|
|
c := SeatChunks{Seat: "mesh-broker", Verb: "snapshot-get", Call: "0123456789abcdef0123456789abcdef"}
|
|
if c.Chunk(2) != "mesh.seat.mesh-broker.chunk.snapshot-get.0123456789abcdef0123456789abcdef.2" ||
|
|
c.Ack(2) != "mesh.seat.mesh-broker.ack.snapshot-get.0123456789abcdef0123456789abcdef.2" ||
|
|
c.Chunks() != "mesh.seat.mesh-broker.chunk.snapshot-get.0123456789abcdef0123456789abcdef.*" ||
|
|
SeatAcks("mesh-broker", "snapshot-get") != "mesh.seat.mesh-broker.ack.snapshot-get.*.*" {
|
|
t.Fatalf("subjects: %s %s %s", c.Chunk(2), c.Ack(2), c.Chunks())
|
|
}
|
|
type sent struct {
|
|
subject string
|
|
body SeatChunk
|
|
missed bool
|
|
}
|
|
good := []sent{{subject: c.Chunk(1), body: SeatChunk{N: 1, Of: 3, Data: []byte("ab")}},
|
|
{subject: c.Chunk(2), body: SeatChunk{N: 2, Of: 3, Data: []byte("cd")}},
|
|
{subject: c.Chunk(3), body: SeatChunk{N: 3, Of: 3, Data: []byte("e")}}}
|
|
for _, tc := range []struct {
|
|
name string
|
|
sends []sent
|
|
want string
|
|
acks int
|
|
err string
|
|
}{
|
|
{name: "three chunks", sends: good, want: "abcde", acks: 3},
|
|
{name: "a missed chunk", sends: append(good[:1:1], sent{missed: true}), want: "ab", acks: 1, err: "chunk 2 of 3"},
|
|
{name: "out of order", sends: []sent{good[1]}, acks: 0, err: "came on"},
|
|
{name: "another count", sends: []sent{{subject: c.Chunk(1), body: SeatChunk{N: 1, Of: 2, Data: []byte("x")}}}, err: "says it is 1 of 2"},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
i, acks := 0, 0
|
|
next := func() (string, []byte, error) {
|
|
if i >= len(tc.sends) || tc.sends[i].missed {
|
|
return "", nil, ErrSeatChunkMissed
|
|
}
|
|
s := tc.sends[i]
|
|
i++
|
|
body, _ := json.Marshal(s.body)
|
|
return s.subject, body, nil
|
|
}
|
|
ack := func(subject string) error {
|
|
acks++
|
|
if subject != c.Ack(int64(acks)) {
|
|
t.Fatalf("ack %d on %s", acks, subject)
|
|
}
|
|
return nil
|
|
}
|
|
var out bytes.Buffer
|
|
n, err := c.Receive(3, next, ack, &out)
|
|
if tc.err == "" && err != nil || tc.err != "" && (err == nil || !strings.Contains(err.Error(), tc.err)) {
|
|
t.Fatalf("err %v, want %q", err, tc.err)
|
|
}
|
|
if tc.name == "a missed chunk" && !errors.Is(err, ErrSeatChunkMissed) {
|
|
t.Fatalf("a missed chunk is not ErrSeatChunkMissed: %v", err)
|
|
}
|
|
if out.String() != tc.want || n != int64(len(tc.want)) || acks != tc.acks {
|
|
t.Fatalf("wrote %q (%d), %d acks; want %q, %d acks", out.String(), n, acks, tc.want, tc.acks)
|
|
}
|
|
})
|
|
}
|
|
}
|