Feature/add throttling

subscription-service/src/main/kotlin/com/egm/stellio/subscription/model/Subscription.kt

@@ -49,7 +49,8 @@ data class Subscription(
     //  - used to compact entities in notifications
     //  - used when needed to serve contexts in JSON notifications
     @JsonProperty(value = JSONLD_CONTEXT)
-    val contexts: List<ExpandedTerm>
+    val contexts: List<ExpandedTerm>,
+    val throttling: Int? = null
 ) {
 
     @Transient

subscription-service/src/main/kotlin/com/egm/stellio/subscription/service/SubscriptionService.kt

@@ -1,7 +1,10 @@
 package com.egm.stellio.subscription.service
 
-import arrow.core.*
+import arrow.core.Either
+import arrow.core.Option
+import arrow.core.left
 import arrow.core.raise.either
+import arrow.core.right
 import com.egm.stellio.shared.model.*
 import com.egm.stellio.shared.util.*
 import com.egm.stellio.shared.util.JsonLdUtils.JSONLD_TYPE_TERM
@@ -10,19 +13,18 @@ import com.egm.stellio.shared.util.JsonLdUtils.NGSILD_SUBSCRIPTION_TERM
 import com.egm.stellio.shared.util.JsonLdUtils.expandJsonLdTerm
 import com.egm.stellio.subscription.config.SubscriptionProperties
 import com.egm.stellio.subscription.model.*
-import com.egm.stellio.subscription.model.GeoQ
-import com.egm.stellio.subscription.model.Subscription
 import com.egm.stellio.subscription.utils.*
 import com.egm.stellio.subscription.utils.ParsingUtils.endpointInfoMapToString
 import com.egm.stellio.subscription.utils.ParsingUtils.endpointInfoToString
 import com.egm.stellio.subscription.utils.ParsingUtils.parseEndpointInfo
 import com.egm.stellio.subscription.utils.ParsingUtils.parseEntitySelector
 import com.egm.stellio.subscription.utils.ParsingUtils.toSqlColumnName
 import com.egm.stellio.subscription.utils.ParsingUtils.toSqlValue
+import com.egm.stellio.subscription.web.invalidSubscriptiondAttributeMessage
+import com.egm.stellio.subscription.web.unsupportedSubscriptiondAttributeMessage
 import io.r2dbc.postgresql.codec.Json
 import kotlinx.coroutines.reactive.awaitFirst
 import org.locationtech.jts.geom.Geometry
-import org.slf4j.LoggerFactory
 import org.springframework.data.r2dbc.core.R2dbcEntityTemplate
 import org.springframework.data.relational.core.query.Criteria.where
 import org.springframework.data.relational.core.query.Query.query
@@ -44,13 +46,12 @@ class SubscriptionService(
     private val r2dbcEntityTemplate: R2dbcEntityTemplate
 ) {
 
-    private val logger = LoggerFactory.getLogger(javaClass)
-
     suspend fun validateNewSubscription(subscription: Subscription): Either<APIException, Unit> = either {
         checkTypeIsSubscription(subscription).bind()
         checkIdIsValid(subscription).bind()
         checkEntitiesOrWatchedAttributes(subscription).bind()
         checkTimeIntervalGreaterThanZero(subscription).bind()
+        checkThrottlingGreaterThanZero(subscription).bind()
         checkSubscriptionValidity(subscription).bind()
         checkExpiresAtInTheFuture(subscription).bind()
         checkIdPatternIsValid(subscription).bind()
@@ -73,17 +74,31 @@ class SubscriptionService(
         else Unit.right()
 
     private fun checkSubscriptionValidity(subscription: Subscription): Either<APIException, Unit> =
-        if (subscription.watchedAttributes != null && subscription.timeInterval != null)
-            BadRequestDataException(
-                "You can't use 'timeInterval' in conjunction with 'watchedAttributes'"
-            ).left()
-        else Unit.right()
+        when {
+            subscription.watchedAttributes != null && subscription.timeInterval != null -> {
+                BadRequestDataException(
+                    "You can't use 'timeInterval' in conjunction with 'watchedAttributes'"
+                ).left()
+            }
+            subscription.timeInterval != null && subscription.throttling != null -> {
+                BadRequestDataException(
+                    "You can't use 'timeInterval' in conjunction with 'throttling'"
+                ).left()
+            }
+            else ->
+                Unit.right()
+        }
 
     private fun checkTimeIntervalGreaterThanZero(subscription: Subscription): Either<APIException, Unit> =
-        if (subscription.timeInterval != null && subscription.timeInterval < 1)
+        if (subscription.timeInterval != null && subscription.timeInterval < 1 && subscription.throttling == null)
             BadRequestDataException("The value of 'timeInterval' must be greater than zero (int)").left()
         else Unit.right()
 
+    private fun checkThrottlingGreaterThanZero(subscription: Subscription): Either<APIException, Unit> =
+        if (subscription.throttling != null && subscription.throttling < 1)
+            BadRequestDataException("The value of 'throttling' must be greater than zero (int)").left()
+        else Unit.right()
+
     private fun checkExpiresAtInTheFuture(subscription: Subscription): Either<BadRequestDataException, Unit> =
         if (subscription.expiresAt != null && subscription.expiresAt.isBefore(ngsiLdDateTime()))
             BadRequestDataException("'expiresAt' must be in the future").left()
@@ -134,11 +149,11 @@ class SubscriptionService(
             INSERT INTO subscription(id, type, subscription_name, created_at, description, watched_attributes,
                 notification_trigger, time_interval, q, scope_q, notif_attributes, notif_format, endpoint_uri, 
                 endpoint_accept, endpoint_receiver_info, endpoint_notifier_info, times_sent, is_active, 
-                expires_at, sub, contexts)
+                expires_at, sub, contexts, throttling)
             VALUES(:id, :type, :subscription_name, :created_at, :description, :watched_attributes, 
                 :notification_trigger, :time_interval, :q, :scope_q, :notif_attributes, :notif_format, :endpoint_uri, 
                 :endpoint_accept, :endpoint_receiver_info, :endpoint_notifier_info, :times_sent, :is_active, 
-                :expires_at, :sub, :contexts)
+                :expires_at, :sub, :contexts, :throttling)
             """.trimIndent()
 
         databaseClient.sql(insertStatement)
@@ -163,6 +178,7 @@ class SubscriptionService(
             .bind("expires_at", subscription.expiresAt)
             .bind("sub", sub.toStringValue())
             .bind("contexts", subscription.contexts.toTypedArray())
+            .bind("throttling", subscription.throttling)
             .execute().bind()
 
         geoQuery?.let {
@@ -233,7 +249,7 @@ class SubscriptionService(
                 notif_format, endpoint_uri, endpoint_accept, endpoint_receiver_info, endpoint_notifier_info, status, 
                 times_sent, is_active, last_notification, last_failure, last_success, entity_selector.id as entity_id, 
                 id_pattern, entity_selector.type_selection as type_selection, georel, geometry, coordinates, 
-                pgis_geometry, geoproperty, scope_q, expires_at, contexts
+                pgis_geometry, geoproperty, scope_q, expires_at, contexts, throttling
             FROM subscription 
             LEFT JOIN entity_selector ON entity_selector.subscription_id = :id
             LEFT JOIN geometry_query ON geometry_query.subscription_id = :id 
@@ -337,22 +353,21 @@ class SubscriptionService(
                     "q",
                     "scopeQ",
                     "isActive",
-                    "modifiedAt"
+                    "modifiedAt",
+                    "throttling"
                 ).contains(it.key) -> {
                     val columnName = it.key.toSqlColumnName()
                     val value = it.value.toSqlValue(it.key)
                     updateSubscriptionAttribute(subscriptionId, columnName, value).bind()
                 }
 
-                listOf("csf", "throttling", "temporalQ").contains(it.key) -> {
-                    logger.warn("Subscription $subscriptionId has unsupported attribute: ${it.key}")
-                    NotImplementedException("Subscription $subscriptionId has unsupported attribute: ${it.key}")
+                listOf("csf", "temporalQ").contains(it.key) -> {
+                    NotImplementedException(unsupportedSubscriptiondAttributeMessage(subscriptionId, it))
                         .left().bind<Unit>()
                 }
 
                 else -> {
-                    logger.warn("Subscription $subscriptionId has invalid attribute: ${it.key}")
-                    BadRequestDataException("Subscription $subscriptionId has invalid attribute: ${it.key}")
+                    BadRequestDataException(invalidSubscriptiondAttributeMessage(subscriptionId, it))
                         .left().bind<Unit>()
                 }
             }
@@ -416,6 +431,7 @@ class SubscriptionService(
                 }
                 listOf(Pair("notif_attributes", attributes))
             }
+
             "format" -> {
                 val format =
                     if (attribute.value == "keyValues")
@@ -424,6 +440,7 @@ class SubscriptionService(
                         NotificationParams.FormatType.NORMALIZED.name
                 listOf(Pair("notif_format", format))
             }
+
             "endpoint" -> {
                 val endpoint = attribute.value as Map<String, Any>
                 val accept =
@@ -441,6 +458,7 @@ class SubscriptionService(
                     Pair("endpoint_notifier_info", Json.of(endpointInfoMapToString(endpointNotifierInfo)))
                 )
             }
+
             else -> throw BadRequestDataException("Could not update attribute ${attribute.key}")
         }
     }
@@ -479,7 +497,7 @@ class SubscriptionService(
                 notif_format, endpoint_uri, endpoint_accept, endpoint_receiver_info, endpoint_notifier_info, status, 
                 times_sent, is_active, last_notification, last_failure, last_success, entity_selector.id as entity_id,
                 id_pattern, entity_selector.type_selection as type_selection, georel, geometry, coordinates, 
-                pgis_geometry, geoproperty, scope_q, expires_at, contexts
+                pgis_geometry, geoproperty, scope_q, expires_at, contexts, throttling
             FROM subscription 
             LEFT JOIN entity_selector ON entity_selector.subscription_id = subscription.id
             LEFT JOIN geometry_query ON geometry_query.subscription_id = subscription.id
@@ -521,13 +539,15 @@ class SubscriptionService(
                    entity_selector.id as entity_id, entity_selector.id_pattern as id_pattern, 
                    entity_selector.type_selection as type_selection, georel, geometry, coordinates, pgis_geometry,
                    geoproperty, scope_q, notif_attributes, notif_format, endpoint_uri, endpoint_accept, times_sent, 
-                   endpoint_receiver_info, endpoint_notifier_info, contexts
+                   endpoint_receiver_info, endpoint_notifier_info, contexts, throttling
             FROM subscription 
             LEFT JOIN entity_selector on subscription.id = entity_selector.subscription_id
             LEFT JOIN geometry_query on subscription.id = geometry_query.subscription_id
             WHERE is_active
             AND ( expires_at is null OR expires_at >= :date )
             AND time_interval IS NULL
+            AND ( throttling IS NULL 
+                OR (last_notification + throttling * INTERVAL '1 second') > CURRENT_TIMESTAMP )
             AND ( string_to_array(watched_attributes, ',') && string_to_array(:updatedAttributes, ',')
                 OR watched_attributes IS NULL)
             AND CASE
@@ -669,7 +689,8 @@ class SubscriptionService(
                 lastSuccess = toNullableZonedDateTime(row["last_success"])
             ),
             isActive = toBoolean(row["is_active"]),
-            contexts = toList(row["contexts"])
+            contexts = toList(row["contexts"]),
+            throttling = toNullableInt(row["throttling"])
         )
     }
 
@@ -698,7 +719,8 @@ class SubscriptionService(
                 lastFailure = null,
                 lastSuccess = null
             ),
-            contexts = toList(row["contexts"])
+            contexts = toList(row["contexts"]),
+            throttling = toNullableInt(row["throttling"])
         )
     }
 
@@ -734,11 +756,12 @@ class SubscriptionService(
                 scope_q, notif_attributes, notif_format, endpoint_uri, endpoint_accept, endpoint_receiver_info,
                 endpoint_notifier_info, status, times_sent, last_notification, last_failure, last_success, is_active, 
                 entity_selector.id as entity_id, id_pattern, entity_selector.type_selection as type_selection, georel,
-                geometry, coordinates, pgis_geometry, geoproperty, contexts
+                geometry, coordinates, pgis_geometry, geoproperty, contexts, throttling
             FROM subscription
             LEFT JOIN entity_selector ON entity_selector.subscription_id = subscription.id
             LEFT JOIN geometry_query ON geometry_query.subscription_id = subscription.id
             WHERE time_interval IS NOT NULL
+            AND throttling IS NULL
             AND (last_notification IS NULL 
                 OR ((EXTRACT(EPOCH FROM last_notification) + time_interval) < EXTRACT(EPOCH FROM :currentDate))
             )

subscription-service/src/main/kotlin/com/egm/stellio/subscription/web/APIResponses.kt

@@ -6,3 +6,8 @@ fun subscriptionNotFoundMessage(subscriptionId: URI) = "Could not find a subscri
 fun subscriptionAlreadyExistsMessage(subscriptionId: URI) = "A subscription with id $subscriptionId already exists"
 fun subscriptionUnauthorizedMessage(subscriptionId: URI) =
     "User is not authorized to access subscription $subscriptionId"
+fun unsupportedSubscriptiondAttributeMessage(subscriptionId: URI, attribute: Map.Entry<String, Any>) =
+    "Subscription $subscriptionId has unsupported attribute: ${attribute.key}"
+
+fun invalidSubscriptiondAttributeMessage(subscriptionId: URI, attribute: Map.Entry<String, Any>) =
+    "Subscription $subscriptionId has invalid attribute: ${attribute.key}"

subscription-service/src/main/resources/db/migration/V0_26__add_throttling_to_subscription.sql

@@ -0,0 +1,2 @@
+ALTER TABLE subscription
+    ADD throttling integer;

subscription-service/src/test/kotlin/com/egm/stellio/subscription/service/SubscriptionServiceTests.kt

@@ -149,6 +149,43 @@ class SubscriptionServiceTests : WithTimescaleContainer, WithKafkaContainer {
             }
     }
 
+    @Test
+    fun `it should not allow a subscription with timeInterval and throttling`() = runTest {
+        val payload = mapOf(
+            "id" to "urn:ngsi-ld:Beehive:1234567890".toUri(),
+            "type" to NGSILD_SUBSCRIPTION_TERM,
+            "timeInterval" to 10,
+            "entities" to listOf(mapOf("type" to BEEHIVE_TYPE)),
+            "notification" to mapOf("endpoint" to mapOf("uri" to "http://my.endpoint/notifiy")),
+            "throttling" to 30
+        )
+
+        val subscription = ParsingUtils.parseSubscription(payload, emptyList()).shouldSucceedAndResult()
+        subscriptionService.validateNewSubscription(subscription)
+            .shouldFailWith {
+                it is BadRequestDataException &&
+                    it.message == "You can't use 'timeInterval' in conjunction with 'throttling'"
+            }
+    }
+
+    @Test
+    fun `it should not allow a subscription with a negative throttling`() = runTest {
+        val payload = mapOf(
+            "id" to "urn:ngsi-ld:Beehive:1234567890".toUri(),
+            "type" to NGSILD_SUBSCRIPTION_TERM,
+            "entities" to listOf(mapOf("type" to BEEHIVE_TYPE)),
+            "notification" to mapOf("endpoint" to mapOf("uri" to "http://my.endpoint/notifiy")),
+            "throttling" to -30
+        )
+
+        val subscription = ParsingUtils.parseSubscription(payload, emptyList()).shouldSucceedAndResult()
+        subscriptionService.validateNewSubscription(subscription)
+            .shouldFailWith {
+                it is BadRequestDataException &&
+                    it.message == "The value of 'throttling' must be greater than zero (int)"
+            }
+    }
+
     @Test
     fun `it should not allow a subscription with a negative timeInterval`() = runTest {
         val payload = mapOf(
@@ -285,7 +322,8 @@ class SubscriptionServiceTests : WithTimescaleContainer, WithKafkaContainer {
                         Endpoint.AcceptType.JSON,
                         listOf(EndpointInfo("Authorization-token", "Authorization-token-value"))
                     ) &&
-                    it.expiresAt == ZonedDateTime.parse("2100-01-01T00:00:00Z")
+                    it.expiresAt == ZonedDateTime.parse("2100-01-01T00:00:00Z") &&
+                    it.throttling == 60
             }
     }
 
@@ -678,7 +716,8 @@ class SubscriptionServiceTests : WithTimescaleContainer, WithKafkaContainer {
                 "geometry" to "Point",
                 "coordinates" to "[100.0, 0.0]",
                 "geoproperty" to "https://uri.etsi.org/ngsi-ld/observationSpace"
-            )
+            ),
+            "throttling" to 50
         )
 
         subscriptionService.update(subscription.id, parsedInput, APIC_COMPOUND_CONTEXTS)
@@ -694,7 +733,8 @@ class SubscriptionServiceTests : WithTimescaleContainer, WithKafkaContainer {
                     it.geoQ!!.georel == "equals" &&
                     it.geoQ!!.geometry == "Point" &&
                     it.geoQ!!.coordinates == "[100.0, 0.0]" &&
-                    it.geoQ!!.geoproperty == "https://uri.etsi.org/ngsi-ld/observationSpace"
+                    it.geoQ!!.geoproperty == "https://uri.etsi.org/ngsi-ld/observationSpace" &&
+                    it.throttling == 50
             }
     }
 
@@ -887,12 +927,12 @@ class SubscriptionServiceTests : WithTimescaleContainer, WithKafkaContainer {
         val subscription = loadAndDeserializeSubscription("subscription_minimal_entities.json")
         subscriptionService.create(subscription, mockUserSub).shouldSucceed()
 
-        val parsedInput = mapOf("type" to NGSILD_SUBSCRIPTION_TERM, "throttling" to "someValue")
+        val parsedInput = mapOf("type" to NGSILD_SUBSCRIPTION_TERM, "csf" to "someValue")
 
         subscriptionService.update(subscription.id, parsedInput, APIC_COMPOUND_CONTEXTS)
             .shouldFailWith {
                 it is NotImplementedException &&
-                    it.message == "Subscription urn:ngsi-ld:Subscription:1 has unsupported attribute: throttling"
+                    it.message == "Subscription urn:ngsi-ld:Subscription:1 has unsupported attribute: csf"
             }
     }
 
@@ -1062,6 +1102,29 @@ class SubscriptionServiceTests : WithTimescaleContainer, WithKafkaContainer {
             .shouldSucceedWith { assertEquals(0, it.size) }
     }
 
+    @Test
+    fun `it should not return a subscription if throttling has not elapsed yet`() = runTest {
+        val expandedEntity = expandJsonLdEntity(entity, APIC_COMPOUND_CONTEXTS)
+
+        val payload = mapOf(
+            "id" to "urn:ngsi-ld:Beehive:1234567890".toUri(),
+            "type" to NGSILD_SUBSCRIPTION_TERM,
+            "watchedAttributes" to listOf(INCOMING_COMPACT_PROPERTY),
+            "notification" to mapOf(
+                "endpoint" to mapOf("uri" to "http://my.endpoint/notifiy"),
+                "lastNotification" to ngsiLdDateTime()
+            ),
+            "throttling" to 300
+        )
+
+        val subscription = ParsingUtils.parseSubscription(payload, emptyList()).shouldSucceedAndResult()
+
+        subscriptionService.create(subscription, mockUserSub).shouldSucceed()
+
+        subscriptionService.getMatchingSubscriptions(expandedEntity, setOf(NGSILD_LOCATION_PROPERTY), ATTRIBUTE_UPDATED)
+            .shouldSucceedWith { assertEquals(0, it.size) }
+    }
+
     @ParameterizedTest
     @CsvSource(
         "near;minDistance==1000, Polygon, '[[[100.0, 0.0], [101.0, 0.0], [101.0, -1.0], [100.0, 0.0]]]', 0",

subscription-service/src/test/kotlin/com/egm/stellio/subscription/support/WithTimescaleContainer.kt

@@ -25,6 +25,7 @@ interface WithTimescaleContainer {
             withEnv("POSTGRES_MULTIPLE_EXTENSIONS", "postgis,timescaledb,pgcrypto")
             withExposedPorts(5432)
             setWaitStrategy(Wait.forLogMessage(".*database system is ready to accept connections.*", 2))
+            withReuse(true)
         }
 
         @JvmStatic

subscription-service/src/test/resources/ngsild/subscription_full.json

@@ -35,5 +35,6 @@
       ]
     }
   },
-  "expiresAt": "2100-01-01T00:00:00Z"
+  "expiresAt": "2100-01-01T00:00:00Z",
+  "throttling": 60
 }