Terminate state on error

aklivity · Dec 18, 2024 · b63aef5 · b63aef5
1 parent 469afdd
commit b63aef5
Show file tree

Hide file tree

Showing 13 changed files with 120 additions and 143 deletions.
diff --git a/.../aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveAlterStreamMacro.java b/.../aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveAlterStreamMacro.java
@@ -18,21 +18,19 @@
 import io.aklivity.zilla.runtime.binding.risingwave.internal.stream.RisingwaveCompletionCommand;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.types.stream.PgsqlFlushExFW;
 
-public class RisingwaveAlterStreamMacro
+public class RisingwaveAlterStreamMacro extends RisingwaveMacroBase
 {
     private final StringBuilder fieldBuilder;
-    private final String sql;
     private final Alter command;
-    private final RisingwaveMacroHandler handler;
 
     public RisingwaveAlterStreamMacro(
         String sql,
         Alter command,
         RisingwaveMacroHandler handler)
     {
-        this.sql = sql;
+        super(sql, handler);
+
         this.command = command;
-        this.handler = handler;
         this.fieldBuilder = new StringBuilder();
     }
 

diff --git a/.../aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveAlterZtableMacro.java b/.../aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveAlterZtableMacro.java
@@ -18,21 +18,19 @@
 import io.aklivity.zilla.runtime.binding.risingwave.internal.stream.RisingwaveCompletionCommand;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.types.stream.PgsqlFlushExFW;
 
-public class RisingwaveAlterZtableMacro
+public class RisingwaveAlterZtableMacro extends RisingwaveMacroBase
 {
     private final StringBuilder fieldBuilder;
-    private final String sql;
     private final Alter command;
-    private final RisingwaveMacroHandler handler;
 
     public RisingwaveAlterZtableMacro(
         String sql,
         Alter command,
         RisingwaveMacroHandler handler)
     {
-        this.sql = sql;
+        super(sql, handler);
+
         this.command = command;
-        this.handler = handler;
         this.fieldBuilder = new StringBuilder();
     }
 
@@ -84,7 +82,8 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+
+            return errorState();
         }
     }
 
@@ -142,5 +141,4 @@ public RisingwaveMacroState onError(
             return this;
         }
     }
-
 }
diff --git a/...livity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveCreateFunctionMacro.java b/...livity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveCreateFunctionMacro.java
@@ -22,7 +22,7 @@
 import io.aklivity.zilla.runtime.binding.risingwave.internal.stream.RisingwaveCompletionCommand;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.types.stream.PgsqlFlushExFW;
 
