keymemimg.kt

interface Command { fun applyTo(system: Any) }

interface Query { fun extractFrom(system: Any): Any?}

interface EventSourcing {
    fun append(event: Any)
    fun <E> replay(eventConsumer: (E) -> Unit)
}

interface TxManager {
  fun begin()
  fun <T> remember(who: Any, what: String, value: T, undo: (T) -> Unit)
  fun rollback()
  companion object: TxManager { ... }
}
​
class MemoryImageProcessor(private val system: Any, 
                           private val eventSourcing: EventSourcing)
{
    init {
        synchronized(this) {
            // Replay serialized events to restore in-memory state
            eventSourcing.replay<Command> { command -> command.applyTo(system) }
        }
    }
​
    // Apply incoming command to system, single-threaded
    fun execute(command: Command): Unit = synchronized(this) { 
        TxManager.begin()
        try {
            command.applyTo(system) // Try and apply command
            try {
                // Serialize; retry internally if needed
                eventSourcing.append(command) 
            } catch (e: Exception) {
                // No attempt to rollback; irrecoverable
                logger.severe("Error persisting: ${e.message}")
                throw e
            }
        } catch (e: Exception) {
            TxManager.rollback() // Undo any partial mutation
            // It's (kinda) ok for a command to fail; rethrow
            throw CommandApplicationException("Command error", e)
        }
    }
​
    // Run incoming query on system, multi-threaded
    fun execute(query: Query): Any? = query.extractFrom(system)
}