test(model-server): IStoreClient transactions

Author: slisson

.gitignore

@@ -1,6 +1,7 @@
 .gradle/
 /build/
 /*/build/
+/*/ignite/
 .DS_Store
 .gradletasknamecache
 .idea/

model-server/build.gradle.kts

@@ -62,6 +62,7 @@ dependencies {
     testImplementation(libs.junit)
     testImplementation(libs.cucumber.java)
     testImplementation(libs.ktor.server.test.host)
+    testImplementation(libs.kotlin.coroutines.test)
     testImplementation(kotlin("test"))
     testImplementation(project(":modelql-untyped"))
 }

model-server/src/main/kotlin/org/modelix/model/server/store/IStoreClient.kt

@@ -21,10 +21,11 @@ import kotlinx.coroutines.withTimeoutOrNull
 import org.modelix.model.IKeyListener
 import kotlin.time.Duration.Companion.seconds
 
-interface IStoreClient {
+interface IStoreClient : AutoCloseable {
     operator fun get(key: String): String?
     fun getAll(keys: List<String>): List<String?>
     fun getAll(keys: Set<String>): Map<String, String?>
+    fun getAll(): Map<String, String?>
     fun put(key: String, value: String?, silent: Boolean = false)
     fun putAll(entries: Map<String, String?>, silent: Boolean = false)
     fun listen(key: String, listener: IKeyListener)

model-server/src/main/kotlin/org/modelix/model/server/store/IgniteStoreClient.kt

@@ -26,7 +26,7 @@ import java.util.*
 import java.util.concurrent.Executors
 import java.util.stream.Collectors
 
-class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
+class IgniteStoreClient(jdbcConfFile: File? = null, inmemory: Boolean = false) : IStoreClient, AutoCloseable {
     private val ignite: Ignite
     private val cache: IgniteCache<String, String?>
     private val timer = Executors.newScheduledThreadPool(1)
@@ -63,7 +63,7 @@ class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
                 )
             }
         }
-        ignite = Ignition.start(javaClass.getResource("ignite.xml"))
+        ignite = Ignition.start(javaClass.getResource(if (inmemory) "ignite-inmemory.xml" else "ignite.xml"))
         cache = ignite.getOrCreateCache("model")
         //        timer.scheduleAtFixedRate(() -> {
         //            System.out.println("stats: " + cache.metrics());
@@ -83,6 +83,10 @@ class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
         return cache.getAll(keys)
     }
 
+    override fun getAll(): Map<String, String?> {
+        return cache.associate { it.key to it.value }
+    }
+
     override fun put(key: String, value: String?, silent: Boolean) {
         putAll(Collections.singletonMap(key, value), silent)
     }
@@ -151,4 +155,8 @@ class IgniteStoreClient(jdbcConfFile: File?) : IStoreClient {
     fun dispose() {
         ignite.close()
     }
+
+    override fun close() {
+        dispose()
+    }
 }

model-server/src/main/kotlin/org/modelix/model/server/store/InMemoryStoreClient.kt

@@ -20,7 +20,7 @@ import java.io.BufferedReader
 import java.io.FileReader
 import java.io.FileWriter
 import java.io.IOException
