diff --git a/cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/ColloboqueServer.kt b/cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/ColloboqueServer.kt index a85bb1f05c..d52d31c414 100644 --- a/cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/ColloboqueServer.kt +++ b/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) diff --git a/cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/DevServer.kt b/cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/DevServer.kt index ee8f64a53d..9f6ff22bc4 100644 --- a/cloud.ganttproject.colloboque/src/main/kotlin/cloud/ganttproject/colloboque/DevServer.kt +++ b/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) } } } diff --git a/ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProject.java b/ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProject.java index 2ff84c2f79..00a5efb449 100644 --- a/ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProject.java +++ b/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 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 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 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; } } diff --git a/ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProjectBase.java b/ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProjectBase.java index e02472d5d7..2a54028d97 100644 --- a/ganttproject/src/main/java/net/sourceforge/ganttproject/GanttProjectBase.java +++ b/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()); diff --git a/ganttproject/src/main/java/net/sourceforge/ganttproject/storage/SqlProjectDatabaseImpl.kt b/ganttproject/src/main/java/net/sourceforge/ganttproject/storage/SqlProjectDatabaseImpl.kt index 4bcb66583c..34b31ec644 100644 --- a/ganttproject/src/main/java/net/sourceforge/ganttproject/storage/SqlProjectDatabaseImpl.kt +++ b/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 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 { diff --git a/ganttproject/src/main/java/net/sourceforge/ganttproject/storage/Xlog.kt b/ganttproject/src/main/java/net/sourceforge/ganttproject/storage/Xlog.kt index f9ba20911e..2b0b2b0446 100644 --- a/ganttproject/src/main/java/net/sourceforge/ganttproject/storage/Xlog.kt +++ b/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" \ No newline at end of file +const val SERVER_COMMIT_ERROR_TYPE = "ServerCommitError" + +/** Base txn ID for the empty log state. */ +const val EMPTY_LOG_BASE_TXN_ID = "abacaba" \ No newline at end of file