# Spring WebFlux

@Component
class Router(private val handler: Handler) {
    @Bean
    fun route() = router {
        GET("/hoge", handler::hoge)
        "users/".nest {
            GET("/", handler::getUsers)
            POST("/", handler::addUser)
            PATCH("/", handler::updateUser)
        }
    }
}

@Component
class ApiHandler {
    fun hello(request: ServerRequest): Mono<ServerResponse> =
        request.toMono()spring
            .flatMap { ok().contentType(MediaType.APPLICATION_JSON).body(Mono.just("hello")) }
            .onErrorResume { handleError(it) }

    private fun handleError(e: Throwable): Mono<ServerResponse> {
        return when (e) {
            is HogeException -> badRequest().body(Mono.just(e.message.orEmpty()))
            else -> badRequest().body(Mono.just("Unknown error. ${e.message}"))
        }
    }
}

@Test
fun test(){
    webTestClient.
        .get()
        .url("/hoge")
        .exchange()
        .expectBody()
        .xpath("/root/header/code").isEqualTo("hoge)
}

Mono.just(listOf(1, 2, 3))
    .flatMapMany { Flux.fromIterable(it) }
    // Flux<Int> になる

Flux.just(1, 2, 3)
    .collectList()
    // Mono<List<Int>> になる

1.toMono()
    .doOnNext{ throw Exception() }
    .onErrorResume { 
        2.toMono()
        .flatMap { reactiveProcess.someThing() } // エラー処理の中でMonoを使う場合は`onErrorResume()`でないとsubscribe()されない
    }

user.toMono()
    .doOnNext{ throw MyException() }
    .retry(3)
    // リトライ失敗のときはその原因となった例外に変換
    .onErrorMap { if(it is RetryExhaustedException) it.cause!! else it }

Flux.just(User(name = 'aaa', age = 10), User('bbb', 10), User('ccc', 12))
    .groupBy(User::age) // 年齢ごとにグループ化
    .flatMap { group -> // 1要素はgroupedFlux(`key()`つきの`Flux<User>`)になる
        // 各グループFluxごとの処理。ここでMonoかFluxを返さないと処理が実行されないので注意。返さないならsubscribe()する。
        group.map(OptOutLog::name)
            .collectList()
            .doOnNext { doSomething(it) } // 年齢ごとの名前リストに対して任意の処理
            .map { it.toString }
    }
    .collectList() // Mono<List<User>>に変換
    .flatMap { status(HttpStatus.OK).contentType(MediaType.APPLICATION_JSON).body(it.toMono()) }

1.toMono()
    .doOnNext {
        // このMonoはsubscribe()されない
        1.toMono()
            .doOnNext{ println("hello...") }
     }
     .flatMap { noContent().build() }

1.toMono()
    .flatMap {
        // このMonoはsubscribe()される
        1.toMono()
            .doOnNext{ println("hello...") }
     }
     .flatMap { noContent().build() }

// bad
hoge.toMono()
    .map { repo.read("id") } // blocking process
    .flatMap { ok().body(Mono.just("OK")) }

// good
hoge.toMono()
    .flatMap { Mono.fromCallable { repo.read("id") } }
    .flatMap { ok().body(Mono.just("OK")) }

const val MIN_BACKOFF = 3
const val MAX_BACKOFF = 10
const val MAX_COUNT = 5

mono.doOnNext(::send)
    .retryWhen(Retry.anyOf<Any>(MyException::class.java)
        .exponentialBackoffWithJitter(Duration.ofSeconds(MIN_BACKOFF), Duration.ofSeconds(MAX_BACKOFF))
        .retryMax(MAX_COUNT))

@Configuration
class WebClientConfig() {
    // `http://`は省略しないこと
    @Bean
    fun webClient(): WebClient = WebClient.create("http://localhost:3000")
}

WebClient
    .builder()
    .defaultHeader("key", "hoge")
    .build()

client.get()
    .uri("/users/{id}", 1)
    .exchange()
    .subscribe()

@JsonFormat(shape = JsonFormat.Shape.Number)
enum class Hoge(val value: Int) {
    VAL1(1),
    VAL2(2),
    VAL3(3)
}

@Table // テーブル名はデフォルトでクラス名と同じ（user）で紐づく
@Entity
class User {
    @Id
    var id: Int = 0
    @Column
    var name: String = ""
    @Column
    var age: Int = 0
    @Column
    var version: Int = 0
}

@Repository
interface UserRepository : JpaRepository<User, Int>

private const val HOGE = "\${aaa.bbb.ccc}"

val hogeService = mock<HogeService> {
    // デフォルト値がある関数の場合も最後の引数まで省略しないで書く
    on{ hello(eq("aaa"), any())} doReturn 123
    // 引数を利用して戻り値を返す場合は`doAnswer()`
    on { hello(any()) } doAnswer { (it.arguments[0] as Hoge) } // もらった引数の0番目をそのまま返す
}

// `hello()`が1回呼ばれたか
verify(hogeService, times(1)).hello(any())

// `hello("hoge")`が1回呼ばれたか
verify(hogeService, times(1)).hello(eq("hoge"))

// `hello()`が1回呼ばれたか検証し、さらに呼ばれたときの引数を`actual`に代入
val actual = argumentCaptor<String>().apply { verify(hogeService, times(1)).hello(capture()) }.firstValue
Assertions.assertEquals("hoge", actual)

@Autowired
private lateinit var webTestClient: WebTestClient

fun test() {
    webTestClient.get().uri("/hello").exchange().expectStatus().isOk
                .expectBody()
                .jsonPath("\$.results[0].name").isEqualTo("aaa")
                .jsonPath("\$.results[0].age").isEqualTo(12)
}

data class Hoge(val a: Int)

fun hello(request: ServerRequest): Mono<ServerResponse> {
        val stream = Stream.iterate(0) {it + 1}
        val flux = Flux.fromStream(stream)
            .zipWith(Flux.interval(Duration.ofSeconds(1)))
            .map { Hoge(it.t1) }
            .doOnNext { println(it) }

        return request.toMono()
            .flatMap { ok().contentType(MediaType.APPLICATION_STREAM_JSON).body(flux, Hoge::class.java) }
    }

data class Hoge(val a: Int)

fun hello(request: ServerRequest): Mono<ServerResponse> {
        val s = Flux.create<Hoge> { sink ->
            var i = 0
            while(true){
                sink.next(Hoge(i))
                Thread.sleep(1000L)
                i++
                if(5 < i) break
            }
            sink.complete()
        }

        return request.toMono()
            .flatMap { ok().contentType(MediaType.APPLICATION_STREAM_JSON).body(s, Hoge::class.java) }
    }

@Repository
interface HogeRepository : JpaRepository<Hoge, UUID>

val repositoryParallel: Scheduler = Schedulers.newParallel("repository", 10)
fun <T> parallel(f: () -> T): Mono<T> = Mono.fromCallable(f)
  .subscribeOn(repositoryParallel)
  .publishOn(Schedulers.elastic())

1.toMono()
  .flatMap { parallel { hogeRepo.findById("id") } }

Flux.just(1, 2, 3)
  .filter{ it -> it < 4 }
  .switchIfEmpty { Mono.error(RuntimeException()) }

data class Request(val count: Int, val offset: Int)

Request(count = 100, offset = 0)
    .toMono()
    .expand { (req, totalCount) ->
        if(totalCount < req.count){
            Mono.empty()
        } else {
            val res = api.users(limit = 10, offset = req.offset)
            Mono.just(Pair(req, totalCount + res.count))
        }
    }