diff --git a/NOTICE b/NOTICE
index 10fe1231ce80f633b95b66e803f3060ff69849e6..df5728fc75152173c0784064d1daf1b1495f095f 100644
--- a/NOTICE
+++ b/NOTICE
@@ -34,7 +34,6 @@ Apache-2.0
========================================================================
The following software have components provided under the terms of this license:
-- AHC/Client (from https://repo1.maven.org/maven2/org/asynchttpclient/async-http-client)
- AMQP 1.0 JMS Spring Boot AutoConfiguration (from https://repo1.maven.org/maven2/org/amqphub/spring/amqp-10-jms-spring-boot-autoconfigure)
- AMQP 1.0 JMS Spring Boot Starter (from https://repo1.maven.org/maven2/org/amqphub/spring/amqp-10-jms-spring-boot-starter)
- ASM based accessors helper used by json-smart (from https://urielch.github.io/)
@@ -321,10 +320,11 @@ The following software have components provided under the terms of this license:
- Apache Log4j SLF4J Binding (from https://repo1.maven.org/maven2/org/apache/logging/log4j/log4j-slf4j-impl)
- Apache Log4j to SLF4J Adapter (from https://repo1.maven.org/maven2/org/apache/logging/log4j/log4j-to-slf4j)
- AssertJ Core (from ${project.organization.url}#${project.artifactId})
+- Asynchronous Http Client (from https://repo1.maven.org/maven2/org/asynchttpclient/async-http-client)
- Asynchronous Http Client Netty Utils (from https://repo1.maven.org/maven2/org/asynchttpclient/async-http-client-netty-utils)
- AutoValue Annotations (from https://github.com/google/auto/tree/master/value, https://repo1.maven.org/maven2/com/google/auto/value/auto-value-annotations)
- BSON (from http://bsonspec.org, https://bsonspec.org)
-- BSON Record Codec (from https://www.mongodb.com/)
+- BSON Record Codec (from <https://www.mongodb.com/>, https://www.mongodb.com/)
- Bean Validation API (from http://beanvalidation.org)
- Brave (from https://repo1.maven.org/maven2/io/zipkin/brave/brave)
- Brave Instrumentation: Http Adapters (from https://repo1.maven.org/maven2/io/zipkin/brave/brave-instrumentation-http)
@@ -347,7 +347,7 @@ The following software have components provided under the terms of this license:
- GSON extensions to the Google HTTP Client Library for Java. (from https://repo1.maven.org/maven2/com/google/http-client/google-http-client-gson)
- Google APIs Client Library for Java (from https://repo1.maven.org/maven2/com/google/api-client/google-api-client)
- Google App Engine extensions to the Google HTTP Client Library for Java. (from https://repo1.maven.org/maven2/com/google/http-client/google-http-client-appengine)
-- Google Cloud Core (from https://github.com/googleapis/google-cloud-java/tree/master/google-cloud-clients/google-cloud-core, https://github.com/googleapis/java-core)
+- Google Cloud Core (from https://github.com/googleapis/google-cloud-java, https://github.com/googleapis/google-cloud-java/tree/master/google-cloud-clients/google-cloud-core, https://github.com/googleapis/java-core)
- Google Cloud Core HTTP (from https://github.com/GoogleCloudPlatform/google-cloud-java/tree/master/google-cloud-core-http, https://github.com/googleapis/google-cloud-java/tree/master/google-cloud-clients/google-cloud-core-http, https://github.com/googleapis/java-core)
- Google Cloud Core gRPC (from https://github.com/googleapis/google-cloud-java/tree/master/google-cloud-clients/google-cloud-core-grpc, https://github.com/googleapis/java-core)
- Google Cloud Datastore (from https://github.com/googleapis/google-cloud-java/tree/master/google-cloud-clients/google-cloud-datastore, https://github.com/googleapis/java-datastore)
@@ -387,6 +387,7 @@ The following software have components provided under the terms of this license:
- Jackson datatype: Joda (from http://wiki.fasterxml.com/JacksonModuleJoda, https://github.com/FasterXML/jackson-datatype-joda)
- Jackson datatype: jdk8 (from https://repo1.maven.org/maven2/com/fasterxml/jackson/datatype/jackson-datatype-jdk8)
- Jackson module: Afterburner (from http://wiki.fasterxml.com/JacksonHome, https://github.com/FasterXML/jackson-modules-base)
+- Jackson module: JAXB Annotations (from http://github.com/FasterXML/jackson-module-jaxb-annotations, http://wiki.fasterxml.com/JacksonJAXBAnnotations, https://github.com/FasterXML/jackson-modules-base)
- Jackson-annotations (from http://github.com/FasterXML/jackson, http://wiki.fasterxml.com/JacksonHome, https://github.com/FasterXML/jackson)
- Jackson-core (from http://wiki.fasterxml.com/JacksonHome, https://github.com/FasterXML/jackson-core)
- Jackson-dataformat-XML (from http://wiki.fasterxml.com/JacksonExtensionXmlDataBinding, https://github.com/FasterXML/jackson-dataformat-xml)
@@ -399,7 +400,7 @@ The following software have components provided under the terms of this license:
- Java Native Access (from https://github.com/java-native-access/jna, https://github.com/twall/jna)
- Java Native Access Platform (from https://github.com/java-native-access/jna)
- Java UUID Generator (from http://wiki.fasterxml.com/JugHome)
-- JavaBeans Activation Framework (from https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
+- JavaBeans Activation Framework (from <http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp>, http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp, https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
- Javassist (from http://www.javassist.org/)
- JetBrains Java Annotations (from https://github.com/JetBrains/java-annotations)
- Joda-Time (from http://joda-time.sourceforge.net, http://www.joda.org/joda-time/, https://www.joda.org/joda-time/)
@@ -411,8 +412,8 @@ The following software have components provided under the terms of this license:
- KeePassJava2 :: Simple (from https://repo1.maven.org/maven2/org/linguafranca/pwdb/KeePassJava2-simple)
- Kotlin Stdlib (from https://kotlinlang.org/, https://repo1.maven.org/maven2/org/jetbrains/kotlin/kotlin-stdlib)
- Kotlin Stdlib Common (from https://kotlinlang.org/)
-- Kotlin Stdlib Jdk7 (from https://kotlinlang.org/)
-- Kotlin Stdlib Jdk8 (from https://kotlinlang.org/)
+- Kotlin Stdlib Jdk7 (from <https://kotlinlang.org/>, https://kotlinlang.org/)
+- Kotlin Stdlib Jdk8 (from <https://kotlinlang.org/>, https://kotlinlang.org/)
- LatencyUtils (from http://latencyutils.github.io/LatencyUtils/)
- 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)
@@ -425,6 +426,7 @@ The following software have components provided under the terms of this license:
- Mockito (from http://mockito.org, http://www.mockito.org, https://github.com/mockito/mockito)
- MongoDB Driver (from https://www.mongodb.com/)
- MongoDB Java Driver (from http://mongodb.org/, http://www.mongodb.org, https://www.mongodb.com/)
+- NanoHttpd-Core (from https://repo1.maven.org/maven2/org/nanohttpd/nanohttpd)
- Netty Reactive Streams Implementation (from https://repo1.maven.org/maven2/com/typesafe/netty/netty-reactive-streams)
- Netty/Buffer (from https://repo1.maven.org/maven2/io/netty/netty-buffer)
- Netty/Codec (from https://repo1.maven.org/maven2/io/netty/netty-codec)
@@ -435,10 +437,11 @@ The following software have components provided under the terms of this license:
- Netty/Common (from https://repo1.maven.org/maven2/io/netty/netty-common)
- Netty/Handler (from https://repo1.maven.org/maven2/io/netty/netty-handler)
- Netty/Handler/Proxy (from https://repo1.maven.org/maven2/io/netty/netty-handler-proxy)
-- Netty/Incubator/Codec/Classes/Quic (from https://repo1.maven.org/maven2/io/netty/incubator/netty-incubator-codec-classes-quic)
+- Netty/Incubator/Codec/Classes/Quic (from <https://repo1.maven.org/maven2/io/netty/incubator/netty-incubator-codec-classes-quic>, https://repo1.maven.org/maven2/io/netty/incubator/netty-incubator-codec-classes-quic)
- Netty/Resolver (from https://repo1.maven.org/maven2/io/netty/netty-resolver)
- Netty/Resolver/DNS (from https://repo1.maven.org/maven2/io/netty/netty-resolver-dns)
- Netty/Resolver/DNS/Classes/MacOS (from https://repo1.maven.org/maven2/io/netty/netty-resolver-dns-classes-macos)
+- Netty/TomcatNative [BoringSSL - Static] (from https://github.com/netty/netty-tcnative/netty-tcnative-boringssl-static/)
- Netty/Transport (from https://repo1.maven.org/maven2/io/netty/netty-transport)
- Netty/Transport/Classes/Epoll (from https://repo1.maven.org/maven2/io/netty/netty-transport-classes-epoll)
- Netty/Transport/Classes/KQueue (from https://repo1.maven.org/maven2/io/netty/netty-transport-classes-kqueue)
@@ -511,7 +514,7 @@ The following software have components provided under the terms of this license:
- Spring Web (from http://www.springframework.org, https://github.com/spring-projects/spring-framework, https://repo1.maven.org/maven2/org/springframework/spring-web)
- Spring Web MVC (from https://github.com/spring-projects/spring-framework, https://repo1.maven.org/maven2/org/springframework/spring-webmvc)
- Spring WebFlux (from https://github.com/spring-projects/spring-framework)
-- Swagger UI (from http://webjars.org)
+- Swagger UI (from <http://webjars.org>, http://webjars.org)
- Vavr (from http://vavr.io, https://www.vavr.io)
- Vavr Match (from http://vavr.io)
- Woodstox (from https://github.com/FasterXML/woodstox)
@@ -560,14 +563,13 @@ The following software have components provided under the terms of this license:
- proto-google-common-protos (from https://github.com/googleapis/api-client-staging, https://github.com/googleapis/gapic-generator-java, https://github.com/googleapis/googleapis, https://github.com/googleapis/java-iam/proto-google-common-protos)
- proto-google-iam-v1 (from https://github.com/googleapis/gapic-generator-java, https://github.com/googleapis/googleapis, https://github.com/googleapis/java-iam/proto-google-iam-v1)
- resilience4j (from https://github.com/resilience4j/resilience4j, https://resilience4j.readme.io, ttps://resilience4j.readme.io)
-- spring-security-crypto (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
- spring-security-oauth2-client (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
- spring-security-oauth2-core (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
- spring-security-oauth2-jose (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
- spring-security-test (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
-- springdoc-openapi-common (from https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-common)
-- springdoc-openapi-ui (from https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-ui)
-- springdoc-openapi-webmvc-core (from https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-webmvc-core)
+- springdoc-openapi-common (from <https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-common>, https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-common)
+- springdoc-openapi-ui (from https://github.com/springdoc/springdoc-openapi, https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-ui)
+- springdoc-openapi-webmvc-core (from <https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-webmvc-core>, https://repo1.maven.org/maven2/org/springdoc/springdoc-openapi-webmvc-core)
- swagger-annotations (from https://repo1.maven.org/maven2/io/swagger/core/v3/swagger-annotations, https://repo1.maven.org/maven2/io/swagger/swagger-annotations)
- swagger-core (from https://repo1.maven.org/maven2/io/swagger/core/v3/swagger-core, https://repo1.maven.org/maven2/io/swagger/swagger-core)
- swagger-jaxrs (from https://repo1.maven.org/maven2/io/swagger/swagger-jaxrs)
@@ -583,9 +585,9 @@ BSD-2-Clause
The following software have components provided under the terms of this license:
- API Common (from https://github.com/googleapis, https://github.com/googleapis/api-common-java, https://repo1.maven.org/maven2/com/google/api/api-common)
-- GAX (Google Api eXtensions) for Java (Core) (from https://repo1.maven.org/maven2/com/google/api/gax)
-- GAX (Google Api eXtensions) for Java (HTTP JSON) (from https://repo1.maven.org/maven2/com/google/api/gax-httpjson)
-- GAX (Google Api eXtensions) for Java (gRPC) (from https://repo1.maven.org/maven2/com/google/api/gax-grpc)
+- GAX (Google Api eXtensions) for Java (Core) (from https://github.com/googleapis, https://github.com/googleapis/gax-java, https://repo1.maven.org/maven2/com/google/api/gax)
+- GAX (Google Api eXtensions) for Java (HTTP JSON) (from <https://repo1.maven.org/maven2/com/google/api/gax-httpjson>, https://repo1.maven.org/maven2/com/google/api/gax-httpjson)
+- GAX (Google Api eXtensions) for Java (gRPC) (from <https://repo1.maven.org/maven2/com/google/api/gax-grpc>, https://repo1.maven.org/maven2/com/google/api/gax-grpc)
- Hamcrest (from http://hamcrest.org/JavaHamcrest/)
- Hamcrest Core (from http://hamcrest.org/, http://hamcrest.org/JavaHamcrest/, https://repo1.maven.org/maven2/org/hamcrest/hamcrest-core)
- HdrHistogram (from http://hdrhistogram.github.io/HdrHistogram/)
@@ -602,9 +604,9 @@ The following software have components provided under the terms of this license:
- API Common (from https://github.com/googleapis, https://github.com/googleapis/api-common-java, https://repo1.maven.org/maven2/com/google/api/api-common)
- ASM Core (from http://asm.ow2.io/, http://asm.ow2.org/)
- AspectJ Weaver (from http://www.aspectj.org, https://www.eclipse.org/aspectj/)
-- GAX (Google Api eXtensions) for Java (Core) (from https://repo1.maven.org/maven2/com/google/api/gax)
-- GAX (Google Api eXtensions) for Java (HTTP JSON) (from https://repo1.maven.org/maven2/com/google/api/gax-httpjson)
-- GAX (Google Api eXtensions) for Java (gRPC) (from https://repo1.maven.org/maven2/com/google/api/gax-grpc)
+- GAX (Google Api eXtensions) for Java (Core) (from https://github.com/googleapis, https://github.com/googleapis/gax-java, https://repo1.maven.org/maven2/com/google/api/gax)
+- GAX (Google Api eXtensions) for Java (HTTP JSON) (from <https://repo1.maven.org/maven2/com/google/api/gax-httpjson>, https://repo1.maven.org/maven2/com/google/api/gax-httpjson)
+- GAX (Google Api eXtensions) for Java (gRPC) (from <https://repo1.maven.org/maven2/com/google/api/gax-grpc>, https://repo1.maven.org/maven2/com/google/api/gax-grpc)
- Google APIs Client Library for Java (from https://repo1.maven.org/maven2/com/google/api-client/google-api-client)
- Google Auth Library for Java - Credentials (from https://repo1.maven.org/maven2/com/google/auth/google-auth-library-credentials)
- Google Auth Library for Java - OAuth2 HTTP (from https://repo1.maven.org/maven2/com/google/auth/google-auth-library-oauth2-http)
@@ -620,6 +622,7 @@ The following software have components provided under the terms of this license:
- Microsoft Application Insights Java SDK Web Module (from https://github.com/Microsoft/ApplicationInsights-Java)
- Microsoft Application Insights Log4j 2 Appender (from https://github.com/Microsoft/ApplicationInsights-Java)
- Mockito (from http://mockito.org, http://www.mockito.org, https://github.com/mockito/mockito)
+- NanoHttpd-Core (from https://repo1.maven.org/maven2/org/nanohttpd/nanohttpd)
- Netty/Codec/HTTP (from https://repo1.maven.org/maven2/io/netty/netty-codec-http)
- PostgreSQL JDBC Driver
- Protocol Buffer Java API (from http://code.google.com/p/protobuf, https://repo1.maven.org/maven2/com/google/protobuf/protobuf-java)
@@ -687,7 +690,7 @@ The following software have components provided under the terms of this license:
- HK2 Implementation Utilities (from https://repo1.maven.org/maven2/org/glassfish/hk2/hk2-utils)
- Jakarta Activation API (from https://github.com/eclipse-ee4j/jaf, https://github.com/jakartaee/jaf-api, https://repo1.maven.org/maven2/jakarta/activation/jakarta.activation-api)
- Java Architecture for XML Binding (from http://jaxb.java.net/, https://repo1.maven.org/maven2/javax/xml/bind/jaxb-api)
-- JavaBeans Activation Framework (from https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
+- JavaBeans Activation Framework (from <http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp>, http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp, https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
- OSGi resource locator (from https://repo1.maven.org/maven2/org/glassfish/hk2/osgi-resource-locator)
- ServiceLocator Default Implementation (from https://repo1.maven.org/maven2/org/glassfish/hk2/hk2-locator)
- aopalliance-repackaged (from https://repo1.maven.org/maven2/org/glassfish/hk2/external/aopalliance-repackaged)
@@ -703,7 +706,7 @@ The following software have components provided under the terms of this license:
- HK2 Implementation Utilities (from https://repo1.maven.org/maven2/org/glassfish/hk2/hk2-utils)
- Java Architecture for XML Binding (from http://jaxb.java.net/, https://repo1.maven.org/maven2/javax/xml/bind/jaxb-api)
-- JavaBeans Activation Framework (from https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
+- JavaBeans Activation Framework (from <http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp>, http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp, https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
- ServiceLocator Default Implementation (from https://repo1.maven.org/maven2/org/glassfish/hk2/hk2-locator)
- aopalliance-repackaged (from https://repo1.maven.org/maven2/org/glassfish/hk2/external/aopalliance-repackaged)
- javax.annotation-api (from http://jcp.org/en/jsr/detail?id=250)
@@ -770,7 +773,7 @@ The following software have components provided under the terms of this license:
- Jakarta Annotations API (from https://projects.eclipse.org/projects/ee4j.ca)
- Java Architecture for XML Binding (from http://jaxb.java.net/, https://repo1.maven.org/maven2/javax/xml/bind/jaxb-api)
-- JavaBeans Activation Framework (from https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
+- JavaBeans Activation Framework (from <http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp>, http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp, https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
- OSGi resource locator (from https://repo1.maven.org/maven2/org/glassfish/hk2/osgi-resource-locator)
- RabbitMQ Java Client (from http://www.rabbitmq.com, https://www.rabbitmq.com)
- tomcat-embed-core (from http://tomcat.apache.org/)
@@ -786,7 +789,7 @@ The following software have components provided under the terms of this license:
- Jakarta Bean Validation API (from https://beanvalidation.org)
- Jakarta XML Binding API (from https://repo1.maven.org/maven2/jakarta/xml/bind/jakarta.xml.bind-api, https://repo1.maven.org/maven2/org/jboss/spec/javax/xml/bind/jboss-jaxb-api_2.3_spec)
- Java Architecture for XML Binding (from http://jaxb.java.net/, https://repo1.maven.org/maven2/javax/xml/bind/jaxb-api)
-- JavaBeans Activation Framework (from https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
+- JavaBeans Activation Framework (from <http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp>, http://java.sun.com/javase/technologies/desktop/javabeans/jaf/index.jsp, https://repo1.maven.org/maven2/com/sun/activation/javax.activation)
- OSGi resource locator (from https://repo1.maven.org/maven2/org/glassfish/hk2/osgi-resource-locator)
- ServiceLocator Default Implementation (from https://repo1.maven.org/maven2/org/glassfish/hk2/hk2-locator)
- aopalliance-repackaged (from https://repo1.maven.org/maven2/org/glassfish/hk2/external/aopalliance-repackaged)
@@ -818,7 +821,7 @@ ISC
The following software have components provided under the terms of this license:
- Java Native Access Platform (from https://github.com/java-native-access/jna)
-- spring-security-crypto (from http://spring.io/spring-security, https://spring.io/projects/spring-security, https://spring.io/spring-security)
+- 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)
========================================================================
ImageMagick
@@ -922,7 +925,7 @@ The following software have components provided under the terms of this license:
- msal4j (from https://github.com/AzureAD/microsoft-authentication-library-for-java)
- msal4j-persistence-extension (from https://github.com/AzureAD/microsoft-authentication-extensions-for-java)
- qpid-proton-j-extensions (from https://github.com/Azure/qpid-proton-j-extensions)
-- webjars-locator-core (from http://webjars.org)
+- webjars-locator-core (from <http://webjars.org>, http://webjars.org)
========================================================================
MPL-1.1
@@ -990,6 +993,7 @@ The following software have components provided under the terms of this license:
- JBoss Logging 3 (from http://www.jboss.org)
- JSON in Java (from https://github.com/douglascrockford/JSON-java)
- LatencyUtils (from http://latencyutils.github.io/LatencyUtils/)
+- Microsoft Azure client library for Blob Storage (from https://github.com/Azure/azure-sdk-for-java)
- PostgreSQL JDBC Driver
========================================================================
diff --git a/docs/tutorial/CollaborationContext.md b/docs/tutorial/CollaborationContext.md
index 5ce368373905fb7a6efc7c8c0e1f05bb5bb73ee1..400000a36cb1cb6cee809909c2cdbefb3f9e7853 100644
--- a/docs/tutorial/CollaborationContext.md
+++ b/docs/tutorial/CollaborationContext.md
@@ -24,7 +24,7 @@ Please refer to this MR for [implementation of Azure](https://community.opengrou
Consumers who want to integrate with record change messages that include changes made within a collaboration context need to register the records to the new topic "recordstopic-v2". Refer the [DataNotification.md](https://community.opengroup.org/osdu/platform/system/notification/-/blob/master/docs/tutorial/DataNotification.md) file for details about the recordstopics-v2.
-This topic replaces the current record changed topic and receives both collaboration and non collaboration messages when the collaborations feature flag is enabled.
+This topic exists in addition to the current record changed topic and receives both collaboration and non collaboration messages when the collaborations feature flag is enabled.
The current record changed topic however does not receive messages when collaboration context is provided. Meaning, the original functionality of storage should not be changed if collaboration context is not provided.
@@ -45,6 +45,8 @@ The message contains the collaboration context header as an atribute when a chan
"data": [
{
"id": "opendes:wellbore:f213e42d5fa848f592917a8df7fed132",
+ "version": "1617915304347525",
+ "modifiedBy": "abc@xyz.com",
"kind": "common:welldb:wellbore:1.0.0",
"op": "create"
}
diff --git a/pom.xml b/pom.xml
index b718a551e1f4120a73bd2b8529e86283b6862e95..5b0d660b90e2774a4e53605c03c74ad169cf5b14 100644
--- a/pom.xml
+++ b/pom.xml
@@ -27,7 +27,7 @@
<maven.compiler.source>1.8</maven.compiler.source>
<docker.image.prefix>opendes</docker.image.prefix>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <os-core-common.version>0.19.0-rc3</os-core-common.version>
+ <os-core-common.version>0.19.0-rc8</os-core-common.version>
<jackson-databind.version>2.14.0</jackson-databind.version>
<jackson.version>2.14.0</jackson.version>
<netty.version>4.1.51.Final</netty.version>
diff --git a/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/MessageBusImpl.java b/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/MessageBusImpl.java
index 40c9d017d362cc1e8732c5e8d059d0a3658c4025..9127ceee6f91672a7373b0af2fb8004e7438c944 100644
--- a/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/MessageBusImpl.java
+++ b/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/MessageBusImpl.java
@@ -22,6 +22,7 @@ import com.amazonaws.services.sns.AmazonSNS;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.aws.sns.AmazonSNSConfig;
@@ -57,7 +58,7 @@ public class MessageBusImpl implements IMessageBus {
}
@Override
- public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages) {
+ public void publishMessage(DpsHeaders headers, PubSubInfo... messages) {
final int BATCH_SIZE = 50;
Gson gson = new Gson();
for (int i =0; i < messages.length; i+= BATCH_SIZE){
@@ -91,4 +92,9 @@ public class MessageBusImpl implements IMessageBus {
}
}
+
+ @Override
+ public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, RecordChangedV2... messages) {
+ //TODO: To be implemented by aws provider
+ }
}
diff --git a/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/jobs/LegalComplianceChangeServiceAWSImpl.java b/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/jobs/LegalComplianceChangeServiceAWSImpl.java
index 08ec7303a0ab0512bf0712c4efc382266a77f8c4..3bc8403b89cea810821b3cb70b6db6fbb3ffde6b 100644
--- a/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/jobs/LegalComplianceChangeServiceAWSImpl.java
+++ b/provider/storage-aws/src/main/java/org/opengroup/osdu/storage/provider/aws/jobs/LegalComplianceChangeServiceAWSImpl.java
@@ -98,7 +98,7 @@ public class LegalComplianceChangeServiceAWSImpl implements ILegalComplianceChan
this.auditLogger.updateRecordsComplianceStateSuccess(
singletonList("[" + recordsId.toString() + "]"));
- this.storageMessageBus.publishMessage(Optional.empty(), headers, pubsubInfos);
+ this.storageMessageBus.publishMessage(headers, pubsubInfos);
} while (cursor != null);
}
diff --git a/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/LegalComplianceChangeServiceAWSImplTest.java b/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/LegalComplianceChangeServiceAWSImplTest.java
index 394a5a2fd6eabf21b0d811d9865a93304a591822..1b250ef7eb5fa7cc9a35e712bc71808950ed7d23 100644
--- a/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/LegalComplianceChangeServiceAWSImplTest.java
+++ b/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/LegalComplianceChangeServiceAWSImplTest.java
@@ -151,7 +151,7 @@ public class LegalComplianceChangeServiceAWSImplTest {
// that storageMessageBus publishMessage is called with the right pubsubinfos
Mockito.verify(storageMessageBus, Mockito.times(2))
- .publishMessage(Mockito.eq(Optional.empty()), Mockito.any(), pubSubArg.capture());
+ .publishMessage(Mockito.any(), pubSubArg.capture());
List<PubSubInfo[]> captured = pubSubArg.getAllValues();
Object incompliantPubSubObj = captured.get(0);
PubSubInfo incompliantPubSub = (PubSubInfo) incompliantPubSubObj;
diff --git a/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/MessageBusImplTest.java b/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/MessageBusImplTest.java
index 8989f5f81e0f5134de61a4df92964f690fa1590c..5411b9758653431457da838a48d19d8dccebaa8f 100644
--- a/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/MessageBusImplTest.java
+++ b/provider/storage-aws/src/test/java/org/opengroup/osdu/storage/provider/aws/api/MessageBusImplTest.java
@@ -106,7 +106,7 @@ public class MessageBusImplTest {
}
// act
- messageBus.publishMessage(Optional.empty(), headers, message);
+ messageBus.publishMessage(headers, message);
// assert
Mockito.verify(snsClient, Mockito.times(1)).publish(Mockito.eq(publishRequest));
diff --git a/provider/storage-azure/pom.xml b/provider/storage-azure/pom.xml
index d99dee01c6b789987a27bcae918dfa05ac661662..6ebdc2fd5776f65ba80aadad67409703673e15e7 100644
--- a/provider/storage-azure/pom.xml
+++ b/provider/storage-azure/pom.xml
@@ -156,7 +156,7 @@
<dependency>
<groupId>org.opengroup.osdu</groupId>
<artifactId>os-core-common</artifactId>
- <version>0.19.0-rc3</version>
+ <version>0.19.0-rc8</version>
</dependency>
<dependency>
<groupId>org.opengroup.osdu</groupId>
diff --git a/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/MessageBusImpl.java b/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/MessageBusImpl.java
index 58425ccd51ea21cc6ef6a04486e36fe2eb289ef0..069b488b1312720e3b94cd4606bfc19c6dc732b4 100644
--- a/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/MessageBusImpl.java
+++ b/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/MessageBusImpl.java
@@ -16,10 +16,10 @@ package org.opengroup.osdu.storage.provider.azure;
import org.opengroup.osdu.azure.publisherFacade.MessagePublisher;
import org.opengroup.osdu.azure.publisherFacade.PublisherInfo;
-import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.azure.di.EventGridConfig;
import org.opengroup.osdu.storage.provider.azure.di.ServiceBusConfig;
import org.opengroup.osdu.storage.provider.azure.di.PublisherConfig;
@@ -39,52 +39,41 @@ public class MessageBusImpl implements IMessageBus {
private MessagePublisher messagePublisher;
@Autowired
private PublisherConfig publisherConfig;
- @Autowired
- private IFeatureFlag iCollaborationFeatureFlag;
- private static final String COLLABORATIONS_FEATURE_NAME = "collaborations-enabled";
-
@Override
- public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages) {
- if (iCollaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
- publishMessageToRecordsTopicV2(collaborationContext, headers, messages);
- if (collaborationContext.isPresent()) {
- return;
- }
- }
+ public void publishMessage(DpsHeaders headers, PubSubInfo... messages) {
+
// The batch size is same for both Event grid and Service bus.
final int BATCH_SIZE = Integer.parseInt(publisherConfig.getPubSubBatchSize());
for(int i = 0; i < messages.length; i += BATCH_SIZE) {
PubSubInfo[] batch = Arrays.copyOfRange(messages, i, Math.min(messages.length, i + BATCH_SIZE));
- PublisherInfo publisherInfo = PublisherInfo.builder()
- .batch(batch)
- .eventGridTopicName(eventGridConfig.getEventGridTopic())
- .eventGridEventSubject(eventGridConfig.getEventSubject())
- .eventGridEventType(eventGridConfig.getEventType())
- .eventGridEventDataVersion(eventGridConfig.getEventDataVersion())
- .serviceBusTopicName(serviceBusConfig.getServiceBusTopic())
- .build();
-
- messagePublisher.publishMessage(headers, publisherInfo, collaborationContext);
+ PublisherInfo publisherInfo = getPartialPublisherInfo();
+ publisherInfo.setBatch(batch);
+ publisherInfo.setServiceBusTopicName(serviceBusConfig.getServiceBusTopic());
+ messagePublisher.publishMessage(headers, publisherInfo, Optional.empty());
}
}
- public void publishMessageToRecordsTopicV2(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages) {
+ public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, RecordChangedV2... messages) {
// The batch size is same for both Event grid and Service bus.
final int BATCH_SIZE = Integer.parseInt(publisherConfig.getPubSubBatchSize());
for (int i = 0; i < messages.length; i += BATCH_SIZE) {
String messageId = String.format("%s-%d",headers.getCorrelationId(), i);
- PubSubInfo[] batch = Arrays.copyOfRange(messages, i, Math.min(messages.length, i + BATCH_SIZE));
- PublisherInfo publisherInfo = PublisherInfo.builder()
- .batch(batch)
- .eventGridTopicName(eventGridConfig.getEventGridTopic())
- .eventGridEventSubject(eventGridConfig.getEventSubject())
- .eventGridEventType(eventGridConfig.getEventType())
- .eventGridEventDataVersion(eventGridConfig.getEventDataVersion())
- .serviceBusTopicName(serviceBusConfig.getServiceBusRecordsEventTopic())
- .messageId(messageId)
- .build();
+ RecordChangedV2[] batch = Arrays.copyOfRange(messages, i, Math.min(messages.length, i + BATCH_SIZE));
+ PublisherInfo publisherInfo = getPartialPublisherInfo();
+ publisherInfo.setBatch(batch);
+ publisherInfo.setMessageId(messageId);
+ publisherInfo.setServiceBusTopicName(serviceBusConfig.getServiceBusRecordsEventTopic());
messagePublisher.publishMessage(headers, publisherInfo, collaborationContext);
}
}
+
+ private PublisherInfo getPartialPublisherInfo() {
+ return PublisherInfo.builder()
+ .eventGridTopicName(eventGridConfig.getEventGridTopic())
+ .eventGridEventSubject(eventGridConfig.getEventSubject())
+ .eventGridEventType(eventGridConfig.getEventType())
+ .eventGridEventDataVersion(eventGridConfig.getEventDataVersion())
+ .build();
+ }
}
diff --git a/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/service/LegalComplianceChangeServiceAzureImpl.java b/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/service/LegalComplianceChangeServiceAzureImpl.java
index b9294337f9f88bd9b24c6509da25fac1c6841bc2..930791edc7ebe4ba81ddb660e48877a1d90f0276 100644
--- a/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/service/LegalComplianceChangeServiceAzureImpl.java
+++ b/provider/storage-azure/src/main/java/org/opengroup/osdu/storage/provider/azure/service/LegalComplianceChangeServiceAzureImpl.java
@@ -15,7 +15,9 @@
package org.opengroup.osdu.storage.provider.azure.service;
import org.apache.http.HttpStatus;
+import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.model.http.AppException;
+import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.indexer.OperationType;
import org.opengroup.osdu.core.common.model.legal.LegalCompliance;
@@ -27,6 +29,7 @@ import org.opengroup.osdu.core.common.model.legal.jobs.LegalTagChangedCollection
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
import org.opengroup.osdu.core.common.model.storage.RecordMetadata;
import org.opengroup.osdu.core.common.model.storage.RecordState;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.azure.MessageBusImpl;
import org.opengroup.osdu.storage.provider.azure.cache.LegalTagCache;
import org.opengroup.osdu.storage.provider.interfaces.IRecordsMetadataRepository;
@@ -37,11 +40,12 @@ import org.springframework.stereotype.Component;
import java.util.*;
+import static org.opengroup.osdu.storage.util.StringConstants.COLLABORATIONS_FEATURE_NAME;
+
@Component
public class LegalComplianceChangeServiceAzureImpl implements ILegalComplianceChangeService {
private static final String LEGAL_STATUS_INVALID = "Invalid";
-
- private final static Logger LOGGER = LoggerFactory.getLogger(LegalComplianceChangeServiceAzureImpl.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(LegalComplianceChangeServiceAzureImpl.class);
@Autowired
private IRecordsMetadataRepository recordsRepo;
@Autowired
@@ -50,12 +54,15 @@ public class LegalComplianceChangeServiceAzureImpl implements ILegalComplianceCh
private LegalTagCache legalTagCache;
@Autowired
private MessageBusImpl pubSubclient;
+ @Autowired
+ private IFeatureFlag collaborationFeatureFlag;
+
@Override
public Map<String, LegalCompliance> updateComplianceOnRecords(LegalTagChangedCollection legalTagsChanged,
DpsHeaders headers) throws ComplianceUpdateStoppedException {
Map<String, LegalCompliance> output = new HashMap<>();
-
+ Optional<CollaborationContext> collaborationContext = Optional.empty();
for (LegalTagChanged lt : legalTagsChanged.getStatusChangedTags()) {
ComplianceChangeInfo complianceChangeInfo = this.getComplianceChangeInfo(lt);
if (complianceChangeInfo == null) {
@@ -71,6 +78,7 @@ public class LegalComplianceChangeServiceAzureImpl implements ILegalComplianceCh
if (results.getValue() != null && !results.getValue().isEmpty()) {
List<RecordMetadata> recordsMetadata = results.getValue();
PubSubInfo[] pubsubInfos = this.updateComplianceStatus(complianceChangeInfo, recordsMetadata, output);
+ RecordChangedV2[] recordsChangedV2s = this.updateComplianceStatusRecordsChangedV2(complianceChangeInfo, recordsMetadata, output);
try {
this.recordsRepo.createOrUpdate(recordsMetadata, Optional.empty());
} catch (Exception e) {
@@ -81,7 +89,12 @@ public class LegalComplianceChangeServiceAzureImpl implements ILegalComplianceCh
for (RecordMetadata recordMetadata : recordsMetadata) {
recordIds.add(recordMetadata.getId());
}
- this.pubSubclient.publishMessage(Optional.empty(), headers, pubsubInfos);
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ this.pubSubclient.publishMessage(collaborationContext, headers, recordsChangedV2s);
+ }
+ if (!collaborationContext.isPresent()) {
+ this.pubSubclient.publishMessage(headers, pubsubInfos);
+ }
logOnSucceedUpdateRecords(lt, recordIds);
}
} while (cursor != null);
@@ -106,6 +119,29 @@ public class LegalComplianceChangeServiceAzureImpl implements ILegalComplianceCh
return pubsubInfo;
}
+ private RecordChangedV2[] updateComplianceStatusRecordsChangedV2(ComplianceChangeInfo complianceChangeInfo,
+ List<RecordMetadata> recordMetadata, Map<String, LegalCompliance> output) {
+
+ RecordChangedV2[] recordChangedV2 = new RecordChangedV2[recordMetadata.size()];
+
+ int i = 0;
+ for (RecordMetadata rm : recordMetadata) {
+ rm.getLegal().setStatus(complianceChangeInfo.getNewState());
+ rm.setStatus(complianceChangeInfo.getNewRecordState());
+ recordChangedV2[i] = RecordChangedV2.builder()
+ .id(rm.getId())
+ .version(rm.getLatestVersion())
+ .modifiedBy(rm.getModifyUser())
+ .kind(rm.getKind())
+ .op(complianceChangeInfo.getPubSubEvent())
+ .build();
+ output.put(rm.getId(), complianceChangeInfo.getNewState());
+ i++;
+ }
+
+ return recordChangedV2;
+ }
+
private ComplianceChangeInfo getComplianceChangeInfo(LegalTagChanged lt) {
ComplianceChangeInfo output = null;
@@ -115,7 +151,7 @@ public class LegalComplianceChangeServiceAzureImpl implements ILegalComplianceCh
this.legalTagCache.delete(lt.getChangedTagName());
output = new ComplianceChangeInfo(LegalCompliance.incompliant, OperationType.delete, RecordState.deleted);
} else {
- this.LOGGER.warn(String.format("Unknown LegalTag compliance status received %s %s",
+ LOGGER.warn(String.format("Unknown LegalTag compliance status received %s %s",
lt.getChangedTagStatus(), lt.getChangedTagName()));
}
return output;
diff --git a/provider/storage-azure/src/test/java/org/opengroup/osdu/storage/provider/azure/MessageBusImplTest.java b/provider/storage-azure/src/test/java/org/opengroup/osdu/storage/provider/azure/MessageBusImplTest.java
index 8c753467efaec91d9efaa8bdb1822697507e75be..3b0e4bc5b0b2fefa06839c7b49d369af87fdd8d1 100644
--- a/provider/storage-azure/src/test/java/org/opengroup/osdu/storage/provider/azure/MessageBusImplTest.java
+++ b/provider/storage-azure/src/test/java/org/opengroup/osdu/storage/provider/azure/MessageBusImplTest.java
@@ -27,6 +27,7 @@ import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.azure.di.EventGridConfig;
import org.opengroup.osdu.storage.provider.azure.di.ServiceBusConfig;
import org.opengroup.osdu.storage.provider.azure.di.PublisherConfig;
@@ -71,7 +72,6 @@ public class MessageBusImplTest {
initMocks(this);
doReturn("10").when(publisherConfig).getPubSubBatchSize();
doReturn(TOPIC_NAME).when(eventGridConfig).getEventGridTopic();
- doReturn(TOPIC_NAME).when(serviceBusConfig).getServiceBusTopic();
doReturn(RECORDS_CHANGED_EVENT_SUBJECT).when(eventGridConfig).getEventSubject();
doReturn(RECORDS_CHANGED_EVENT_DATA_VERSION).when(eventGridConfig).getEventDataVersion();
doReturn(RECORDS_CHANGED_EVENT_TYPE).when(eventGridConfig).getEventType();
@@ -88,24 +88,21 @@ public class MessageBusImplTest {
for (int i = 0; i < ids.length; ++i) {
pubSubInfo[i] = getPubsInfo(ids[i], kinds[i]);
}
- sut.publishMessage(Optional.empty(), dpsHeaders, pubSubInfo);
+ sut.publishMessage(dpsHeaders, pubSubInfo);
}
@Test
public void should_publishToOnlyRecordsEventTopic_WhenCollaborationContextIsProvided() {
- PubSubInfo[] pubSubInfo = setup();
- when(iCollaborationFeatureFlag.isFeatureEnabled(FEATURE_NAME)).thenReturn(true);
- sut.publishMessage(COLLABORATION_CONTEXT, dpsHeaders, pubSubInfo);
- verify(messagePublisher, times(1)).publishMessage(any(), any(), any());
+ RecordChangedV2[] recordChangedV2s = setUpRecordsChangedV2();
+ sut.publishMessage(COLLABORATION_CONTEXT, dpsHeaders, recordChangedV2s);
+ verify(messagePublisher, times(1)).publishMessage(any(), any(), eq(COLLABORATION_CONTEXT));
}
@Test
public void should_publishToBothTopics_WhenCollaborationContextIsNotProvided() {
- PubSubInfo[] pubSubInfo = setup();
- when(iCollaborationFeatureFlag.isFeatureEnabled(FEATURE_NAME)).thenReturn(true);
-
- sut.publishMessage(Optional.empty(), dpsHeaders, pubSubInfo);
- verify(messagePublisher, times(2)).publishMessage(any(), any(), any());
+ PubSubInfo[] pubSubInfo = setupPubSubInfo();
+ sut.publishMessage(dpsHeaders, pubSubInfo);
+ verify(messagePublisher, times(1)).publishMessage(any(), any(), eq(Optional.empty()));
}
private PubSubInfo getPubsInfo(String id, String kind) {
@@ -115,14 +112,32 @@ public class MessageBusImplTest {
return pubSubInfo;
}
- private PubSubInfo[] setup() {
+ private RecordChangedV2 getRecordChangedV2(String id, String kind) {
+ RecordChangedV2 recordChangedV2 = new RecordChangedV2();
+ recordChangedV2.setId(id);
+ recordChangedV2.setKind(kind);
+ recordChangedV2.setVersion(1L);
+ return recordChangedV2;
+ }
+
+ private PubSubInfo[] setupPubSubInfo() {
String[] ids = {"id1", "id2", "id3", "id4", "id5"};
String[] kinds = {"kind1", "kind2", "kind3", "kind4", "kind5"};
- doReturn("id").when(dpsHeaders).getCorrelationId();
PubSubInfo[] pubSubInfo = new PubSubInfo[5];
for (int i = 0; i < ids.length; ++i) {
pubSubInfo[i] = getPubsInfo(ids[i], kinds[i]);
}
return pubSubInfo;
}
+
+ private RecordChangedV2[] setUpRecordsChangedV2() {
+ String[] ids = {"id1", "id2", "id3", "id4", "id5"};
+ String[] kinds = {"kind1", "kind2", "kind3", "kind4", "kind5"};
+ doReturn("id").when(dpsHeaders).getCorrelationId();
+ RecordChangedV2[] recordChangedV2s = new RecordChangedV2[5];
+ for (int i = 0; i < ids.length; ++i) {
+ recordChangedV2s[i] = getRecordChangedV2(ids[i], kinds[i]);
+ }
+ return recordChangedV2s;
+ }
}
diff --git a/provider/storage-byoc/src/main/java/org/opengroup/osdu/storage/provider/byoc/MessageBusImpl.java b/provider/storage-byoc/src/main/java/org/opengroup/osdu/storage/provider/byoc/MessageBusImpl.java
index edce3197e5a56a535010ccb4c17759139405cb53..a65d19979ca6b52bc8ad51d932a18f8ebe10c4f0 100644
--- a/provider/storage-byoc/src/main/java/org/opengroup/osdu/storage/provider/byoc/MessageBusImpl.java
+++ b/provider/storage-byoc/src/main/java/org/opengroup/osdu/storage/provider/byoc/MessageBusImpl.java
@@ -18,6 +18,7 @@ import com.google.gson.Gson;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.springframework.stereotype.Component;
@@ -32,7 +33,7 @@ public class MessageBusImpl implements IMessageBus {
private static PriorityQueue<String> memQueue = new PriorityQueue<>();
- public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages)
+ public void publishMessage(DpsHeaders headers, PubSubInfo... messages)
{
final int BATCH_SIZE = 50;
Map<String, String> message = new HashMap<>();
@@ -50,5 +51,10 @@ public class MessageBusImpl implements IMessageBus {
memQueue.add(gson.toJson(message));
}
}
+
+ @Override
+ public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, RecordChangedV2... messages) {
+ //TODO: to be implemented for byoc provider
+ }
}
diff --git a/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/LegalComplianceChangeServiceGcpImpl.java b/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/LegalComplianceChangeServiceGcpImpl.java
index 4619215ac496f5e9b3be81607d9daf5450ea07d5..5ce1ec17b84e6822bdbae2a207b8ce1f49400bf4 100644
--- a/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/LegalComplianceChangeServiceGcpImpl.java
+++ b/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/LegalComplianceChangeServiceGcpImpl.java
@@ -86,7 +86,7 @@ public class LegalComplianceChangeServiceGcpImpl implements ILegalComplianceChan
recordsId.append(", ").append(recordMetadata.getId());
}
this.recordsRepo.createOrUpdate(recordsMetadata, Optional.empty());
- this.messageBus.publishMessage(Optional.empty(), headers, pubsubInfos);
+ this.messageBus.publishMessage(headers, pubsubInfos);
this.auditLogger.updateRecordsComplianceStateSuccess(
singletonList("[" + recordsId.substring(2) + "]"));
results = this.recordsRepo.queryByLegal(lt.getChangedTagName(), complianceChangeInfo.getCurrent(), 500);
diff --git a/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/web/pubsub/OqmPubSub.java b/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/web/pubsub/OqmPubSub.java
index e8b7a86d0bea3c7e72031bb98ddbd0c71e7500eb..aec2108663466d262ec92c394d08549a734d0021 100644
--- a/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/web/pubsub/OqmPubSub.java
+++ b/provider/storage-gc/src/main/java/org/opengroup/osdu/storage/provider/gcp/web/pubsub/OqmPubSub.java
@@ -32,6 +32,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.OqmMessage;
import org.opengroup.osdu.core.gcp.oqm.model.OqmTopic;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.gcp.web.config.GcpAppServiceConfig;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.springframework.stereotype.Repository;
@@ -53,7 +54,7 @@ public class OqmPubSub implements IMessageBus {
}
@Override
- public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages) {
+ public void publishMessage(DpsHeaders headers, PubSubInfo... messages) {
OqmDestination oqmDestination = OqmDestination.builder().partitionId(headers.getPartitionId()).build();
@@ -75,4 +76,9 @@ public class OqmPubSub implements IMessageBus {
driver.publish(oqmMessage, oqmTopic, oqmDestination);
}
}
+
+ @Override
+ public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, RecordChangedV2... messages) {
+ //TODO: to be implemented by gcp provider
+ }
}
diff --git a/provider/storage-gc/src/test/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/stub/OqmPubSubStub.java b/provider/storage-gc/src/test/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/stub/OqmPubSubStub.java
index 1dc3935774ca7c071ec6fc5d8fb8b395a5a85347..acd86722cb4cb29186559e3b466aacfac61c0d7f 100644
--- a/provider/storage-gc/src/test/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/stub/OqmPubSubStub.java
+++ b/provider/storage-gc/src/test/java/org/opengroup/osdu/storage/provider/gcp/messaging/jobs/stub/OqmPubSubStub.java
@@ -27,6 +27,7 @@ import lombok.Getter;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
@Getter
@@ -35,7 +36,12 @@ public class OqmPubSubStub implements IMessageBus {
private final List<Map<DpsHeaders, PubSubInfo[]>> collector = new ArrayList<>();
@Override
- public synchronized void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages) {
+ public synchronized void publishMessage(DpsHeaders headers, PubSubInfo... messages) {
collector.add(ImmutableMap.of(DpsHeaders.createFromMap(headers.getHeaders()), messages));
}
+
+ @Override
+ public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, RecordChangedV2... messages) {
+ //TODO: to be implemented by gcp provider
+ }
}
diff --git a/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/MessageBusImpl.java b/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/MessageBusImpl.java
index 434f18b8d1b9019edc0ba8cf9b2e813a23b77044..3656fbf757475b510f39d1d34647860240cba054 100644
--- a/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/MessageBusImpl.java
+++ b/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/MessageBusImpl.java
@@ -14,6 +14,7 @@ import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
import org.opengroup.osdu.core.ibm.messagebus.IMessageFactory;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.springframework.stereotype.Component;
@@ -25,7 +26,7 @@ public class MessageBusImpl implements IMessageBus {
@Inject
IMessageFactory mq;
- public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages) {
+ public void publishMessage(DpsHeaders headers, PubSubInfo... messages) {
final int BATCH_SIZE = 50;
Map<String, String> message = new HashMap<>();
@@ -46,4 +47,9 @@ public class MessageBusImpl implements IMessageBus {
}
+ @Override
+ public void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, RecordChangedV2... messages) {
+ //TODO: to be implemented by ibm provider
+ }
+
}
diff --git a/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/jobs/LegalComplianceChangeServiceImpl.java b/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/jobs/LegalComplianceChangeServiceImpl.java
index 9b75ffdefe8ae91fc91ff67c95e8f39c3475cfd4..32d0a7011287f2fabe676e5203233512b31ae118 100644
--- a/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/jobs/LegalComplianceChangeServiceImpl.java
+++ b/provider/storage-ibm/src/main/java/org/opengroup/osdu/storage/provider/ibm/jobs/LegalComplianceChangeServiceImpl.java
@@ -93,7 +93,7 @@ public class LegalComplianceChangeServiceImpl implements ILegalComplianceChangeS
this.auditLogger.updateRecordsComplianceStateSuccess(
singletonList("[" + recordsId.toString().substring(2) + "]"));
- this.storageMessageBus.publishMessage(Optional.empty(), headers, pubsubInfos);
+ this.storageMessageBus.publishMessage(headers, pubsubInfos);
} while (cursor != null);
}
diff --git a/storage-core/pom.xml b/storage-core/pom.xml
index cc955c3e89a45ed90d78bc3e4c6185868e4c414a..24fafb548212a82290d7703c705aada2c416688e 100644
--- a/storage-core/pom.xml
+++ b/storage-core/pom.xml
@@ -34,7 +34,7 @@
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
<springfox-version>2.7.0</springfox-version>
- <os-core-common.version>0.19.0-rc3</os-core-common.version>
+ <os-core-common.version>0.19.0-rc8</os-core-common.version>
<!--
<springfox-version>2.9.2</springfox-version>
-->
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/model/RecordChangedV2.java b/storage-core/src/main/java/org/opengroup/osdu/storage/model/RecordChangedV2.java
new file mode 100644
index 0000000000000000000000000000000000000000..188d8de48ee108e79f5018e122a836f04992b1d1
--- /dev/null
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/model/RecordChangedV2.java
@@ -0,0 +1,32 @@
+package org.opengroup.osdu.storage.model;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+import org.opengroup.osdu.core.common.model.indexer.OperationType;
+
+@Data
+@SuperBuilder
+@AllArgsConstructor
+@NoArgsConstructor
+public class RecordChangedV2 {
+ private String id;
+ private Long version;
+ private String modifiedBy;
+ private String kind;
+ private OperationType op;
+
+ /**
+ * This specifies the changes that have been made to the record
+ * e.g. "data" "data metadata" "data metadata+" "metadata-" ...
+ */
+ private String recordBlocks;
+
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @Builder.Default
+ private String previousVersionKind = null;
+
+}
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/model/RecordChangedV2Delete.java b/storage-core/src/main/java/org/opengroup/osdu/storage/model/RecordChangedV2Delete.java
new file mode 100644
index 0000000000000000000000000000000000000000..54c8a71108d8abbaf846c0f3f3bd59e69e088d7f
--- /dev/null
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/model/RecordChangedV2Delete.java
@@ -0,0 +1,16 @@
+package org.opengroup.osdu.storage.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+import org.opengroup.osdu.core.common.model.indexer.DeletionType;
+
+@Data
+@SuperBuilder
+@AllArgsConstructor
+@NoArgsConstructor
+public class RecordChangedV2Delete extends RecordChangedV2 {
+ private DeletionType deletionType;
+
+}
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/provider/interfaces/IMessageBus.java b/storage-core/src/main/java/org/opengroup/osdu/storage/provider/interfaces/IMessageBus.java
index a22da0d63f0ec10203ec299faeb42f76657268c4..3995362c8cf3beec69c0da05e39a37e6469c2f49 100644
--- a/storage-core/src/main/java/org/opengroup/osdu/storage/provider/interfaces/IMessageBus.java
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/provider/interfaces/IMessageBus.java
@@ -17,10 +17,13 @@ package org.opengroup.osdu.storage.provider.interfaces;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import java.util.Optional;
public interface IMessageBus {
- void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, PubSubInfo... messages);
+ void publishMessage(DpsHeaders headers, PubSubInfo... messages);
+
+ void publishMessage(Optional<CollaborationContext> collaborationContext, DpsHeaders headers, RecordChangedV2... messages);
}
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/service/PersistenceServiceImpl.java b/storage-core/src/main/java/org/opengroup/osdu/storage/service/PersistenceServiceImpl.java
index e8366e171dcd6e10c8560ea808cc707dc577cb1d..79045124dd9c671362aed2761d32e0c64fc78399 100644
--- a/storage-core/src/main/java/org/opengroup/osdu/storage/service/PersistenceServiceImpl.java
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/service/PersistenceServiceImpl.java
@@ -17,6 +17,7 @@ package org.opengroup.osdu.storage.service;
import com.google.common.base.Strings;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.http.HttpStatus;
+import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
import org.opengroup.osdu.core.common.model.entitlements.Acl;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
@@ -24,6 +25,7 @@ import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.indexer.OperationType;
import org.opengroup.osdu.core.common.model.storage.*;
import org.opengroup.osdu.core.common.model.http.AppException;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.interfaces.ICloudStorage;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.opengroup.osdu.storage.provider.interfaces.IRecordsMetadataRepository;
@@ -36,6 +38,8 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
+import static org.opengroup.osdu.storage.util.StringConstants.COLLABORATIONS_FEATURE_NAME;
+
@Service
public class PersistenceServiceImpl implements PersistenceService {
@@ -53,7 +57,8 @@ public class PersistenceServiceImpl implements PersistenceService {
@Autowired
private JaxRsDpsLog logger;
-
+ @Autowired
+ private IFeatureFlag collaborationFeatureFlag;
@Override
public void persistRecordBatch(TransferBatch transfer, Optional<CollaborationContext> collaborationContext) {
@@ -61,26 +66,38 @@ public class PersistenceServiceImpl implements PersistenceService {
List<RecordMetadata> recordsMetadata = new ArrayList<>(recordsProcessing.size());
PubSubInfo[] pubsubInfo = new PubSubInfo[recordsProcessing.size()];
+ RecordChangedV2[] recordChangedV2 = new RecordChangedV2[recordsProcessing.size()];
for (int i = 0; i < recordsProcessing.size(); i++) {
RecordProcessing processing = recordsProcessing.get(i);
RecordMetadata recordMetadata = processing.getRecordMetadata();
recordsMetadata.add(recordMetadata);
if(processing.getOperationType() == OperationType.create) {
- pubsubInfo[i] = PubSubInfo.builder().id(recordMetadata.getId()).kind(recordMetadata.getKind()).op(OperationType.create).build();
+ pubsubInfo[i] = getPubSubInfo(recordMetadata, OperationType.create);
+ recordChangedV2[i] = getRecordChangedV2(recordMetadata, OperationType.create);
} else {
- pubsubInfo[i] = PubSubInfo.builder().id(recordMetadata.getId()).kind(recordMetadata.getKind()).op(OperationType.update).recordBlocks(processing.getRecordBlocks()).build();
+ pubsubInfo[i] = getPubSubInfo(recordMetadata, OperationType.update);
+ pubsubInfo[i].setRecordBlocks(processing.getRecordBlocks());
+ recordChangedV2[i] = getRecordChangedV2(recordMetadata, OperationType.update);
+ recordChangedV2[i].setRecordBlocks(processing.getRecordBlocks());
+
if (!Strings.isNullOrEmpty(processing.getRecordMetadata().getPreviousVersionKind())) {
pubsubInfo[i].setPreviousVersionKind(processing.getRecordMetadata().getPreviousVersionKind());
+ recordChangedV2[i].setPreviousVersionKind(processing.getRecordMetadata().getPreviousVersionKind());
}
}
}
this.commitBatch(recordsProcessing, recordsMetadata, collaborationContext);
- this.pubSubClient.publishMessage(collaborationContext, this.headers, pubsubInfo);
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ this.pubSubClient.publishMessage(collaborationContext, this.headers, recordChangedV2);
+ }
+ if (!collaborationContext.isPresent()) {
+ this.pubSubClient.publishMessage(this.headers, pubsubInfo);
+ }
}
- private void commitBatch(List<RecordProcessing> recordsProcessing, List<RecordMetadata> recordsMetadata, Optional<CollaborationContext> collaborationContext) {
+ private void commitBatch(List<RecordProcessing> recordsProcessing, List<RecordMetadata> recordsMetadata, Optional<CollaborationContext> collaborationContext) {
try {
this.commitCloudStorageTransaction(recordsProcessing);
@@ -120,14 +137,40 @@ public class PersistenceServiceImpl implements PersistenceService {
throw e;
}
PubSubInfo[] pubsubInfo = new PubSubInfo[recordMetadata.size()];
+ RecordChangedV2[] recordChangedV2 = new RecordChangedV2[recordMetadata.size()];
for (int i = 0; i < recordMetadata.size(); i++) {
RecordMetadata metadata = recordMetadata.get(i);
- pubsubInfo[i] = new PubSubInfo(metadata.getId(), metadata.getKind(), OperationType.update);
+ pubsubInfo[i] = getPubSubInfo(metadata, OperationType.update);
+ recordChangedV2[i] = getRecordChangedV2(metadata, OperationType.update);
+ }
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ this.pubSubClient.publishMessage(collaborationContext, this.headers, recordChangedV2);
}
- this.pubSubClient.publishMessage(collaborationContext, this.headers, pubsubInfo);
+ if (!collaborationContext.isPresent()) {
+ this.pubSubClient.publishMessage(this.headers, pubsubInfo);
+ }
+
return lockedRecords;
}
+ private PubSubInfo getPubSubInfo(RecordMetadata recordMetadata, OperationType operationType) {
+ return PubSubInfo.builder()
+ .id(recordMetadata.getId())
+ .kind(recordMetadata.getKind())
+ .op(operationType)
+ .build();
+ }
+
+ private RecordChangedV2 getRecordChangedV2(RecordMetadata recordMetadata, OperationType operationType) {
+ return RecordChangedV2.builder()
+ .id(recordMetadata.getId())
+ .version(recordMetadata.getLatestVersion())
+ .modifiedBy(recordMetadata.getModifyUser())
+ .kind(recordMetadata.getKind())
+ .op(operationType)
+ .build();
+ }
+
private void tryCleanupCloudStorage(List<RecordProcessing> recordsProcessing) {
recordsProcessing.forEach(r -> this.cloudStorage.deleteVersion(r.getRecordMetadata(), r.getRecordMetadata().getLatestVersion()));
}
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/service/RecordServiceImpl.java b/storage-core/src/main/java/org/opengroup/osdu/storage/service/RecordServiceImpl.java
index 6e9f575a3efbbb4098b8b1d962b55fd5785b275d..8a80fbcc73e4e33a45aa62bb9e4e985458451e41 100644
--- a/storage-core/src/main/java/org/opengroup/osdu/storage/service/RecordServiceImpl.java
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/service/RecordServiceImpl.java
@@ -18,6 +18,7 @@ import com.google.common.collect.Lists;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.http.HttpStatus;
+import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
@@ -30,6 +31,7 @@ import org.opengroup.osdu.core.common.model.storage.RecordState;
import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
import org.opengroup.osdu.storage.exception.DeleteRecordsException;
import org.opengroup.osdu.storage.logging.StorageAuditLogger;
+import org.opengroup.osdu.storage.model.RecordChangedV2Delete;
import org.opengroup.osdu.storage.provider.interfaces.ICloudStorage;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.opengroup.osdu.storage.provider.interfaces.IRecordsMetadataRepository;
@@ -47,6 +49,7 @@ import java.util.stream.Collectors;
import static java.util.Collections.singletonList;
import static java.util.stream.Collectors.toList;
+import static org.opengroup.osdu.storage.util.StringConstants.COLLABORATIONS_FEATURE_NAME;
@Service
public class RecordServiceImpl implements RecordService {
@@ -74,6 +77,8 @@ public class RecordServiceImpl implements RecordService {
@Autowired
private RecordUtil recordUtil;
+ @Autowired
+ private IFeatureFlag collaborationFeatureFlag;
@Override
public void purgeRecord(String recordId, Optional<CollaborationContext> collaborationContext) {
@@ -105,9 +110,12 @@ public class RecordServiceImpl implements RecordService {
}
this.auditLogger.purgeRecordSuccess(singletonList(recordId));
- this.pubSubClient.publishMessage(collaborationContext, this.headers,
- new PubSubDeleteInfo(recordId, recordMetadata.getKind(), DeletionType.hard));
-
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ this.pubSubClient.publishMessage(collaborationContext, this.headers, getRecordChangedV2Delete(recordId, recordMetadata, DeletionType.hard));
+ }
+ if (!collaborationContext.isPresent()) {
+ this.pubSubClient.publishMessage(this.headers, new PubSubDeleteInfo(recordId, recordMetadata.getKind(), DeletionType.hard));
+ }
}
@Override
@@ -127,8 +135,14 @@ public class RecordServiceImpl implements RecordService {
this.recordRepository.createOrUpdate(recordsMetadata, collaborationContext);
this.auditLogger.deleteRecordSuccess(singletonList(recordId));
- PubSubDeleteInfo pubSubDeleteInfo = new PubSubDeleteInfo(recordId, recordMetadata.getKind(), DeletionType.soft);
- this.pubSubClient.publishMessage(collaborationContext, this.headers, pubSubDeleteInfo);
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ RecordChangedV2Delete recordChangedV2Delete = getRecordChangedV2Delete(recordId, recordMetadata, DeletionType.soft);
+ this.pubSubClient.publishMessage(collaborationContext, this.headers, recordChangedV2Delete);
+ }
+ if (!collaborationContext.isPresent()) {
+ PubSubDeleteInfo pubSubDeleteInfo = new PubSubDeleteInfo(recordId, recordMetadata.getKind(), DeletionType.soft);
+ this.pubSubClient.publishMessage(this.headers, pubSubDeleteInfo);
+ }
}
@Override
@@ -165,11 +179,31 @@ public class RecordServiceImpl implements RecordService {
}
}
+ private RecordChangedV2Delete getRecordChangedV2Delete(String recordId, RecordMetadata recordMetadata, DeletionType deletionType) {
+ return RecordChangedV2Delete.builder()
+ .id(recordId)
+ .version(recordMetadata.getLatestVersion())
+ .modifiedBy(recordMetadata.getModifyUser())
+ .kind(recordMetadata.getKind())
+ .op(OperationType.delete)
+ .deletionType(deletionType)
+ .build();
+ }
+
private void publishDeletedRecords(Optional<CollaborationContext> collaborationContext, List<RecordMetadata> records) {
- List<PubSubDeleteInfo> messages = records.stream()
- .map(recordMetadata -> new PubSubDeleteInfo(recordMetadata.getId(), recordMetadata.getKind(), DeletionType.soft))
- .collect(Collectors.toList());
- pubSubClient.publishMessage(collaborationContext, headers, messages.toArray(new PubSubDeleteInfo[messages.size()]));
+
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ List<RecordChangedV2Delete> messages = records.stream()
+ .map(recordMetadata -> getRecordChangedV2Delete(recordMetadata.getId(), recordMetadata, DeletionType.soft))
+ .collect(Collectors.toList());
+ pubSubClient.publishMessage(collaborationContext, headers, messages.toArray(new RecordChangedV2Delete[messages.size()]));
+ }
+ if (!collaborationContext.isPresent()) {
+ List<PubSubDeleteInfo> messages = records.stream()
+ .map(recordMetadata -> new PubSubDeleteInfo(recordMetadata.getId(), recordMetadata.getKind(), DeletionType.soft))
+ .collect(Collectors.toList());
+ pubSubClient.publishMessage(headers, messages.toArray(new PubSubDeleteInfo[messages.size()]));
+ }
}
private RecordMetadata getRecordMetadata(String recordId, boolean isPurgeRequest, Optional<CollaborationContext> collaborationContext) {
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/service/SchemaServiceImpl.java b/storage-core/src/main/java/org/opengroup/osdu/storage/service/SchemaServiceImpl.java
index b090398c0b5867b0708054a4569d12f59a0a069f..1419ac273b288e68f9be1d8c21ca0f8d984a77fa 100644
--- a/storage-core/src/main/java/org/opengroup/osdu/storage/service/SchemaServiceImpl.java
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/service/SchemaServiceImpl.java
@@ -17,8 +17,10 @@ package org.opengroup.osdu.storage.service;
import com.lambdaworks.redis.RedisException;
import org.apache.http.HttpStatus;
import org.opengroup.osdu.core.common.cache.ICache;
+import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
import org.opengroup.osdu.core.common.model.http.AppException;
+import org.opengroup.osdu.core.common.model.http.CollaborationContext;
import org.opengroup.osdu.core.common.model.http.DpsHeaders;
import org.opengroup.osdu.core.common.model.indexer.OperationType;
import org.opengroup.osdu.core.common.model.storage.PubSubInfo;
@@ -28,6 +30,7 @@ import org.opengroup.osdu.core.common.model.storage.validation.ValidationDoc;
import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
import org.opengroup.osdu.core.common.util.Crc32c;
import org.opengroup.osdu.storage.logging.StorageAuditLogger;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.opengroup.osdu.storage.provider.interfaces.ISchemaRepository;
import org.springframework.beans.factory.annotation.Autowired;
@@ -41,6 +44,7 @@ import java.util.Optional;
import java.util.TreeMap;
import static java.util.Collections.singletonList;
+import static org.opengroup.osdu.storage.util.StringConstants.COLLABORATIONS_FEATURE_NAME;
@Service
public class SchemaServiceImpl implements SchemaService {
@@ -85,12 +89,14 @@ public class SchemaServiceImpl implements SchemaService {
@Autowired
private StorageAuditLogger auditLogger;
+ @Autowired
+ private IFeatureFlag collaborationFeatureFlag;
@Override
public void createSchema(Schema inputSchema) {
this.validateKindFromTenant(inputSchema.getKind());
this.validateCircularReference(inputSchema, null);
-
+ Optional<CollaborationContext> collaborationContext = Optional.empty();
Schema schema = this.validateSchema(inputSchema);
try {
@@ -99,8 +105,15 @@ public class SchemaServiceImpl implements SchemaService {
this.auditLogger.createSchemaSuccess(singletonList(inputSchema.getKind()));
this.cache.put(this.getSchemaCacheKey(inputSchema.getKind()), schema);
- this.pubSubClient.publishMessage(Optional.empty(), this.headers,
- new PubSubInfo(null, inputSchema.getKind(), OperationType.create_schema));
+
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ this.pubSubClient.publishMessage(Optional.empty(), this.headers,
+ getRecordChangedV2(inputSchema.getKind(), OperationType.create_schema));
+ }
+ if (!collaborationContext.isPresent()) {
+ this.pubSubClient.publishMessage(this.headers,
+ new PubSubInfo(null, inputSchema.getKind(), OperationType.create_schema));
+ }
} catch (IllegalArgumentException e) {
throw new AppException(HttpStatus.SC_CONFLICT, "Schema already registered",
@@ -119,6 +132,8 @@ public class SchemaServiceImpl implements SchemaService {
@Override
public void deleteSchema(String kind) {
+ Optional<CollaborationContext> collaborationContext = Optional.empty();
+
this.validateKindFromTenant(kind);
Schema schema = this.schemaRepository.get(kind);
@@ -131,8 +146,14 @@ public class SchemaServiceImpl implements SchemaService {
this.auditLogger.deleteSchemaSuccess(singletonList(schema.getKind()));
this.cache.delete(this.getSchemaCacheKey(kind));
- this.pubSubClient.publishMessage(Optional.empty(), this.headers,
- new PubSubInfo(null, schema.getKind(), OperationType.purge_schema));
+ if (collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ this.pubSubClient.publishMessage(Optional.empty(), this.headers,
+ getRecordChangedV2(schema.getKind(), OperationType.purge_schema));
+ }
+ if (!collaborationContext.isPresent()) {
+ this.pubSubClient.publishMessage(this.headers,
+ new PubSubInfo(null, schema.getKind(), OperationType.purge_schema));
+ }
}
@Override
@@ -198,6 +219,13 @@ public class SchemaServiceImpl implements SchemaService {
return new Schema(schema.getKind(), items.toArray(new SchemaItem[items.size()]), schema.getExt());
}
+ private RecordChangedV2 getRecordChangedV2(String kind, OperationType operationType) {
+ return RecordChangedV2.builder()
+ .kind(kind)
+ .op(operationType)
+ .build();
+ }
+
private Schema fetchSchema(String kind) {
String key = this.getSchemaCacheKey(kind);
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/util/CollaborationFilter.java b/storage-core/src/main/java/org/opengroup/osdu/storage/util/CollaborationFilter.java
index f5027b3fa38371cb419d6f68928c866490306b7c..bee6cbd2a6adceee545bb1bb88cb705ba761cceb 100644
--- a/storage-core/src/main/java/org/opengroup/osdu/storage/util/CollaborationFilter.java
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/util/CollaborationFilter.java
@@ -17,22 +17,23 @@ import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
+import static org.opengroup.osdu.storage.util.StringConstants.COLLABORATIONS_FEATURE_NAME;
+
@Component
public class CollaborationFilter implements Filter {
public static final String X_COLLABORATION_HEADER_NAME = "x-collaboration";
private static final String DATA_PARTITION_ID = "data-partition-id";
- private static final String COLLABORATIONS_FEATURE_NAME = "collaborations-enabled";
@Autowired
- public IFeatureFlag iCollaborationFeatureFlag;
+ public IFeatureFlag collaborationFeatureFlag;
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
throws IOException, ServletException {
HttpServletResponse httpResponse = (HttpServletResponse) response;
- if (!iCollaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
+ if (!collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)) {
String collaborationHeader = ((HttpServletRequest) request).getHeader(X_COLLABORATION_HEADER_NAME);
if (!Strings.isNullOrEmpty(collaborationHeader)) {
httpResponse.setContentType(MediaType.APPLICATION_JSON_VALUE);
diff --git a/storage-core/src/main/java/org/opengroup/osdu/storage/util/StringConstants.java b/storage-core/src/main/java/org/opengroup/osdu/storage/util/StringConstants.java
new file mode 100644
index 0000000000000000000000000000000000000000..02bec43fa7478ca4fc8c693849d9e7feb40d106e
--- /dev/null
+++ b/storage-core/src/main/java/org/opengroup/osdu/storage/util/StringConstants.java
@@ -0,0 +1,6 @@
+package org.opengroup.osdu.storage.util;
+
+public class StringConstants {
+ public static final String COLLABORATIONS_FEATURE_NAME = "collaborations-enabled";
+
+}
diff --git a/storage-core/src/test/java/org/opengroup/osdu/storage/service/PersistenceServiceImplTest.java b/storage-core/src/test/java/org/opengroup/osdu/storage/service/PersistenceServiceImplTest.java
index f47ce60746b02d0f16ca8ab87d98a91e53987b52..d56eaf4dd0e5453981cdd70058848471eda021e2 100644
--- a/storage-core/src/test/java/org/opengroup/osdu/storage/service/PersistenceServiceImplTest.java
+++ b/storage-core/src/test/java/org/opengroup/osdu/storage/service/PersistenceServiceImplTest.java
@@ -22,6 +22,7 @@ import org.mockito.ArgumentCaptor;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
+import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
import org.opengroup.osdu.core.common.model.entitlements.Acl;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
@@ -30,12 +31,11 @@ import org.opengroup.osdu.core.common.model.tenant.TenantInfo;
import org.opengroup.osdu.core.common.model.indexer.OperationType;
import org.opengroup.osdu.core.common.model.storage.*;
import org.opengroup.osdu.core.common.model.http.AppException;
+import org.opengroup.osdu.storage.model.RecordChangedV2;
import org.opengroup.osdu.storage.provider.interfaces.ICloudStorage;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.opengroup.osdu.storage.provider.interfaces.IRecordsMetadataRepository;
-//import com.google.cloud.datastore.DatastoreException;
import org.apache.http.HttpStatus;
-import org.springframework.beans.factory.annotation.Autowired;
import java.util.*;
@@ -43,12 +43,14 @@ import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.*;
+import static org.opengroup.osdu.storage.util.StringConstants.COLLABORATIONS_FEATURE_NAME;
@RunWith(MockitoJUnitRunner.class)
public class PersistenceServiceImplTest {
private static final Integer BATCH_SIZE = 48;
private static final String BUCKET = "anyBucket";
+ private static final String MODIFIED_BY = "modifyUser";
private final Optional<CollaborationContext> COLLABORATION_CONTEXT = Optional.ofNullable(CollaborationContext.builder().id(UUID.fromString("9e1c4e74-3b9b-4b17-a0d5-67766558ec65")).application("TestApp").build());
@Mock
@@ -69,6 +71,9 @@ public class PersistenceServiceImplTest {
@Mock
private JaxRsDpsLog logger;
+ @Mock
+ private IFeatureFlag collaborationFeatureFlag;
+
@InjectMocks
private PersistenceServiceImpl sut;
@@ -91,6 +96,44 @@ public class PersistenceServiceImplTest {
@SuppressWarnings({ "unchecked", "rawtypes" })
public void should_persistRecords_when_noExceptionIsThrown() {
+ when(collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)).thenReturn(false);
+
+ this.setupRecordRepository(25, 23, 25);
+
+ TransferBatch batch = this.createBatchTransfer();
+
+ this.sut.persistRecordBatch(batch, Optional.empty());
+
+ for (int i = 0; i < BATCH_SIZE; i++) {
+ verify(this.cloudStorage)
+ .write(batch.getRecords().toArray(new RecordProcessing[batch.getRecords().size()]));
+ }
+
+ ArgumentCaptor<List> datastoreCaptor = ArgumentCaptor.forClass(List.class);
+
+ verify(this.recordRepository, times(1)).createOrUpdate(datastoreCaptor.capture(), any());
+
+ List<List> capturedDatastoreList = datastoreCaptor.getAllValues();
+
+ assertEquals(1, capturedDatastoreList.size());
+
+ List list1 = capturedDatastoreList.get(0);
+ assertEquals(48, list1.size());
+
+ ArgumentCaptor<PubSubInfo[]> pubsubCaptor = ArgumentCaptor.forClass(PubSubInfo[].class);
+
+ verify(this.pubSubClient).publishMessage(eq(this.headers), pubsubCaptor.capture());
+
+ this.assertPubsubInfo(48, pubsubCaptor.getAllValues());
+ verify(this.cloudStorage, times(0)).delete(any(RecordMetadata.class));
+ }
+
+ @Test
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ public void should_persistRecords_when_noExceptionIsThrown_when_collaborationIsEmptyAndFFIsEnabled() {
+
+ when(collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)).thenReturn(true);
+
this.setupRecordRepository(25, 23, 25);
TransferBatch batch = this.createBatchTransfer();
@@ -114,10 +157,51 @@ public class PersistenceServiceImplTest {
assertEquals(48, list1.size());
ArgumentCaptor<PubSubInfo[]> pubsubCaptor = ArgumentCaptor.forClass(PubSubInfo[].class);
+ ArgumentCaptor<RecordChangedV2[]> recordChangedV2Captor = ArgumentCaptor.forClass(RecordChangedV2[].class);
- verify(this.pubSubClient).publishMessage(eq(Optional.empty()), eq(this.headers), pubsubCaptor.capture());
+ verify(this.pubSubClient).publishMessage(eq(Optional.empty()), eq(this.headers), recordChangedV2Captor.capture());
+ verify(this.pubSubClient).publishMessage(eq(this.headers), pubsubCaptor.capture());
this.assertPubsubInfo(48, pubsubCaptor.getAllValues());
+ this.assertRecordChangedV2Info(48, recordChangedV2Captor.getAllValues());
+ verify(this.cloudStorage, times(0)).delete(any(RecordMetadata.class));
+ }
+
+ @Test
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ public void should_persistRecords_when_noExceptionIsThrown_when_collaborationIsPresentAndFFIsEnabled() {
+
+ when(collaborationFeatureFlag.isFeatureEnabled(COLLABORATIONS_FEATURE_NAME)).thenReturn(true);
+
+ this.setupRecordRepository(25, 23, 25);
+
+ TransferBatch batch = this.createBatchTransfer();
+
+ this.sut.persistRecordBatch(batch, COLLABORATION_CONTEXT);
+
+ for (int i = 0; i < BATCH_SIZE; i++) {
+ verify(this.cloudStorage)
+ .write(batch.getRecords().toArray(new RecordProcessing[batch.getRecords().size()]));
+ }
+
+ ArgumentCaptor<List> datastoreCaptor = ArgumentCaptor.forClass(List.class);
+
+ verify(this.recordRepository, times(1)).createOrUpdate(datastoreCaptor.capture(), any());
+
+ List<List> capturedDatastoreList = datastoreCaptor.getAllValues();
+
+ assertEquals(1, capturedDatastoreList.size());
+
+ List list1 = capturedDatastoreList.get(0);
+ assertEquals(48, list1.size());
+
+ ArgumentCaptor<RecordChangedV2[]> recordChangedV2Captor = ArgumentCaptor.forClass(RecordChangedV2[].class);
+ ArgumentCaptor<PubSubInfo[]> pubSubInfoCaptor = ArgumentCaptor.forClass(PubSubInfo[].class);
+
+ verify(this.pubSubClient).publishMessage(eq(COLLABORATION_CONTEXT), eq(this.headers), recordChangedV2Captor.capture());
+ verify(this.pubSubClient, never()).publishMessage(eq(this.headers), pubSubInfoCaptor.capture());
+
+ this.assertRecordChangedV2Info(48, recordChangedV2Captor.getAllValues());
verify(this.cloudStorage, times(0)).delete(any(RecordMetadata.class));
}
@@ -299,6 +383,7 @@ public class PersistenceServiceImplTest {
recordMetadata.setKind("anyKind");
recordMetadata.setAcl(this.acl);
recordMetadata.setUser("createUser");
+ recordMetadata.setModifyUser("modifyUser");
Date date = new Date();
recordMetadata.setGcsVersionPaths(Arrays.asList(String.format("%s/%s/%s", "anyKind", ("ID" + i), date.getTime())));
@@ -334,6 +419,23 @@ public class PersistenceServiceImplTest {
}
}
+ private void assertRecordChangedV2Info(int successfullRecords, Object capturedRecordChangedV2List) {
+
+ LinkedList<RecordChangedV2> recordChangedV2s = (LinkedList<RecordChangedV2>) capturedRecordChangedV2List;
+
+ assertEquals(successfullRecords, recordChangedV2s.size());
+
+ for (int i = 0; i < recordChangedV2s.size(); i++) {
+ RecordChangedV2 recordChangedV2 = recordChangedV2s.get(i);
+ assertEquals("anyKind", recordChangedV2.getKind());
+ assertEquals(i % 2 == 0 ? OperationType.create : OperationType.update, recordChangedV2.getOp());
+ assertNull(recordChangedV2.getPreviousVersionKind());
+ assertTrue(recordChangedV2.getId().startsWith("ID"));
+ assertNotNull(recordChangedV2.getVersion());
+ assertEquals(MODIFIED_BY, recordChangedV2.getModifiedBy());
+ }
+ }
+
private List<RecordMetadata> createListOfRecordMetadata() {
List<RecordMetadata> recordMetadataList = new ArrayList<>();
diff --git a/storage-core/src/test/java/org/opengroup/osdu/storage/service/RecordServiceImplTest.java b/storage-core/src/test/java/org/opengroup/osdu/storage/service/RecordServiceImplTest.java
index 75af5bd1bc5df25dc74d41306001ba20c36ac823..49619fc80e6087404efae0a7f3cdf8b380a2b382 100644
--- a/storage-core/src/test/java/org/opengroup/osdu/storage/service/RecordServiceImplTest.java
+++ b/storage-core/src/test/java/org/opengroup/osdu/storage/service/RecordServiceImplTest.java
@@ -31,6 +31,7 @@ import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.opengroup.osdu.core.common.entitlements.IEntitlementsAndCacheService;
+import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.model.entitlements.Acl;
import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.core.common.model.http.CollaborationContext;
@@ -45,9 +46,11 @@ import org.opengroup.osdu.core.common.provider.interfaces.ITenantFactory;
import org.opengroup.osdu.core.common.storage.PersistenceHelper;
import org.opengroup.osdu.storage.exception.DeleteRecordsException;
import org.opengroup.osdu.storage.logging.StorageAuditLogger;
+import org.opengroup.osdu.storage.model.RecordChangedV2Delete;
import org.opengroup.osdu.storage.provider.interfaces.ICloudStorage;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.opengroup.osdu.storage.provider.interfaces.IRecordsMetadataRepository;
+import org.opengroup.osdu.storage.util.StringConstants;
import org.opengroup.osdu.storage.util.api.RecordUtil;
import java.util.Arrays;
@@ -108,6 +111,9 @@ public class RecordServiceImplTest {
@Mock
private DataAuthorizationService dataAuthorizationService;
+ @Mock
+ private IFeatureFlag collaborationFeatureFlag;
+
@Before
public void setup() {
mock(PersistenceHelper.class);
@@ -161,7 +167,7 @@ public class RecordServiceImplTest {
PubSubDeleteInfo pubSubDeleteInfo = new PubSubDeleteInfo(RECORD_ID, "any kind", DeletionType.hard);
- verify(this.pubSubClient).publishMessage(Optional.empty(), this.headers, pubSubDeleteInfo);
+ verify(this.pubSubClient).publishMessage(this.headers, pubSubDeleteInfo);
}
@@ -312,7 +318,7 @@ public class RecordServiceImplTest {
ArgumentCaptor<PubSubDeleteInfo> pubsubMessageCaptor = ArgumentCaptor.forClass(PubSubDeleteInfo.class);
- verify(this.pubSubClient).publishMessage(eq(Optional.empty()), eq(this.headers), pubsubMessageCaptor.capture());
+ verify(this.pubSubClient).publishMessage(eq(this.headers), pubsubMessageCaptor.capture());
PubSubDeleteInfo capturedMessage = pubsubMessageCaptor.getValue();
assertEquals(RECORD_ID, capturedMessage.getId());
@@ -372,12 +378,13 @@ public class RecordServiceImplTest {
}
@Test
- public void shouldDeleteRecords_successfully() {
+ public void shouldDeleteRecords_successfully_when_collaborationFFIsDisabled() {
RecordMetadata record = buildRecordMetadata();
Map<String, RecordMetadata> expectedRecordMetadataMap = new HashMap<String, RecordMetadata>() {{
put(RECORD_ID, record);
}};
+ when(collaborationFeatureFlag.isFeatureEnabled(StringConstants.COLLABORATIONS_FEATURE_NAME)).thenReturn(false);
when(recordRepository.get(singletonList(RECORD_ID), Optional.empty())).thenReturn(expectedRecordMetadataMap);
when(dataAuthorizationService.validateOwnerAccess(record, OperationType.delete)).thenReturn(true);
@@ -387,7 +394,7 @@ public class RecordServiceImplTest {
verify(dataAuthorizationService, only()).validateOwnerAccess(record, OperationType.delete);
verify(recordRepository, times(1)).createOrUpdate(singletonList(record), Optional.empty());
verify(auditLogger, only()).deleteRecordSuccess(singletonList(RECORD_ID));
- verifyPubSubPublished(Optional.empty());
+ verifyPubSubPublished(Optional.empty(), false);
assertEquals(RecordState.deleted, record.getStatus());
assertEquals(USER_NAME, record.getModifyUser());
@@ -396,12 +403,13 @@ public class RecordServiceImplTest {
}
@Test
- public void shouldBulkDeleteRecords_successfully_inCollaborationContext() {
+ public void shouldBulkDeleteRecords_successfully_when_collaborationFFIsEnabledAndContextIsPresent() {
RecordMetadata record = buildRecordMetadata();
Map<String, RecordMetadata> expectedRecordMetadataMap = new HashMap<String, RecordMetadata>() {{
put(COLLABORATION_CONTEXT.get().getId() + RECORD_ID, record);
}};
+ when(collaborationFeatureFlag.isFeatureEnabled(StringConstants.COLLABORATIONS_FEATURE_NAME)).thenReturn(true);
when(recordRepository.get(singletonList(RECORD_ID), COLLABORATION_CONTEXT)).thenReturn(expectedRecordMetadataMap);
when(dataAuthorizationService.validateOwnerAccess(record, OperationType.delete)).thenReturn(true);
@@ -411,7 +419,32 @@ public class RecordServiceImplTest {
verify(dataAuthorizationService, only()).validateOwnerAccess(record, OperationType.delete);
verify(recordRepository, times(1)).createOrUpdate(singletonList(record), COLLABORATION_CONTEXT);
verify(auditLogger, only()).deleteRecordSuccess(singletonList(RECORD_ID));
- verifyPubSubPublished(COLLABORATION_CONTEXT);
+ verifyPubSubPublished(COLLABORATION_CONTEXT, true);
+
+ assertEquals(RecordState.deleted, record.getStatus());
+ assertEquals(USER_NAME, record.getModifyUser());
+ assertNotNull(record.getModifyTime());
+ assertTrue(record.getModifyTime() != 0);
+ }
+
+ @Test
+ public void shouldBulkDeleteRecords_successfully_when_collaborationFFIsEnabledAndContextIsNotPresent() {
+ RecordMetadata record = buildRecordMetadata();
+ Map<String, RecordMetadata> expectedRecordMetadataMap = new HashMap<String, RecordMetadata>() {{
+ put(RECORD_ID, record);
+ }};
+
+ when(collaborationFeatureFlag.isFeatureEnabled(StringConstants.COLLABORATIONS_FEATURE_NAME)).thenReturn(true);
+ when(recordRepository.get(singletonList(RECORD_ID), Optional.empty())).thenReturn(expectedRecordMetadataMap);
+ when(dataAuthorizationService.validateOwnerAccess(record, OperationType.delete)).thenReturn(true);
+
+ sut.bulkDeleteRecords(singletonList(RECORD_ID), USER_NAME, Optional.empty());
+
+ verify(recordRepository, times(1)).get(singletonList(RECORD_ID), Optional.empty());
+ verify(dataAuthorizationService, only()).validateOwnerAccess(record, OperationType.delete);
+ verify(recordRepository, times(1)).createOrUpdate(singletonList(record), Optional.empty());
+ verify(auditLogger, only()).deleteRecordSuccess(singletonList(RECORD_ID));
+ verifyPubSubPublished(Optional.empty(), true);
assertEquals(RecordState.deleted, record.getStatus());
assertEquals(USER_NAME, record.getModifyUser());
@@ -422,13 +455,14 @@ public class RecordServiceImplTest {
@Test
public void shouldSoftDeleteRecords_successfully_inCollaborationContext() {
RecordMetadata record = buildRecordMetadata();
+ when(collaborationFeatureFlag.isFeatureEnabled(StringConstants.COLLABORATIONS_FEATURE_NAME)).thenReturn(true);
when(recordRepository.get(RECORD_ID, COLLABORATION_CONTEXT)).thenReturn(record);
when(dataAuthorizationService.validateOwnerAccess(record, OperationType.delete)).thenReturn(true);
sut.deleteRecord(RECORD_ID, USER_NAME, COLLABORATION_CONTEXT);
verify(recordRepository, times(1)).get(RECORD_ID, COLLABORATION_CONTEXT);
verify(dataAuthorizationService, only()).validateOwnerAccess(record, OperationType.delete);
verify(auditLogger, only()).deleteRecordSuccess(singletonList(RECORD_ID));
- verifyPubSubPublished(COLLABORATION_CONTEXT);
+ verifyPubSubPublished(COLLABORATION_CONTEXT, true);
assertEquals(RecordState.deleted, record.getStatus());
assertEquals(USER_NAME, record.getModifyUser());
@@ -489,7 +523,7 @@ public class RecordServiceImplTest {
verify(recordRepository, times(1)).createOrUpdate(singletonList(record), Optional.empty());
verify(auditLogger, times(1)).deleteRecordSuccess(singletonList(RECORD_ID));
verify(auditLogger, times(1)).deleteRecordFail(singletonList(expectedErrorMessage));
- verifyPubSubPublished(Optional.empty());
+ verifyPubSubPublished(Optional.empty(), false);
assertEquals(RecordState.deleted, record.getStatus());
assertEquals(USER_NAME, record.getModifyUser());
@@ -525,20 +559,28 @@ public class RecordServiceImplTest {
}
}
- private void verifyPubSubPublished(Optional<CollaborationContext> collaborationContext) {
+ private void verifyPubSubPublished(Optional<CollaborationContext> collaborationContext, boolean isCollaborationFFEnabled) {
ArgumentCaptor<PubSubDeleteInfo> pubsubMessageCaptor = ArgumentCaptor.forClass(PubSubDeleteInfo.class);
-
- if (collaborationContext.isPresent()){
- verify(this.pubSubClient).publishMessage(eq(collaborationContext), eq(this.headers), pubsubMessageCaptor.capture());
- } else {
- verify(this.pubSubClient).publishMessage(eq(Optional.empty()), eq(this.headers), pubsubMessageCaptor.capture());
+ ArgumentCaptor<RecordChangedV2Delete> recordChangedV2DeleteArgumentCaptor = ArgumentCaptor.forClass(RecordChangedV2Delete.class);
+
+ if(isCollaborationFFEnabled) {
+ verify(this.pubSubClient).publishMessage(eq(collaborationContext), eq(this.headers), recordChangedV2DeleteArgumentCaptor.capture());
+ RecordChangedV2Delete capturedMessage = recordChangedV2DeleteArgumentCaptor.getValue();
+ assertEquals(RECORD_ID, capturedMessage.getId());
+ assertEquals(KIND, capturedMessage.getKind());
+ assertEquals(OperationType.delete, capturedMessage.getOp());
+ assertEquals(DeletionType.soft, capturedMessage.getDeletionType());
+ assertEquals(USER_NAME, capturedMessage.getModifiedBy());
}
- PubSubDeleteInfo capturedMessage = pubsubMessageCaptor.getValue();
- assertEquals(RECORD_ID, capturedMessage.getId());
- assertEquals(KIND, capturedMessage.getKind());
- assertEquals(OperationType.delete, capturedMessage.getOp());
- assertEquals(DeletionType.soft, capturedMessage.getDeletionType());
+ if (!collaborationContext.isPresent()) {
+ verify(this.pubSubClient).publishMessage(eq(this.headers), pubsubMessageCaptor.capture());
+ PubSubDeleteInfo capturedMessage = pubsubMessageCaptor.getValue();
+ assertEquals(RECORD_ID, capturedMessage.getId());
+ assertEquals(KIND, capturedMessage.getKind());
+ assertEquals(OperationType.delete, capturedMessage.getOp());
+ assertEquals(DeletionType.soft, capturedMessage.getDeletionType());
+ }
}
private RecordMetadata buildRecordMetadata() {
diff --git a/storage-core/src/test/java/org/opengroup/osdu/storage/service/SchemaServiceImplTest.java b/storage-core/src/test/java/org/opengroup/osdu/storage/service/SchemaServiceImplTest.java
index f6c2f1a86d12b99faef7fb45a9ec12c2c8f3b89b..895b924c1c34def6f13085357fd6bfc0ce4f2a6f 100644
--- a/storage-core/src/test/java/org/opengroup/osdu/storage/service/SchemaServiceImplTest.java
+++ b/storage-core/src/test/java/org/opengroup/osdu/storage/service/SchemaServiceImplTest.java
@@ -25,6 +25,7 @@ import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.Optional;
+import org.opengroup.osdu.core.common.feature.IFeatureFlag;
import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.storage.provider.interfaces.IMessageBus;
import org.opengroup.osdu.storage.provider.interfaces.ISchemaRepository;
@@ -74,6 +75,9 @@ public class SchemaServiceImplTest {
@Mock
private StorageAuditLogger auditLogger;
+ @Mock
+ private IFeatureFlag collaborationFeatureFlag;
+
@InjectMocks
private SchemaServiceImpl sut;
@@ -139,7 +143,7 @@ public class SchemaServiceImplTest {
verify(this.schemaRepository).add(schema, USER);
verify(this.cacheService).put("EUerYg==", schema);
- verify(this.pubSubClient).publishMessage(Optional.empty(), this.headers,
+ verify(this.pubSubClient).publishMessage(this.headers,
new PubSubInfo(null, KIND, OperationType.create_schema));
}