redis-daemon/dev.ust.zinc.grader.daemon/MessagingManager

MessagingManager

object MessagingManager : Messenger, CoroutineScope

Contains methods for processing jobs from worker queues and subscribed channels

Properties

open override val apiQueue: String

coroutineContext

open override val coroutineContext: CoroutineContext

open override val graderQueue: String

Names of the queues required

graderReservedQueue

open override val graderReservedQueue: String

private val logger: Logger

val sleepFor: Duration

Adjustable sleeping duration between polling in queueUp

waitUntilReconnect

open override val waitUntilReconnect: Duration

How long to wait until attempting a reconnection

Functions

open suspend override fun cleanReserved()

Cleans any reserved jobs from reserved queue

open override fun close()

decompressingStatus

open override fun decompressingStatus(assignmentConfigId: Long): String

Names of the statuses required

open suspend override fun deleteReserved(jobName: String)

Deletes job of jobName from graderReservedQueue

getDecompressing

open suspend override fun getDecompressing(assignmentConfigId: Long): Set<Long>

Retrieve the set of all submission ids under assignmentConfigId that are still undergoing decompression

open suspend override fun getNextJob(): String?

Gets next job from graderQueue

open override fun gradingStatus(assignmentConfigId: Long): String

logFailedConnection

open override fun logFailedConnection(type: String, tr: Throwable)

Common logging for failing to connect to the messenger

open suspend override fun <R> loopingTry(type: String, resetBlock: suspend () -> Unit, block: suspend () -> R): R

Runs block, retrying every period set by waitUntilReconnect until the operation succeeds.

open suspend override fun loopRecover(type: String, resetBlock: suspend () -> Unit, block: suspend () -> Unit)

Always run block, and executes resetBlock when an exception occurs.

makeGradingStatusString

open override fun makeGradingStatusString(stageName: String, details: String): String

Generate the status string for the grading-in-progress state

private fun pop(reservedRawJobMessage: String?, jobParamWithRaw: (String) -> JobParam): JobRequest?

Pops and parse a queued job to a JobRequest

open suspend override fun publish(jobName: String, payload: Payload)

Publish payload to jobName

open suspend override fun push(queueName: String, jobMessage: JobMessage)

Push jobs as jobMessage to queueName for external system components to consume

suspend fun pushToApi(jobMessage: JobMessage)

Push jobs as jobMessage to Messenger.apiQueue

suspend fun pushToGrader(jobMessage: JobMessage)

Push jobs as jobMessage to Messenger.graderQueue

open suspend override fun queueUp()

Blocking queue to consume incoming jobs on Messenger.graderQueue

suspend fun runOnce(): JobRequest?

Step function to process one job from the Messenger.graderQueue

fun runOnce(channel: String?, message: String?)

Step function to process one provided message from the subscription channels

setGradingProgress

open suspend override fun setGradingProgress(assignmentConfigId: Long, submissionId: Long, status: String?)

Update the status string of submissionId under assignmentConfigId to status if it is not-null, or delete the key submissionId to nil if status is null, to indicate this submission is no longer active

open suspend override fun subscribe()

Blocking subscribe to Job.subscriptions channels on Redis