-import java.util.*
+import kotlin.collections.HashMap
 
 fun generateId(idStr: String?): Long {
     return try {
@@ -41,26 +41,32 @@ class InMemoryStoreClient : IStoreClient {
     }
 
     private val values: MutableMap<String, String?> = HashMap()
+    private var transactionValues: MutableMap<String, String?>? = null
     private val listeners: MutableMap<String?, MutableSet<IKeyListener>> = HashMap()
 
     @Synchronized
     override fun get(key: String): String? {
-        return values[key]
+        return if (transactionValues?.contains(key) == true) transactionValues!![key] else values[key]
     }
 
     @Synchronized
     override fun getAll(keys: List<String>): List<String?> {
-        return keys.map { values[it] }
+        return keys.map { get(it) }
+    }
+
+    @Synchronized
+    override fun getAll(): Map<String, String?> {
+        return values + (transactionValues ?: emptyMap())
     }
 
     @Synchronized
     override fun getAll(keys: Set<String>): Map<String, String?> {
-        return keys.associateWith { values[it] }
+        return keys.associateWith { get(it) }
     }
 
     @Synchronized
     override fun put(key: String, value: String?, silent: Boolean) {
-        values[key] = value
+        (transactionValues ?: values)[key] = value
         if (!silent) {
             listeners[key]?.toList()?.forEach {
                 try {
@@ -124,6 +130,20 @@ class InMemoryStoreClient : IStoreClient {
 
     @Synchronized
     override fun <T> runTransaction(body: () -> T): T {
-        return body()
+        if (transactionValues == null) {
+            try {
+                transactionValues = HashMap()
+                val result = body()
+                values.putAll(transactionValues!!)
+                return result
+            } finally {
+                transactionValues = null
+            }
+        } else {
+            return body()
+        }
+    }
+
+    override fun close() {
     }
 }

model-server/src/main/resources/org/modelix/model/server/store/ignite-inmemory.xml

@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+-->
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:util="http://www.springframework.org/schema/util"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans
+                           http://www.springframework.org/schema/beans/spring-beans.xsd
+                           http://www.springframework.org/schema/util
+                           http://www.springframework.org/schema/util/spring-util.xsd">
+
+    <bean id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
+        <property name="consistentId" value="inmemory"/>
+        <property name="cacheConfiguration">
+            <list>
+                <bean class="org.apache.ignite.configuration.CacheConfiguration">
+                    <property name="name" value="model"/>
+                    <property name="atomicityMode" value="TRANSACTIONAL"/>
+                    <property name="backups" value="0"/>
+                    <property name="statisticsEnabled" value="true"/>
+                </bean>
+            </list>
+        </property>
+    </bean>
+</beans>

model-server/src/test/kotlin/org/modelix/model/server/StoreClientTest.kt

@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) 2023.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.modelix.model.server
+
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import org.modelix.model.server.store.IStoreClient
+import org.modelix.model.server.store.IgniteStoreClient
+import org.modelix.model.server.store.InMemoryStoreClient
+import kotlin.random.Random
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertNull
+
+class MapBasedStoreClientTest : StoreClientTest(InMemoryStoreClient())
+class IgniteStoreClientTest : StoreClientTest(IgniteStoreClient(inmemory = true))
+
+abstract class StoreClientTest(val store: IStoreClient) {
+
+    @AfterTest
+    fun cleanup() {
+        store.close()
+    }
+
+    @Test
+    fun `transaction can be started from inside a transaction`() {
+        store.runTransaction {
+            store.runTransaction {
+                store.put("abc", "def")
+            }
+        }
+    }
+
+    @Test
+    fun `allow put without transaction`() {
+        val key = "ljnrdlfkesmgf"
+        val value = "izujztdrsew"
+        assertNull(store.get(key))
+        store.put(key, value)
+        assertEquals(value, store.get(key))
+    }
+
+    @Test
+    fun `transactions are isolated`() = runTest {
+        val key = "kjhndsweret"
+        repeat(2) {
+            val rand = Random(it)
+            launch {
+                store.runTransaction {
+                    repeat(10) {
+                        val value = rand.nextInt().toString()
+                        store.put(key, value)
+                        assertEquals(value, store.get(key))
+                        Thread.sleep(rand.nextLong(5, 10))
+                        assertEquals(value, store.get(key))
+                    }
+                }
+            }
+        }
+    }
+
+    @Test
+    fun `transactions are atomic`() = runTest {
+        val key = "ioudgbnr"
+        val value1 = "a"
+        val value2 = "b"
+
+        store.put(key, value1)
+        assertEquals(value1, store.get(key))
+        assertFailsWith(NullPointerException::class) {
+            store.runTransaction {
+                store.put(key, value2)
+                assertEquals(value2, store.get(key))
+                throw NullPointerException()
+            }
+        }
+        assertEquals(value1, store.get(key)) // failed transaction should be rolled back
+    }
+}