jschoubben
6e208f7b3e
The outbound half went behind `Bus` and a node's two statements stopped naming a transport. This is the other half, and where the transport reached furthest: the run loop selected on a channel of the client library's own delivery type, so every part of holding a node in its mesh knew which bus it was on. `Link` is dialling, hearing and saying in one interface, because dialling is where the transport is chosen and choosing it twice is how one half of a node ends up on a different bus from the other. `Declaration` has one way of being done rather than two: a declaration set aside for a newer one is settled exactly as an applied one is, on both buses, and the difference is a fact the report carries. Four things this settled. **The host declares nothing on the new bus.** On the bus the mesh has it declares its own queue, because a queue that is not there means a node that hears nothing. Here it binds to a consumer the mesh made when the node enrolled, and a missing one is said as the mesh's to answer rather than quietly created with whatever this client happens to default to. **The pin is easier here than in the tool runtime, not harder.** The Go client takes a *tls.Config, so the same PinnedConfig with the same VerifyPeerCertificate does the work — the subject-alternative-name constraint recorded against the runtime's client is that client's, because it takes PEM strings with no verify hook. A host checks the fingerprint and nothing else. **Binding needs the subject as well as the consumer.** An empty subject is refused rather than taken to mean "whatever that consumer delivers", which the server said plainly and only when asked. **Reconnection stays the caller's.** Hold already decides when to try again and how long to wait; a client reconnecting underneath it would make that reasoning a duplicate of the library's. The drain keeps its live half and loses its catch-up half, as it said it would: verified that three declarations pushed to an absent node leave one on the stream, and it is the newest. One test-harness lesson worth the comment it got: delete-then-add is not a reset. A test that did that inherited the previous test's messages, and the symptom was a declaration counted as delivered twice — which reads as a redelivery bug in the code under test rather than as a dirty stream.
64 lines
2.1 KiB
Go
64 lines
2.1 KiB
Go
package link
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// said is one declaration as a test hands it over, with no transport under it — which is what the
|
|
// seam bought: the drain's reasoning was reachable only through a real broker before.
|
|
type said struct {
|
|
body []byte
|
|
handled bool
|
|
}
|
|
|
|
func (s *said) Body() []byte { return s.body }
|
|
func (s *said) Handled() error { s.handled = true; return nil }
|
|
|
|
func arriving(bodies ...string) chan Declaration {
|
|
ch := make(chan Declaration, 8)
|
|
for _, b := range bodies {
|
|
ch <- &said{body: []byte(b)}
|
|
}
|
|
return ch
|
|
}
|
|
|
|
// A machine asked to be five things becomes the last one: what is already waiting supersedes what
|
|
// arrived first, and everything set aside is named so it can be reported.
|
|
func TestWhatIsAlreadyWaitingSupersedesWhatArrivedFirst(t *testing.T) {
|
|
waiting := arriving("two", "three", "four")
|
|
apply, superseded := newest(waiting, &said{body: []byte("one")}, 50*time.Millisecond)
|
|
if string(apply.Body()) != "four" {
|
|
t.Fatalf("applied %q, not the newest", apply.Body())
|
|
}
|
|
if len(superseded) != 3 || string(superseded[0].Body()) != "one" ||
|
|
string(superseded[2].Body()) != "three" {
|
|
t.Fatalf("set aside %d: %v", len(superseded), superseded)
|
|
}
|
|
}
|
|
|
|
// One declaration with nothing behind it is applied as it always was, after the window.
|
|
func TestALoneDeclarationIsAppliedAfterTheWindow(t *testing.T) {
|
|
began := time.Now()
|
|
apply, superseded := newest(arriving(), &said{body: []byte("only")}, 30*time.Millisecond)
|
|
if string(apply.Body()) != "only" || len(superseded) != 0 {
|
|
t.Fatalf("got %q with %d set aside", apply.Body(), len(superseded))
|
|
}
|
|
if time.Since(began) < 30*time.Millisecond {
|
|
t.Fatal("did not wait the window for a straggler")
|
|
}
|
|
}
|
|
|
|
// A straggler within the window is taken; one after it is the next push.
|
|
func TestAStragglerWithinTheWindowIsTaken(t *testing.T) {
|
|
waiting := arriving()
|
|
go func() {
|
|
time.Sleep(20 * time.Millisecond)
|
|
waiting <- &said{body: []byte("late")}
|
|
}()
|
|
apply, superseded := newest(waiting, &said{body: []byte("first")}, 100*time.Millisecond)
|
|
if string(apply.Body()) != "late" || len(superseded) != 1 {
|
|
t.Fatalf("got %q with %d set aside", apply.Body(), len(superseded))
|
|
}
|
|
}
|