Add LogRecord table update callbacks

cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/ColloboqueServer.kt

@@ -91,7 +91,7 @@ class ColloboqueServer(
               )
             }
         }
-        refidToBaseTxnId[projectRefid] = "abacaba"  // TODO: get from the database
+        refidToBaseTxnId[projectRefid] = EMPTY_LOG_BASE_TXN_ID  // TODO: get from the database
       }
     } catch (e: Exception) {
       throw ColloboqueServerException("Failed to init project $projectRefid", e)

cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/DevServer.kt

@@ -123,8 +123,8 @@ class ColloboqueWebSocketServer(port: Int, private val colloboqueServer: Collobo
     }
 
     override fun onMessage(message: WebSocketFrame) {
-      LOG.debug("Message received\n {}", message.textPayload)
       val inputXlog = parseInputXlog(message.textPayload) ?: return
+      LOG.debug("Xlog received\n {}", inputXlog)
       if (inputXlog.transactions.size != 1) {
         // TODO: add multiple transactions support.
         LOG.error("Only single transaction commit supported")
@@ -138,7 +138,7 @@ class ColloboqueWebSocketServer(port: Int, private val colloboqueServer: Collobo
     override fun onPong(pong: WebSocketFrame?) {}
 
     override fun onException(exception: IOException) {
-      LOG.error("WebSocket exception", exception)
+      LOG.error("WebSocket exception", exception = exception)
     }
   }
 }

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

@@ -69,9 +69,9 @@
 import net.sourceforge.ganttproject.storage.InputXlog;
 import net.sourceforge.ganttproject.storage.ProjectDatabaseException;
 import net.sourceforge.ganttproject.storage.ServerCommitResponse;
+import net.sourceforge.ganttproject.storage.XlogRecord;
 import net.sourceforge.ganttproject.task.CustomColumnsStorage;
 import net.sourceforge.ganttproject.task.Task;
-import net.sourceforge.ganttproject.task.event.TaskListenerAdapter;
 import net.sourceforge.ganttproject.undo.GPUndoListener;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.jetbrains.annotations.NotNull;
@@ -82,15 +82,14 @@
 import java.awt.*;
 import java.awt.event.*;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
+import java.util.*;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 import static biz.ganttproject.storage.cloud.GPCloudHttpImplKt.getWebSocket;
 import static biz.ganttproject.storage.cloud.GPCloudHttpImplKt.isColloboqueLocalTest;
+import static net.sourceforge.ganttproject.storage.XlogKt.EMPTY_LOG_BASE_TXN_ID;
 
 /**
  * Main frame of the project
@@ -167,7 +166,7 @@ private static class TxnCommitInfo {
     /** If `oldTxnId` is currently being hold, sets the txn ID to `newTxnId` and moves the local ID ahead by `committedNum`. */
     void update(String oldTxnId, String newTxnId, int committedNum) {
       myTxnId.updateAndGet(oldValue -> {
-        if (oldValue.left.equals(oldTxnId)) {
+        if (Objects.equals(oldValue.left, oldTxnId)) {
           return new ImmutablePair<>(newTxnId, oldValue.right + committedNum);
         } else {
           return oldValue;
@@ -178,9 +177,13 @@ void update(String oldTxnId, String newTxnId, int committedNum) {
     ImmutablePair<String, Integer> get() {
       return myTxnId.get();
     }
+
+    void reset() {
+      myTxnId.set(new ImmutablePair<>(null, 0));
+    }
   }
 
-  private final TxnCommitInfo myBaseTxnCommitInfo = new TxnCommitInfo("", 0);
+  private final TxnCommitInfo myBaseTxnCommitInfo = new TxnCommitInfo(null, 0);
 
 
   public GanttProject(boolean isOnlyViewer) {
@@ -208,10 +211,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,33 +955,73 @@ public void refresh() {
     super.repaint();
   }
 
-  // TODO: Accumulate changes instead of sending it every time.
+  private interface TxnSendListener {
+    void onSendCompleted();
+  }
+
+  private final AtomicReference<TxnSendListener> txnSendingListener = new AtomicReference<>();
+
   private Unit sendProjectStateLogs() {
     gpLogger.debug("Sending project state logs");
+    if (txnSendingListener.get() != null) return Unit.INSTANCE;
+    var baseTxnCommitInfo = myBaseTxnCommitInfo.get();
+    if (baseTxnCommitInfo.left == null) {
+      // Connection with the server was not established.
+      return Unit.INSTANCE;
+    }
+    List<XlogRecord> transactions;
     try {
-      var baseTxnCommitInfo = myBaseTxnCommitInfo.get();
-      var txns = myProjectDatabase.fetchTransactions(baseTxnCommitInfo.right + 1, 1);
-      if (!txns.isEmpty()) {
+      transactions = myProjectDatabase.fetchTransactions(baseTxnCommitInfo.right + 1, 1);
+    } catch (ProjectDatabaseException e) {
+      gpLogger.error("Failed to send logs", new Object[]{}, ImmutableMap.of(), e);
+      return Unit.INSTANCE;
+    }
+    if (!transactions.isEmpty()) {
+      var listener = new TxnSendListener() {
+        @Override
+        public void onSendCompleted() {
+          txnSendingListener.compareAndSet(this, null);
+        }
+      };
+      if (txnSendingListener.compareAndSet(null, listener)) {
         getWebSocket().sendLogs(new InputXlog(
           baseTxnCommitInfo.left,
           "userId",
           "refid",
-          txns
+          transactions
         ));
+      } else {
+        // Logs were sent by another thread, no action required.
       }
-    } catch (ProjectDatabaseException e) {
-      gpLogger.error("Failed to send logs", new Object[]{}, ImmutableMap.of(), e);
+    }
+    return Unit.INSTANCE;
+  }
+
+  @Override
+  protected Unit onProjectLogUpdate() {
+    if (isColloboqueLocalTest()) {
+      super.onProjectLogUpdate();
+      sendProjectStateLogs();
     }
     return Unit.INSTANCE;
   }
 
   private Unit fireXlogReceived(ServerCommitResponse response) {
     myBaseTxnCommitInfo.update(response.getBaseTxnId(), response.getNewBaseTxnId(), 1);
+    txnSendingListener.get().onSendCompleted();
+    sendProjectStateLogs();
     return Unit.INSTANCE;
   }
 
+  // TODO: sync logs with the server.
   private Unit onBaseTxnIdReceived(String baseTxnId) {
-    myBaseTxnCommitInfo.update("", baseTxnId, 0);
+    // Websocket is [re-]started. Previous messages are discarded.
+    txnSendingListener.set(null);
+    if (EMPTY_LOG_BASE_TXN_ID.equals(baseTxnId)) {
+      myBaseTxnCommitInfo.reset();
+      myBaseTxnCommitInfo.update(null, 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,7 @@ of the License, or (at your option) any later version.
 import java.util.Map;
 import java.util.function.Supplier;
 
+
 /**
  * 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 +194,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 +263,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 {

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

@@ -81,4 +81,7 @@ data class ServerCommitError(
 )
 
 const val SERVER_COMMIT_RESPONSE_TYPE = "ServerCommitResponse"
-const val SERVER_COMMIT_ERROR_TYPE = "ServerCommitError"
+const val SERVER_COMMIT_ERROR_TYPE = "ServerCommitError"
+
+/** Base txn ID for the empty log state. */
+const val EMPTY_LOG_BASE_TXN_ID = "abacaba"