diff --git a/go/buscalls/seatchunks.go b/go/buscalls/seatchunks.go new file mode 100644 index 00000000..94a2648c --- /dev/null +++ b/go/buscalls/seatchunks.go @@ -0,0 +1,87 @@ +package buscalls + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "strconv" + "time" +) + +// SeatChunks is the send of one seat verb's answer file in chunks (design 47 G §681, A-64, A-139), a verb whose seat +// definition says `chunks: true`: the holder's process sends the nth chunk on mesh.seat..chunk..., +// and the caller acknowledges it on mesh.seat..ack... within SeatAckWithin before the next is sent. +// `call` is the 128 random bits the caller chose. The holder sends with it and the caller receives with it: one form. +type SeatChunks struct { + Seat, Verb, Call string +} + +// SeatChunkBytes is at most what one chunk carries: 512 KiB. +const SeatChunkBytes = 512 << 10 + +// SeatAckWithin is how long the holder waits for a chunk's acknowledgement, and so how long a caller waits for the next chunk. +const SeatAckWithin = 30 * time.Second + +// SeatChunk is the body of one chunk: its number from 1, how many there are, and its bytes. +type SeatChunk struct { + N int64 `json:"n"` + Of int64 `json:"of"` + Data []byte `json:"data"` +} + +// Chunk is the subject of the nth chunk. +func (c SeatChunks) Chunk(n int64) string { + return "mesh.seat." + c.Seat + ".chunk." + c.Verb + "." + c.Call + "." + strconv.FormatInt(n, 10) +} + +// Ack is the subject the nth chunk is acknowledged on. +func (c SeatChunks) Ack(n int64) string { + return "mesh.seat." + c.Seat + ".ack." + c.Verb + "." + c.Call + "." + strconv.FormatInt(n, 10) +} + +// Chunks is every chunk subject of this call: what a caller subscribes to before it asks the verb. +func (c SeatChunks) Chunks() string { + return "mesh.seat." + c.Seat + ".chunk." + c.Verb + "." + c.Call + ".*" +} + +// SeatAcks is every acknowledgement of a verb's sends: what the holder subscribes to. +func SeatAcks(seat, verb string) string { + return "mesh.seat." + seat + ".ack." + verb + ".*.*" +} + +// ErrSeatChunkMissed is a chunk that did not come within its bound: the holder ended the send. +var ErrSeatChunkMissed = errors.New("a chunk did not come within its bound, so the send ended") + +// Receive reads `of` chunks in order, writes each one's bytes to w and acknowledges it before the next is sent. next +// yields the subject and body of the next chunk heard on Chunks(), or ErrSeatChunkMissed at its bound (SeatAckWithin); +// ack publishes on a subject. A chunk out of order, of another count or on another subject ends the receive, and so the +// send, since it is never acknowledged. It answers the bytes written. +func (c SeatChunks) Receive(of int64, next func() (subject string, body []byte, err error), ack func(subject string) error, w io.Writer) (int64, error) { + var written int64 + for n := int64(1); n <= of; n++ { + subject, body, err := next() + if err != nil { + return written, fmt.Errorf("chunk %d of %d: %w", n, of, err) + } + if subject != c.Chunk(n) { + return written, fmt.Errorf("chunk %d of %d came on %s, not %s", n, of, subject, c.Chunk(n)) + } + var ch SeatChunk + if err := json.Unmarshal(body, &ch); err != nil { + return written, fmt.Errorf("chunk %d of %d does not read: %w", n, of, err) + } + if ch.N != n || ch.Of != of { + return written, fmt.Errorf("chunk %d of %d says it is %d of %d", n, of, ch.N, ch.Of) + } + k, err := w.Write(ch.Data) + written += int64(k) + if err != nil { + return written, fmt.Errorf("chunk %d of %d cannot be written: %w", n, of, err) + } + if err := ack(c.Ack(n)); err != nil { + return written, fmt.Errorf("chunk %d of %d cannot be acknowledged: %w", n, of, err) + } + } + return written, nil +} diff --git a/go/buscalls/seatchunks_test.go b/go/buscalls/seatchunks_test.go new file mode 100644 index 00000000..4b1a995c --- /dev/null +++ b/go/buscalls/seatchunks_test.go @@ -0,0 +1,72 @@ +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) + } + }) + } +}