Add LogRecord table update callbacks

ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProject.java

@@ -86,6 +86,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
@@ -208,10 +209,6 @@ public GanttProject(boolean isOnlyViewer) {
       getWebSocket().register(null);
       getWebSocket().onCommitResponseReceived(this::fireXlogReceived);
       getWebSocket().onBaseTxnIdReceived(this::onBaseTxnIdReceived);
-      var taskListenerAdapter = new TaskListenerAdapter();
-      // TODO: add listeners sensibly.
-      taskListenerAdapter.setTaskAddedHandler(event -> this.sendProjectStateLogs());
-      getTaskManager().addTaskListener(taskListenerAdapter);
     }
 
     area = new GanttGraphicArea(this, getTaskManager(), getZoomManager(), getUndoManager(),
@@ -956,7 +953,8 @@ public void refresh() {
     super.repaint();
   }
 
-  // TODO: Accumulate changes instead of sending it every time.
+  private final AtomicBoolean isSendingInProgress = new AtomicBoolean();
+
   private Unit sendProjectStateLogs() {
     gpLogger.debug("Sending project state logs");
     try {
@@ -969,20 +967,34 @@ private Unit sendProjectStateLogs() {
           "refid",
           txns
         ));
+        isSendingInProgress.set(true);
       }
     } catch (ProjectDatabaseException e) {
       gpLogger.error("Failed to send logs", new Object[]{}, ImmutableMap.of(), e);
+      isSendingInProgress.set(false);
+    }
+    return Unit.INSTANCE;
+  }
+
+  @Override
+  protected Unit onProjectLogUpdate() {
+    if (isColloboqueLocalTest()) {
+      super.onProjectLogUpdate();
+      if (!isSendingInProgress.get()) sendProjectStateLogs();
     }
     return Unit.INSTANCE;
   }
 
   private Unit fireXlogReceived(ServerCommitResponse response) {
     myBaseTxnCommitInfo.update(response.getBaseTxnId(), response.getNewBaseTxnId(), 1);
+    isSendingInProgress.set(false);
+    sendProjectStateLogs();
     return Unit.INSTANCE;
   }
 
   private Unit onBaseTxnIdReceived(String baseTxnId) {
     myBaseTxnCommitInfo.update("", baseTxnId, 0);
+    sendProjectStateLogs();
     return Unit.INSTANCE;
   }
 }

ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProjectBase.java

@@ -40,6 +40,7 @@ of the License, or (at your option) any later version.
 import javafx.beans.property.SimpleIntegerProperty;
 import javafx.beans.property.SimpleObjectProperty;
 import javafx.collections.FXCollections;
+import kotlin.Unit;
 import kotlin.jvm.functions.Function0;
 import net.sourceforge.ganttproject.chart.Chart;
 import net.sourceforge.ganttproject.chart.ChartModelBase;
@@ -79,6 +80,8 @@ of the License, or (at your option) any later version.
 import java.util.Map;
 import java.util.function.Supplier;
 
+import static biz.ganttproject.storage.cloud.GPCloudHttpImplKt.isColloboqueLocalTest;
+
 /**
  * This class is designed to be a GanttProject-after-refactorings. I am going to
  * refactor GanttProject in order to make true view communicating with other
@@ -192,7 +195,10 @@ GPOptionGroup getTaskOptions() {
 
   protected GanttProjectBase() {
     super("GanttProject");
-    var databaseProxy = new LazyProjectDatabaseProxy(SqlProjectDatabaseImpl.Factory::createInMemoryDatabase, this::getTaskManager);
+    var databaseProxy = new LazyProjectDatabaseProxy(
+      () -> SqlProjectDatabaseImpl.Factory.createInMemoryDatabase(this::onProjectLogUpdate),
+      this::getTaskManager
+    );
 
     myProjectDatabase = databaseProxy;
     myTaskManagerConfig = new TaskManagerConfigImpl();
@@ -258,6 +264,9 @@ protected ParserFactory getParserFactory() {
   protected GanttProjectImpl getProjectImpl() {
     return myProjectImpl;
   }
+
+  protected Unit onProjectLogUpdate() { return Unit.INSTANCE; }
+
   @Override
   public void restore(@NotNull Document fromDocument) throws Document.DocumentException, IOException {
     GanttProjectImplKt.restoreProject(this, fromDocument, myProjectImpl.getListeners());

ganttproject/src/main/java/net/sourceforge/ganttproject/storage/SqlProjectDatabaseImpl.kt

@@ -47,12 +47,33 @@ class SqlProjectDatabaseImpl(private val dataSource: DataSource) : ProjectDataba
       dataSource.setURL(H2_IN_MEMORY_URL)
       return SqlProjectDatabaseImpl(dataSource)
     }
+
+    fun createInMemoryDatabase(logUpdateCallback: () -> Unit): ProjectDatabase {
+      val dataSource = JdbcDataSource()
+      dataSource.setURL(H2_IN_MEMORY_URL)
+      val database = SqlProjectDatabaseImpl(dataSource)
+      database.addLogUpdateCallback(logUpdateCallback)
+      return database
+    }
   }
 
   /** Queries which belong to the current transaction. Null if each statement should be committed separately. */
   private var currentTxn: TransactionImpl? = null
   private var localTxnId: Int = 1
 
+  private val logUpdateCallbacks: MutableList<() -> Unit> = mutableListOf()
+
+  /** Log update callbacks are invoked when a new log record is added. */
+  fun addLogUpdateCallback(listener: () -> Unit) = logUpdateCallbacks.add(listener)
+
+  private fun onLogUpdate() = logUpdateCallbacks.forEach {
+    try {
+      it.invoke()
+    } catch (e: Exception) {
+      LOG.error("Failed to execute update callback", e)
+    }
+  }
+
   private fun <T> withDSL(
     errorMessage: () -> String = { "Failed to execute query" },
     body: (dsl: DSLContext) -> T
@@ -90,6 +111,7 @@ class SqlProjectDatabaseImpl(private val dataSource: DataSource) : ProjectDataba
         }
       }
     }
+    onLogUpdate()
   }
 
   /** Add a query to the current txn. Executes immediately if no transaction started. */
@@ -155,6 +177,7 @@ class SqlProjectDatabaseImpl(private val dataSource: DataSource) : ProjectDataba
   @Throws(ProjectDatabaseException::class)
   internal fun commitTransaction(txn: TransactionImpl) {
     try {
+      if (txn.statements.isEmpty()) return
       executeAndLog(txn.statements, localTxnId)
       localTxnId++  // Increment only on success.
     } finally {