refactor: extract function

Author: ali-ince

sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/strategy/cdc/apoc/ApocCdcHandler.kt

@@ -121,27 +121,15 @@ abstract class ApocCdcHandler(
             )
             appendLine("WITH k, $EVENT WHERE $EVENT.offset > k.offset")
             appendLine("WITH k, $EVENT ORDER BY $EVENT.offset ASC")
-            if (canIUse(Cypher.callSubqueryWithVariableScopeClause()).withNeo4j(neo4j))
-                appendLine("CALL ($EVENT) {")
-            else appendLine("CALL { WITH $EVENT")
-            appendLine(
-                "  CALL apoc.cypher.doIt($EVENT.stmt, $EVENT.params) YIELD value $termination"
-            )
-            appendLine("}")
+            appendCallSubquery(termination)
             appendLine("WITH k, max($EVENT.offset) AS newOffset SET k.offset = newOffset")
             append(termination)
           }
         } else {
           buildString {
             appendLine("UNWIND \$events AS $EVENT")
             appendLine("WITH $EVENT ORDER BY $EVENT.offset ASC")
-            if (canIUse(Cypher.callSubqueryWithVariableScopeClause()).withNeo4j(neo4j))
-                appendLine("CALL ($EVENT) {")
-            else appendLine("CALL { WITH $EVENT")
-            appendLine(
-                "  CALL apoc.cypher.doIt($EVENT.stmt, $EVENT.params) YIELD value $termination"
-            )
-            appendLine("}")
+            appendCallSubquery(termination)
             append(termination)
           }
         }
@@ -157,6 +145,14 @@ abstract class ApocCdcHandler(
     )
   }
 
+  private fun StringBuilder.appendCallSubquery(termination: String) {
+    if (canIUse(Cypher.callSubqueryWithVariableScopeClause()).withNeo4j(neo4j))
+        appendLine("CALL ($EVENT) {")
+    else appendLine("CALL { WITH $EVENT")
+    appendLine("  CALL apoc.cypher.doIt($EVENT.stmt, $EVENT.params) YIELD value $termination")
+    appendLine("}")
+  }
+
   protected abstract fun transformCreate(event: NodeEvent): CdcData
 
   protected abstract fun transformUpdate(event: NodeEvent): CdcData