Give each enrolment stand-in its own durable and wait for it to be gone (issue 604)
The stand-in shared one durable name across tests and let go with Unsubscribe alone, so a test starting before the server dropped the binding was refused with "consumer is already bound to a subscription".
This commit is contained in:
@@ -3,6 +3,8 @@ package link
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -48,6 +50,19 @@ func theMeshAnswers(t *testing.T, conn *nats.Conn, js nats.JetStreamContext,
|
|||||||
reply EnrolReply) <-chan string {
|
reply EnrolReply) <-chan string {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
sawReplyField := make(chan string, 1)
|
sawReplyField := make(chan string, 1)
|
||||||
|
// **A consumer of this test's own, gone before it starts and gone when it ends.** The stand-in
|
||||||
|
// used to be one durable name shared by every test, released by Unsubscribe alone. Unsubscribe
|
||||||
|
// only asks the server to let go: it deletes the consumer only when this subscription made it,
|
||||||
|
// and the server drops the binding on its own time. A test that started in that window (or one
|
||||||
|
// that found the consumer left by an earlier run on a long-lived bus) was refused with "consumer
|
||||||
|
// is already bound to a subscription". Per-test names remove the sharing; the delete-and-wait
|
||||||
|
// before and after removes the dependence on the server's timing.
|
||||||
|
durable := standinName(t)
|
||||||
|
stream, err := js.StreamNameBySubject(EnrolSubject)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
standinGone(t, js, stream, durable)
|
||||||
sub, err := js.Subscribe(EnrolSubject, func(msg *nats.Msg) {
|
sub, err := js.Subscribe(EnrolSubject, func(msg *nats.Msg) {
|
||||||
select {
|
select {
|
||||||
case sawReplyField <- msg.Reply:
|
case sawReplyField <- msg.Reply:
|
||||||
@@ -65,14 +80,46 @@ func theMeshAnswers(t *testing.T, conn *nats.Conn, js nats.JetStreamContext,
|
|||||||
// send it to whatever the transport's reply field holds, and that is the point.
|
// send it to whatever the transport's reply field holds, and that is the point.
|
||||||
_ = conn.Publish(addressed.ReplyTo, body)
|
_ = conn.Publish(addressed.ReplyTo, body)
|
||||||
_ = msg.Ack()
|
_ = msg.Ack()
|
||||||
}, nats.Durable("controller-standin"), nats.ManualAck())
|
}, nats.Durable(durable), nats.ManualAck())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
t.Cleanup(func() {
|
||||||
|
_ = sub.Unsubscribe()
|
||||||
|
standinGone(t, js, stream, durable)
|
||||||
|
})
|
||||||
return sawReplyField
|
return sawReplyField
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// standinName is the stand-in's durable name for one test: the test's name with every character a
|
||||||
|
// consumer name may not hold replaced.
|
||||||
|
func standinName(t *testing.T) string {
|
||||||
|
t.Helper()
|
||||||
|
return "controller-standin-" + strings.Map(func(r rune) rune {
|
||||||
|
if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' {
|
||||||
|
return r
|
||||||
|
}
|
||||||
|
return '_'
|
||||||
|
}, t.Name())
|
||||||
|
}
|
||||||
|
|
||||||
|
// standinGone deletes the durable consumer if it exists and waits until the server no longer reports
|
||||||
|
// it, so that nothing after this call can meet its binding.
|
||||||
|
func standinGone(t *testing.T, js nats.JetStreamContext, stream, durable string) {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.Now().Add(5 * time.Second)
|
||||||
|
for {
|
||||||
|
if _, err := js.ConsumerInfo(stream, durable); errors.Is(err, nats.ErrConsumerNotFound) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
_ = js.DeleteConsumer(stream, durable)
|
||||||
|
if time.Now().After(deadline) {
|
||||||
|
t.Fatalf("consumer %s on %s still exists 5s after it was deleted", durable, stream)
|
||||||
|
}
|
||||||
|
time.Sleep(20 * time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// An enrolment is answered on the address the request carried, and the transport's own reply field
|
// An enrolment is answered on the address the request carried, and the transport's own reply field
|
||||||
// held something else entirely.
|
// held something else entirely.
|
||||||
func TestNatsAnEnrolmentIsAnsweredOnTheAddressInItsPayload(t *testing.T) {
|
func TestNatsAnEnrolmentIsAnsweredOnTheAddressInItsPayload(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user