Oracle Cloud Streaming and the Micronaut Framework - Event-Driven Applications at Scale

version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper
    ports:
      - 2181:2181 # 
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  kafka:
    image: confluentinc/cp-kafka
    depends_on:
      - zookeeper
    ports:
      - 9092:9092 # 
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

package example.micronaut

import example.micronaut.chess.dto.Player
import example.micronaut.chess.dto.GameDTO
import example.micronaut.chess.dto.GameStateDTO
import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.core.type.Argument
import io.micronaut.http.HttpRequest
import io.micronaut.http.client.HttpClient
import io.micronaut.http.client.annotation.Client
import io.micronaut.test.extensions.spock.annotation.MicronautTest
import io.micronaut.test.support.TestPropertyProvider
import org.testcontainers.kafka.KafkaContainer
import org.testcontainers.utility.DockerImageName
import spock.lang.Specification
import spock.util.concurrent.PollingConditions
import io.micronaut.http.uri.UriBuilder
import jakarta.inject.Inject
import java.util.concurrent.ConcurrentLinkedDeque
import static io.micronaut.configuration.kafka.annotation.OffsetReset.EARLIEST
import static io.micronaut.http.MediaType.APPLICATION_FORM_URLENCODED_TYPE

@MicronautTest //  
class GameReporterSpec extends Specification implements TestPropertyProvider { // 

    private static final Collection<GameDTO> receivedGames = new ConcurrentLinkedDeque<>()
    private static final Collection<GameStateDTO> receivedMoves = new ConcurrentLinkedDeque<>()

    private static final PollingConditions pollingConditions = new PollingConditions(timeout: 30)

    static KafkaContainer kafka = new KafkaContainer(
            DockerImageName.parse('apache/kafka:latest')) // 

    @Inject
    ChessListener chessListener // 

    @Inject
    @Client('/')
    HttpClient client // 

    void 'test game ending in checkmate'() {

        given:
        String blackName = 'b_name'
        String whiteName = 'w_name'

        when: 'start game'
        Optional<String> result = startGame(blackName, whiteName)
        String gameId = result.orElseThrow(() -> new RuntimeException('Expected GameDTO id'))

        then:
        pollingConditions.eventually { // 
            !receivedGames.empty
        }

        1 == receivedGames.size()
        0 == receivedMoves.size()

        when:
        GameDTO game = receivedGames[0]

        then:
        gameId == game.id
        blackName == game.blackName
        whiteName == game.whiteName
        !game.draw
        !game.winner

        when: 'make moves'
        receivedGames.clear()

        makeMove(gameId, Player.WHITE, 'f3', 'rnbqkbnr/pppppppp/8/8/8/5P2/PPPPP1PP/RNBQKBNR b KQkq - 0 1', '1. f3')
        makeMove(gameId, Player.BLACK, 'e6', 'rnbqkbnr/pppp1ppp/4p3/8/8/5P2/PPPPP1PP/RNBQKBNR w KQkq - 0 2', '1. f3 e6')
        makeMove(gameId, Player.WHITE, 'g4', 'rnbqkbnr/pppp1ppp/4p3/8/6P1/5P2/PPPPP2P/RNBQKBNR b KQkq g3 0 2', '1. f3 e6 2. g4')
        makeMove(gameId, Player.BLACK, 'Qh4#', 'rnb1kbnr/pppp1ppp/4p3/8/6Pq/5P2/PPPPP2P/RNBQKBNR w KQkq - 1 3', '1. f3 e6 2. g4 Qh4#')

        then:
        pollingConditions.eventually {
            receivedMoves.size() > 3
        }

        0 == receivedGames.size()
        4 == receivedMoves.size()

        when:
        List<GameStateDTO> moves = new ArrayList<>(receivedMoves)

        then:
        Player.WHITE == moves[0].player
        'f3' == moves[0].move

        Player.BLACK == moves[1].player
        'e6' == moves[1].move

        Player.WHITE == moves[2].player
        'g4' == moves[2].move

        Player.BLACK == moves[3].player
        'Qh4#' == moves[3].move

        when: 'end game'

        receivedMoves.clear()

        endGame(gameId, Player.BLACK)

        then:
        pollingConditions.eventually {
            !receivedGames.empty
        }

        1 == receivedGames.size()
        0 == receivedMoves.size()

        when:
        game = receivedGames[0]

        then:
        gameId == game.id
        !game.blackName
        !game.whiteName
        !game.draw
        Player.BLACK == game.winner
    }

