fix: restore behavior of failed events handler, so that it only works…

… with sse events
telekom · Sep 23, 2024 · 01f9fd7 · 01f9fd7
1 parent 0590949
commit 01f9fd7
Showing 1 changed file with 14 additions and 4 deletions.
diff --git a/internal/handler/failed.go b/internal/handler/failed.go
@@ -55,13 +55,23 @@ func CheckFailedEvents() {
 		log.Debug().Msgf("Found %d FAILED messages in database", len(dbMessages))
 
 		for _, dbMessage := range dbMessages {
-			if err := republish.RepublishEvent(picker, &dbMessage, nil); err != nil {
-				log.Error().Err(err).Msgf("Error while republishing message for subscriptionId %s", dbMessage.SubscriptionId)
+			subscriptionId := dbMessage.SubscriptionId
 
-				continue
+			subscription, err := cache.SubscriptionCache.Get(config.Current.Hazelcast.Caches.SubscriptionCache, subscriptionId)
+			if err != nil {
+				log.Error().Err(err).Msgf("Error while fetching republishing entry for subscriptionId %s", subscriptionId)
+				return
 			}
 
-			log.Debug().Msgf("Successfully republished message in state FAILED for subscriptionId %s", dbMessage.SubscriptionId)
+			if subscription != nil && (subscription.Spec.Subscription.DeliveryType == "sse" || subscription.Spec.Subscription.DeliveryType == "server_sent_event") {
+				if err := republish.RepublishEvent(picker, &dbMessage, subscription); err != nil {
+					log.Error().Err(err).Msgf("Error while republishing message for subscriptionId %s", dbMessage.SubscriptionId)
+
+					continue
+				}
+
+				log.Debug().Msgf("Successfully republished message in state FAILED for subscriptionId %s", dbMessage.SubscriptionId)
+			}
 		}
 
 		if len(dbMessages) < int(config.Current.Republishing.BatchSize) {