feat: 增加 reporter

core/src/main/java/de/xab/porter/core/Session.java

@@ -8,7 +8,7 @@
 public class Session {
 
     public void start(Context context) {
-        final Task task = new Task(context);
+        Task task = new Task(context);
         task.init();
         task.start();
     }

core/src/main/java/de/xab/porter/core/Task.java

@@ -8,6 +8,7 @@
 import de.xab.porter.transfer.channel.Channel;
 import de.xab.porter.transfer.exception.ConnectionException;
 import de.xab.porter.transfer.reader.Reader;
+import de.xab.porter.transfer.reporter.Reporter;
 import de.xab.porter.transfer.writer.Writer;
 
 import java.util.ArrayList;
@@ -44,6 +45,8 @@ public void init() {
      */
     public void register() {
         List<SinkConnection> sinkConnections = context.getSinkConnections();
+        Reporter reporter = ExtensionLoader.getExtensionLoader(Reporter.class).
+                loadExtension(null, "default");
         writers = sinkConnections.stream().
                 map(sink -> {
                     Writer<?> writer = ExtensionLoader.getExtensionLoader(Writer.class).
@@ -56,7 +59,10 @@ public void register() {
                     }
                     Channel channel = ExtensionLoader.getExtensionLoader(Channel.class).
                             loadExtension(null, this.context.getProperties().getChannel());
-                    channel.setOnReadListener(writer::write);
+                    channel.setOnReadListener(data -> {
+                        writer.write(data);
+                        reporter.report(data);
+                    });
                     reader.getChannels().add(channel);
                     return Map.entry(writer, channel);
                 }).collect(Collectors.toList());

transfer/jdbc/src/main/java/de/xab/porter/transfer/jdbc/reader/JDBCReader.java

@@ -39,9 +39,7 @@ public void doRead(Map<String, Column> columnMap) {
         Instant start = Instant.now();
         long batch = 0L;
         try {
-            if (!connection.isReadOnly()) {
-                connection.setReadOnly(true);
-            }
+            connection.setReadOnly(true);
             statement = getStatement();
             resultSet = statement.executeQuery(properties.getSql());
             ResultSetMetaData metaData = resultSet.getMetaData();

transfer/src/main/java/de/xab/porter/transfer/channel/Channel.java

@@ -24,6 +24,4 @@ public interface Channel {
      * register a listener for channel, eligible channel wil notify writer to read data from it
      */
     void setOnReadListener(Consumer<Result<?>> listener);
-
-    void setType(String type);
 }

transfer/src/main/java/de/xab/porter/transfer/channel/DefaultChannel.java

@@ -12,7 +12,6 @@
  */
 public class DefaultChannel implements Channel {
     private final BlockingQueue<Result<?>> resultQueue = new LinkedBlockingQueue<>();
-    private String type;
     private Consumer<Result<?>> onReadListener;
 
     public void push(Result<?> result) {
@@ -41,9 +40,4 @@ public void notifyWriter() {
     public void setOnReadListener(Consumer<Result<?>> listener) {
         this.onReadListener = listener;
     }
-
-    @Override
-    public void setType(String type) {
-        this.type = type;
-    }
 }

transfer/src/main/java/de/xab/porter/transfer/reporter/DefaultReporter.java

@@ -0,0 +1,17 @@
+package de.xab.porter.transfer.reporter;
+
+import de.xab.porter.api.Result;
+import de.xab.porter.common.util.Loggers;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class DefaultReporter implements Reporter {
+    Logger logger = Loggers.getLogger("REPORTER");
+
+    @Override
+    public <T> void report(T t) {
+        Result<?> result = (Result<?>) t;
+        logger.log(Level.INFO, String.format("wrote %sth batch", result.getSequenceNum()));
+    }
+}

transfer/src/main/java/de/xab/porter/transfer/reporter/Reporter.java

@@ -0,0 +1,5 @@
+package de.xab.porter.transfer.reporter;
+
+public interface Reporter {
+    <T> void report(T t);
+}

transfer/src/main/java/module-info.java

@@ -5,9 +5,11 @@
 
     exports de.xab.porter.transfer.connector;
     exports de.xab.porter.transfer.channel;
+    exports de.xab.porter.transfer.reporter;
     exports de.xab.porter.transfer.reader;
     exports de.xab.porter.transfer.writer;
     exports de.xab.porter.transfer.exception;
 
     opens de.xab.porter.transfer.channel to de.xab.porter.common;
+    opens de.xab.porter.transfer.reporter to de.xab.porter.common;
 }

transfer/src/main/resources/META-INF/porter/de.xab.porter.transfer.reporter.Reporter

@@ -0,0 +1 @@
+default=de.xab.porter.transfer.reporter.DefaultReporter