Function TestJetStreamPublishAsync

test/js_test.go:8052–8204

(t *testing.T)

}

func TestJetStreamPublishAsync(t *testing.T) {
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

	nc, js := jsClient(t, s)
	defer nc.Close()

	// Make sure we get a proper failure when no stream is present.
	paf, err := js.PublishAsync("foo", []byte("Hello JS"))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	select {
	case <-paf.Ok():
		t.Fatalf("Did not expect to get PubAck")
	case err := <-paf.Err():
		if err != nats.ErrNoResponders {
			t.Fatalf("Expected a ErrNoResponders error, got %v", err)
		}
		// Should be able to get the message back to resend, etc.
		m := paf.Msg()
		if m == nil {
			t.Fatalf("Expected to be able to retrieve the message")
		}
		if m.Subject != "foo" || string(m.Data) != "Hello JS" {
			t.Fatalf("Wrong message: %+v", m)
		}
	case <-time.After(time.Second):
		t.Fatalf("Did not receive an error in time")
	}

	// Now create a stream and expect a PubAck from <-OK().
	if _, err := js.AddStream(&nats.StreamConfig{Name: "TEST"}); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	paf, err = js.PublishAsync("TEST", []byte("Hello JS ASYNC PUB"))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	select {
	case pa := <-paf.Ok():
		if pa.Stream != "TEST" || pa.Sequence != 1 {
			t.Fatalf("Bad PubAck: %+v", pa)
		}
	case err := <-paf.Err():
		t.Fatalf("Did not expect to get an error: %v", err)
	case <-time.After(time.Second):
		t.Fatalf("Did not receive a PubAck in time")
	}

	errCh := make(chan error, 1)

	// Make sure we can register an async err handler for these.
	errHandler := func(js nats.JetStream, originalMsg *nats.Msg, err error) {
		if originalMsg == nil {