Conformance Test: PubSub - Ignore already processed messages (#722)

* Conformance Test: PubSub - Ignore already processed messages

* Fixing a race condition in StartHTTPServer

tests/conformance/pubsub/pubsub.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -112,13 +113,15 @@ func ConformanceTests(t *testing.T, props map[string]string, ps pubsub.PubSub, c
 	// from being considered as part of this test.
 	runID := uuid.Must(uuid.NewRandom()).String()
 	awaitingMessages := make(map[string]struct{}, 20)
+	var mu sync.Mutex
+	processedMessages := make(map[int]struct{}, 20)
 	processedC := make(chan string, config.messageCount*2)
 	errorCount := 0
 	dataPrefix := "message-" + runID + "-"
 	var outOfOrder bool
 
 	// Subscribe
-	if config.HasOperation("subscribe") {
+	if config.HasOperation("subscribe") { // nolint: nestif
 		t.Run("subscribe", func(t *testing.T) {
 			var counter int
 			var lastSequence int
@@ -133,8 +136,6 @@ func ConformanceTests(t *testing.T, props map[string]string, ps pubsub.PubSub, c
 					return nil
 				}
 
-				counter++
-
 				sequence, err := strconv.Atoi(dataString[len(dataPrefix):])
 				if err != nil {
 					t.Logf("Message did not contain a sequence number")
@@ -143,6 +144,20 @@ func ConformanceTests(t *testing.T, props map[string]string, ps pubsub.PubSub, c
 					return err
 				}
 
+				// Ignore already processed messages
+				// in case we receive a redelivery from the broker
+				// during retries.
+				mu.Lock()
+				_, alreadyProcessed := processedMessages[sequence]
+				mu.Unlock()
+				if alreadyProcessed {
+					t.Logf("Message was already processed: %d", sequence)
+
+					return nil
+				}
+
+				counter++
+
 				if sequence < lastSequence {
 					outOfOrder = true
 					t.Logf("Message received out of order: expected sequence >= %d, got %d", lastSequence, sequence)
@@ -165,6 +180,10 @@ func ConformanceTests(t *testing.T, props map[string]string, ps pubsub.PubSub, c
 				t.Logf("Simulating subscriber success")
 				actualReadCount++
 
+				mu.Lock()
+				processedMessages[sequence] = struct{}{}
+				mu.Unlock()
+
 				processedC <- dataString
 
 				return nil

tests/conformance/utils/utils.go

@@ -6,13 +6,13 @@
 package utils
 
 import (
-	"context"
 	"fmt"
 	"io/ioutil"
+	"net"
 	"net/http"
+	"net/http/httptest"
 	"os"
 	"os/signal"
-	"time"
 
 	"github.com/dapr/dapr/pkg/logger"
 	"github.com/gorilla/mux"
@@ -50,32 +50,33 @@ func (cc CommonConfig) CopyMap(config map[string]string) map[string]string {
 }
 
 func StartHTTPServer(port int, ready chan bool) {
-	s = server{}
-	srv := http.Server{
-		Addr:    fmt.Sprintf(":%d", port),
-		Handler: appRouter(),
+	l, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	if err != nil {
+		testLogger.Errorf("Error starting test HTTP serer: %v", err)
+
+		return
 	}
 
 	testLogger.Info(("Starting HTTP Server"))
-	go func() {
-		if err := srv.ListenAndServe(); err != nil {
-			testLogger.Errorf("Error with http server: %s", err.Error())
-		}
-	}()
+	ts := httptest.NewUnstartedServer(appRouter())
+	// NewUnstartedServer creates a listener. Close that listener and replace
+	// with the one we created.
+	ts.Listener.Close()
+	ts.Listener = l
+
+	// Start the server.
+	ts.Start()
+	defer ts.Close()
+
+	ready <- true
 
 	testLogger.Info(("Registering Signal"))
 	stop := make(chan os.Signal, 1)
 	signal.Notify(stop, os.Interrupt)
-	ready <- true
+
 	testLogger.Info(("Waiting to stop Server"))
 	<-stop
 	testLogger.Info(("Stopping Server"))
-
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	if err := srv.Shutdown(ctx); err != nil {
-		testLogger.Errorf("Error shutting down http server: %s", err.Error())
-	}
 }
 
 func appRouter() *mux.Router {