    void 'test game ending in draw'() {

        given:
        String blackName = 'b_name'
        String whiteName = 'w_name'

        when: 'start game'
        Optional<String> result = startGame(blackName, whiteName)
        String gameId = result.orElseThrow(() -> new RuntimeException('Expected GameDTO id'))

        then:
        pollingConditions.eventually {
            !receivedGames.empty
        }

        1 == receivedGames.size()
        0 == receivedMoves.size()

        when:
        GameDTO game = receivedGames[0]

        then:
        gameId == game.id
        blackName == game.blackName
        whiteName == game.whiteName
        !game.draw
        !game.winner

        when: 'make moves'
        receivedGames.clear()

        makeMove(gameId, Player.WHITE, 'f3', 'rnbqkbnr/pppppppp/8/8/8/5P2/PPPPP1PP/RNBQKBNR b KQkq - 0 1', '1. f3')
        makeMove(gameId, Player.BLACK, 'e6', 'rnbqkbnr/pppp1ppp/4p3/8/8/5P2/PPPPP1PP/RNBQKBNR w KQkq - 0 2', '1. f3' +
                'Player')

        then:
        pollingConditions.eventually {
            receivedMoves.size() > 1
        }

        0 == receivedGames.size()
        2 == receivedMoves.size()

        when: 'end game'
        receivedMoves.clear()
        endGame(gameId, null)

        then:
        pollingConditions.eventually {
            !receivedGames.empty
        }

        1 == receivedGames.size()
        0 == receivedMoves.size()

        when:
        game = receivedGames[0]

        then:
        gameId == game.id
        !game.blackName
        !game.whiteName
        game.draw
        !game.winner
        Player
    }


    @Override
    Map<String, String> getProperties() {
        kafka.start()
        ['kafka.bootstrap.servers': kafka.bootstrapServers] // 
    }

    void cleanup() {
        receivedGames.clear()
        receivedMoves.clear()
    }

    @KafkaListener(offsetReset = EARLIEST)
    static class ChessListener {

        @Topic('chessGame')
        void onGame(GameDTO game) {
            receivedGames << game
        }

        @Topic('chessGameState')
        void onGameState(GameStateDTO gameState) {
            receivedMoves << gameState
        }
    }

    private Optional<String> startGame(String blackName, String whiteName) {
        Map body = [b: blackName, w: whiteName] // 

        HttpRequest<?> request = HttpRequest.POST('/game/start', body)
                .contentType(APPLICATION_FORM_URLENCODED_TYPE)
        return client.toBlocking().retrieve(request,
                Argument.of(Optional, String)) // 
    }

    private void makeMove(String gameId,
                          Player player,
                          String move,
                          String fen, String pgn) {
        Map body = [player: player, move: move, fen: fen, pgn: pgn]

        HttpRequest<?> request = HttpRequest.POST('/game/move/' + gameId, body)
                .contentType(APPLICATION_FORM_URLENCODED_TYPE)
        client.toBlocking().exchange(request) // 
    }

    private void endGame(String gameId, Player winner) {
        UriBuilder uriBuilder = UriBuilder.of("/game").path(winner == null ? "draw" : "checkmate").path(gameId)
        if (winner != null) {
            uriBuilder = uriBuilder.path(winner.toString())
        }
        URI uri = uriBuilder.build()
        HttpRequest<Object> request = HttpRequest.POST(uri, null)
        client.toBlocking().exchange(request) // 
    }
}

package example.micronaut

