fix: use new procedure names (#48)

neo4j · Aug 27, 2024 · 2070ee2 · 2070ee2
1 parent 86cd34d
commit 2070ee2
Show file tree

Hide file tree

Showing 5 changed files with 76 additions and 192 deletions.
diff --git a/src/main/java/org/neo4j/cdc/client/CDCClient.java b/src/main/java/org/neo4j/cdc/client/CDCClient.java
@@ -42,9 +42,9 @@
 public class CDCClient implements CDCService {
     private final Logger log = LoggerFactory.getLogger(CDCClient.class);
 
-    private static final String CDC_EARLIEST_STATEMENT = "call cdc.earliest()";
-    private static final String CDC_CURRENT_STATEMENT = "call cdc.current()";
-    private static final String CDC_QUERY_STATEMENT = "call cdc.query($from, $selectors)";
+    private static final String CDC_EARLIEST_STATEMENT = "call db.cdc.earliest()";
+    private static final String CDC_CURRENT_STATEMENT = "call db.cdc.current()";
+    private static final String CDC_QUERY_STATEMENT = "call db.cdc.query($from, $selectors)";
     private final Driver driver;
     private final List<Selector> selectors;
     private final SessionConfigSupplier sessionConfigSupplier;
@@ -114,12 +114,12 @@ public CDCClient(
 
     @Override
     public Mono<ChangeIdentifier> earliest() {
-        return queryForChangeIdentifier(CDC_EARLIEST_STATEMENT, "cdc.earliest");
+        return queryForChangeIdentifier(CDC_EARLIEST_STATEMENT, "db.cdc.earliest");
     }
 
     @Override
     public Mono<ChangeIdentifier> current() {
-        return queryForChangeIdentifier(CDC_CURRENT_STATEMENT, "cdc.current");
+        return queryForChangeIdentifier(CDC_CURRENT_STATEMENT, "db.cdc.current");
     }
 
     @Override
@@ -133,7 +133,7 @@ public Flux<ChangeEvent> query(ChangeIdentifier from) {
                                     "selectors",
                                     selectors.stream().map(Selector::asMap).collect(Collectors.toList()));
 
-                            log.trace("running cdc.query using parameters {}", params);
+                            log.trace("running db.cdc.query using parameters {}", params);
                             RxResult result = tx.run(CDC_QUERY_STATEMENT, params);
 
                             return Flux.from(result.records())
@@ -152,7 +152,7 @@ public Flux<ChangeEvent> stream(ChangeIdentifier from) {
         var query = Flux.usingWhen(
                 Mono.fromSupplier(() -> driver.rxSession(sessionConfigSupplier.sessionConfig())),
                 (RxSession session) -> Flux.from(session.readTransaction(tx -> {
-                    var current = Mono.from(tx.run("CALL cdc.current()").records())
+                    var current = Mono.from(tx.run("CALL db.cdc.current()").records())
                             .map(MapAccessor::asMap)
                             .map(ResultMapper::parseChangeIdentifier);
 
@@ -162,7 +162,7 @@ public Flux<ChangeEvent> stream(ChangeIdentifier from) {
                             "selectors",
                             selectors.stream().map(Selector::asMap).collect(Collectors.toList()));
 
-                    log.trace("running cdc.query using parameters {}", params);
+                    log.trace("running db.cdc.query using parameters {}", params);
                     RxResult result = tx.run(CDC_QUERY_STATEMENT, params);
 
                     return current.flatMapMany(changeId -> Flux.from(result.records())

diff --git a/src/test/java/org/neo4j/cdc/client/CDCClient514IT.java b/src/test/java/org/neo4j/cdc/client/CDCClient514IT.java