Function TestJetStreamConsumerConfigReplicasAndMemStorage

test/js_test.go:9763–9804

(t *testing.T)

}

func TestJetStreamConsumerConfigReplicasAndMemStorage(t *testing.T) {
	withJSCluster(t, "CR", 3, func(t *testing.T, nodes ...*jsServer) {
		nc, js := jsClient(t, nodes[0].Server)
		defer nc.Close()

		if _, err := js.AddStream(&nats.StreamConfig{
			Name:     "CR",
			Subjects: []string{"foo"},
			Replicas: 3,
		}); err != nil {
			t.Fatalf("Error adding stream: %v", err)
		}

		// We can't really check if the consumer ends-up with memory storage or not.
		// We are simply going to create a NATS subscription on the request subject
		// and make sure that the request contains "mem_storage:true".
		sub, err := nc.SubscribeSync("$JS.API.CONSUMER.CREATE.CR.dur")
		if err != nil {
			t.Fatalf("Error on subscribe: %v", err)
		}

		ci, err := js.AddConsumer("CR", &nats.ConsumerConfig{
			Durable:        "dur",
			DeliverSubject: "bar",
			Replicas:       1,
			MemoryStorage:  true,
		})
		if err != nil {
			t.Fatalf("Error adding consumer: %v", err)
		}
		if n := len(ci.Cluster.Replicas); n > 0 {
			t.Fatalf("Expected replicas to be 1, got %+v", ci.Cluster)
		}
		msg, err := sub.NextMsg(time.Second)
		if err != nil {
			t.Fatalf("Error on next msg: %v", err)
		}
		if str := string(msg.Data); !strings.Contains(str, "mem_storage\":true") {
			t.Fatalf("Does not look like the request asked for memory storage: %s", str)
		}
	})
}

func TestJetStreamRePublish(t *testing.T) {
	s := RunBasicJetStreamServer()