feat(horizon.core): add new mongo query for pulsar to get events betw…

…een from and to timestamp
telekom · Sep 19, 2024 · f956e07 · f956e07
2 parents e83a636 + 9b34db4
commit f956e07
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 1 deletion.
diff --git a/...on-core/src/main/java/de/telekom/eni/pandora/horizon/metrics/HorizonMetricsConstants.java b/...on-core/src/main/java/de/telekom/eni/pandora/horizon/metrics/HorizonMetricsConstants.java
@@ -9,6 +9,7 @@ public class HorizonMetricsConstants {
     public static final String METRIC_PUBLISHED_EVENTS = "published_events";
     public static final String METRIC_MULTIPLEXED_EVENTS = "multiplexed_events";
     public static final String METRIC_SUBSCRIPTION_COUNT = "subscription_count";
+    public static final String METRIC_SENT_SSE_EVENTS = "sent_sse_events";
     public static final String METRIC_OPEN_SSE_CONNECTIONS = "open_sse_connections";
     public static final String METRIC_OPEN_CIRCUIT_BREAKERS = "open_circuit_breakers";
     public static final String METRIC_CALLBACK_HTTP_CODE_COUNT = "callback_http_code_count";

diff --git a/.../src/main/java/de/telekom/eni/pandora/horizon/mongo/repository/MessageStateMongoRepo.java b/.../src/main/java/de/telekom/eni/pandora/horizon/mongo/repository/MessageStateMongoRepo.java
@@ -12,6 +12,7 @@
 import org.springframework.data.mongodb.repository.MongoRepository;
 import org.springframework.data.mongodb.repository.Query;
 
+import java.time.Instant;
 import java.util.Date;
 import java.util.List;
 
@@ -42,9 +43,14 @@ public interface MessageStateMongoRepo extends MongoRepository<MessageStateMongo
     @Query(value = "{status: {$in:  ?0}, deliveryType: ?1, subscriptionId: ?2}", sort = "{timestamp: 1}")
     Slice<MessageStateMongoDocument> findByStatusInAndDeliveryTypeAndSubscriptionIdAsc(List<Status> status, DeliveryType deliveryType, String subscriptionId, Pageable pageable);
 
+    @Query(value = "{status: {$in:  ?0}, deliveryType: ?1, subscriptionId: ?2, timestamp: { $gte: ?3, $lte: ?4 }}", sort = "{timestamp: 1}")
+    List<MessageStateMongoDocument> findByStatusInAndDeliveryTypeAndSubscriptionIdAndTimestampBetweenAsc(List<Status> status, DeliveryType deliveryType, String subscriptionId, Instant redeliveryFrom, Instant redeliveryTo);
+    @Query(value = "{status: {$in:  ?0}, deliveryType: ?1, subscriptionId: ?2, timestamp: { $gte: ?3, $lte: ?4 }}", sort = "{timestamp: 1}")
+    Slice<MessageStateMongoDocument> findByStatusInAndDeliveryTypeAndSubscriptionIdAndTimestampBetweenAsc(List<Status> status, DeliveryType deliveryType, String subscriptionId, Instant redeliveryFrom, Instant redeliveryTo, Pageable pageable);
+
+
     @Query(value = "{status: {$in:  ?0}, deliveryType: ?1, subscriptionId: {$in:  ?2}}", sort = "{timestamp: 1}")
     List<MessageStateMongoDocument> findByStatusInAndDeliveryTypeAndSubscriptionIdsAsc(List<Status> status, DeliveryType deliveryType, List<String> subscriptionIds);
-
     @Query(value = "{status: {$in:  ?0}, deliveryType: ?1, subscriptionId: {$in:  ?2}}", sort = "{timestamp: 1}")
     Slice<MessageStateMongoDocument> findByStatusInAndDeliveryTypeAndSubscriptionIdsAsc(List<Status> status, DeliveryType deliveryType, List<String> subscriptionIds, Pageable pageable);