Function TestConsumerConfigMatches

jetstream/test/jetstream_test.go:2136–2189

(t *testing.T)

}

func TestConsumerConfigMatches(t *testing.T) {
	srv := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, srv)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	s, err := js.CreateStream(context.Background(), jetstream.StreamConfig{
		Name:     "FOO",
		Subjects: []string{"foo.*"},
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	cfg := jetstream.ConsumerConfig{
		Name:               "cons",
		Durable:            "cons",
		Description:        "test",
		DeliverPolicy:      jetstream.DeliverByStartSequencePolicy,
		OptStartSeq:        5,
		AckPolicy:          jetstream.AckAllPolicy,
		AckWait:            1 * time.Second,
		MaxDeliver:         5,
		BackOff:            []time.Duration{1 * time.Second, 2 * time.Second, 3 * time.Second},
		ReplayPolicy:       jetstream.ReplayOriginalPolicy,
		SampleFrequency:    "50%",
		MaxWaiting:         100,
		MaxAckPending:      1000,
		HeadersOnly:        true,
		MaxRequestBatch:    100,
		MaxRequestExpires:  10 * time.Second,
		MaxRequestMaxBytes: 1000,
		InactiveThreshold:  20 * time.Second,
		Replicas:           1,
		MemoryStorage:      true,
		FilterSubjects:     []string{"foo.1", "foo.2"},
	}

	c, err := s.CreateConsumer(context.Background(), cfg)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	// server will set metadata values, so we need to clear them
	c.CachedInfo().Config.Metadata = nil
	if !reflect.DeepEqual(c.CachedInfo().Config, cfg) {
		t.Fatalf("ConsumerConfig doesn't match")
	}
}

func TestJetStreamCleanupPublisher(t *testing.T) {