Function TestBarrier

test/conn_test.go:2285–2496

(t *testing.T)

}

func TestBarrier(t *testing.T) {
	s := RunDefaultServer()
	defer s.Shutdown()

	nc := NewDefaultConnection(t)
	defer nc.Close()

	pubMsgs := int32(0)
	ch := make(chan bool, 1)

	sub1, err := nc.Subscribe("pub", func(_ *nats.Msg) {
		atomic.AddInt32(&pubMsgs, 1)
		time.Sleep(250 * time.Millisecond)
	})
	if err != nil {
		t.Fatalf("Error on subscribe: %v", err)
	}

	sub2, err := nc.Subscribe("close", func(_ *nats.Msg) {
		// The "close" message was sent/received lat, but
		// because we are dealing with different subscriptions,
		// which are dispatched by different dispatchers, and
		// because the "pub" subscription is delayed, this
		// callback is likely to be invoked before the sub1's
		// second callback is invoked. Using the Barrier call
		// here will ensure that the given function will be invoked
		// after the preceding messages have been dispatched.
		nc.Barrier(func() {
			res := atomic.LoadInt32(&pubMsgs) == 2
			ch <- res
		})
	})
	if err != nil {
		t.Fatalf("Error on subscribe: %v", err)
	}

	// Send 2 "pub" messages followed by a "close" message
	for i := 0; i < 2; i++ {
		if err := nc.Publish("pub", []byte("pub msg")); err != nil {
			t.Fatalf("Error on publish: %v", err)
		}
	}
	if err := nc.Publish("close", []byte("closing")); err != nil {
		t.Fatalf("Error on publish: %v", err)
	}

	select {
	case ok := <-ch:
		if !ok {
			t.Fatal("The barrier function was invoked before the second message")
		}
	case <-time.After(2 * time.Second):
		t.Fatal("Waited for too long...")
	}

	// Remove all subs
	sub1.Unsubscribe()
	sub2.Unsubscribe()