Handling errors on pubsub

Author: fcostaoliveira

.gitignore

@@ -87,6 +87,8 @@ Thumbs.db
 # Coverage Results #
 ####################
 coverage.txt
+.coverdata/
+pubsub-sub-bench-test
 
 # Profiler Results #
 ####################

Makefile

@@ -33,6 +33,11 @@ build-race:
 	$(GOBUILDRACE) \
         -ldflags=$(LDFLAGS) .
 
+build-cover:
+	@echo "Building binary with coverage instrumentation..."
+	$(GOBUILD) -cover \
+        -ldflags=$(LDFLAGS) .
+
 checkfmt:
 	@echo 'Checking gofmt';\
  	bash -c "diff -u <(echo -n) <(go fmt .)";\
@@ -52,9 +57,21 @@ fmt:
 get:
 	$(GOGET) -t -v ./...
 
-test: get
+test: get build-cover
 	$(GOFMT) ./...
-	$(GOTEST) -race -covermode=atomic ./...
+	@rm -rf .coverdata
+	@mkdir -p .coverdata
+	$(GOTEST) -v -race -covermode=atomic ./...
 
-coverage: get test
-	$(GOTEST) -race -coverprofile=coverage.txt -covermode=atomic .
+coverage: get build-cover
+	$(GOFMT) ./...
+	@rm -rf .coverdata
+	@mkdir -p .coverdata
+	$(GOTEST) -v -race -covermode=atomic .
+	@if [ -d .coverdata ] && [ -n "$$(ls -A .coverdata 2>/dev/null)" ]; then \
+		echo "Converting coverage data..."; \
+		go tool covdata textfmt -i=.coverdata -o coverage.txt; \
+	else \
+		echo "No coverage data found, creating empty coverage file"; \
+		touch coverage.txt; \
+	fi

subscriber.go

@@ -126,7 +126,8 @@ func publisherRoutine(clientName string, channels []string, mode string, measure
 				publishLatency := time.Now().UnixNano() - startPublish
 
 				if err != nil {
-					log.Printf("Error publishing to channel %s: %v", ch, err)
+					log.Printf("Publisher %s: error publishing to channel %s: %v", clientName, ch, err)
+					// Don't send metrics on error, but still count the message attempt
 				} else {
 					// Send metrics to channels
 					publishLatencyChannel <- publishLatency
@@ -218,11 +219,34 @@ func subscriberRoutine(clientName, mode string, channels []string, verbose bool,
 			// Handle messages
 			msg, err := pubsub.ReceiveMessage(ctx)
 			if err != nil {
-				// Handle Redis connection errors, e.g., reconnect immediately
+				// Handle Redis connection errors
 				if err == redis.Nil || err == context.DeadlineExceeded || err == context.Canceled {
 					continue
 				}
-				panic(err)
+				// Connection error (EOF, network error, etc.) - attempt to reconnect
+				log.Printf("Subscriber %s: connection error: %v - attempting to reconnect\n", clientName, err)
+
+				// Close the bad connection
+				if pubsub != nil {
+					pubsub.Close()
+					atomic.AddInt64(&totalSubscribedChannels, int64(-len(channels)))
+				}
+
+				// Wait a bit before reconnecting
+				time.Sleep(100 * time.Millisecond)
+
+				// Resubscribe
+				switch mode {
+				case "ssubscribe":
+					pubsub = client.SSubscribe(ctx, channels...)
+				default:
+					pubsub = client.Subscribe(ctx, channels...)
+				}
+				atomic.AddInt64(&totalSubscribedChannels, int64(len(channels)))
+				atomic.AddUint64(&totalConnects, 1)
+
+				log.Printf("Subscriber %s: reconnected successfully\n", clientName)
+				continue
 			}
 			if verbose {
 				log.Println(fmt.Sprintf("received message in channel %s. Message: %s", msg.Channel, msg.Payload))