Hello, I'm excited to have found this library. Thanks for doing this!

This is not really issue, but more of an example of getting this to work using Kotlin and with a CoroutineScope.

Background

I am building an app to allow me to keep track of vegetable growing activities
It is modeled after Square Foot Gardening in which a bed is modeled as individual cells of 1 sq ft. For example, a 4 by 8 bed has four rows, eight columns and is 32 thus 32 cells
I want to allow any number of clients to observe activities against a given Bed
When a gardener in the fields is tending a given Bed and invokes the 'Water Bed', indicating specifically that only Cells 1 - 4 were watered, that produces a "BedWatered" event flowing from those four cells
Using sse-eventbus, I publish those four events to each client observing the bed, and update its UI representation of the cell with a water droplet colored blue. (The droplet's opacity decays at a percentage rate equal to the number of days between recommended waterings for the given Cell's plant type)

Code

First, read this issue about sending to specific clients only

This issue was very informative and worked exactly as I needed. Thanks!

Controller

This is a WIP and I am by no means an expert on Coroutines yet, so any feedback is welcome!

Command handler function

package org.starbornag.api.rest.bed

import ch.rasc.sse.eventbus.SseEventBus
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import org.springframework.http.MediaType
import org.springframework.http.ResponseEntity
import org.springframework.web.bind.annotation.*
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter
import org.starbornag.api.domain.bed.BedRepository
import java.util.*
import java.util.concurrent.ConcurrentHashMap

@RestController
class BedCommandHandler(
    private val bedCommandMapper: BedCommandMapper,
    private val sseEventBus: SseEventBus
) {
    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)

    @PostMapping("/api/beds/{bedId}/{action}", consumes = [MediaType.APPLICATION_JSON_VALUE])
    suspend fun handle(
        @PathVariable bedId: UUID,
        @PathVariable action: String,
        @RequestBody commandPayload: Any
    ): ResponseEntity<BedResourceWithCurrentState>  {
        try {
            val bed = BedRepository.getBed(bedId)
            scope.launch {
                val command = bedCommandMapper.convertCommand(action, commandPayload)
                bed?.execute(command, sseEventBus) // Execute the command directly
            }
            val resource = BedResourceWithCurrentState.from(bed!!)
            val response = ResponseEntity.ok(resource)
            return response
        } catch(e: Exception) {
            println("Here is the exception: " + e.stackTraceToString())
            throw e
        }
    }

    @GetMapping("/api/beds/{bedId}/events", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
    fun events(@PathVariable bedId: UUID,
               @RequestParam clientId: UUID
               ): SseEmitter =
        sseEventBus.createSseEmitter(clientId.toString(), 60_000L, bedId.toString())
}

I declare a CoroutineScope with the supplied SupervisorJob and Dispatchers.IO configuration
- I'm still learning this stuff, but I recommend chapters 14 and 15 of Kotlin in Action 2nd Edition for an amazing treatment of these subjects
Declare the handle function as suspend
- See Urs Peter's lectures for equally informative treatment of how doing this allows you to avoid dealing with Mono<> and Flux<>
- Headache-Free Reactive Programming With Spring Boot and Kotlin Coroutines
- I have not attempted to instead adapt my code to use Virtual Threads via Project Loom, but he also discusses this in his next lecture: Reactive Spring Boot With Kotlin Coroutines: Adding Virtual Threads
- Execute the bed.execute function within the CoroutineScope using launch, because I don't care about waiting around for any results.
- We pass in the sseEventBus instance here. Earlier, I had to pass in a basic SseEmitter to ensure no issues with thread state. I assume this may also be needed here, but open to alternatives if anyone knows.
- The BedResourceWithCurrentState using Spring Boot HATEOAS to simply return the bed's current state, which may not reflect the processed results of each dispatched water command, and that is more than fine, because that's what the event stream is all about. I specifically do not want the client app to have to wait for each cell to process the command. I want it to get an immediate result and keep it moving.

events fun for streaming events

Bed.execute to dispatch to each Cell

Here, we provide the stream to interested clients. The clientId, as suggested in the linked issue above, allows us to register distinct clients for the single bed, which is passed in as its UUID for the event type.

    @GetMapping("/api/beds/{bedId}/events", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
    fun events(@PathVariable bedId: UUID,
               @RequestParam clientId: UUID
               ): SseEmitter =
        sseEventBus.createSseEmitter(clientId.toString(), 60_000L, bedId.toString())

BedAggregate

execute function

Within this class, most commands simply get dispatched to the associate BedCellAggregate instances. That is the case for the WaterBedCommand we will see soon.

Note this is also declared as suspend:

    // Generic command handler dispatcher
    suspend fun <T : BedCommand> execute(command: T, sseEventBus: SseEventBus) {
        when (command) {
            is PlantSeedlingCommand -> execute(command, sseEventBus)
            else ->  {
                dispatchCommandToAllCells(command, sseEventBus)
            }
        }
    }

dispatchCommandToAllCells

Again, we are in a suspend function, and we use coroutineScope and then launch on the call to get the cell and call its own execute, again passing the bus:

    private suspend fun dispatchCommandToAllCells(command: BedCommand, sseEventBus: SseEventBus) {
        coroutineScope {
            rows.forEach { row ->
                row.cells.forEach { cellId ->
                    launch {
                        val cell = BedCellRepository.getBedCell(cellId)
                        cell.execute(command, sseEventBus)
                    }
                }
            }
        }
    }

BedCellAggregate

Within here, the generic execute fun is also suspend, but the one that handles the WaterCommand specifically is not. It is where we actually call sseEventBus.handleEvent().

We pass the bedId to it and the wateredEvent
Since we subscribed a client via the /events route earlier, it will send these events to each connected client registered for this specific Bed.

    // Generic command handler dispatcher
    suspend fun <T : BedCommand> execute(command: T, sseEventBus: SseEventBus) {
        // Simulate latency
        delay(10.milliseconds)
        when (command) {
            is PlantSeedlingCommand -> execute(command)
            is WaterCommand -> execute(command, sseEventBus)
            is FertilizeCommand -> execute(command)
            is HarvestCommand -> execute(command)
            else -> throw IllegalArgumentException("Unsupported command type")
        }
    }

    private fun execute(command: WaterCommand, sseEventBus: SseEventBus) {
        val wateredEvent = BedWatered(command.started, command.volume)
        events.add(wateredEvent)
        sseEventBus.handleEvent(SseEvent.of(command.bedId.toString(), wateredEvent))
    }

CLI Demo

Service is up and running

Start two listeners

#!/bin/sh
clientId=0AF7867B-CFC0-48C4-A7A9-0BAFBCDC5569
curl -N http://localhost:8080/api/beds/2fbda883-d49d-4067-8e16-2b04cc523111/events?clientId=$clientId

#!/bin/sh
clientId=B59D7469-6989-49EF-AA1D-2D646EF8B06B
curl -N http://localhost:8080/api/beds/2fbda883-d49d-4067-8e16-2b04cc523111/events?clientId=$clientId

Run the commands

Note that the "lasterWatered" value for the bed comes back initially as null, as expected because there is not yet any historical data here due to the simulated latency in the code above with delay(10):

Observe events in the first listener

Observe events in the second listener

I can continue sending the water bed commands and continue seeing the events flow in as well!

I hope this example proves helpful to anyone. If you know more about coroutines than I do, please feel free to give me some tips.

ralscha / sse-eventbus

Example of using see-event bus with Kotlin and CoroutineScope #26