import example.micronaut.chess.dto.Player
import example.micronaut.chess.dto.GameDTO
import example.micronaut.chess.dto.GameStateDTO
import example.micronaut.chess.entity.Game
import example.micronaut.chess.entity.GameState
import example.micronaut.chess.repository.GameRepository
import example.micronaut.chess.repository.GameStateRepository
import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.test.extensions.spock.annotation.MicronautTest
import io.micronaut.test.support.TestPropertyProvider
import org.testcontainers.kafka.KafkaContainer
import org.testcontainers.utility.DockerImageName
import reactor.core.publisher.Mono
import spock.lang.Specification
import spock.util.concurrent.PollingConditions

import jakarta.inject.Inject

@MicronautTest //  
class GameServiceSpec extends Specification implements TestPropertyProvider { // 

    private static final PollingConditions pollingConditions = new PollingConditions(timeout: 30)

    static KafkaContainer kafka = new KafkaContainer(
            DockerImageName.parse('apache/kafka:latest')) // 

    @Inject
    GameReporter gameReporter // 

    @Inject
    GameRepository gameRepository

    @Inject
    GameStateRepository gameStateRepository

    void setup() {
        gameStateRepository.deleteAll()
        gameRepository.deleteAll()
    }

    void 'test game ending in checkmate'() {
        given:
        String blackName = 'b_name'
        String whiteName = 'w_name'

        when: 'start game'

        UUID gameId = UUID.randomUUID()
        String gameIdString = gameId
        GameDTO gameDto = new GameDTO(gameIdString, blackName, whiteName)

        gameReporter.game(gameIdString, gameDto).subscribe()

        then:
        pollingConditions.eventually { // 
            gameRepository.count() > 0
        }

        1 == gameRepository.count()
        0 == gameStateRepository.count()

        when:
        Game game = gameRepository.findById(gameId).orElseThrow(() ->
                new IllegalStateException('Unable to find expected Game'))

        then:
        gameId == game.id
        blackName == game.blackName
        whiteName == game.whiteName
        !game.draw
        !game.winner

        when: 'make moves'

        List<UUID> gameStateIds = []

        UUID gameStateId = makeMove(gameIdString, Player.WHITE, 'f3', 'rnbqkbnr/pppppppp/8/8/8/5P2/PPPPP1PP/RNBQKBNR b KQkq - 0 1', '1. f3')
        gameStateIds << gameStateId

        gameStateId = makeMove(gameIdString, Player.BLACK, 'e6', 'rnbqkbnr/pppp1ppp/4p3/8/8/5P2/PPPPP1PP/RNBQKBNR w KQkq - 0 2', '1. f3 e6')
        gameStateIds << gameStateId

        gameStateId = makeMove(gameIdString, Player.WHITE, 'g4', 'rnbqkbnr/pppp1ppp/4p3/8/6P1/5P2/PPPPP2P/RNBQKBNR b KQkq g3 0 2', '1. f3 e6 2. g4')
        gameStateIds << gameStateId

        gameStateId = makeMove(gameIdString, Player.BLACK, 'Qh4#', 'rnb1kbnr/pppp1ppp/4p3/8/6Pq/5P2/PPPPP2P/RNBQKBNR w KQkq - 1 3', '1. f3 e6 2. g4 Qh4#')
        gameStateIds << gameStateId

        then:
        pollingConditions.eventually {
            gameStateRepository.count() > 3
        }

        1 == gameRepository.count()
        4 == gameStateRepository.count()

        when:
        List<GameState> moves = gameStateIds.collect { UUID id -> gameStateRepository.findById(id).orElse(null) }

        then:
        Player.WHITE == moves[0].player
        'f3' == moves[0].move

        Player.BLACK == moves[1].player
        'e6' == moves[1].move

        Player.WHITE == moves[2].player
        'g4' == moves[2].move

        Player.BLACK == moves[3].player
        'Qh4#' == moves[3].move

        when: 'end game'

        gameDto = new GameDTO(gameIdString, false, Player.BLACK)
        gameReporter.game(gameIdString, gameDto).subscribe()

        then:
        pollingConditions.eventually {
            Game g = gameRepository.findById(gameId).orElse(null)
            if (!g) return false
            g.winner
        }

        1 == gameRepository.count()
        4 == gameStateRepository.count()

        when:
        game = gameRepository.findById(gameId).orElseThrow(() ->
                new IllegalStateException('Unable to find expected Game'))

        then:
        gameId == game.id
        blackName == game.blackName
        whiteName == game.whiteName
        !game.draw
        Player.BLACK == game.winner
    }

