CAMEL-QUARKUS-1720: Added Postgres replication slot native support

docs/modules/ROOT/pages/reference/extensions/pg-replication-slot.adoc

@@ -2,15 +2,15 @@
 // This file was generated by camel-quarkus-maven-plugin:update-extension-doc-page
 = PostgresSQL Replication Slot
 :cq-artifact-id: camel-quarkus-pg-replication-slot
-:cq-native-supported: false
-:cq-status: Preview
+:cq-native-supported: true
+:cq-status: Stable
 :cq-description: Poll for PostgreSQL Write-Ahead Log (WAL) records using Streaming Replication Slots.
 :cq-deprecated: false
 :cq-jvm-since: 1.1.0
-:cq-native-since: n/a
+:cq-native-since: 1.2.0
 
 [.badges]
-[.badge-key]##JVM since##[.badge-supported]##1.1.0## [.badge-key]##Native##[.badge-unsupported]##unsupported##
+[.badge-key]##JVM since##[.badge-supported]##1.1.0## [.badge-key]##Native since##[.badge-supported]##1.2.0##
 
 Poll for PostgreSQL Write-Ahead Log (WAL) records using Streaming Replication Slots.
 

docs/modules/ROOT/partials/reference/components/pg-replication-slot.adoc

@@ -2,11 +2,11 @@
 // This file was generated by camel-quarkus-maven-plugin:update-extension-doc-page
 :cq-artifact-id: camel-quarkus-pg-replication-slot
 :cq-artifact-id-base: pg-replication-slot
-:cq-native-supported: false
-:cq-status: Preview
+:cq-native-supported: true
+:cq-status: Stable
 :cq-deprecated: false
 :cq-jvm-since: 1.1.0
-:cq-native-since: n/a
+:cq-native-since: 1.2.0
 :cq-camel-part-name: pg-replication-slot
 :cq-camel-part-title: PostgresSQL Replication Slot
 :cq-camel-part-description: Poll for PostgreSQL Write-Ahead Log (WAL) records using Streaming Replication Slots.

extensions-jvm/pom.xml

@@ -116,7 +116,6 @@
         <module>ognl</module>
         <module>openstack</module>
         <module>optaplanner</module>
-        <module>pg-replication-slot</module>
         <module>pgevent</module>
         <module>printer</module>
         <module>protobuf</module>

extensions-jvm/pg-replication-slot/deployment/pom.xml → extensions/pg-replication-slot/deployment/pom.xml

@@ -34,11 +34,15 @@
     <dependencies>
         <dependency>
             <groupId>org.apache.camel.quarkus</groupId>
-            <artifactId>camel-quarkus-core-deployment</artifactId>
+            <artifactId>camel-quarkus-pg-replication-slot</artifactId>
         </dependency>
         <dependency>
             <groupId>org.apache.camel.quarkus</groupId>
-            <artifactId>camel-quarkus-pg-replication-slot</artifactId>
+            <artifactId>camel-quarkus-core-deployment</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-jdbc-postgresql-deployment</artifactId>
         </dependency>
     </dependencies>
 

extensions-jvm/pg-replication-slot/deployment/src/main/java/org/apache/camel/quarkus/component/pg/replication/slot/deployment/PgReplicationSlotProcessor.java → extensions/pg-replication-slot/deployment/src/main/java/org/apache/camel/quarkus/component/pg/replication/slot/deployment/PgReplicationSlotProcessor.java

@@ -20,27 +20,20 @@
 import io.quarkus.deployment.annotations.ExecutionTime;
 import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