-public class RisingwaveCreateFunctionMacro
+public class RisingwaveCreateFunctionMacro extends RisingwaveMacroBase
 {
     private final String javaServer;
     private final String pythonServer;
@@ -31,7 +31,6 @@ public class RisingwaveCreateFunctionMacro
     private final String user;
     private final String sql;
     private final Function command;
-    private final RisingwaveMacroHandler handler;
     private final StringBuilder fieldBuilder;
 
     public RisingwaveCreateFunctionMacro(
@@ -42,6 +41,8 @@ public RisingwaveCreateFunctionMacro(
         Function command,
         RisingwaveMacroHandler handler)
     {
+        super(sql, handler);
+
         String javaServer = null;
         String pythonServer = null;
 
@@ -66,7 +67,6 @@ else if (udf.language.equalsIgnoreCase("python"))
         this.user = user;
         this.sql = sql;
         this.command = command;
-        this.handler = handler;
         this.fieldBuilder = new StringBuilder();
     }
 

diff --git a/...aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveCreateStreamMacro.java b/...aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveCreateStreamMacro.java
@@ -23,7 +23,7 @@
 import io.aklivity.zilla.runtime.binding.risingwave.internal.stream.RisingwaveCompletionCommand;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.types.stream.PgsqlFlushExFW;
 
-public class RisingwaveCreateStreamMacro
+public class RisingwaveCreateStreamMacro extends RisingwaveMacroBase
 {
     //TODO: Remove after implementing zstream
     private static final String ZILLA_CORRELATION_ID_OLD = "zilla_correlation_id";
@@ -47,10 +47,7 @@ public class RisingwaveCreateStreamMacro
     private final long scanStartupMil;
     private final String systemSchema;
     private final String user;
-    private final String sql;
     private final CreateStream command;
-    private final RisingwaveMacroHandler handler;
-    private final RisingwaveOnReadyHandler terminateState = this::terminateState;
 
     public RisingwaveCreateStreamMacro(
         String bootstrapServer,
@@ -62,12 +59,12 @@ public RisingwaveCreateStreamMacro(
         CreateStream command,
         RisingwaveMacroHandler handler)
     {
+        super(sql, handler);
+
         this.scanStartupMil = scanStartupMil;
         this.systemSchema = systemSchema;
         this.user = user;
-        this.sql = sql;
         this.command = command;
-        this.handler = handler;
 
         this.bootstrapServer = bootstrapServer;
         this.schemaRegistry = schemaRegistry;
@@ -78,32 +75,11 @@ public RisingwaveMacroState start()
         return new CreateTopicState();
     }
 
-    private RisingwaveMacroState terminateState(
-        long traceId,
-        long authorization)
-    {
-        handler.doReady(traceId, authorization, sql.length());
-
-        return null;
-    }
-
     private final class CreateTopicState implements RisingwaveMacroState
     {
         private final String sqlFormat = """
             CREATE TOPIC IF NOT EXISTS %s (%s%s);\u0000""";
 
-        private RisingwaveOnReadyHandler onReadyHandler = this::transition;
-
-        private RisingwaveMacroState transition(
-            long traceId,
-            long authorization)
-        {
-            CreateSourceState state = new CreateSourceState();
-            state.onStarted(traceId, authorization);
-
-            return state;
-        }
-
         @Override
         public void onStarted(
             long traceId,
@@ -133,7 +109,10 @@ public RisingwaveMacroState onReady(
             long authorization,
             PgsqlFlushExFW flushEx)
         {
-            return onReadyHandler.onReady(traceId, authorization);
+            CreateSourceState state = new CreateSourceState();
+            state.onStarted(traceId, authorization);
+
+            return state;
         }
 
         @Override
@@ -143,9 +122,8 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            onReadyHandler = terminateState;
 
-            return this;
+            return errorState();
         }
     }
 
@@ -163,18 +141,6 @@ private final class CreateSourceState implements RisingwaveMacroState
                schema.registry = '%s'
             );\u0000""";
 
-        private RisingwaveOnReadyHandler onReadyHandler = this::transition;
-
-        private RisingwaveMacroState transition(
-            long traceId,
-            long authorization)
-        {
-            GrantResourceState state = new GrantResourceState();
-            state.onStarted(traceId, authorization);
-
-            return state;
-        }
-
         @Override
         public void onStarted(
             long traceId,
@@ -206,7 +172,10 @@ public RisingwaveMacroState onReady(
             long authorization,
             PgsqlFlushExFW flushEx)
         {
-            return onReadyHandler.onReady(traceId, authorization);
+            GrantResourceState state = new GrantResourceState();
+            state.onStarted(traceId, authorization);
+
+            return state;
         }
 
         @Override
@@ -216,8 +185,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            onReadyHandler = terminateState;
-            return this;
+            return errorState();
         }
     }
 

diff --git a/...aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveCreateZtableMacro.java b/...aklivity/zilla/runtime/binding/risingwave/internal/macro/RisingwaveCreateZtableMacro.java
@@ -27,7 +27,7 @@
 import io.aklivity.zilla.runtime.binding.risingwave.internal.stream.RisingwaveCompletionCommand;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.types.stream.PgsqlFlushExFW;
 
-public class RisingwaveCreateZtableMacro
+public class RisingwaveCreateZtableMacro extends RisingwaveMacroBase
 {
     private static final String ZTABLE_NAME = "ztables";
     private static final String TABLE_NAME = "tables";
@@ -49,9 +49,7 @@ public class RisingwaveCreateZtableMacro
     private final StringBuilder includeBuilder;
     private final String systemSchema;
     private final String user;
-    private final String sql;
     private final CreateTable command;
-    private final RisingwaveMacroHandler handler;
 
     public RisingwaveCreateZtableMacro(
         String bootstrapServer,
@@ -63,11 +61,11 @@ public RisingwaveCreateZtableMacro(
         CreateTable command,
         RisingwaveMacroHandler handler)
     {
+        super(sql, handler);
+
         this.systemSchema = systemSchema;
         this.user = user;
-        this.sql = sql;
         this.command = command;
-        this.handler = handler;
         this.fieldBuilder = new StringBuilder();
         this.includeBuilder = new StringBuilder();
 
@@ -140,7 +138,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+            return errorState();
         }
     }
 
@@ -224,7 +222,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+            return errorState();
         }
     }
 
@@ -312,7 +310,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+            return errorState();
         }
     }
 
@@ -364,7 +362,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+            return errorState();
         }
     }
 
@@ -402,7 +400,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+            return errorState();
         }
     }
 
@@ -441,7 +439,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+            return errorState();
         }
     }
 
@@ -498,7 +496,7 @@ public RisingwaveMacroState onError(
             PgsqlFlushExFW flushEx)
         {
             handler.doFlushProxy(traceId, authorization, flushEx);
-            return this;
+            return errorState();
         }
 
         private final class InsertIntoCatalogState implements RisingwaveMacroState
@@ -552,5 +550,4 @@ public RisingwaveMacroState onError(
             }
         }
     }
-
 }