Implement RhtPQMap (#10)

buildSrc/src/main/kotlin/Versions.kt

@@ -3,4 +3,5 @@ object Versions {
     const val grpc = "1.49.0"
     const val grpcKotlin = "1.3.0"
     const val composeUi = "1.2.1"
+    const val coroutines = "1.6.4"
 }

yorkie/build.gradle.kts

@@ -97,9 +97,10 @@ dependencies {
     implementation("io.grpc:grpc-android:${Versions.grpc}")
     implementation("io.grpc:grpc-okhttp:${Versions.grpc}")
 
-    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.4")
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:${Versions.coroutines}")
 
     testImplementation("junit:junit:4.13.2")
+    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:${Versions.coroutines}")
     androidTestImplementation("androidx.test.ext:junit:1.1.3")
     androidTestImplementation("androidx.test.espresso:espresso-core:3.4.0")
 }

yorkie/src/main/kotlin/dev/yorkie/document/crdt/CrdtElement.kt

@@ -31,10 +31,11 @@ internal abstract class CrdtElement(
         }
 
     fun remove(removedAt: TimeTicket?): Boolean {
-        if (removedAt != null && createdAt < removedAt &&
-            (this.removedAt == null || checkNotNull(this.removedAt) < removedAt)
+        val removedAtTicket = removedAt ?: return false
+        if (createdAt < removedAtTicket &&
+            (this.removedAt == null || checkNotNull(this.removedAt) < removedAtTicket)
         ) {
-            this.removedAt = removedAt
+            this.removedAt = removedAtTicket
             return true
         }
         return false

yorkie/src/main/kotlin/dev/yorkie/document/crdt/Primitive.kt

@@ -1,6 +1,7 @@
 package dev.yorkie.document.crdt
 
 import dev.yorkie.document.time.TimeTicket
+import java.util.Date
 
 internal class Primitive(
     val value: Any?,
@@ -16,7 +17,7 @@ internal class Primitive(
             is Double -> PrimitiveType.Double
             is String -> PrimitiveType.String
             is ByteArray -> PrimitiveType.Bytes
-            // TODO(daeyounglnc): support Date type
+            is Date -> PrimitiveType.Date
             else -> PrimitiveType.Null
         }
     }

yorkie/src/main/kotlin/dev/yorkie/document/crdt/RhtPQMap.kt

@@ -0,0 +1,156 @@
+package dev.yorkie.document.crdt
+
+import dev.yorkie.document.time.TimeTicket
+import dev.yorkie.util.MaxPriorityQueue
+import dev.yorkie.util.PQNode
+import dev.yorkie.util.YorkieLogger
+
+/**
+ * [RhtPQMap] is replicated hash table with a priority queue by creation time.
+ */
+internal class RhtPQMap {
+    private val logTag = "RhtPQMap"
+
+    private val elementQueueMapByKey:
+        MutableMap<String, MaxPriorityQueue<PQNode<TimeTicket, CrdtElement>>> = mutableMapOf()
+
+    private val nodeMapByCreatedAt: MutableMap<TimeTicket, RhtPQMapNode> = mutableMapOf()
+
+    /**
+     * Sets the [value] using the given [key].
+     * If the object exists in [elementQueueMapByKey] by same [key] then return [CrdtElement], otherwise null.
+     */
+    fun set(key: String, value: CrdtElement): CrdtElement? {
+        var removed: CrdtElement? = null
+        val queue = elementQueueMapByKey[key]
+        if (queue != null && queue.size > 0) {
+            val node = queue.peek() as RhtPQMapNode
+            if (!node.isRemoved() && node.remove(value.createdAt)) {
+                removed = node.value
+            }
+        }
+        setInternal(key, value)
+        return removed
+    }
+
+    /**
+     * Sets the [value] using the given [key] internally.
+     */
+    private fun setInternal(key: String, value: CrdtElement) {
+        val queueMap = elementQueueMapByKey
+        if (!queueMap.contains(key)) {
+            queueMap[key] = MaxPriorityQueue()
+        }
+
+        val node = RhtPQMapNode.of(key, value)
+        val queue = queueMap[key]
+            ?: throw IllegalStateException("The MaxPriorityQueue by $key doesn't exist.")
+        queue.add(node)
+        nodeMapByCreatedAt[value.createdAt] = node
+    }
+
+    /**
+     * Deletes the Element in [nodeMapByCreatedAt] using the given key [createdAt],
+     * and TimeTicket will be removed by [executedAt].
+     * If the object exists in [nodeMapByCreatedAt] by same key [createdAt], it will return.
+     *
+     * @throws IllegalStateException if RHTPQMapNode doesn't exist.
+     */
+    fun delete(createdAt: TimeTicket, executedAt: TimeTicket): CrdtElement {
+        val nodeMap = nodeMapByCreatedAt
+        if (!nodeMap.contains(createdAt)) {
+            YorkieLogger.e(logTag, "fail to find $createdAt")
+        }
+
+        val node = nodeMap[createdAt]
+            ?: throw IllegalStateException("The RHTPQMapNode by $createdAt doesn't exist.")
+        node.remove(executedAt)
+        return node.value
+    }
+
+    /**
+     * Returns [RhtPQMapNode.strKey] of node based on creation time.
+     * The node will be found in [nodeMapByCreatedAt] using [createdAt]
+     *
+     * @throws IllegalStateException if RHTPQMapNode doesn't exist.
+     */
+    fun subPathOf(createdAt: TimeTicket): String {
+        return nodeMapByCreatedAt[createdAt]?.strKey
+            ?: throw IllegalStateException("RHTPQMapNode's strKey by $createdAt doesn't exist")
+    }
+
+    /**
+     * Physically purges child [element] from [nodeMapByCreatedAt] and [elementQueueMapByKey]
+     */
+    fun purge(element: CrdtElement) {
+        val nodeMap = nodeMapByCreatedAt
+        val node = nodeMap[element.createdAt]
+
+        if (node == null) {
+            YorkieLogger.e(logTag, "fail to find ${element.createdAt}")
+            return
+        }
+
+        val queue = elementQueueMapByKey[node.strKey]
+        if (queue == null) {
+            YorkieLogger.e(logTag, "fail to find queue of ${node.strKey}")
+            return
+        }
+
+        queue.remove(node)
+        nodeMap.remove(node.value.createdAt)
+    }
+
+    /**
+     * Deletes the element in [elementQueueMapByKey] using the given [key] and [removedAt].
+     * If the element removed successfully, removed [CrdtElement] will return.
+     *
+     * @throws IllegalStateException if MaxPriorityQueue doesn't exist.
+     */
+    fun deleteByKey(key: String, removedAt: TimeTicket): CrdtElement {
+        val queue = elementQueueMapByKey[key]
+            ?: throw IllegalStateException("MaxPriorityQueue by $key doesn't exist")
+        val node = queue.peek() as RhtPQMapNode
+        node.remove(removedAt)
+        return node.value
+    }
+
+    /**
+     * Checks the element exists in [elementQueueMapByKey] using [key].
+     * If the RHTPQMapNode is exist, then returns true, otherwise false.
+     */
+    fun has(key: String): Boolean {
+        val queue = elementQueueMapByKey[key] ?: return false
+        val node = queue.peek() as RhtPQMapNode
+        return !node.isRemoved()
+    }
+
+    /**
+     * Returns the [CrdtElement] using given [key].
+     *
+     * @throws IllegalStateException if MaxPriorityQueue doesn't exist.
+     */
+    fun get(key: String): CrdtElement {
+        return elementQueueMapByKey[key]?.peek()?.value
+            ?: throw IllegalStateException("MaxPriorityQueue by $key doesn't exist")
+    }
+
+    /**
+     * Returns the sequence of [elementQueueMapByKey]'s values
+     */
+    fun getKeyOfQueue(): Sequence<RhtPQMapNode> {
+        return elementQueueMapByKey.values
+            .asSequence()
+            .map { it.element() as RhtPQMapNode }
+    }
+
+    companion object {
+        /**
+         * Creates a instance of [RhtPQMap]
+         */
+        @JvmStatic
+        fun create(): RhtPQMap {
+            return RhtPQMap()
+        }
+    }
+}

yorkie/src/main/kotlin/dev/yorkie/document/crdt/RhtPQMapNode.kt

@@ -0,0 +1,39 @@
+package dev.yorkie.document.crdt
+
+import dev.yorkie.document.time.TimeTicket
+import dev.yorkie.util.PQNode
+
+/**
+ * [RhtPQMapNode] is a node of RHTPQMap.
+ */
+internal class RhtPQMapNode(val strKey: String, value: CrdtElement) :
+    PQNode<TimeTicket, CrdtElement>(value.createdAt, value) {
+
+    /**
+     * Checks whether this value was removed.
+     */
+    fun isRemoved(): Boolean {
+        return value.isRemoved
+    }
+
+    /**
+     * Removes a value base on removing time.
+     */
+    fun remove(removedAt: TimeTicket): Boolean {
+        return value.remove(removedAt)
+    }
+
+    companion object {
+        /**
+         * Creates an instance of [RhtPQMapNode].
+         */
+        @JvmStatic
+        fun of(strKey: String, value: CrdtElement): RhtPQMapNode {
+            return RhtPQMapNode(strKey, value)
+        }
+    }
+
+    override fun compareTo(other: PQNode<TimeTicket, CrdtElement>): Int {
+        return key.compareTo(other.key)
+    }
+}

yorkie/src/main/kotlin/dev/yorkie/document/time/TimeTicket.kt

@@ -9,13 +9,39 @@ internal data class TimeTicket(
     val actorID: ActorID,
 ) : Comparable<TimeTicket> {
 
+    fun toIDString(): String {
+        return "$lamport:$actorID:$delimiter"
+    }
+
+    override fun toString(): String {
+        return toIDString()
+    }
+
     override fun compareTo(other: TimeTicket): Int {
         return lamport.compareTo(other.lamport).takeUnless { it == 0 }
             ?: actorID.compareTo(other.actorID).takeUnless { it == 0 }
             ?: delimiter.compareTo(delimiter)
     }
 
-    fun toIDString() = "$lamport:$actorID$delimiter"
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (javaClass != other?.javaClass) return false
+
+        other as TimeTicket
+
+        if (lamport != other.lamport) return false
+        if (delimiter != other.delimiter) return false
+        if (actorID != other.actorID) return false
+
+        return true
+    }
+
+    override fun hashCode(): Int {
+        var result = lamport.hashCode()
+        result = 31 * result + delimiter
+        result = 31 * result + actorID.hashCode()
+        return result
+    }
 
     companion object {
         const val INITIAL_DELIMITER = 0

yorkie/src/main/kotlin/dev/yorkie/util/MaxPriorityQueue.kt

@@ -14,3 +14,23 @@ internal class MaxPriorityQueue<E : Comparable<E>>(
         private const val DEFAULT_INITIAL_CAPACITY = 11
     }
 }
+
+/**
+ * [PQNode] is a node of [MaxPriorityQueue].
+ */
+internal abstract class PQNode<K, V>(internal val key: K, internal val value: V) :
+    Comparable<PQNode<K, V>> {
+    /**
+     * Returns the key of this [PQNode].
+     */
+    fun getKey(): K {
+        return key
+    }
+
+    /**
+     * Returns the value of [PQNode].
+     */
+    fun getValue(): V {
+        return this.value
+    }
+}

yorkie/src/main/kotlin/dev/yorkie/util/YorkieLogger.kt

@@ -0,0 +1,21 @@
+package dev.yorkie.util
+
+internal object YorkieLogger {
+    private const val TAG_PREFIX = "Yorkie."
+
+    var logger: Logger? = null
+
+    fun d(tag: String, message: String) {
+        logger?.d("$TAG_PREFIX$tag", message)
+    }
+
+    fun e(tag: String, message: String) {
+        logger?.e("$TAG_PREFIX$tag", message)
+    }
+}
+
+interface Logger {
+    fun d(tag: String, message: String)
+
+    fun e(tag: String, message: String)
+}