From 406a220f14728f4c1fa0b6adbddd36352fe5f1c0 Mon Sep 17 00:00:00 2001
From: "Riabokon Stanislav(EPAM)[GCP]" <stanislav_riabokon@epam.com>
Date: Wed, 20 Dec 2023 14:17:25 +0000
Subject: [PATCH] Merge branch 'gc-fix-cache' into 'master'
Fix GC caching, and enable baremetal int tests.
See merge request osdu/platform/system/notification!462
(cherry picked from commit 962d55c13dd9d57df3dd92d753945fb67c711d10)
b0108ff2 fix cache properties
8eea6c1b Merge branch 'master' into gc-fix-cache
d9d9b19c restructure cache handling, code cleanups
5b70a3c4 added handling for unavailable subscriptions
cb2c50cb unit and integration testing
f59dd0d4 add test resources
15febd49 update README
2b248bdd Updating NOTICE
a5dc7407 add sleep in test after sub created
f51b1846 remove ttl for subs cache
e88de9dc remove not necessary while loop
---
NOTICE | 9 +-
.../notification-gc/docs/baremetal/README.md | 6 +-
provider/notification-gc/docs/gc/README.md | 6 +-
provider/notification-gc/pom.xml | 10 ++
.../provider/gcp/config/CacheConfig.java | 68 +++++--
.../config/ExternalSubscriptionsManager.java | 118 ------------
.../provider/gcp/config/RedisProperties.java | 2 +
.../gcp/pubsub/MessageBrokerProvider.java | 3 +-
...va => OqmNotificationDeliveryService.java} | 45 +++--
.../gcp/pubsub/OqmSubscriberManager.java | 21 +--
.../OqmConfigurationEventReceiver.java | 88 +++++++++
.../receiver/OqmControlTopicReceiver.java | 103 -----------
...cReceiver.java => OqmEventReplicator.java} | 43 +++--
...r.java => OqmReplicatedEventReceiver.java} | 12 +-
.../gcp/repo/SubscriptionCacheRepo.java | 132 ++++++++++++++
.../service/ExternalSubscriptionsManager.java | 102 +++++++++++
.../RegisterSubscriptionService.java} | 15 +-
.../gcp/service/SubscriptionServiceGc.java | 2 +-
.../src/main/resources/application.properties | 12 +-
...=> OqmConfigurationEventReceiverTest.java} | 52 ++----
...rTest.java => OqmEventReplicatorTest.java} | 33 ++--
...va => OqmReplicatedEventReceiverTest.java} | 8 +-
.../gcp/pubsub/OqmSubscriberManagerTest.java | 12 +-
.../ExternalSubscriptionsManagerTest.java | 168 ++++++++++++++++++
.../notification/api/TestPushEndpointGsa.java | 149 ++++++++++++++++
.../api/TestPushEndpointHMAC.java | 135 ++++++++++++++
.../osdu/notification/util/Constants.java | 6 +
.../osdu/notification/util/FileUtils.java | 31 ++--
.../osdu/notification/util/HttpClient.java | 108 +++++++++++
.../osdu/notification/util/ServicesUtils.java | 96 ++++++++++
.../src/test/resources/LegalTag.json | 16 ++
.../src/test/resources/StorageRecord.json | 33 ++++
.../notification/api/TestPushEndpointGsa.java | 1 +
.../api/TestPushEndpointHMAC.java | 1 +
34 files changed, 1264 insertions(+), 382 deletions(-)
delete mode 100644 provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/ExternalSubscriptionsManager.java
rename provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/{OqmNotificationHandler.java => OqmNotificationDeliveryService.java} (70%)
create mode 100644 provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmConfigurationEventReceiver.java
delete mode 100644 provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmControlTopicReceiver.java
rename provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/{OqmServiceTopicReceiver.java => OqmEventReplicator.java} (78%)
rename provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/{OqmPublishTopicReceiver.java => OqmReplicatedEventReceiver.java} (83%)
create mode 100644 provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/repo/SubscriptionCacheRepo.java
create mode 100644 provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManager.java
rename provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/{pubsub/OqmSubscriptionHandler.java => service/RegisterSubscriptionService.java} (87%)
rename provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/{OqmControlTopicReceiverTest.java => OqmConfigurationEventReceiverTest.java} (53%)
rename provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/{OqmServiceTopicReceiverTest.java => OqmEventReplicatorTest.java} (72%)
rename provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/{OqmPublishTopicReceiverTest.java => OqmReplicatedEventReceiverTest.java} (94%)
create mode 100644 provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManagerTest.java
create mode 100644 testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java
create mode 100644 testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java
create mode 100644 testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/Constants.java
rename provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/model/ExternalSubscriptions.java => testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/FileUtils.java (50%)
create mode 100644 testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/HttpClient.java
create mode 100644 testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/ServicesUtils.java
create mode 100644 testing/notification-test-baremetal/src/test/resources/LegalTag.json
create mode 100644 testing/notification-test-baremetal/src/test/resources/StorageRecord.json
diff --git a/NOTICE b/NOTICE
index c98d48db0..350e57837 100644
--- a/NOTICE
+++ b/NOTICE
@@ -155,6 +155,7 @@ The following software have components provided under the terms of this license:
- Kotlin Stdlib Common (from https://kotlinlang.org/)
- Kotlin Stdlib Jdk7 (from <https://kotlinlang.org/>, https://kotlinlang.org/)
- Kotlin Stdlib Jdk8 (from <https://kotlinlang.org/>, https://kotlinlang.org/)
+- Lettuce (from http://github.com/lettuce-io/lettuce-core)
- Lucene Core (from https://repo1.maven.org/maven2/org/apache/lucene/lucene-core)
- Metrics Core (from https://repo1.maven.org/maven2/io/dropwizard/metrics/metrics-core)
- Microsoft Application Insights Java Agent (from https://github.com/Microsoft/ApplicationInsights-Java)
@@ -236,12 +237,16 @@ The following software have components provided under the terms of this license:
- Spring Boot WebFlux Starter (from https://projects.spring.io/spring-boot/#/spring-boot-parent/spring-boot-starters/spring-boot-starter-webflux, https://spring.io/projects/spring-boot)
- Spring Commons Logging Bridge (from https://github.com/spring-projects/spring-framework)
- Spring Context (from http://www.springframework.org, https://github.com/spring-projects/spring-framework, https://repo1.maven.org/maven2/org/springframework/spring-context)
+- Spring Context Support (from https://github.com/spring-projects/spring-framework)
- Spring Core (from http://www.springframework.org, https://github.com/spring-projects/spring-framework, https://repo1.maven.org/maven2/org/springframework/spring-core)
- Spring Data Core (from https://spring.io/projects/spring-data)
+- Spring Data KeyValue (from https://repo1.maven.org/maven2/org/springframework/data/spring-data-keyvalue)
- Spring Data MongoDB - Core (from https://repo1.maven.org/maven2/org/springframework/data/spring-data-mongodb)
+- Spring Data Redis (from http://github.com/spring-projects/spring-data-redis, https://repo1.maven.org/maven2/org/springframework/data/spring-data-redis, https://spring.io/projects/spring-data-redis)
- Spring Expression Language (SpEL) (from https://github.com/SpringSource/spring-framework, https://github.com/spring-projects/spring-framework, https://repo1.maven.org/maven2/org/springframework/spring-expression)
- Spring JMS (from http://www.springframework.org, https://github.com/SpringSource/spring-framework, https://github.com/spring-projects/spring-framework, https://repo1.maven.org/maven2/org/springframework/spring-jms)
- Spring Messaging (from https://github.com/spring-projects/spring-framework)
+- Spring Object/XML Marshalling (from https://github.com/spring-projects/spring-framework)
- Spring Plugin - Metadata Extension (from https://repo1.maven.org/maven2/org/springframework/plugin/spring-plugin-metadata)
- Spring Plugin Core (from https://github.com/spring-projects/spring-plugin/spring-plugin-core, https://repo1.maven.org/maven2/org/springframework/plugin/spring-plugin-core)
- Spring Security - Core (from http://spring.io/spring-security, https://repo1.maven.org/maven2/org/springframework/security/spring-security-core, https://spring.io/projects/spring-security, https://spring.io/spring-security)
@@ -364,6 +369,7 @@ The following software have components provided under the terms of this license:
- RE2/J (from http://github.com/google/re2j)
- Redisson (from http://redisson.org)
- Spring Core (from http://www.springframework.org, https://github.com/spring-projects/spring-framework, https://repo1.maven.org/maven2/org/springframework/spring-core)
+- Spring Data KeyValue (from https://repo1.maven.org/maven2/org/springframework/data/spring-data-keyvalue)
- ThreeTen backport (from https://github.com/ThreeTen/threetenbp, https://www.threeten.org/threetenbp)
========================================================================
@@ -628,7 +634,7 @@ The following software have components provided under the terms of this license:
- QpidJMS Client (from https://repo1.maven.org/maven2/org/apache/qpid/qpid-jms-client)
- SLF4J API Module (from http://www.slf4j.org)
- Spongy Castle (from http://rtyley.github.io/spongycastle/)
-- Spring Data for Azure Cosmos DB SQL API (from https://github.com/Azure/azure-sdk-for-java/tree/master/sdk/cosmos/azure-spring-data-cosmos)
+- Spring Data for Azure Cosmos DB SQL API (from https://github.com/Azure/azure-sdk-for-java/tree/master/sdk/cosmos/azure-spring-data-cosmos, https://github.com/Azure/azure-sdk-for-java/tree/master/sdk/spring/azure-spring-data-cosmos)
- ThreeTen backport (from https://github.com/ThreeTen/threetenbp, https://www.threeten.org/threetenbp)
- adal4j (from https://github.com/AzureAD/azure-activedirectory-library-for-java)
- micrometer-core (from https://github.com/micrometer-metrics/micrometer)
@@ -693,7 +699,6 @@ The following software have components provided under the terms of this license:
- Undertow Core (from <https://repo1.maven.org/maven2/io/undertow/undertow-core>, https://repo1.maven.org/maven2/io/undertow/undertow-core)
- Undertow WebSockets JSR356 implementations (from <https://repo1.maven.org/maven2/io/undertow/undertow-websockets-jsr>, https://repo1.maven.org/maven2/io/undertow/undertow-websockets-jsr)
- XNIO API (from <http://www.jboss.org/xnio>, http://www.jboss.org/xnio)
-- XNIO NIO Implementation (from <https://repo1.maven.org/maven2/org/jboss/xnio/xnio-nio>, https://repo1.maven.org/maven2/org/jboss/xnio/xnio-nio)
========================================================================
unknown
diff --git a/provider/notification-gc/docs/baremetal/README.md b/provider/notification-gc/docs/baremetal/README.md
index 0d168b61e..809766233 100644
--- a/provider/notification-gc/docs/baremetal/README.md
+++ b/provider/notification-gc/docs/baremetal/README.md
@@ -82,9 +82,9 @@ After the service has started it should be accessible via a web browser by visit
**Entitlements configuration for integration accounts**
-| DE_OPS_TESTER | DE_ADMIN_TESTER | DE_EDITOR_TESTER | DE_NO_ACCESS_TESTER |
-| --- | --- | --- | --- |
-|notification.pubsub<br/>service.entitlements.user<br/>users<br/>users.datalake.ops</br>| service.entitlements.user<br/>users<br/>users.datalake.admins</br> | service.entitlements.user<br/>users<br/>users.datalake.editors</br> | service.entitlements.user<br/>users<br/>|
+| DE_OPS_TESTER | DE_ADMIN_TESTER | DE_EDITOR_TESTER | DE_NO_ACCESS_TESTER |
+|-----------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------|---------------------------------------------------------------------|------------------------------------------|
+| notification.pubsub<br/>service.entitlements.user<br/>users<br/>users.datalake.ops</br> | service.entitlements.user<br/>users<br/>users.datalake.admins</br>service.legal.admin | service.entitlements.user<br/>users<br/>users.datalake.editors</br> | service.entitlements.user<br/>users<br/> |
Above variables should be configured in the release pipeline to run integration tests. You should also replace them with proper values if you wish to run tests locally.
diff --git a/provider/notification-gc/docs/gc/README.md b/provider/notification-gc/docs/gc/README.md
index b2c8bb3ff..eaa1e5ad4 100644
--- a/provider/notification-gc/docs/gc/README.md
+++ b/provider/notification-gc/docs/gc/README.md
@@ -84,9 +84,9 @@ After the service has started it should be accessible via a web browser by visit
**Entitlements configuration for integration accounts**
-| DE_OPS_TESTER | DE_ADMIN_TESTER | DE_EDITOR_TESTER | DE_NO_ACCESS_TESTER |
-| --- | --- | --- | --- |
-|notification.pubsub<br/>service.entitlements.user<br/>users<br/>users.datalake.ops</br>| service.entitlements.user<br/>users<br/>users.datalake.admins</br> | service.entitlements.user<br/>users<br/>users.datalake.editors</br> | service.entitlements.user<br/>users<br/>|
+| DE_OPS_TESTER | DE_ADMIN_TESTER | DE_EDITOR_TESTER | DE_NO_ACCESS_TESTER |
+|-----------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------|---------------------------------------------------------------------|------------------------------------------|
+| notification.pubsub<br/>service.entitlements.user<br/>users<br/>users.datalake.ops</br> | service.entitlements.user<br/>users<br/>users.datalake.admins</br>service.legal.admin | service.entitlements.user<br/>users<br/>users.datalake.editors</br> | service.entitlements.user<br/>users<br/> |
Above variables should be configured in the release pipeline to run integration tests. You should also replace them with proper values if you wish to run tests locally.
diff --git a/provider/notification-gc/pom.xml b/provider/notification-gc/pom.xml
index 53b5e20ff..cb43f8e29 100644
--- a/provider/notification-gc/pom.xml
+++ b/provider/notification-gc/pom.xml
@@ -48,6 +48,16 @@
</dependencyManagement>
<dependencies>
+ <dependency>
+ <groupId>org.springframework.data</groupId>
+ <artifactId>spring-data-redis</artifactId>
+ <version>2.7.18</version>
+ </dependency>
+ <dependency>
+ <groupId>io.lettuce</groupId>
+ <artifactId>lettuce-core</artifactId>
+ <version>6.3.0.RELEASE</version>
+ </dependency>
<dependency>
<groupId>org.opengroup.osdu</groupId>
<artifactId>core-lib-gc</artifactId>
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/CacheConfig.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/CacheConfig.java
index 24673cf69..a5d0155a2 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/CacheConfig.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/CacheConfig.java
@@ -17,34 +17,70 @@
package org.opengroup.osdu.notification.provider.gcp.config;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Objects;
import lombok.RequiredArgsConstructor;
import org.opengroup.osdu.core.common.cache.ICache;
-import org.opengroup.osdu.core.common.cache.IRedisCache;
import org.opengroup.osdu.core.common.cache.VmCache;
+import org.opengroup.osdu.core.common.model.notification.Subscription;
import org.opengroup.osdu.core.common.partition.PartitionInfo;
-import org.opengroup.osdu.core.gcp.cache.RedisCacheBuilder;
-import org.opengroup.osdu.notification.provider.gcp.model.ExternalSubscriptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConfiguration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
@Configuration
@RequiredArgsConstructor
public class CacheConfig {
- private final RedisCacheBuilder<String, ExternalSubscriptions> builder;
-
- @Bean
- public IRedisCache<String, ExternalSubscriptions> subscriptionCache(RedisProperties properties) {
- return builder.buildRedisCache(
- properties.getRedisHost(),
- properties.getRedisPort(),
- properties.getRedisPassword(),
- properties.getRedisExpiration(),
- properties.getRedisWithSsl(),
- String.class,
- ExternalSubscriptions.class
- );
+ @Bean
+ public RedisTemplate<String, Subscription> setsTemplate(RedisConnectionFactory redisConnectionFactory) {
+ RedisTemplate<String, Subscription> template = new RedisTemplate<>();
+ Jackson2JsonRedisSerializer<Subscription> jsonRedisSerializer = new Jackson2JsonRedisSerializer<>(
+ Subscription.class);
+ ObjectMapper mapper = new ObjectMapper();
+ jsonRedisSerializer.setObjectMapper(mapper);
+ template.setValueSerializer(jsonRedisSerializer);
+ template.setKeySerializer(new StringRedisSerializer());
+ template.setHashKeySerializer(new StringRedisSerializer());
+ template.setConnectionFactory(redisConnectionFactory);
+ return template;
+ }
+
+ @Bean
+ public RedisConnectionFactory redisConnectionFactory(RedisConfiguration redisConfiguration,
+ RedisProperties redisProperties) {
+ LettuceClientConfigurationBuilder configurationBuilder = LettuceClientConfiguration.builder();
+
+ if (Boolean.TRUE.equals(redisProperties.getRedisWithSsl())) {
+ configurationBuilder.useSsl();
}
+ configurationBuilder.commandTimeout(redisProperties.getTimeOut());
+ LettuceClientConfiguration lettuceClientConfiguration = configurationBuilder.build();
+
+ return new LettuceConnectionFactory(
+ redisConfiguration,
+ lettuceClientConfiguration
+ );
+ }
+
+ @Bean
+ public RedisConfiguration redisConfiguration(RedisProperties redisProperties) {
+ RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
+ redisStandaloneConfiguration.setHostName(redisProperties.getRedisHost());
+ redisStandaloneConfiguration.setPort(redisProperties.getRedisPort());
+ if (Objects.nonNull(redisProperties.getRedisPassword()) && !redisProperties.getRedisPassword().isEmpty()) {
+ redisStandaloneConfiguration.setPassword(redisProperties.getRedisPassword());
+ }
+ return redisStandaloneConfiguration;
+ }
@Bean
public ICache<String, PartitionInfo> partitionInfoCache() {
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/ExternalSubscriptionsManager.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/ExternalSubscriptionsManager.java
deleted file mode 100644
index 7889a4333..000000000
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/ExternalSubscriptionsManager.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Copyright 2020-2023 Google LLC
- * Copyright 2020-2023 EPAM Systems, Inc
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.opengroup.osdu.notification.provider.gcp.config;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.stream.Collectors;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.http.HttpStatus;
-import org.opengroup.osdu.core.common.cache.IRedisCache;
-import org.opengroup.osdu.core.common.model.http.AppException;
-import org.opengroup.osdu.core.common.model.notification.Subscription;
-import org.opengroup.osdu.notification.provider.gcp.model.ExternalSubscriptions;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.OqmSubscriptionHandler;
-import org.springframework.stereotype.Component;
-
-@Component
-@RequiredArgsConstructor
-@Slf4j
-public class ExternalSubscriptionsManager {
-
- private final IRedisCache<String, ExternalSubscriptions> subscriptionInfoCache;
- private final OqmSubscriptionHandler subscriptionHandler;
-
- public ExternalSubscriptions getExternalSubscriptions(String dataPartitionId) {
- if (subscriptionInfoCache.get(dataPartitionId) == null) {
- log.debug("Subscription info cache wasn't found for tenant {}", dataPartitionId);
- reloadSubscriptionInfoCache(dataPartitionId);
- }
- return subscriptionInfoCache.get(dataPartitionId);
- }
-
- public Subscription getSubscription(String dataPartitionId, String subscriptionId, String serviceTopic) {
- List<Subscription> cachedInfos = Optional.ofNullable(getExternalSubscriptions(dataPartitionId))
- .orElseThrow(() -> new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Server error", "OQM | Subscription info cache was not initialized"))
- .getSubscriptions();
-
- Subscription subscription = getFilteredSubscription(cachedInfos, dataPartitionId,
- subscriptionId, serviceTopic);
-
- if (Objects.nonNull(subscription) && Objects.isNull(subscription.getSecret())) {
- return sendGetSubscriptionRequest(dataPartitionId, subscriptionId, serviceTopic, cachedInfos);
- }
- return subscription;
- }
-
- private Subscription getFilteredSubscription(List<Subscription> subscriptions, String dataPartitionId,
- String subscriptionId, String serviceTopic) {
- List<Subscription> filteredInfos = filterSubscriptionInfosByTopic(subscriptions, subscriptionId, serviceTopic);
- if (filteredInfos.isEmpty()) {
- return sendGetSubscriptionRequest(dataPartitionId, subscriptionId, serviceTopic, subscriptions);
- } else {
- log.debug("Register client cache | `{}` subscriptions info found. The first was taken.", filteredInfos.size());
- return filteredInfos.get(0);
- }
- }
-
- public void updateExternalSubscriptionsCache(String dataPartitionId, ExternalSubscriptions externalSubscriptions) {
- subscriptionInfoCache.put(dataPartitionId, externalSubscriptions);
- log.debug("Subscription info cache was updated for tenant {} with {} values", dataPartitionId, externalSubscriptions.getSubscriptions().size());
- }
-
- /**
- * Get all subscription infos from Register service and enrich each of them with secret by additional request
- *
- * @param dataPartitionId partition id
- */
- private void reloadSubscriptionInfoCache(String dataPartitionId) {
- List<Subscription> subscriptionInfos = subscriptionHandler.getAllSubscriptionInfos(dataPartitionId);
- List<Subscription> enrichedSubscriptionInfos = subscriptionInfos.stream()
- .map(subscription -> getFilteredSubscription(subscriptionInfos, dataPartitionId, subscription.getNotificationId(), subscription.getTopic()))
- .collect(Collectors.toList());
- subscriptionInfoCache.put(dataPartitionId, ExternalSubscriptions.builder().subscriptions(enrichedSubscriptionInfos).build());
- log.debug("Subscription info cache PRELOADED for tenant: {}. Size is: {}.", dataPartitionId, enrichedSubscriptionInfos.size());
- }
-
- private Subscription sendGetSubscriptionRequest(String dataPartitionId, String subscriptionId, String serviceTopic, List<Subscription> cachedInfos) {
- List<Subscription> freshInfos = subscriptionHandler.getSubscriptionsById(dataPartitionId, subscriptionId);
- if (freshInfos.isEmpty()) {
- log.warn("Subscription info with sub ID: `{}` not found", subscriptionId);
- return null;
- }
-
- List<Subscription> filteredFreshInfos = filterSubscriptionInfosByTopic(freshInfos, subscriptionId, serviceTopic);
- if (filteredFreshInfos.isEmpty()) {
- log.warn("Subscription info with sub ID: `{}` not found", subscriptionId);
- return null;
- }
- cachedInfos.addAll(filteredFreshInfos);
- updateExternalSubscriptionsCache(dataPartitionId, ExternalSubscriptions.builder().subscriptions(cachedInfos).build());
-
- log.debug("Register client | `{}` subscriptions info found. The first was taken.", filteredFreshInfos.size());
- return filteredFreshInfos.get(0);
- }
-
- private static List<Subscription> filterSubscriptionInfosByTopic(List<Subscription> infos, String subscriptionId, String serviceTopic) {
- return infos.stream()
- .filter(info -> serviceTopic.equals(info.getTopic()) && subscriptionId.equals(info.getNotificationId()))
- .collect(Collectors.toList());
- }
-}
\ No newline at end of file
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/RedisProperties.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/RedisProperties.java
index fb7ae8978..6ebea2825 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/RedisProperties.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/config/RedisProperties.java
@@ -1,6 +1,7 @@
package org.opengroup.osdu.notification.provider.gcp.config;
+import java.time.Duration;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -16,4 +17,5 @@ public class RedisProperties {
private String redisPassword;
private Integer redisExpiration = 300;
private Boolean redisWithSsl = false;
+ private Duration timeOut = Duration.ofSeconds(30L);
}
\ No newline at end of file
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/MessageBrokerProvider.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/MessageBrokerProvider.java
index ddaedeefe..cdd935b2a 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/MessageBrokerProvider.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/MessageBrokerProvider.java
@@ -8,6 +8,7 @@ import org.opengroup.osdu.core.gcp.oqm.driver.OqmDriver;
import org.opengroup.osdu.core.gcp.oqm.model.OqmDestination;
import org.opengroup.osdu.core.gcp.oqm.model.OqmSubscription;
import org.opengroup.osdu.core.gcp.oqm.model.OqmTopic;
+import org.opengroup.osdu.notification.provider.gcp.service.RegisterSubscriptionService;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
@@ -24,7 +25,7 @@ import static org.opengroup.osdu.notification.provider.gcp.pubsub.OqmSubscriberM
public class MessageBrokerProvider {
private final OqmDriver driver;
- private final OqmSubscriptionHandler subscriptionHandler;
+ private final RegisterSubscriptionService subscriptionHandler;
public void validateMessageBrokerConfiguration(TenantInfo tenantInfo) {
List<OqmTopic> topics = subscriptionHandler.getTopics(tenantInfo.getDataPartitionId());
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmNotificationHandler.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmNotificationDeliveryService.java
similarity index 70%
rename from provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmNotificationHandler.java
rename to provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmNotificationDeliveryService.java
index a4ecd5296..ad4d33a25 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmNotificationHandler.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmNotificationDeliveryService.java
@@ -16,6 +16,7 @@
package org.opengroup.osdu.notification.provider.gcp.pubsub;
+import java.util.Map;
import java.util.Objects;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -23,24 +24,21 @@ import org.apache.http.HttpStatus;
import org.opengroup.osdu.core.common.http.HttpClient;
import org.opengroup.osdu.core.common.http.HttpRequest;
import org.opengroup.osdu.core.common.http.HttpResponse;
-import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.notification.Secret;
import org.opengroup.osdu.core.common.model.notification.Subscription;
import org.opengroup.osdu.notification.auth.factory.AuthFactory;
import org.opengroup.osdu.notification.auth.interfaces.SecretAuth;
-import org.opengroup.osdu.notification.provider.gcp.config.ExternalSubscriptionsManager;
+import org.opengroup.osdu.notification.provider.gcp.service.ExternalSubscriptionsManager;
import org.opengroup.osdu.notification.provider.gcp.config.OqmConfigurationProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
-import java.util.Map;
-
@Component
@ConditionalOnProperty(name = "oqmDriver")
@RequiredArgsConstructor
@Slf4j
-public class OqmNotificationHandler {
+public class OqmNotificationDeliveryService {
private final OqmConfigurationProperties oqmConfigurationProperties;
private final HttpClient httpClient;
@@ -53,17 +51,27 @@ public class OqmNotificationHandler {
String correlationId = headerAttributes.get(DpsHeaders.CORRELATION_ID);
if (serviceTopic == null || dataPartitionId == null || correlationId == null) {
- log.warn("Service topic: {}, dataPartitionId: {}, correlationId: {}.", serviceTopic, dataPartitionId, correlationId);
- throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Server error", "Missed header attributes");
+ log.warn("Missing vital properties, event cannot be delivered. Service topic: {}, dataPartitionId: {}, correlationId: {}.", serviceTopic, dataPartitionId, correlationId);
+ return getStubResponse();
}
- Subscription subscription = externalSubscriptionsManager.getSubscription(dataPartitionId, subscriptionId, serviceTopic);
+ Subscription subscription = externalSubscriptionsManager.getSubscription(
+ dataPartitionId,
+ serviceTopic,
+ subscriptionId
+ );
+
if (Objects.isNull(subscription)) {
- throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR,
- "Subscriber config not found.",
- String.format("Subscriber with id: %s not found in cache", subscriptionId)
- );
+ log.warn("Warning! The event was replicated previously but there is no configuration for subId: {}, skipping the event.", subscriptionId);
+ return getStubResponse();
+ }
+
+ if(!isValidSub(subscription)){
+ externalSubscriptionsManager.removeSubscription(dataPartitionId, serviceTopic, subscriptionId);
+ log.warn("Warning! Subscriber configuration is not valid: {}, skipping the event. Evicting sub from cache.", subscriptionId);
+ return getStubResponse();
}
+
Secret secret = subscription.getSecret();
SecretAuth secretAuth = authFactory.getSecretAuth(secret.getSecretType());
@@ -85,4 +93,17 @@ public class OqmNotificationHandler {
log.debug("Notification handler | Sending notification to Sub ID: `{}` Endpoint: `{}`.", subscriptionId, subscription.getPushEndpoint());
return response;
}
+
+ private boolean isValidSub(Subscription subscription) {
+ boolean endpointIsValid = Objects.nonNull(subscription.getPushEndpoint()) && !subscription.getPushEndpoint()
+ .isEmpty();
+ boolean secretIsValid = Objects.nonNull(subscription.getSecret());
+ return endpointIsValid && secretIsValid;
+ }
+
+ private static HttpResponse getStubResponse() {
+ HttpResponse response = new HttpResponse();
+ response.setResponseCode(HttpStatus.SC_OK);
+ return response;
+ }
}
\ No newline at end of file
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManager.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManager.java
index f7908a742..2c51db098 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManager.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManager.java
@@ -23,10 +23,11 @@ import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
import org.opengroup.osdu.core.common.provider.interfaces.ITenantFactory;
import org.opengroup.osdu.core.gcp.oqm.driver.OqmDriver;
import org.opengroup.osdu.core.gcp.oqm.model.*;
-import org.opengroup.osdu.notification.provider.gcp.config.ExternalSubscriptionsManager;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmControlTopicReceiver;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmPublishTopicReceiver;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmServiceTopicReceiver;
+import org.opengroup.osdu.notification.provider.gcp.service.ExternalSubscriptionsManager;
+import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmConfigurationEventReceiver;
+import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmReplicatedEventReceiver;
+import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmEventReplicator;
+import org.opengroup.osdu.notification.provider.gcp.service.RegisterSubscriptionService;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
@@ -51,10 +52,10 @@ public class OqmSubscriberManager {
public static final String SERVICE_SUFFIX = "-service";
public static final String PUBLISH_SUFFIX = "-publish";
public static final String CONTROL_SUBSCRIPTION = NOTIFICATION_PREFIX + "control-sub";
- private final OqmSubscriptionHandler subscriptionHandler;
+ private final RegisterSubscriptionService subscriptionHandler;
private final ITenantFactory tenantInfoFactory;
private final OqmDriver driver;
- private final OqmNotificationHandler notificationHandler;
+ private final OqmNotificationDeliveryService notificationHandler;
private final MessageBrokerProvider messageBrokerProvider;
private final ExternalSubscriptionsManager externalSubscriptionsManager;
@@ -73,7 +74,7 @@ public class OqmSubscriberManager {
*/
void provisionSubscriptionInfoCache() {
for (TenantInfo tenantInfo : tenantInfoFactory.listTenantInfo()) {
- externalSubscriptionsManager.getExternalSubscriptions(tenantInfo.getDataPartitionId());
+ externalSubscriptionsManager.invokeTenantSubscribers(tenantInfo.getDataPartitionId());
}
}
@@ -104,7 +105,7 @@ public class OqmSubscriberManager {
private void registerControlTopicSubscriber(TenantInfo tenantInfo, OqmSubscription controlTopicSubscription) {
OqmSubscriber subscriber = OqmSubscriber.builder()
.subscription(controlTopicSubscription)
- .messageReceiver(new OqmControlTopicReceiver(externalSubscriptionsManager))
+ .messageReceiver(new OqmConfigurationEventReceiver(externalSubscriptionsManager))
.build();
driver.subscribe(subscriber, getDestination(tenantInfo));
}
@@ -119,7 +120,7 @@ public class OqmSubscriberManager {
void registerServiceTopicSubscriber(TenantInfo tenantInfo, OqmSubscription serviceSubscription) {
OqmSubscriber subscriber = OqmSubscriber.builder()
.subscription(serviceSubscription)
- .messageReceiver(new OqmServiceTopicReceiver(serviceSubscription, driver, externalSubscriptionsManager))
+ .messageReceiver(new OqmEventReplicator(serviceSubscription, driver, externalSubscriptionsManager))
.build();
driver.subscribe(subscriber, getDestination(tenantInfo));
}
@@ -135,7 +136,7 @@ public class OqmSubscriberManager {
private void registerPublishTopicSubscriber(TenantInfo tenantInfo, OqmSubscription subscription) {
OqmSubscriber subscriber = OqmSubscriber.builder()
.subscription(subscription)
- .messageReceiver(new OqmPublishTopicReceiver(subscription, notificationHandler))
+ .messageReceiver(new OqmReplicatedEventReceiver(subscription, notificationHandler))
.build();
driver.subscribe(subscriber, getDestination(tenantInfo));
}
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmConfigurationEventReceiver.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmConfigurationEventReceiver.java
new file mode 100644
index 000000000..aee8d2e7a
--- /dev/null
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmConfigurationEventReceiver.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.provider.gcp.pubsub.receiver;
+
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmAckReplier;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmMessage;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmMessageReceiver;
+import org.opengroup.osdu.notification.provider.gcp.service.ExternalSubscriptionsManager;
+import org.opengroup.osdu.notification.provider.gcp.thread.ThreadScopeContextHolder;
+
+@Slf4j
+public class OqmConfigurationEventReceiver implements OqmMessageReceiver {
+
+ private static final String SUBSCRIPTION_CREATED = "Subscription Created";
+ private static final String SUBSCRIPTION_UPDATED = "Subscription Updated";
+ private static final String SUBSCRIPTION_DELETED = "Subscription Deleted";
+
+ private final ExternalSubscriptionsManager externalSubscriptionsManager;
+
+ public OqmConfigurationEventReceiver(ExternalSubscriptionsManager externalSubscriptionsManager) {
+ this.externalSubscriptionsManager = externalSubscriptionsManager;
+ }
+
+ @Override
+ public void receiveMessage(OqmMessage message, OqmAckReplier replier) {
+ try {
+ handleMessage(message);
+ } catch (Exception e) {
+ log.error("OQM | Control topic | Message not acknowledged by client", e);
+ replier.nack();
+ } finally {
+ ThreadScopeContextHolder.currentThreadScopeAttributes().clear();
+ }
+ log.debug("OQM | Control topic | Message acknowledged by client");
+ replier.ack();
+ }
+
+ private void handleMessage(OqmMessage oqmMessage) {
+ String pubsubMessage = oqmMessage.getData();
+ Map<String, String> headerAttributes = oqmMessage.getAttributes();
+ String subscriptionId = headerAttributes.get("subscription-id");
+ String serviceTopic = headerAttributes.get("topic");
+ String dataPartitionId = headerAttributes.get("data-partition-id");
+ log.debug("OQM | Control topic | Received message: `{}` for service topic: `{}` with Sub ID: `{}`", pubsubMessage, serviceTopic, subscriptionId);
+
+ switch (pubsubMessage) {
+ case SUBSCRIPTION_CREATED, SUBSCRIPTION_UPDATED -> {
+ externalSubscriptionsManager.updateSubscription(dataPartitionId, subscriptionId);
+ log.debug(
+ "OQM | Control topic | Subscription info with Sub ID: `{}` and topic: `{}` was added to cache.",
+ subscriptionId, serviceTopic);
+ }
+ case SUBSCRIPTION_DELETED -> {
+ boolean deleted = externalSubscriptionsManager.removeSubscription(dataPartitionId,
+ serviceTopic, subscriptionId);
+ if (deleted) {
+ log.debug(
+ "OQM | Control topic | Subscription info with Sub ID: `{}` and topic: `{}` was removed from cache.",
+ subscriptionId, serviceTopic);
+ } else {
+ log.warn(
+ "OQM | Control topic requested to delete subscription with id: `{}`, but it was not found in cache.",
+ subscriptionId);
+ }
+ }
+ default -> log.warn(
+ "Not supported operation! OQM | Control topic | Received unsupported control topic operation: {}",
+ pubsubMessage);
+ }
+ }
+}
\ No newline at end of file
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmControlTopicReceiver.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmControlTopicReceiver.java
deleted file mode 100644
index 45891a9ae..000000000
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmControlTopicReceiver.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * Copyright 2020-2023 Google LLC
- * Copyright 2020-2023 EPAM Systems, Inc
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.opengroup.osdu.notification.provider.gcp.pubsub.receiver;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.http.HttpStatus;
-import org.opengroup.osdu.core.common.model.http.AppException;
-import org.opengroup.osdu.core.common.model.notification.Subscription;
-import org.opengroup.osdu.core.gcp.oqm.model.OqmAckReplier;
-import org.opengroup.osdu.core.gcp.oqm.model.OqmMessage;
-import org.opengroup.osdu.core.gcp.oqm.model.OqmMessageReceiver;
-import org.opengroup.osdu.notification.provider.gcp.config.ExternalSubscriptionsManager;
-import org.opengroup.osdu.notification.provider.gcp.model.ExternalSubscriptions;
-import org.opengroup.osdu.notification.provider.gcp.thread.ThreadScopeContextHolder;
-
-@Slf4j
-public class OqmControlTopicReceiver implements OqmMessageReceiver {
-
- private static final String SUBSCRIPTION_CREATED = "Subscription Created";
- private static final String SUBSCRIPTION_UPDATED = "Subscription Updated";
- private static final String SUBSCRIPTION_DELETED = "Subscription Deleted";
-
- private final ExternalSubscriptionsManager externalSubscriptionsManager;
-
- public OqmControlTopicReceiver(ExternalSubscriptionsManager externalSubscriptionsManager) {
- this.externalSubscriptionsManager = externalSubscriptionsManager;
- }
-
- @Override
- public void receiveMessage(OqmMessage message, OqmAckReplier replier) {
- try {
- handleMessage(message);
- } catch (Exception e) {
- log.error("OQM | Control topic | Message not acknowledged by client", e);
- replier.nack();
- } finally {
- ThreadScopeContextHolder.currentThreadScopeAttributes().clear();
- }
-
- log.debug("OQM | Control topic | Message acknowledged by client");
- replier.ack();
- }
-
- private void handleMessage(OqmMessage oqmMessage) {
- String pubsubMessage = oqmMessage.getData();
- Map<String, String> headerAttributes = oqmMessage.getAttributes();
- String subscriptionId = headerAttributes.get("subscription-id");
- String serviceTopic = headerAttributes.get("topic");
- String dataPartitionId = headerAttributes.get("data-partition-id");
- log.debug("OQM | Control topic | Received message: `{}` for service topic: `{}` with Sub ID: `{}`",
- pubsubMessage, serviceTopic, subscriptionId);
-
- List<Subscription> subscriptionInfos = Optional.ofNullable(
- externalSubscriptionsManager.getExternalSubscriptions(dataPartitionId))
- .orElseThrow(() -> new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Server error", "OQM | Subscription info cache was not initialized"))
- .getSubscriptions();
- Optional<Subscription> cachedInfo = subscriptionInfos.stream()
- .filter(info -> subscriptionId.equals(info.getNotificationId()) && serviceTopic.equals(info.getTopic()))
- .findFirst();
-
- switch (pubsubMessage) {
- case SUBSCRIPTION_CREATED:
- case SUBSCRIPTION_UPDATED:
- cachedInfo.ifPresent(subscriptionInfos::remove);
- Subscription freshSubscriptionInfo = externalSubscriptionsManager.getSubscription(dataPartitionId, subscriptionId, serviceTopic);
- if (freshSubscriptionInfo != null) {
- subscriptionInfos.add(freshSubscriptionInfo);
- externalSubscriptionsManager.updateExternalSubscriptionsCache(dataPartitionId, ExternalSubscriptions.builder().subscriptions(subscriptionInfos).build());
- log.debug("OQM | Control topic | Subscription info with Sub ID: `{}` and topic: `{}` was added to cache.", subscriptionId, serviceTopic);
- }
- break;
- case SUBSCRIPTION_DELETED:
- if (cachedInfo.isPresent()) {
- subscriptionInfos.remove(cachedInfo.get());
- externalSubscriptionsManager.updateExternalSubscriptionsCache(dataPartitionId, ExternalSubscriptions.builder().subscriptions(subscriptionInfos).build());
- log.debug("OQM | Control topic | Subscription info with Sub ID: `{}` and topic: `{}` was removed from cache.", subscriptionId, serviceTopic);
- } else {
- log.warn("OQM | Control topic requested to delete subscription with id: `{}`, but is was not found in cache.", subscriptionId);
- }
- break;
- default:
- throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Not supported", "OQM | Control topic | Received unsupported control topic operation: " + pubsubMessage);
- }
- }
-}
\ No newline at end of file
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmServiceTopicReceiver.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmEventReplicator.java
similarity index 78%
rename from provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmServiceTopicReceiver.java
rename to provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmEventReplicator.java
index 770bc5788..9b98e761b 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmServiceTopicReceiver.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmEventReplicator.java
@@ -17,31 +17,35 @@
package org.opengroup.osdu.notification.provider.gcp.pubsub.receiver;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.http.HttpStatus;
-import org.opengroup.osdu.core.common.model.http.AppException;
-import org.opengroup.osdu.core.common.model.notification.Subscription;
-import org.opengroup.osdu.core.gcp.oqm.driver.OqmDriver;
-import org.opengroup.osdu.core.gcp.oqm.model.*;
-import org.opengroup.osdu.notification.provider.gcp.config.ExternalSubscriptionsManager;
-import org.opengroup.osdu.notification.provider.gcp.thread.ThreadScopeContextHolder;
+import static org.opengroup.osdu.notification.provider.gcp.pubsub.OqmSubscriberManager.PUBLISH_SUFFIX;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.Optional;
+import java.util.Set;
import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.opengroup.osdu.core.common.model.notification.Subscription;
+import org.opengroup.osdu.core.gcp.oqm.driver.OqmDriver;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmAckReplier;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmDestination;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmMessage;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmMessageReceiver;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmSubscription;
+import org.opengroup.osdu.core.gcp.oqm.model.OqmTopic;
+import org.opengroup.osdu.notification.provider.gcp.service.ExternalSubscriptionsManager;
+import org.opengroup.osdu.notification.provider.gcp.thread.ThreadScopeContextHolder;
-import static org.opengroup.osdu.notification.provider.gcp.pubsub.OqmSubscriberManager.*;
-
+/**
+ * Used for event replication among existing subscribers in the Register service.
+ */
@Slf4j
-public class OqmServiceTopicReceiver implements OqmMessageReceiver {
+public class OqmEventReplicator implements OqmMessageReceiver {
private final OqmSubscription subscription;
private final OqmDriver driver;
private final ExternalSubscriptionsManager externalSubscriptionsManager;
- public OqmServiceTopicReceiver(OqmSubscription subscription,
+ public OqmEventReplicator(OqmSubscription subscription,
OqmDriver driver,
ExternalSubscriptionsManager externalSubscriptionsManager) {
this.subscription = subscription;
@@ -73,7 +77,8 @@ public class OqmServiceTopicReceiver implements OqmMessageReceiver {
.name(serviceTopic + PUBLISH_SUFFIX)
.build();
- List<Subscription> cachedSubscriptionInfos = getCachedSubscriptionInfosByTopic(dataPartitionId, serviceTopic);
+ Set<Subscription> cachedSubscriptionInfos = externalSubscriptionsManager.getSubscriptionsForTopic(
+ dataPartitionId, serviceTopic);
for (Subscription info : cachedSubscriptionInfos) {
Map<String, String> attributes = new HashMap<>(refineAttributes(message));
attributes.put("subscription-id", info.getNotificationId());
@@ -98,12 +103,4 @@ public class OqmServiceTopicReceiver implements OqmMessageReceiver {
log.debug("Message attributes for message id: `{}` were refined: `{}`", message.getId(), refinedAttributes);
return refinedAttributes;
}
-
- private List<Subscription> getCachedSubscriptionInfosByTopic(String dataPartitionId, String serviceTopic) {
- return Optional.ofNullable(externalSubscriptionsManager.getExternalSubscriptions(dataPartitionId))
- .orElseThrow(() -> new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Server error", "OQM | Subscription info cache was not initialized"))
- .getSubscriptions().stream()
- .filter(info -> serviceTopic.equals(info.getTopic()))
- .collect(Collectors.toList());
- }
}
\ No newline at end of file
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmPublishTopicReceiver.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmReplicatedEventReceiver.java
similarity index 83%
rename from provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmPublishTopicReceiver.java
rename to provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmReplicatedEventReceiver.java
index 49f985eb1..63115eb46 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmPublishTopicReceiver.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/receiver/OqmReplicatedEventReceiver.java
@@ -25,18 +25,18 @@ import org.opengroup.osdu.core.gcp.oqm.model.OqmAckReplier;
import org.opengroup.osdu.core.gcp.oqm.model.OqmMessage;
import org.opengroup.osdu.core.gcp.oqm.model.OqmMessageReceiver;
import org.opengroup.osdu.core.gcp.oqm.model.OqmSubscription;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.OqmNotificationHandler;
+import org.opengroup.osdu.notification.provider.gcp.pubsub.OqmNotificationDeliveryService;
import org.opengroup.osdu.notification.provider.gcp.thread.ThreadScopeContextHolder;
@Slf4j
-public class OqmPublishTopicReceiver implements OqmMessageReceiver {
+public class OqmReplicatedEventReceiver implements OqmMessageReceiver {
private final OqmSubscription subscription;
- private final OqmNotificationHandler notificationHandler;
+ private final OqmNotificationDeliveryService notificationDeliveryService;
- public OqmPublishTopicReceiver(OqmSubscription subscription, OqmNotificationHandler notificationHandler) {
+ public OqmReplicatedEventReceiver(OqmSubscription subscription, OqmNotificationDeliveryService notificationDeliveryService) {
this.subscription = subscription;
- this.notificationHandler = notificationHandler;
+ this.notificationDeliveryService = notificationDeliveryService;
}
@Override
@@ -62,7 +62,7 @@ public class OqmPublishTopicReceiver implements OqmMessageReceiver {
Map<String, String> headerAttributes = oqmMessage.getAttributes();
String subscriptionId = headerAttributes.get("subscription-id");
- HttpResponse response = notificationHandler.notifySubscriber(subscriptionId, pubsubMessage, headerAttributes);
+ HttpResponse response = notificationDeliveryService.notifySubscriber(subscriptionId, pubsubMessage, headerAttributes);
if (!response.isSuccessCode()) {
throw new AppException(response.getResponseCode(), "Notification failed", response.getBody());
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/repo/SubscriptionCacheRepo.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/repo/SubscriptionCacheRepo.java
new file mode 100644
index 000000000..d9788610e
--- /dev/null
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/repo/SubscriptionCacheRepo.java
@@ -0,0 +1,132 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.provider.gcp.repo;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.opengroup.osdu.core.common.model.notification.Subscription;
+import org.opengroup.osdu.notification.provider.gcp.config.RedisProperties;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class SubscriptionCacheRepo {
+
+ public static final String POISON = ":poison";
+ private final RedisProperties redisProperties;
+ private final RedisTemplate<String, Subscription> redisTemplate;
+
+ public void writeSubscription(String dataPartitionId, Subscription subscription) {
+ String subscriberCacheKey = getSubscriberCacheKey(
+ dataPartitionId,
+ subscription.getTopic(),
+ subscription.getNotificationId()
+ );
+ redisTemplate.opsForValue().set(
+ subscriberCacheKey,
+ subscription
+ );
+ }
+
+ public Subscription readSubscription(String dataPartitionId, String topic,
+ String subscriptionId) {
+ String subscriberCacheKey = getSubscriberCacheKey(
+ dataPartitionId,
+ topic,
+ subscriptionId
+ );
+ return redisTemplate.opsForValue().get(subscriberCacheKey);
+ }
+
+ /**
+ * To avoid Register service spamming, if a subscriber was deleted, but replicated events are
+ * still being processed.
+ *
+ * @param dataPartitionId
+ * @param topic
+ * @param subscriptionId
+ */
+ public void poisonSub(String dataPartitionId, String topic, String subscriptionId) {
+ String subPoisonKey = getSubscriberCacheKey(dataPartitionId, topic, subscriptionId) + POISON;
+
+ redisTemplate.opsForValue().set(
+ subPoisonKey,
+ new Subscription(),
+ redisProperties.getRedisExpiration(),
+ TimeUnit.SECONDS
+ );
+ }
+
+ /**
+ * Used to check if subscribers were not found in Register.
+ *
+ * @param dataPartitionId
+ * @param topic
+ * @param subscriptionId
+ * @return empty Subscription
+ */
+ public Subscription checkIfSubIsPoisoned(String dataPartitionId, String topic,
+ String subscriptionId) {
+ String subPoisonKey = getSubscriberCacheKey(dataPartitionId, topic, subscriptionId) + POISON;
+ return redisTemplate.opsForValue().get(subPoisonKey);
+ }
+
+ public boolean deleteSubscription(String dataPartitionId, String topic, String subscriptionId) {
+ String subscriberCacheKey = getSubscriberCacheKey(dataPartitionId, topic, subscriptionId);
+ Boolean delete = redisTemplate.delete(subscriberCacheKey);
+ return Boolean.TRUE.equals(delete);
+ }
+
+ public Set<Subscription> getDataPartitionSubscribers(String dataPartitionId) {
+ return getSubscriptions(dataPartitionId + ":*");
+ }
+
+ public Set<Subscription> getTopicSubscribers(String dataPartitionId, String topic) {
+ return getSubscriptions(dataPartitionId + ":" + topic + ":*");
+ }
+
+ private HashSet<Subscription> getSubscriptions(String pattern) {
+ ScanOptions options = ScanOptions.scanOptions()
+ .match(pattern)
+ .build();
+
+ HashSet<String> keys = new HashSet<>();
+
+ try (Cursor<String> cursor = redisTemplate.scan(options)) {
+ while (cursor.hasNext()) {
+ keys.add(cursor.next());
+ }
+ }
+
+ return new HashSet<>(Optional.ofNullable(redisTemplate.opsForValue().multiGet(keys))
+ .orElse(Collections.emptyList()));
+ }
+
+ private static String getSubscriberCacheKey(String dataPartitionId, String topic, String subId) {
+ return dataPartitionId + ":" + topic + ":" + subId;
+ }
+}
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManager.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManager.java
new file mode 100644
index 000000000..2be560afb
--- /dev/null
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManager.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.provider.gcp.service;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.opengroup.osdu.core.common.model.notification.Subscription;
+import org.opengroup.osdu.notification.provider.gcp.repo.SubscriptionCacheRepo;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class ExternalSubscriptionsManager {
+
+ private final SubscriptionCacheRepo subscriptionCacheRepo;
+ private final RegisterSubscriptionService subscriptionService;
+
+ public void invokeTenantSubscribers(String dataPartitionId) {
+ Set<Subscription> members = subscriptionCacheRepo.getDataPartitionSubscribers(dataPartitionId);
+ if (Objects.isNull(members) || members.isEmpty()) {
+ log.debug("Subscription info cache wasn't found for tenant {}", dataPartitionId);
+ reloadSubscriptionInfoCache(dataPartitionId);
+ }
+ }
+
+ public Set<Subscription> getSubscriptionsForTopic(String dataPartitionId, String topicName) {
+ return subscriptionCacheRepo.getTopicSubscribers(dataPartitionId, topicName);
+ }
+
+ public Subscription getSubscription(String dataPartitionId, String topic, String subscriptionId) {
+ Subscription subscription = subscriptionCacheRepo.readSubscription(dataPartitionId, topic, subscriptionId);
+ if (subscription != null) {
+ return subscription;
+ }
+
+ log.debug("Subscription: {} not in cache, checking if it available in Register.", subscriptionId);
+ subscription = subscriptionCacheRepo.checkIfSubIsPoisoned(dataPartitionId, topic, subscriptionId);
+ if (subscription != null) {
+ log.debug("Subscription:{} marked as not available in cache.", subscriptionId);
+ return null;
+ }
+ return retrieveSubscriptionFromHandler(dataPartitionId, subscriptionId, topic);
+ }
+
+ private Subscription retrieveSubscriptionFromHandler(String dataPartitionId, String subscriptionId, String topic) {
+ Subscription subscription = subscriptionService.getSubscriptionsByNotificationId(dataPartitionId,subscriptionId);
+ if (subscription == null) {
+ log.debug("Subscription:{} not available in Register. Marking it as not available in cache.", subscriptionId);
+ subscriptionCacheRepo.poisonSub(dataPartitionId, topic, subscriptionId);
+ return null;
+ }
+ subscriptionCacheRepo.writeSubscription(dataPartitionId, subscription);
+ return subscription;
+ }
+
+ public boolean removeSubscription(String dataPartitionId, String topic, String subscriptionId) {
+ return subscriptionCacheRepo.deleteSubscription(dataPartitionId, topic, subscriptionId);
+ }
+
+ public void updateSubscription(String dataPartitionId, String subscriptionId) {
+ Subscription subscription = subscriptionService.getSubscriptionsByNotificationId(
+ dataPartitionId,
+ subscriptionId
+ );
+ if(Objects.isNull(subscription)){
+ log.warn("Cannot update sub config with id:{} in the cache. Skipping it.", subscriptionId);
+ }else {
+ subscriptionCacheRepo.writeSubscription(dataPartitionId, subscription);
+ }
+ }
+
+ private void reloadSubscriptionInfoCache(String dataPartitionId) {
+ List<Subscription> fragmentarySubInfos = subscriptionService.getAllSubscriptionInfos(dataPartitionId);
+ for (Subscription freagmentedSubscription : fragmentarySubInfos) {
+ Subscription subscription = subscriptionService.getSubscriptionsByNotificationId(dataPartitionId, freagmentedSubscription.getNotificationId());
+ if(Objects.isNull(subscription)){
+ log.warn("Cannot init cache for sub id:{}. Empty response from Register. Skipping it.", freagmentedSubscription.getNotificationId());
+ }else {
+ subscriptionCacheRepo.writeSubscription(dataPartitionId, subscription);
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriptionHandler.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/RegisterSubscriptionService.java
similarity index 87%
rename from provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriptionHandler.java
rename to provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/RegisterSubscriptionService.java
index fa45de097..b41976eec 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriptionHandler.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/RegisterSubscriptionService.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.opengroup.osdu.notification.provider.gcp.pubsub;
+package org.opengroup.osdu.notification.provider.gcp.service;
import java.util.List;
import java.util.stream.Collectors;
@@ -43,7 +43,7 @@ import org.springframework.stereotype.Component;
@RequiredArgsConstructor
@ConditionalOnProperty(name = "oqmDriver")
@Slf4j
-public class OqmSubscriptionHandler {
+public class RegisterSubscriptionService {
private final ISubscriptionFactory registerClientFactory;
private final DpsHeadersProvider dpsHeadersProvider;
@@ -52,7 +52,6 @@ public class OqmSubscriptionHandler {
public List<OqmTopic> getTopics(String dataPartitionId) {
DpsHeaders headers = dpsHeadersProvider.getDpsHeaders(dataPartitionId);
ISubscriptionService registerClient = registerClientFactory.create(headers);
-
try {
return registerClient.getTopics()
.stream()
@@ -75,11 +74,17 @@ public class OqmSubscriptionHandler {
}
}
- public List<Subscription> getSubscriptionsById(String dataPartitionId, String subscriptionId) {
+ public Subscription getSubscriptionsByNotificationId(String dataPartitionId, String subscriptionId) {
try {
DpsHeaders headers = dpsHeadersProvider.getDpsHeaders(dataPartitionId);
ISubscriptionService registerClient = registerClientFactory.create(headers);
- return registerClient.query(subscriptionId);
+ List<Subscription> subscriptions = registerClient.query(subscriptionId);
+ if(subscriptions.isEmpty()){
+ log.warn("Empty response from Register for subId: {}, config fetch not possible.", subscriptionId);
+ return null;
+ }else {
+ return subscriptions.get(0);
+ }
} catch (SubscriptionException se) {
throw new AppException(HttpStatus.SC_INTERNAL_SERVER_ERROR, "Server error", "Unexpected error while sending request", se);
}
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/SubscriptionServiceGc.java b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/SubscriptionServiceGc.java
index ae1914887..a89b06e7b 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/SubscriptionServiceGc.java
+++ b/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/service/SubscriptionServiceGc.java
@@ -45,7 +45,7 @@ public class SubscriptionServiceGc {
if (response.isSuccessCode()) {
try {
ObjectMapper objectMapper = new ObjectMapper();
- return objectMapper.readValue(response.getBody(), new TypeReference<List<Subscription>>(){});
+ return objectMapper.readValue(response.getBody(), new TypeReference<>() {});
} catch (IOException ex) {
throw new SubscriptionException("Exception in deserializing response", response, ex);
}
diff --git a/provider/notification-gc/src/main/resources/application.properties b/provider/notification-gc/src/main/resources/application.properties
index 8cc65690e..4c65e44ec 100644
--- a/provider/notification-gc/src/main/resources/application.properties
+++ b/provider/notification-gc/src/main/resources/application.properties
@@ -43,10 +43,12 @@ partition-auth-enabled=true
oqmDriver=pubsub
# Redis config
-redis-host=${REDIS_USER_INFO_HOST:127.0.0.1}
-redis-port=${REDIS_USER_INFO_PORT:6379}
-redis-password=${REDIS_USER_INFO_PASSWORD:}
-redis-with-ssl=${REDIS_USER_INFO_WITH_SSL:false}
+redis-host=${REDIS_HOST:127.0.0.1}
+redis-port=${REDIS_PORT:6379}
+redis-password=${REDIS_PASSWORD:}
+redis-with-ssl=${REDIS_WITH_SSL:false}
cache.codec=jackson
-propertyResolver.strategy=partition
\ No newline at end of file
+propertyResolver.strategy=partition
+
+spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration,org.springframework.boot.autoconfigure.data.redis.RedisRepositoriesAutoConfiguration
\ No newline at end of file
diff --git a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmControlTopicReceiverTest.java b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmConfigurationEventReceiverTest.java
similarity index 53%
rename from provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmControlTopicReceiverTest.java
rename to provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmConfigurationEventReceiverTest.java
index 40a172819..813574980 100644
--- a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmControlTopicReceiverTest.java
+++ b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmConfigurationEventReceiverTest.java
@@ -20,11 +20,9 @@ package org.opengroup.osdu.notification.provider.gcp.pubsub;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
import java.util.HashMap;
import java.util.List;
-import java.util.stream.Stream;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
@@ -33,70 +31,58 @@ import org.mockito.junit.MockitoJUnitRunner;
import org.opengroup.osdu.core.common.model.notification.Subscription;
import org.opengroup.osdu.core.gcp.oqm.model.OqmAckReplier;
import org.opengroup.osdu.core.gcp.oqm.model.OqmMessage;
-import org.opengroup.osdu.notification.provider.gcp.config.ExternalSubscriptionsManager;
-import org.opengroup.osdu.notification.provider.gcp.model.ExternalSubscriptions;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmControlTopicReceiver;
+import org.opengroup.osdu.notification.provider.gcp.service.ExternalSubscriptionsManager;
+import org.opengroup.osdu.notification.provider.gcp.repo.SubscriptionCacheRepo;
+import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmConfigurationEventReceiver;
@RunWith(MockitoJUnitRunner.class)
-public class OqmControlTopicReceiverTest {
+public class OqmConfigurationEventReceiverTest {
@InjectMocks
- private OqmControlTopicReceiver sut;
+ private OqmConfigurationEventReceiver sut;
@Mock
private OqmAckReplier replier;
@Mock
private Subscription subscription;
@Mock
- private ExternalSubscriptions externalSubscriptions;
- @Mock
private List<Subscription> subscriptionInfos;
@Mock
private ExternalSubscriptionsManager externalSubscriptionsManager;
+ @Mock
+ private SubscriptionCacheRepo subscriptionCacheRepo;
@Test
public void testReceiveMessageSubscriptionCreated() {
- when(externalSubscriptionsManager.getSubscription(any(), any(), any())).thenReturn(subscription);
- when(externalSubscriptions.getSubscriptions()).thenReturn(subscriptionInfos);
- when(externalSubscriptionsManager.getExternalSubscriptions(any())).thenReturn(externalSubscriptions);
OqmMessage createMessage = OqmMessage.builder().data("Subscription Created").attributes(new HashMap<>()).build();
sut.receiveMessage(createMessage, replier);
- verify(subscriptionInfos, times(1)).add(subscription);
- verify(externalSubscriptionsManager, times(1)).getExternalSubscriptions(any());
- verify(externalSubscriptionsManager, times(1)).updateExternalSubscriptionsCache(any(), any());
+ verify(externalSubscriptionsManager, times(1)).updateSubscription(any(), any());
verify(replier, times(1)).ack();
}
@Test
public void testReceiveMessageSubscriptionUpdated() {
- when(externalSubscriptionsManager.getSubscription(any(), any(), any())).thenReturn(subscription);
- when(externalSubscriptions.getSubscriptions()).thenReturn(subscriptionInfos);
- when(externalSubscriptionsManager.getExternalSubscriptions(any())).thenReturn(externalSubscriptions);
OqmMessage updatedMessage = OqmMessage.builder().data("Subscription Updated").attributes(new HashMap<>()).build();
sut.receiveMessage(updatedMessage, replier);
- verify(subscriptionInfos, times(1)).add(subscription);
- verify(externalSubscriptionsManager, times(1)).getExternalSubscriptions(any());
- verify(externalSubscriptionsManager, times(1)).updateExternalSubscriptionsCache(any(), any());
+ verify(externalSubscriptionsManager, times(1)).updateSubscription(any(), any());
verify(replier, times(1)).ack();
}
@Test
public void testReceiveMessageSubscriptionDeleted() {
- when(externalSubscriptions.getSubscriptions()).thenReturn(subscriptionInfos);
- when(externalSubscriptionsManager.getExternalSubscriptions(any())).thenReturn(externalSubscriptions);
- when(subscriptionInfos.stream()).thenReturn(Stream.of(subscription));
- when(subscription.getNotificationId()).thenReturn("4");
- when(subscription.getTopic()).thenReturn("topic1");
+ String partitionId = "test";
+ String topic = "topic1";
+ String subId = "4";
+
OqmMessage deletedMessage = OqmMessage.builder()
.data("Subscription Deleted")
- .attributes(new HashMap<String, String>() {{
- put("subscription-id", "4");
- put("topic", "topic1");
+ .attributes(new HashMap<>() {{
+ put("subscription-id", subId);
+ put("topic", topic);
+ put("data-partition-id", partitionId);
}})
.build();
sut.receiveMessage(deletedMessage, replier);
- verify(subscriptionInfos, times(1)).remove(subscription);
- verify(externalSubscriptionsManager, times(1)).getExternalSubscriptions(any());
- verify(externalSubscriptionsManager, times(1)).updateExternalSubscriptionsCache(any(), any());
+ verify(externalSubscriptionsManager, times(1)).removeSubscription(partitionId, topic, subId);
verify(replier, times(1)).ack();
}
@@ -104,6 +90,6 @@ public class OqmControlTopicReceiverTest {
public void testReceiveMessageUnsupported() {
OqmMessage unsupportedMessage = OqmMessage.builder().data("Unsupported").attributes(new HashMap<>()).build();
sut.receiveMessage(unsupportedMessage, replier);
- verify(replier, times(1)).nack();
+ verify(replier, times(1)).ack();
}
}
\ No newline at end of file
diff --git a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmServiceTopicReceiverTest.java b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmEventReplicatorTest.java
similarity index 72%
rename from provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmServiceTopicReceiverTest.java
rename to provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmEventReplicatorTest.java
index c859a72ea..776c169b3 100644
--- a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmServiceTopicReceiverTest.java
+++ b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmEventReplicatorTest.java
@@ -17,33 +17,34 @@
package org.opengroup.osdu.notification.provider.gcp.pubsub;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
-import org.opengroup.osdu.core.common.cache.IRedisCache;
import org.opengroup.osdu.core.common.model.notification.Subscription;
import org.opengroup.osdu.core.gcp.oqm.driver.OqmDriver;
import org.opengroup.osdu.core.gcp.oqm.model.*;
-import org.opengroup.osdu.notification.provider.gcp.config.ExternalSubscriptionsManager;
-import org.opengroup.osdu.notification.provider.gcp.model.ExternalSubscriptions;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmServiceTopicReceiver;
+import org.opengroup.osdu.notification.provider.gcp.service.ExternalSubscriptionsManager;
+import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmEventReplicator;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
@RunWith(MockitoJUnitRunner.class)
-public class OqmServiceTopicReceiverTest {
+public class OqmEventReplicatorTest {
+ public static final String TEST_TENANT = "tenant1";
+ public static final String TEST_TOPIC = "topic1";
@InjectMocks
- private OqmServiceTopicReceiver sut;
+ private OqmEventReplicator sut;
@Mock
private OqmAckReplier replier;
@Mock
@@ -51,31 +52,31 @@ public class OqmServiceTopicReceiverTest {
@Mock
private OqmSubscription oqmSubscription;
@Mock
- private ExternalSubscriptions externalSubscriptions;
- @Mock
private ExternalSubscriptionsManager externalSubscriptionsManager;
- private List<Subscription> subscriptionInfos;
+ private Set<Subscription> subscriptionInfos;
private OqmTopic topic;
@Before
public void setUp() {
- topic = OqmTopic.builder().name("topic1").build();
+ topic = OqmTopic.builder().name(TEST_TOPIC).build();
Subscription subscription1 = new Subscription();
subscription1.setTopic(topic.getName());
+ subscription1.setId("1");
Subscription subscription2 = new Subscription();
subscription2.setTopic(topic.getName());
+ subscription2.setId("2");
Subscription subscription3 = new Subscription();
subscription3.setTopic(topic.getName());
- subscriptionInfos = Arrays.asList(subscription1, subscription2, subscription3);
+ subscription3.setId("3");
+ subscriptionInfos = Stream.of(subscription1, subscription2, subscription3).collect(Collectors.toSet());
}
@Test
public void testReceiveMessageSubscriptionCreated() {
when(oqmSubscription.getTopics()).thenReturn(Collections.singletonList(topic));
- when(externalSubscriptions.getSubscriptions()).thenReturn(subscriptionInfos);
- when(externalSubscriptionsManager.getExternalSubscriptions(any())).thenReturn(externalSubscriptions);
+ when(externalSubscriptionsManager.getSubscriptionsForTopic(TEST_TENANT, TEST_TOPIC)).thenReturn(subscriptionInfos);
OqmMessage message = OqmMessage.builder().data("Message").attributes(new HashMap<String, String>() {{
- put("data-partition-id", "tenant1");
+ put("data-partition-id", TEST_TENANT);
}}).build();
sut.receiveMessage(message, replier);
verify(driver, times(3)).publish(any(OqmMessage.class), any(), any());
diff --git a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmPublishTopicReceiverTest.java b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmReplicatedEventReceiverTest.java
similarity index 94%
rename from provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmPublishTopicReceiverTest.java
rename to provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmReplicatedEventReceiverTest.java
index dc1a08318..c5cf438c0 100644
--- a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmPublishTopicReceiverTest.java
+++ b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmReplicatedEventReceiverTest.java
@@ -27,7 +27,7 @@ import org.opengroup.osdu.core.gcp.oqm.model.OqmAckReplier;
import org.opengroup.osdu.core.gcp.oqm.model.OqmMessage;
import org.opengroup.osdu.core.gcp.oqm.model.OqmSubscription;
import org.opengroup.osdu.core.gcp.oqm.model.OqmTopic;
-import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmPublishTopicReceiver;
+import org.opengroup.osdu.notification.provider.gcp.pubsub.receiver.OqmReplicatedEventReceiver;
import java.util.Collections;
import java.util.HashMap;
@@ -37,14 +37,14 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
@RunWith(MockitoJUnitRunner.class)
-public class OqmPublishTopicReceiverTest {
+public class OqmReplicatedEventReceiverTest {
@InjectMocks
- private OqmPublishTopicReceiver sut;
+ private OqmReplicatedEventReceiver sut;
@Mock
private OqmAckReplier replier;
@Mock
- private OqmNotificationHandler notificationHandler;
+ private OqmNotificationDeliveryService notificationHandler;
@Mock
private HttpResponse httpResponse;
@Mock
diff --git a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManagerTest.java b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManagerTest.java
index a3d33bddd..025bad77c 100644
--- a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManagerTest.java
+++ b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/pubsub/OqmSubscriberManagerTest.java
@@ -22,20 +22,16 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
-import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;
-import org.opengroup.osdu.core.common.cache.IRedisCache;
-import org.opengroup.osdu.core.common.model.notification.Subscription;
import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
import org.opengroup.osdu.core.common.provider.interfaces.ITenantFactory;
import org.opengroup.osdu.core.gcp.oqm.driver.OqmDriver;
import org.opengroup.osdu.core.gcp.oqm.model.OqmTopic;
-import org.opengroup.osdu.notification.provider.gcp.config.ExternalSubscriptionsManager;
-import org.opengroup.osdu.notification.provider.gcp.config.OqmConfigurationProperties;
-import org.opengroup.osdu.notification.provider.gcp.model.ExternalSubscriptions;
+import org.opengroup.osdu.notification.provider.gcp.service.ExternalSubscriptionsManager;
import java.util.Arrays;
import java.util.List;
+import org.opengroup.osdu.notification.provider.gcp.service.RegisterSubscriptionService;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
@@ -52,7 +48,7 @@ public class OqmSubscriberManagerTest {
@Mock
private MessageBrokerProvider subscriptionProvider;
@Mock
- private OqmSubscriptionHandler subscriptionHandler;
+ private RegisterSubscriptionService subscriptionHandler;
@Mock
private ExternalSubscriptionsManager externalSubscriptionsManager;
@@ -105,6 +101,6 @@ public class OqmSubscriberManagerTest {
public void testProvisioningSubscriptionInfoCache() {
when(tenantFactory.listTenantInfo()).thenReturn(tenantInfos);
sut.provisionSubscriptionInfoCache();
- verify(externalSubscriptionsManager, times(3)).getExternalSubscriptions(any());
+ verify(externalSubscriptionsManager, times(3)).invokeTenantSubscribers(any());
}
}
\ No newline at end of file
diff --git a/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManagerTest.java b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManagerTest.java
new file mode 100644
index 000000000..0e0ac0b97
--- /dev/null
+++ b/provider/notification-gc/src/test/java/org/opengroup/osdu/notification/provider/gcp/service/ExternalSubscriptionsManagerTest.java
@@ -0,0 +1,168 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.provider.gcp.service;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.opengroup.osdu.core.common.model.notification.Subscription;
+import org.opengroup.osdu.notification.provider.gcp.repo.SubscriptionCacheRepo;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ExternalSubscriptionsManagerTest {
+
+ public static final String TEST_TENANT = "test-tenant";
+ private static final String TENANT_WO_SUBS = "no-sub-tenant";
+ public static final String TEST_TOPIC = "topic";
+ public static final String TEST_NOTIF_ID = "test-id";
+
+ @InjectMocks
+ private ExternalSubscriptionsManager manager;
+ @Mock
+ private SubscriptionCacheRepo mockSubCacheRepo;
+ @Mock
+ private RegisterSubscriptionService mockSubService;
+
+ private Subscription subscription;
+
+ private Set<Subscription> subscriptions;
+
+ @Before
+ public void setUp() {
+ subscription = new Subscription();
+ subscription.setNotificationId(TEST_NOTIF_ID);
+ subscription.setTopic(TEST_TOPIC);
+
+ subscriptions = new HashSet<>() {{
+ add(subscription);
+ }};
+ }
+
+ @Test
+ public void testInvokeTenantSubscribers_noActiveSubs() {
+ when(mockSubCacheRepo.getDataPartitionSubscribers(TENANT_WO_SUBS)).thenReturn(
+ Collections.emptySet());
+ manager.invokeTenantSubscribers(TENANT_WO_SUBS);
+ verify(mockSubCacheRepo, times(0)).writeSubscription(anyString(), any());
+ verify(mockSubService, times(1)).getAllSubscriptionInfos(TENANT_WO_SUBS);
+ }
+
+ @Test
+ public void testInvokeTenantSubscribers_withActiveSubs() {
+ when(mockSubCacheRepo.getDataPartitionSubscribers(TEST_TENANT)).thenReturn(subscriptions);
+ manager.invokeTenantSubscribers(TEST_TENANT);
+ verify(mockSubService, times(0)).getAllSubscriptionInfos(TEST_TENANT);
+ }
+
+ @Test
+ public void testInvokeTenantSubscribers_withActiveSubs_emptyCache() {
+ when(mockSubCacheRepo.getDataPartitionSubscribers(TEST_TENANT)).thenReturn(null);
+ when(mockSubService.getAllSubscriptionInfos(TEST_TENANT)).thenReturn(Collections.singletonList(subscription));
+ when(mockSubService.getSubscriptionsByNotificationId(TEST_TENANT, TEST_NOTIF_ID)).thenReturn(subscription);
+
+ manager.invokeTenantSubscribers(TEST_TENANT);
+
+ verify(mockSubService, times(1)).getAllSubscriptionInfos(TEST_TENANT);
+ verify(mockSubCacheRepo, times(1)).writeSubscription(TEST_TENANT, subscription);
+ }
+
+ @Test
+ public void testGetSubscriptionsForTopic() {
+ doReturn(subscriptions).when(mockSubCacheRepo).getTopicSubscribers(TEST_TENANT, TEST_TOPIC);
+ Set<Subscription> actualSubscriptions = manager.getSubscriptionsForTopic(TEST_TENANT,
+ TEST_TOPIC);
+ assertEquals(subscriptions, actualSubscriptions);
+ }
+
+ @Test
+ public void testGetSubscription_inCache() {
+ when(mockSubCacheRepo.readSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID)).thenReturn(
+ subscription);
+ Subscription actualSub = manager.getSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID);
+ assertEquals(subscription, actualSub);
+ verify(mockSubService, times(0)).getSubscriptionsByNotificationId(anyString(), anyString());
+ }
+
+ @Test
+ public void testGetSubscription_notInCache_active() {
+ when(mockSubCacheRepo.readSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID)).thenReturn(null);
+ when(mockSubService.getSubscriptionsByNotificationId(TEST_TENANT, TEST_NOTIF_ID)).thenReturn(
+ subscription);
+
+ Subscription actualSub = manager.getSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID);
+ assertEquals(subscription, actualSub);
+ verify(mockSubCacheRepo, times(1)).writeSubscription(TEST_TENANT, subscription);
+ }
+
+ @Test
+ public void testGetSubscription_notInCache_notAvailableInRegister() {
+ when(mockSubCacheRepo.readSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID)).thenReturn(null);
+ when(mockSubService.getSubscriptionsByNotificationId(TEST_TENANT, TEST_NOTIF_ID)).thenReturn(
+ null);
+
+ Subscription actualSub = manager.getSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID);
+ assertNull(actualSub);
+ verify(mockSubCacheRepo, times(1)).poisonSub(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID);
+ }
+
+ @Test
+ public void testGetSubscription_InCache_Poisoned() {
+ when(mockSubCacheRepo.readSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID)).thenReturn(null);
+ when(mockSubCacheRepo.checkIfSubIsPoisoned(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID)).thenReturn(
+ new Subscription());
+
+ Subscription actualSub = manager.getSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID);
+ assertNull(actualSub);
+ verify(mockSubService, times(0)).getSubscriptionsByNotificationId(anyString(), anyString());
+ }
+
+ @Test
+ public void testRemoveSubscription() {
+ boolean b = manager.removeSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID);
+ verify(mockSubCacheRepo, times(1)).deleteSubscription(TEST_TENANT, TEST_TOPIC, TEST_NOTIF_ID);
+ }
+
+ @Test
+ public void testUpdateSubscription_availableInRegister() {
+ when(mockSubService.getSubscriptionsByNotificationId(TEST_TENANT, TEST_NOTIF_ID)).thenReturn(
+ subscription);
+ manager.updateSubscription(TEST_TENANT, TEST_NOTIF_ID);
+ verify(mockSubCacheRepo, times(1)).writeSubscription(TEST_TENANT, subscription);
+ }
+
+ @Test
+ public void testUpdateSubscription_notAvailableInRegister() {
+ manager.updateSubscription(TEST_TENANT, TEST_NOTIF_ID);
+ verify(mockSubCacheRepo, times(0)).writeSubscription(anyString(), any());
+ }
+}
\ No newline at end of file
diff --git a/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java
new file mode 100644
index 000000000..f5d8f4f2d
--- /dev/null
+++ b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java
@@ -0,0 +1,149 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.api;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.opengroup.osdu.notification.util.Constants.GROUP_ID;
+
+import com.google.common.base.Strings;
+import com.sun.jersey.api.client.ClientResponse;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.notification.GsaSecret;
+import org.opengroup.osdu.core.common.model.notification.GsaSecretValue;
+import org.opengroup.osdu.core.common.model.notification.Secret;
+import org.opengroup.osdu.core.common.model.notification.Subscription;
+import org.opengroup.osdu.core.common.notification.ISubscriptionService;
+import org.opengroup.osdu.core.common.notification.SubscriptionAPIConfig;
+import org.opengroup.osdu.core.common.notification.SubscriptionException;
+import org.opengroup.osdu.core.common.notification.SubscriptionFactory;
+import org.opengroup.osdu.notification.util.AnthosTestUtils;
+import org.opengroup.osdu.notification.util.ServicesUtils;
+import org.opengroup.osdu.notification.util.TestUtils;
+
+public class TestPushEndpointGsa {
+ public static final String REGISTER_BASE_URL = "REGISTER_BASE_URL";
+ public static final String TOPIC_ID = "TOPIC_ID";
+ public static final String INTEGRATION_AUDIENCE = "INTEGRATION_AUDIENCE";
+ public static final String OSDU_TENANT = "OSDU_TENANT";
+ public static final String STORAGE_HOST = "STORAGE_HOST";
+ public static final String LEGAL_HOST = "LEGAL_HOST";
+ public static final String DE_OPS_TESTER = "DE_OPS_TESTER";
+ private String subscriptionId = null;
+ private String notificationId = null;
+ private ISubscriptionService subscriptionService;
+ private static SubscriptionFactory factory;
+ private TestUtils testUtils = new AnthosTestUtils();
+ private final String suffix = String.valueOf(System.currentTimeMillis());
+ private String baseRegisterUrl;
+ private String topic;
+ private String tenant;
+ private String integrationAudience;
+ private String storageHost;
+ private String legalHost;
+ private String groupId;
+ private static final String LEGAL_TAG_NAME = "notification-test-gsa";
+ private ServicesUtils servicesUtils;
+
+ @After
+ public void deleteResource() throws Exception {
+ subscriptionService.delete(subscriptionId);
+ servicesUtils.deleteStorageRecords(3, suffix);
+ servicesUtils.deleteLegalTag(LEGAL_TAG_NAME);
+ }
+
+ @Before
+ public void createResource() throws Exception {
+ baseRegisterUrl = System.getProperty(REGISTER_BASE_URL, System.getenv(REGISTER_BASE_URL));
+ topic = System.getProperty(TOPIC_ID, System.getenv(TOPIC_ID));
+ integrationAudience = System.getProperty(INTEGRATION_AUDIENCE, System.getenv(INTEGRATION_AUDIENCE));
+ tenant = System.getProperty(OSDU_TENANT, System.getenv(OSDU_TENANT));
+ if (Strings.isNullOrEmpty(integrationAudience)) {
+ integrationAudience = tenant;
+ }
+ storageHost = System.getProperty(STORAGE_HOST, System.getenv(STORAGE_HOST));
+ legalHost = System.getProperty(LEGAL_HOST, System.getenv(LEGAL_HOST));
+ groupId = System.getProperty(GROUP_ID, System.getenv(GROUP_ID));
+ servicesUtils = new ServicesUtils(storageHost, legalHost, testUtils, tenant, groupId);
+ servicesUtils.createLegalTag(LEGAL_TAG_NAME);
+ createResourceInPartition(tenant);
+ }
+
+ @Test
+ public void testPushEndpoint() throws Exception {
+ servicesUtils.createStorageRecords(suffix, 3, LEGAL_TAG_NAME);
+ Thread.sleep(10000);
+ assertNotNull(subscriptionId);
+ assertNotNull(notificationId);
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, tenant);
+ ClientResponse clientResponse = testUtils.send(baseRegisterUrl, "/test-gc/state", "GET", testUtils.getOpsToken(), null, "",
+ headers, false);
+ Map<String, Number> response = new HashMap<>();
+ response = testUtils.getResult(clientResponse, 200, response.getClass());
+ assertNotNull(response);
+ assertTrue(response.containsKey(suffix));
+ assertTrue(response.get(suffix).longValue() >= 3);
+ }
+
+ private void createResourceInPartition(String partitionId) throws Exception {
+
+ SubscriptionAPIConfig config = SubscriptionAPIConfig.builder().rootUrl(baseRegisterUrl).build();
+ factory = new SubscriptionFactory(config);
+
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, partitionId);
+ headers.put(DpsHeaders.AUTHORIZATION, testUtils.getOpsToken());
+ DpsHeaders dpsHeaders = DpsHeaders.createFromMap(headers);
+ subscriptionService = factory.create(dpsHeaders);
+
+ Subscription subscription = new Subscription();
+ subscription.setName("subscription-integration-test-gsa-" + suffix);
+ subscription.setDescription("subscription created for gsa integration test " + suffix);
+ subscription.setTopic(topic);
+ subscription.setPushEndpoint(getPushUrl());
+
+ Secret gsaSecret = new GsaSecret();
+ GsaSecretValue gsaSecretValue = new GsaSecretValue();
+ gsaSecretValue.setAudience(integrationAudience);
+ String opsTester = System.getProperty(DE_OPS_TESTER, System.getenv(DE_OPS_TESTER));
+ gsaSecretValue.setKey(new String(Base64.getDecoder().decode(opsTester)));
+ gsaSecret.setSecretType("GSA");
+ ((GsaSecret) gsaSecret).setValue(gsaSecretValue);
+ subscription.setSecret(gsaSecret);
+ try {
+ Subscription subscriptionCreated = subscriptionService.create(subscription);
+ notificationId = subscriptionCreated.getNotificationId();
+ subscriptionId = subscriptionCreated.getId();
+ } catch (SubscriptionException e) {
+ System.out.println("Subscription exception inner response : " + e.getHttpResponse());
+ throw e;
+ }
+ }
+
+ private String getPushUrl() {
+ return baseRegisterUrl + "/test-gc/gsa-challenge/" + suffix;
+ }
+
+}
diff --git a/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java
new file mode 100644
index 000000000..0d5b0458a
--- /dev/null
+++ b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.api;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.opengroup.osdu.notification.util.Constants.GROUP_ID;
+
+import com.sun.jersey.api.client.ClientResponse;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+import org.opengroup.osdu.core.common.model.notification.HmacSecret;
+import org.opengroup.osdu.core.common.model.notification.Subscription;
+import org.opengroup.osdu.core.common.notification.ISubscriptionService;
+import org.opengroup.osdu.core.common.notification.SubscriptionAPIConfig;
+import org.opengroup.osdu.core.common.notification.SubscriptionException;
+import org.opengroup.osdu.core.common.notification.SubscriptionFactory;
+import org.opengroup.osdu.notification.util.AnthosTestUtils;
+import org.opengroup.osdu.notification.util.ServicesUtils;
+import org.opengroup.osdu.notification.util.TestUtils;
+
+public class TestPushEndpointHMAC {
+ public static final String REGISTER_BASE_URL = "REGISTER_BASE_URL";
+ public static final String TOPIC_ID = "TOPIC_ID";
+ public static final String HMAC_SECRET = "HMAC_SECRET";
+ public static final String OSDU_TENANT = "OSDU_TENANT";
+ public static final String STORAGE_HOST = "STORAGE_HOST";
+ public static final String LEGAL_HOST = "LEGAL_HOST";
+ private String subscriptionId = null;
+ private String notificationId = null;
+ private ISubscriptionService subscriptionService;
+ private static SubscriptionFactory factory;
+ private TestUtils testUtils = new AnthosTestUtils();
+ private final String suffix = String.valueOf(System.currentTimeMillis());
+ private String baseRegisterUrl;
+ private String topic;
+ private String hmacSecretValue;
+ private String tenant;
+ private String storageHost;
+ private String legalHost;
+ private String groupId;
+ private static final String LEGAL_TAG_NAME = "notification-test-hmac";
+ private ServicesUtils servicesUtils;
+
+ @After
+ public void deleteResource() throws Exception {
+ subscriptionService.delete(subscriptionId);
+ servicesUtils.deleteStorageRecords(3, suffix);
+ servicesUtils.deleteLegalTag(LEGAL_TAG_NAME);
+ }
+
+ @Before
+ public void createResource() throws Exception {
+ baseRegisterUrl = System.getProperty(REGISTER_BASE_URL, System.getenv(REGISTER_BASE_URL));
+ topic = System.getProperty(TOPIC_ID, System.getenv(TOPIC_ID));
+ hmacSecretValue = System.getProperty(HMAC_SECRET, System.getenv(HMAC_SECRET));
+ tenant = System.getProperty(OSDU_TENANT, System.getenv(OSDU_TENANT));
+ storageHost = System.getProperty(STORAGE_HOST, System.getenv(STORAGE_HOST));
+ legalHost = System.getProperty(LEGAL_HOST, System.getenv(LEGAL_HOST));
+ groupId = System.getProperty(GROUP_ID, System.getenv(GROUP_ID));
+ servicesUtils = new ServicesUtils(storageHost, legalHost, testUtils, tenant, groupId);
+ servicesUtils.createLegalTag(LEGAL_TAG_NAME);
+ createResourceInPartition(tenant);
+ }
+
+ @Test
+ public void testPushEndpoint() throws Exception {
+ servicesUtils.createStorageRecords(suffix, 3, LEGAL_TAG_NAME);
+ Thread.sleep(10000);
+ assertNotNull(subscriptionId);
+ assertNotNull(notificationId);
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, tenant);
+ ClientResponse clientResponse = testUtils.send(baseRegisterUrl, "/test-gc/state", "GET", testUtils.getOpsToken(), null, "",
+ headers, false);
+ Map<String, Number> response = new HashMap<>();
+ response = testUtils.getResult(clientResponse, 200, response.getClass());
+ assertNotNull(response);
+ assertTrue(response.containsKey(suffix));
+ assertTrue(response.get(suffix).longValue() >= 3);
+ }
+
+ private void createResourceInPartition(String partitionId) throws Exception {
+
+ SubscriptionAPIConfig config = SubscriptionAPIConfig.builder().rootUrl(baseRegisterUrl).build();
+ factory = new SubscriptionFactory(config);
+
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, partitionId);
+ headers.put(DpsHeaders.AUTHORIZATION, testUtils.getOpsToken());
+ DpsHeaders dpsHeaders = DpsHeaders.createFromMap(headers);
+ subscriptionService = factory.create(dpsHeaders);
+
+ Subscription subscription = new Subscription();
+ subscription.setName("subscription-integration-test-hmac-" + suffix);
+ subscription.setDescription("subscription created for hmac integration test " + suffix);
+ subscription.setTopic(topic);
+ subscription.setPushEndpoint(getPushUrl());
+ HmacSecret secret = new HmacSecret();
+ secret.setValue(hmacSecretValue);
+
+ subscription.setSecret(secret);
+ try {
+ Subscription subscriptionCreated = subscriptionService.create(subscription);
+ notificationId = subscriptionCreated.getNotificationId();
+ subscriptionId = subscriptionCreated.getId();
+ } catch (SubscriptionException e) {
+ System.out.println("Subscription exception inner response : " + e.getHttpResponse());
+ throw e;
+ }
+ }
+
+ private String getPushUrl() {
+ return baseRegisterUrl + "/test-gc/challenge/" + suffix;
+ }
+}
diff --git a/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/Constants.java b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/Constants.java
new file mode 100644
index 000000000..7889fab81
--- /dev/null
+++ b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/Constants.java
@@ -0,0 +1,6 @@
+package org.opengroup.osdu.notification.util;
+
+public class Constants {
+
+ public static final String GROUP_ID = "GROUP_ID";
+}
diff --git a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/model/ExternalSubscriptions.java b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/FileUtils.java
similarity index 50%
rename from provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/model/ExternalSubscriptions.java
rename to testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/FileUtils.java
index 953d14605..fc0713c16 100644
--- a/provider/notification-gc/src/main/java/org/opengroup/osdu/notification/provider/gcp/model/ExternalSubscriptions.java
+++ b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/FileUtils.java
@@ -15,20 +15,25 @@
* limitations under the License.
*/
-package org.opengroup.osdu.notification.provider.gcp.model;
+package org.opengroup.osdu.notification.util;
-import lombok.*;
-import org.opengroup.osdu.core.common.model.notification.Subscription;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
-import java.util.List;
+public class FileUtils {
+ public String readFromLocalFilePath(String filePath) throws IOException {
-/**
- * Wrapped list of subscriptions to create redis cache bean.
- */
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@Builder
-public class ExternalSubscriptions {
- private List<Subscription> subscriptions;
+ InputStream inStream = this.getClass().getResourceAsStream(filePath);
+ BufferedReader br = new BufferedReader(new InputStreamReader(inStream));
+ StringBuilder stringBuilder = new StringBuilder();
+
+ String eachLine = "";
+ while ((eachLine = br.readLine()) != null) {
+ stringBuilder.append(eachLine);
+ }
+
+ return stringBuilder.toString();
+ }
}
diff --git a/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/HttpClient.java b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/HttpClient.java
new file mode 100644
index 000000000..b14839c91
--- /dev/null
+++ b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/HttpClient.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.util;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.gson.Gson;
+import com.sun.jersey.api.client.Client;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import java.net.URI;
+import java.security.SecureRandom;
+import java.security.cert.X509Certificate;
+import java.util.Map;
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+import javax.ws.rs.core.MediaType;
+import org.apache.commons.lang3.StringUtils;
+
+public class HttpClient {
+ private HttpClient(){
+ }
+
+ public static ClientResponse send(String url, String path, String httpMethod,
+ Map<String, String> headers,
+ String requestBody, String query) throws Exception {
+
+ String normalizedUrl = new URI(String.format("%s/%s", url, path)).normalize().toString();
+ normalizedUrl = StringUtils.removeEnd(normalizedUrl, "/");
+ log(httpMethod, normalizedUrl + query, headers, requestBody);
+ Client client = getClient();
+
+ WebResource webResource = client.resource(normalizedUrl + query);
+ WebResource.Builder builder = webResource.accept(MediaType.APPLICATION_JSON)
+ .type(MediaType.APPLICATION_JSON);
+ headers.forEach(builder::header);
+
+ if ("POST".equals(httpMethod) && StringUtils.isEmpty(requestBody)) {
+ requestBody = "{}"; //solves 411 error when sending empty-body POST request
+ }
+
+ return builder.method(httpMethod, ClientResponse.class, requestBody);
+ }
+
+ private static void log(String method, String url, Map<String, String> headers, String body) {
+ System.out.println(String.format("%s: %s", method, url));
+ System.out.println(body);
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> T getResult(ClientResponse response, int exepectedStatus, Class<T> classOfT) {
+ assertEquals(exepectedStatus, response.getStatus());
+ if (exepectedStatus == 204) {
+ return null;
+ }
+
+ assertEquals("application/json; charset=UTF-8", response.getType().toString());
+ String json = response.getEntity(String.class);
+ if (classOfT == String.class) {
+ return (T) json;
+ }
+
+ Gson gson = new Gson();
+ return gson.fromJson(json, classOfT);
+ }
+
+ protected static Client getClient() {
+ TrustManager[] trustAllCerts = new TrustManager[]{new X509TrustManager() {
+ @Override
+ public X509Certificate[] getAcceptedIssuers() {
+ return null;
+ }
+
+ @Override
+ public void checkClientTrusted(X509Certificate[] certs, String authType) {
+ }
+
+ @Override
+ public void checkServerTrusted(X509Certificate[] certs, String authType) {
+ }
+ }};
+
+ try {
+ SSLContext sc = SSLContext.getInstance("TLS");
+ sc.init(null, trustAllCerts, new SecureRandom());
+ HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory());
+ } catch (Exception e) {
+ }
+ return Client.create();
+ }
+}
diff --git a/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/ServicesUtils.java b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/ServicesUtils.java
new file mode 100644
index 000000000..eabebc585
--- /dev/null
+++ b/testing/notification-test-baremetal/src/test/java/org/opengroup/osdu/notification/util/ServicesUtils.java
@@ -0,0 +1,96 @@
+/*
+ * Copyright 2020-2023 Google LLC
+ * Copyright 2020-2023 EPAM Systems, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.opengroup.osdu.notification.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.sun.jersey.api.client.ClientResponse;
+import java.util.HashMap;
+import java.util.Map;
+import org.opengroup.osdu.core.common.model.http.DpsHeaders;
+
+public class ServicesUtils {
+ private String storageHost;
+ private String legalHost;
+ private TestUtils testUtils;
+ private String partitionId;
+ private String groupId;
+ private FileUtils fileUtils;
+
+ public ServicesUtils(String storageHost, String legalHost, TestUtils testUtils, String partitionId, String groupId) {
+ this.storageHost = storageHost;
+ this.legalHost = legalHost;
+ this.testUtils = testUtils;
+ this.partitionId = partitionId;
+ this.groupId = groupId;
+ this.fileUtils = new FileUtils();
+ }
+
+ public ClientResponse createLegalTag(String tagName) throws Exception {
+ String legalBody = fileUtils.readFromLocalFilePath("/LegalTag.json");
+ legalBody = legalBody.replace("{{tagName}}", tagName);
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, partitionId);
+ headers.put(DpsHeaders.AUTHORIZATION, testUtils.getAdminToken());
+
+ ClientResponse legalResponse = HttpClient.send(legalHost, "legaltags", "POST", headers, legalBody, "");
+
+ boolean createdOrAlreadyExists = legalResponse.getStatus() == 201 || legalResponse.getStatus() == 409;
+ assertTrue(createdOrAlreadyExists);
+ return legalResponse;
+ }
+
+ public void deleteLegalTag(String tagName) throws Exception{
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, partitionId);
+ headers.put(DpsHeaders.AUTHORIZATION, testUtils.getAdminToken());
+ ClientResponse legalResponse = HttpClient.send(legalHost, String.format("legaltags/%s", tagName), "DELETE",
+ headers, "", "");
+ assertEquals(204, legalResponse.getStatus());
+ }
+
+ public void createStorageRecords(String suffix, int count, String legalTag) throws Exception{
+ String body = fileUtils.readFromLocalFilePath("/StorageRecord.json");
+ body = body.replace("{{data-partition-id}}", partitionId);
+ body = body.replace("{{legal-tag}}", partitionId + "-" + legalTag);
+ body = body.replace("{{group_id}}", groupId);
+ for (int i = 0; i < count; i++) {
+ String actualBody = body.replace("{{ids-suffix}}", suffix + String.valueOf(i));
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, partitionId);
+ headers.put(DpsHeaders.AUTHORIZATION, testUtils.getAdminToken());
+ ClientResponse storageResponse = HttpClient.send(storageHost, "records", "PUT",
+ headers, actualBody, "");
+ assertEquals(201, storageResponse.getStatus());
+ }
+ }
+
+ public void deleteStorageRecords(int count, String suffix) throws Exception{
+ for (int i = 0; i < count; i++) {
+ Map<String, String> headers = new HashMap<>();
+ headers.put(DpsHeaders.DATA_PARTITION_ID, partitionId);
+ headers.put(DpsHeaders.AUTHORIZATION, testUtils.getAdminToken());
+ String recordId = partitionId + ":dataset--ConnectedSource.Generic:notification-test-" + suffix + i;
+ ClientResponse storageResponse = HttpClient.send(storageHost, "records/" + recordId, "DELETE",
+ headers, "", "");
+ assertEquals(204, storageResponse.getStatus());
+ }
+ }
+
+}
diff --git a/testing/notification-test-baremetal/src/test/resources/LegalTag.json b/testing/notification-test-baremetal/src/test/resources/LegalTag.json
new file mode 100644
index 000000000..e9629c72a
--- /dev/null
+++ b/testing/notification-test-baremetal/src/test/resources/LegalTag.json
@@ -0,0 +1,16 @@
+{
+ "name": "{{tagName}}",
+ "properties": {
+ "countryOfOrigin": [
+ "US"
+ ],
+ "contractId": "A1234",
+ "expirationDate": 2222222222222,
+ "originator": "Default",
+ "dataType": "Public Domain Data",
+ "securityClassification": "Public",
+ "personalData": "No Personal Data",
+ "exportClassification": "EAR99"
+ },
+ "description": "Test legal tag for notification"
+}
\ No newline at end of file
diff --git a/testing/notification-test-baremetal/src/test/resources/StorageRecord.json b/testing/notification-test-baremetal/src/test/resources/StorageRecord.json
new file mode 100644
index 000000000..c4a8e395f
--- /dev/null
+++ b/testing/notification-test-baremetal/src/test/resources/StorageRecord.json
@@ -0,0 +1,33 @@
+[
+ {
+ "id": "{{data-partition-id}}:dataset--ConnectedSource.Generic:notification-test-{{ids-suffix}}",
+ "kind": "{{data-partition-id}}:wks:dataset--ConnectedSource.Generic:1.0.0",
+ "data": {
+ "Name": "name",
+ "DatasetProperties": {
+ "ConnectedSourceDataJobId": "no-data",
+ "ConnectedSourceRegistryEntryId": "no-data",
+ "SourceDataPartitionId": "no-data",
+ "SourceRecordId": "no-data"
+ }
+ },
+ "namespace": "{{data-partition-id}}:osdu",
+ "legal": {
+ "legaltags": [
+ "{{legal-tag}}"
+ ],
+ "otherRelevantDataCountries": [
+ "US"
+ ],
+ "status": "compliant"
+ },
+ "acl": {
+ "viewers": [
+ "data.default.viewers@{{data-partition-id}}.{{group_id}}"
+ ],
+ "owners": [
+ "data.default.owners@{{data-partition-id}}.{{group_id}}"
+ ]
+ }
+ }
+]
\ No newline at end of file
diff --git a/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java b/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java
index 2245bebf5..c68244d99 100644
--- a/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java
+++ b/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointGsa.java
@@ -85,6 +85,7 @@ public class TestPushEndpointGsa {
servicesUtils = new ServicesUtils(storageHost, legalHost, testUtils, tenant, groupId);
servicesUtils.createLegalTag(LEGAL_TAG_NAME);
createResourceInPartition(tenant);
+ Thread.sleep(10000);
}
@Test
diff --git a/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java b/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java
index a4b5ec17d..9e6da6e80 100644
--- a/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java
+++ b/testing/notification-test-gc/src/test/java/org/opengroup/osdu/notification/api/TestPushEndpointHMAC.java
@@ -81,6 +81,7 @@ public class TestPushEndpointHMAC {
servicesUtils = new ServicesUtils(storageHost, legalHost, testUtils, tenant, groupId);
servicesUtils.createLegalTag(LEGAL_TAG_NAME);
createResourceInPartition(tenant);
+ Thread.sleep(10000);
}
@Test
--
GitLab