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 }