Function wmReceiveMessage

watch_manager_test.go:78–93

wmReceiveMessage receives a message with timeout and asserts expected values

(t *testing.T, receiveChan <-chan *Message, expectBucket, expectKey string, timeout time.Duration)

// wmReceiveMessage receives a message with timeout and asserts expected values
func wmReceiveMessage(t *testing.T, receiveChan <-chan *Message, expectBucket, expectKey string, timeout time.Duration) *Message {
	select {
	case msg, ok := <-receiveChan:
		if !ok {
			t.Logf("the channel with subscriber id %+v is closed", expectKey)
			return nil
		}
		assert.Equal(t, expectBucket, msg.BucketName)
		assert.Equal(t, expectKey, msg.Key)
		assert.NotNil(t, msg.Value)
		return msg
	case <-time.After(timeout):
		t.Fatalf("timeout waiting for message on %s/%s", expectBucket, expectKey)
		return nil
	}
}

// wmReceiveMessages receives count messages and returns the count received
func wmReceiveMessages(t *testing.T, receiveChan <-chan *Message, expectBucket, expectKey string, expectCount int, timeout time.Duration) int {