Wait for receiver to be done (#5509)

This polls, unlike AssertReceivedAll. My theory is that in slow
environments the time between sender being done and receiver being done
might be significant enough to fail a test.

test/rekt/features/broker/source_to_sink.go

@@ -160,6 +160,8 @@ func SourceToTwoSinksWithDLQ(brokerName string) *feature.Feature {
 
 	// After we have finished sending.
 	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("receiver 1 is finished", prober.ReceiverDone("source", "sink1"))
+	f.Requirement("receiver 2 is finished", prober.ReceiverDone("source", "sink2"))
 
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").

test/rekt/features/channel/features.go

@@ -137,6 +137,7 @@ func EventTransformation() *feature.Feature {
 	f.Setup("event library is ready", eventlibrary.IsReady(lib))
 
 	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("receiver is finished", prober.ReceiverDone("source", "sink"))
 
 	f.Assert("sink receives events", prober.AssertReceivedAll("source", "sink"))
 	f.Assert("events have passed through transform service", func(ctx context.Context, t feature.T) {
@@ -177,6 +178,7 @@ func SingleEventWithEncoding(encoding binding.Encoding) *feature.Feature {
 	f.Setup("install source", prober.SenderInstall("source", eventshub.InputEventWithEncoding(event, encoding)))
 
 	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("receiver is finished", prober.ReceiverDone("source", "sink"))
 
 	f.Assert("sink receives events", prober.AssertReceivedAll("source", "sink"))