Try to make sure the Receiver starts before we send events

Doug Davis · Doug Davis · commit 57be3cda29de · 2023-10-28T09:33:34.000Z
Signed-off-by: Doug Davis &lt;dug@microsoft.com&gt;
diff --git a/test/integration/mqtt_paho/mqtt_test.go b/test/integration/mqtt_paho/mqtt_test.go
@@ -37,12 +37,18 @@ func TestSendEvent(t *testing.T) {
 		// start a cloudevents receiver client go to receive the event
 		eventChan := make(chan receiveEvent)
 		defer close(eventChan)
+
+		// Used to try to make sure the receiver is ready before we start to
+		// send events
+		wait := make(chan bool)
+
 		go func() {
 			client, err := cloudevents.NewClient(protocolFactory(ctx, t, topicName))
 			if err != nil {
 				eventChan <- receiveEvent{err: err}
 				return
 			}
+			wait <- true
 			err = client.StartReceiver(ctx, func(event cloudevents.Event) {
 				eventChan <- receiveEvent{event: event}
 			})
@@ -52,6 +58,11 @@ func TestSendEvent(t *testing.T) {
 			}
 		}()
 
+		// Wait until receiver thread starts, and then wait a second to
+		// give the "StartReceive" call a chance to start (finger's crossed)
+		<-wait
+		time.Sleep(time.Second)
+
 		// start a cloudevents sender client go to send the event, set the topic on context
 		client, err := cloudevents.NewClient(protocolFactory(ctx, t, ""))
 		require.NoError(t, err)
diff --git a/test/integration/mqtt_paho_binding/mqtt_test.go b/test/integration/mqtt_paho_binding/mqtt_test.go
@@ -106,13 +106,25 @@ func startReceiver(ctx context.Context, topicName string, messageChan chan recei
 	if err != nil {
 		messageChan <- receiveMessage{err: err}
 	}
+
+	// Used to try to make sure the receiver is ready before we start to
+	// get events
+	wait := make(chan bool)
+
 	go func() {
+		wait <- true
 		err := receiver.OpenInbound(ctx)
 		if err != nil {
 			messageChan <- receiveMessage{err: err}
 		}
 		receiver.Close(ctx)
 	}()
+
+	// Wait for other thread to start and run OpenInbound + sleep a sec
+	// hoping that things will get ready before we call Receive() below
+	<-wait
+	time.Sleep(time.Second)
+
 	go func() {
 		msg, result := receiver.Receive(ctx)
 		messageChan <- receiveMessage{msg, result}
diff --git a/v2/client/test/test.go b/v2/client/test/test.go
@@ -25,13 +25,20 @@ import (
 func SendReceive(t *testing.T, protocolFactory func() interface{}, in event.Event, outAssert func(e event.Event), opts ...client.Option) {
 	t.Helper()
 	pf := protocolFactory()
-	c, err := client.New(pf, opts...)
+
+	// Create a sender and receiver client since we can't assume it's safe
+	// to use the same one for both roles
+
+	sender, err := client.New(pf, opts...)
+	require.NoError(t, err)
+
+	receiver, err := client.New(pf, opts...)
 	require.NoError(t, err)
+
 	wg := sync.WaitGroup{}
 	wg.Add(2)
 
-	// Give time for Kafka client protocol to get setup
-	time.Sleep(2 * time.Second)
+	receiverReady := make(chan bool)
 
 	go func() {
 		ctx, cancel := context.WithCancel(context.TODO())
@@ -42,7 +49,8 @@ func SendReceive(t *testing.T, protocolFactory func() interface{}, in event.Even
 			wg.Done()
 		}(inCh)
 		go func(channel chan event.Event) {
-			err := c.StartReceiver(ctx, func(e event.Event) {
+			receiverReady <- true
+			err := receiver.StartReceiver(ctx, func(e event.Event) {
 				channel <- e
 			})
 			if err != nil {
@@ -53,12 +61,14 @@ func SendReceive(t *testing.T, protocolFactory func() interface{}, in event.Even
 		outAssert(e)
 	}()
 
-	// Give time for the receiever to start
+	// Wait for receiver to be setup. Not 100% perefect but the channel + the
+	// sleep should do it
+	<-receiverReady
 	time.Sleep(2 * time.Second)
 
 	go func() {
 		defer wg.Done()
-		err := c.Send(context.Background(), in)
+		err := sender.Send(context.Background(), in)
 		require.NoError(t, err)
 	}()
 
diff --git a/v2/protocol/test/test.go b/v2/protocol/test/test.go
@@ -25,8 +25,13 @@ func SendReceive(t *testing.T, ctx context.Context, in binding.Message, s protoc
 	wg := sync.WaitGroup{}
 	wg.Add(2)
 
+	// Used to try to make sure the receiver is ready before we start to
+	// send events
+	wait := make(chan bool)
+
 	go func() {
 		defer wg.Done()
+		wait <- true
 		out, result := r.Receive(ctx)
 		if !protocol.IsACK(result) {
 			require.NoError(t, result)
@@ -36,6 +41,11 @@ func SendReceive(t *testing.T, ctx context.Context, in binding.Message, s protoc
 		require.NoError(t, finishErr)
 	}()
 
+	// Wait until receiver thread starts, and then wait a second to
+	// give the "Receive" call a chance to start (finger's crossed)
+	<-wait
+	time.Sleep(time.Second)
+
 	go func() {
 		defer wg.Done()
 		mx := sync.Mutex{}