    void 'test game ending in draw'() {
        given:
        String blackName = 'b_name'
        String whiteName = 'w_name'

        when: 'start game'

        UUID gameId = UUID.randomUUID()
        String gameIdString = gameId
        GameDTO gameDto = new GameDTO(gameIdString, blackName, whiteName)

        gameReporter.game(gameIdString, gameDto).subscribe()

        then:
        pollingConditions.eventually { // 
            gameRepository.count() > 0
        }

        1 == gameRepository.count()
        0 == gameStateRepository.count()

        when:
        Game game = gameRepository.findById(gameId).orElseThrow(() ->
                new IllegalStateException('Unable to find expected Game'))

        then:
        gameId == game.id
        blackName == game.blackName
        whiteName == game.whiteName
        !game.draw
        !game.winner

        when: 'make moves'

        List<UUID> gameStateIds = []

        UUID gameStateId = makeMove(gameIdString, Player.WHITE, 'f3', 'rnbqkbnr/pppppppp/8/8/8/5P2/PPPPP1PP/RNBQKBNR b KQkq - 0 1', '1. f3')
        gameStateIds << gameStateId

        gameStateId = makeMove(gameIdString, Player.BLACK, 'e6', 'rnbqkbnr/pppp1ppp/4p3/8/8/5P2/PPPPP1PP/RNBQKBNR w KQkq - 0 2', '1. f3 e6')
        gameStateIds << gameStateId

        then:
        pollingConditions.eventually {
            gameStateRepository.count() > 1
        }

        1 == gameRepository.count()
        2 == gameStateRepository.count()

        when:
        List<GameState> moves = []
        for (UUID id : gameStateIds) {
            moves << gameStateRepository.findById(id).orElseThrow(() ->
                    new IllegalStateException('Unable to find expected GameState'))
        }

        then:
        Player.WHITE == moves[0].player
        'f3' == moves[0].move

        Player.BLACK == moves[1].player
        'e6' == moves[1].move

        when: 'end game'

        gameDto = new GameDTO(gameIdString, true, null)
        gameReporter.game(gameIdString, gameDto).subscribe()

        then:
        pollingConditions.eventually {
            Game g = gameRepository.findById(gameId).orElse(null)
            if (!g) return false
            g.draw
        }

        1 == gameRepository.count()
        2 == gameStateRepository.count()

        when:
        game = gameRepository.findById(gameId).orElseThrow(() ->
                new IllegalStateException('Unable to find expected Game'))

        then:
        gameId == game.id
        blackName == game.blackName
        whiteName == game.whiteName
        game.draw
        !game.winner
    }

    @Override
    Map<String, String> getProperties() {
        kafka.start()
        ['kafka.bootstrap.servers': kafka.bootstrapServers] // 
    }

    @KafkaClient
    static interface GameReporter {

        @Topic('chessGame')
        Mono<GameDTO> game(@KafkaKey String gameId, GameDTO game)

        @Topic('chessGameState')
        Mono<GameStateDTO> gameState(@KafkaKey String gameId, GameStateDTO gameState)
    }

    private UUID makeMove(String gameId,
                          Player player,
                          String move,
                          String fen,
                          String pgn) {
        UUID gameStateId = UUID.randomUUID()
        gameReporter.gameState(gameId, new GameStateDTO(gameStateId.toString(),
                gameId, player, move, fen, pgn)).subscribe()
        gameStateId
    }
}