-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #18 from Kilemonn/add-support-for-mongo
Add support for mongo
- Loading branch information
Showing
18 changed files
with
593 additions
and
30 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
43 changes: 43 additions & 0 deletions
43
src/main/kotlin/au/kilemon/messagequeue/message/QueueMessageDocument.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package au.kilemon.messagequeue.message | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonIgnore | ||
| import org.springframework.data.annotation.Id | ||
| import org.springframework.data.mongodb.core.mapping.Document | ||
| import org.springframework.util.SerializationUtils | ||
| import java.util.* | ||
| import javax.persistence.* | ||
|
|
||
| /** | ||
| * This is used for `No-SQL` queues. | ||
| * | ||
| * @author github.com/Kilemonn | ||
| */ | ||
| @Document(value = QueueMessageDocument.DOCUMENT_NAME) | ||
| class QueueMessageDocument(var payload: Any?, var type: String, var assignedTo: String? = null) | ||
| { | ||
| companion object | ||
| { | ||
| const val DOCUMENT_NAME: String = "multiqueuemessages" | ||
| } | ||
|
|
||
| var uuid: String = UUID.randomUUID().toString() | ||
|
|
||
| @JsonIgnore | ||
| @Id | ||
| var id: Long? = null | ||
|
|
||
| /** | ||
| * Required for JSON deserialisation. | ||
| */ | ||
| constructor() : this(null, "") | ||
|
|
||
| constructor(queueMessage: QueueMessage) : this() | ||
| { | ||
| val resolvedQueueMessage = queueMessage.resolvePayloadObject() | ||
| this.type = resolvedQueueMessage.type | ||
| this.uuid = resolvedQueueMessage.uuid | ||
| this.id = resolvedQueueMessage.id | ||
| this.payload = resolvedQueueMessage.payload | ||
| this.assignedTo = resolvedQueueMessage.assignedTo | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
174 changes: 174 additions & 0 deletions
174
src/main/kotlin/au/kilemon/messagequeue/queue/nosql/mongo/MongoMultiQueue.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,174 @@ | ||
| package au.kilemon.messagequeue.queue.nosql.mongo | ||
|
|
||
| import au.kilemon.messagequeue.logging.HasLogger | ||
| import au.kilemon.messagequeue.message.QueueMessage | ||
| import au.kilemon.messagequeue.message.QueueMessageDocument | ||
| import au.kilemon.messagequeue.queue.MultiQueue | ||
| import au.kilemon.messagequeue.queue.exception.MessageUpdateException | ||
| import au.kilemon.messagequeue.queue.nosql.mongo.repository.MongoQueueMessageRepository | ||
| import org.slf4j.Logger | ||
| import org.springframework.beans.factory.annotation.Autowired | ||
| import org.springframework.context.annotation.Lazy | ||
| import java.util.* | ||
| import java.util.concurrent.ConcurrentLinkedQueue | ||
| import java.util.concurrent.atomic.AtomicLong | ||
|
|
||
| /** | ||
| * A NoSql mongo backed [MultiQueue]. All operations are performed directly on the database it is the complete source of truth. | ||
| * It allows the messages to never go out of sync in a case where there are multiple [MultiQueue]s working on the same data source. | ||
| * | ||
| * @author github.com/Kilemonn | ||
| */ | ||
| class MongoMultiQueue : MultiQueue, HasLogger | ||
| { | ||
| companion object | ||
| { | ||
| const val INDEX_ID = "index_id" | ||
| } | ||
|
|
||
| override val LOG: Logger = initialiseLogger() | ||
|
|
||
| override lateinit var maxQueueIndex: HashMap<String, AtomicLong> | ||
|
|
||
| @Lazy | ||
| @Autowired | ||
| private lateinit var queueMessageRepository: MongoQueueMessageRepository | ||
|
|
||
| /** | ||
| * Just initialise map, so it's not null, but the SQL [QueueMessage] ID is maintained by the database. | ||
| */ | ||
| override fun initialiseQueueIndex() | ||
| { | ||
| maxQueueIndex = HashMap() | ||
| } | ||
|
|
||
| override fun persistMessage(message: QueueMessage) | ||
| { | ||
| val queueMessageDocument = QueueMessageDocument(message) | ||
| try | ||
| { | ||
| queueMessageRepository.save(queueMessageDocument) | ||
| } | ||
| catch (ex: Exception) | ||
| { | ||
| throw MessageUpdateException(message.uuid, ex) | ||
| } | ||
| } | ||
|
|
||
| override fun getQueueForType(queueType: String): Queue<QueueMessage> | ||
| { | ||
| val entries = queueMessageRepository.findByTypeOrderByIdAsc(queueType) | ||
| return ConcurrentLinkedQueue(entries.map { entry -> QueueMessage(entry) }) | ||
| } | ||
|
|
||
| override fun performHealthCheckInternal() | ||
| { | ||
| queueMessageRepository.existsById(1) | ||
| } | ||
|
|
||
| override fun getMessageByUUID(uuid: String): Optional<QueueMessage> | ||
| { | ||
| val documentMessage = queueMessageRepository.findByUuid(uuid) | ||
| return if (documentMessage.isPresent) | ||
| { | ||
| Optional.of(QueueMessage(documentMessage.get())) | ||
| } | ||
| else | ||
| { | ||
| Optional.empty() | ||
| } | ||
| } | ||
|
|
||
| override fun clearForTypeInternal(queueType: String): Int | ||
| { | ||
| val amountCleared = queueMessageRepository.deleteByType(queueType) | ||
| LOG.debug("Cleared existing queue for type [{}]. Removed [{}] message entries.", queueType, amountCleared) | ||
| return amountCleared | ||
| } | ||
|
|
||
| override fun isEmptyForType(queueType: String): Boolean | ||
| { | ||
| return queueMessageRepository.findByTypeOrderByIdAsc(queueType).isEmpty() | ||
| } | ||
|
|
||
| override fun pollInternal(queueType: String): Optional<QueueMessage> | ||
| { | ||
| val messages = queueMessageRepository.findByTypeOrderByIdAsc(queueType) | ||
| return if (messages.isNotEmpty()) | ||
| { | ||
| return Optional.of(QueueMessage(messages[0])) | ||
| } | ||
| else | ||
| { | ||
| Optional.empty() | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * The [includeEmpty] value makes no difference it is always effectively `false`. | ||
| */ | ||
| override fun keys(includeEmpty: Boolean): Set<String> | ||
| { | ||
| val keySet = queueMessageRepository.getDistinctTypes().toSet() | ||
| LOG.debug("Total amount of queue keys [{}].", keySet.size) | ||
| return keySet | ||
| } | ||
|
|
||
| override fun containsUUID(uuid: String): Optional<String> | ||
| { | ||
| val optionalMessage = queueMessageRepository.findByUuid(uuid) | ||
| return if (optionalMessage.isPresent) | ||
| { | ||
| val message = optionalMessage.get() | ||
| LOG.debug("Found queue type [{}] for UUID: [{}].", message.type, uuid) | ||
| Optional.of(message.type) | ||
| } | ||
| else | ||
| { | ||
| LOG.debug("No queue type exists for UUID: [{}].", uuid) | ||
| Optional.empty() | ||
| } | ||
| } | ||
|
|
||
| override fun addInternal(element: QueueMessage): Boolean | ||
| { | ||
| val queueMessageDocument = QueueMessageDocument(element) | ||
| val saved = queueMessageRepository.save(queueMessageDocument) | ||
| return saved.id != null | ||
| } | ||
|
|
||
| override fun removeInternal(element: QueueMessage): Boolean | ||
| { | ||
| val removedCount = queueMessageRepository.deleteByUuid(element.uuid) | ||
| return removedCount > 0 | ||
| } | ||
|
|
||
| /** | ||
| * Overriding to use the constant [INDEX_ID] for all look-ups since the ID is shared and needs to be assigned to | ||
| * the [QueueMessageDocument] before it is created. | ||
| */ | ||
| override fun getAndIncrementQueueIndex(queueType: String): Optional<Long> | ||
| { | ||
| return super.getAndIncrementQueueIndex(INDEX_ID) | ||
| } | ||
|
|
||
| /** | ||
| * Override to never clear the queue index for the type, since it's a shared index map. | ||
| */ | ||
| override fun clearQueueIndexForType(queueType: String) | ||
| { | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Clear the [maxQueueIndex] if the entire map is cleared. | ||
| * | ||
| * | ||
| * Since [clearQueueIndexForType] is not clearing any of map entries. | ||
| */ | ||
| override fun clear() | ||
| { | ||
| super.clear() | ||
| maxQueueIndex.clear() | ||
| } | ||
| } |
Oops, something went wrong.