-import io.quarkus.deployment.pkg.steps.NativeBuild;
-import org.apache.camel.quarkus.core.JvmOnlyRecorder;
-import org.jboss.logging.Logger;
+import org.apache.camel.quarkus.component.pg.replication.slot.PostgresDriverRegistrationRecorder;
 
 class PgReplicationSlotProcessor {
 
-    private static final Logger LOG = Logger.getLogger(PgReplicationSlotProcessor.class);
     private static final String FEATURE = "camel-pg-replication-slot";
 
     @BuildStep
     FeatureBuildItem feature() {
         return new FeatureBuildItem(FEATURE);
     }
 
-    /**
-     * Remove this once this extension starts supporting the native mode.
-     */
-    @BuildStep(onlyIf = NativeBuild.class)
-    @Record(value = ExecutionTime.RUNTIME_INIT)
-    void warnJvmInNative(JvmOnlyRecorder recorder) {
-        JvmOnlyRecorder.warnJvmInNative(LOG, FEATURE); // warn at build time
-        recorder.warnJvmInNative(FEATURE); // warn at runtime
+    @BuildStep
+    @Record(ExecutionTime.STATIC_INIT)
+    public void registerPostgresDriver(PostgresDriverRegistrationRecorder recorder) {
+        recorder.registerPostgresDriver();
     }
 }

extensions-jvm/pg-replication-slot/pom.xml → extensions/pg-replication-slot/pom.xml

@@ -35,6 +35,5 @@
     <modules>
         <module>deployment</module>
         <module>runtime</module>
-        <module>integration-test</module>
     </modules>
 </project>

extensions-jvm/pg-replication-slot/runtime/pom.xml → extensions/pg-replication-slot/runtime/pom.xml

@@ -34,6 +34,7 @@
 
     <properties>
         <camel.quarkus.jvmSince>1.1.0</camel.quarkus.jvmSince>
+        <camel.quarkus.nativeSince>1.2.0</camel.quarkus.nativeSince>
     </properties>
 
     <dependencyManagement>
@@ -49,13 +50,17 @@
     </dependencyManagement>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-pg-replication-slot</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.camel.quarkus</groupId>
             <artifactId>camel-quarkus-core</artifactId>
         </dependency>
         <dependency>
-            <groupId>org.apache.camel</groupId>
-            <artifactId>camel-pg-replication-slot</artifactId>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-jdbc-postgresql</artifactId>
         </dependency>
     </dependencies>
 

extensions/pg-replication-slot/runtime/src/main/java/org/apache/camel/quarkus/component/pg/replication/slot/PostgresDriverRegistrationRecorder.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.pg.replication.slot;
+
+import io.quarkus.runtime.annotations.Recorder;
+import org.postgresql.Driver;
+
+/**
+ * Forces the Postgres driver registration to workaround https://github.com/quarkusio/quarkus/issues/12116.
+ */
+@Recorder
+public class PostgresDriverRegistrationRecorder {
+
+    public void registerPostgresDriver() {
+        new Driver();
+    }
+
+}

extensions-jvm/pg-replication-slot/runtime/src/main/resources/META-INF/quarkus-extension.yaml → extensions/pg-replication-slot/runtime/src/main/resources/META-INF/quarkus-extension.yaml

@@ -24,9 +24,8 @@
 name: "Camel PostgresSQL Replication Slot"
 description: "Poll for PostgreSQL Write-Ahead Log (WAL) records using Streaming Replication Slots"
 metadata:
-  unlisted: true
   guide: "https://camel.apache.org/camel-quarkus/latest/reference/extensions/pg-replication-slot.html"
   categories:
   - "integration"
   status:
-  - "preview"
+  - "stable"

extensions/pom.xml

@@ -157,6 +157,7 @@
         <module>opentracing</module>
         <module>paho</module>
         <module>pdf</module>
+        <module>pg-replication-slot</module>
         <module>platform-http</module>
         <module>quartz</module>
         <module>qute</module>

extensions-jvm/pg-replication-slot/integration-test/pom.xml → integration-tests/pg-replication-slot/pom.xml

@@ -23,13 +23,12 @@
     <modelVersion>4.0.0</modelVersion>
     <parent>
         <groupId>org.apache.camel.quarkus</groupId>
-        <artifactId>camel-quarkus-build-parent-it</artifactId>
+        <artifactId>camel-quarkus-integration-tests</artifactId>
         <version>1.2.0-SNAPSHOT</version>
-        <relativePath>../../../poms/build-parent-it/pom.xml</relativePath>
     </parent>
 
-    <artifactId>camel-quarkus-pg-replication-slot-integration-test</artifactId>
-    <name>Camel Quarkus :: PostgresSQL Replication Slot :: Integration Test</name>
+    <artifactId>camel-quarkus-integration-test-pg-replication-slot</artifactId>
+    <name>Camel Quarkus :: Integration Tests :: PostgresSQL Replication Slot</name>
     <description>Integration tests for Camel Quarkus PostgresSQL Replication Slot extension</description>
 
     <dependencyManagement>
@@ -53,6 +52,10 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-resteasy-jackson</artifactId>
+        </dependency>
 
         <!-- test dependencies -->
         <dependency>
@@ -65,6 +68,16 @@
             <artifactId>rest-assured</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-integration-testcontainers-support</artifactId>
+            <scope>test</scope>
+        </dependency>
 
         <!-- The following dependencies guarantee that this module is built after them. You can update them by running `mvn process-resources -Pformat -N` from the source tree root directory -->
         <dependency>
@@ -97,4 +110,34 @@
             </plugin>
         </plugins>
     </build>
+
+    <profiles>
+        <profile>
+            <id>native</id>
+            <activation>
+                <property>
+                    <name>native</name>
+                </property>
+            </activation>
+            <properties>
+                <quarkus.package.type>native</quarkus.package.type>
+            </properties>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-failsafe-plugin</artifactId>
+                        <executions>
+                            <execution>
+                                <goals>
+                                    <goal>integration-test</goal>
+                                    <goal>verify</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
 </project>

extensions-jvm/pg-replication-slot/integration-test/src/main/java/org/apache/camel/quarkus/component/pg/replication/slot/it/PgReplicationSlotResource.java → integration-tests/pg-replication-slot/src/main/java/org/apache/camel/quarkus/component/pg/replication/slot/it/PgReplicationSlotResource.java

@@ -16,15 +16,14 @@
  */
 package org.apache.camel.quarkus.component.pg.replication.slot.it;
 
+import java.util.concurrent.ConcurrentLinkedQueue;
+
 import javax.enterprise.context.ApplicationScoped;
-import javax.inject.Inject;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
 
-import org.apache.camel.CamelContext;
 import org.jboss.logging.Logger;
 
 @Path("/pg-replication-slot")
@@ -33,19 +32,18 @@ public class PgReplicationSlotResource {
 
     private static final Logger LOG = Logger.getLogger(PgReplicationSlotResource.class);
 
-    private static final String COMPONENT_PG_REPLICATION_SLOT = "pg-replication-slot";
-    @Inject
-    CamelContext context;
+    private final ConcurrentLinkedQueue<String> replicationEvents = new ConcurrentLinkedQueue<>();
+
+    void logReplicationEvent(String event) {
+        LOG.debugf("Calling logReplicationEvent(\"%s\")", event);
+        replicationEvents.add(event);
+    }
 
-    @Path("/load/component/pg-replication-slot")
+    @Path("/get-events")
     @GET
-    @Produces(MediaType.TEXT_PLAIN)
-    public Response loadComponentPgReplicationSlot() throws Exception {
-        /* This is an autogenerated test */
-        if (context.getComponent(COMPONENT_PG_REPLICATION_SLOT) != null) {
-            return Response.ok().build();
-        }
-        LOG.warnf("Could not load [%s] from the Camel context", COMPONENT_PG_REPLICATION_SLOT);
-        return Response.status(500, COMPONENT_PG_REPLICATION_SLOT + " could not be loaded from the Camel context").build();
+    @Produces(MediaType.APPLICATION_JSON)
+    public ConcurrentLinkedQueue<String> getReplicationEvents() {
+        LOG.debug("Calling getReplicationEvents");
+        return replicationEvents;
     }
 }

integration-tests/pg-replication-slot/src/main/java/org/apache/camel/quarkus/component/pg/replication/slot/it/PgReplicationSlotRoute.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.pg.replication.slot.it;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+import org.apache.camel.builder.RouteBuilder;
+
+@ApplicationScoped
+public class PgReplicationSlotRoute extends RouteBuilder {
+
+    public static final String PG_AUTHORITY_CFG_KEY = "quarkus.camel.pg-replication-slot.test.authority";
+    public static final String PG_DBNAME_CFG_KEY = "quarkus.camel.pg-replication-slot.test.db-name";
+    public static final String PG_PASSRD_CFG_KEY = "quarkus.camel.pg-replication-slot.test.password";
+    public static final String PG_USER_CFG_KEY = "quarkus.camel.pg-replication-slot.test.user";
+
+    private static final String URI_FORMAT = "pg-replication-slot://{{%s}}/{{%s}}/{{%s}}_test_slot:test_decoding?user={{%s}}&password={{%s}}&slotOptions.skip-empty-xacts=true&slotOptions.include-xids=false";
+
+    @Inject
+    PgReplicationSlotResource resource;
+
+    @Override
+    public void configure() {
+        fromF(URI_FORMAT, PG_AUTHORITY_CFG_KEY, PG_DBNAME_CFG_KEY, PG_DBNAME_CFG_KEY, PG_USER_CFG_KEY, PG_PASSRD_CFG_KEY)
+                .process(e -> {
+                    resource.logReplicationEvent(e.getMessage().getBody(String.class));
+                });
+    }
+
+}

extensions-jvm/pg-replication-slot/integration-test/src/test/java/org/apache/camel/quarkus/component/pg/replication/slot/it/PgReplicationSlotTest.java → integration-tests/pg-replication-slot/src/test/java/org/apache/camel/quarkus/component/pg/replication/slot/it/PgReplicationSlotIT.java

@@ -16,19 +16,9 @@
  */
 package org.apache.camel.quarkus.component.pg.replication.slot.it;
 
-import io.quarkus.test.junit.QuarkusTest;
-import io.restassured.RestAssured;
-import org.junit.jupiter.api.Test;
+import io.quarkus.test.junit.NativeImageTest;
 
-@QuarkusTest
-class PgReplicationSlotTest {
-
-    @Test
-    public void loadComponentPgReplicationSlot() {
-        /* A simple autogenerated test */
-        RestAssured.get("/pg-replication-slot/load/component/pg-replication-slot")
-                .then()
-                .statusCode(200);
-    }
+@NativeImageTest
+class PgReplicationSlotIT extends PgReplicationSlotTest {
 
 }