1
2
3
4
5
6
7
8
9
10
11
@Document
data class Coffee(
	@Id
	val id: String?,
	val name: String
)

data class CoffeeOrder(
	val coffeeId: String,
	val whenOrdered: Instant
)

1
2
interface CoffeeRepository: ReactiveCrudRepository<Coffee, String> {
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Component
class DataLoader(private val repo: CoffeeRepository) {
	
	@PostConstruct
	fun loadData() {
		println("### load Datas......")
		repo.deleteAll() // 기존 데이터는 일단 지우고..
			.thenMany(Flux.just("Cafe Cereza", "Don Pablo", "Sumatra", "Kaldi", "Kona") // 이 이름들로 커피 정보를 생성 할 것이다.
				.map { Coffee(null, it) } // Flux를 통해 들어오는 커피이름으로 Coffee 정보를 생성한다.
				.flatMap { repo.save(it) }) // 그리고 MongoDB에 저장
			.thenMany(repo.findAll()) // 저장이 잘 됐는지 전체 데이터를 조회해서
			.subscribe { println("### $it") } // 출력한다.
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Service
class CoffeeService(private val repo: CoffeeRepository) {

	fun getAllCoffees(): Flux<Coffee> {
		return repo.findAll()
	}

	fun getCoffeeById(id: String): Mono<Coffee> {
		return repo.findById(id)
	}

	fun getOrdersForCoffee(coffeeId: String): Flux<CoffeeOrder> {
		return Flux.interval(Duration.ofSeconds(1)) // 1초 간격으로 숫자를 발행한다.
			.onBackpressureDrop() // 처리량이 많아서 처리를 못하는 요소는 버린다.
			.map { CoffeeOrder(coffeeId, Instant.now()) } // 주문정보를 생성한다.
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@RestController
@RequestMapping("/coffees")
class CoffeeController(private val service: CoffeeService) {

	@GetMapping
	fun all(): Flux<Coffee> {
		return service.getAllCoffees()
	}

	@GetMapping("/{id}")
	fun byId(@PathVariable id: String): Mono<Coffee> {
		return service.getCoffeeById(id)
	}

	@GetMapping("/{id}/orders", produces = [MediaType.TEXT_EVENT_STREAM_VALUE]) // Stream으로 데이터를 보낸다.
	fun orders(@PathVariable id: String): Flux<CoffeeOrder> {
		return service.getOrdersForCoffee(id)
	}
}

1
2
3
4
5
6
7
8
9
data class Coffee(
	val id: String,
	val name: String
)

data class CoffeeOrder(
	val coffeeId: String,
	val whenOrdered: Instant
)

1
2
3
4
5
6
7
@SpringBootApplication
class KotlinCoffeeClientApplication {
	@Bean
	fun client(): WebClient {
		return WebClient.create("http://localhost:8080")
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Component
class TestClient(private val client: WebClient) {

	@PostConstruct
	fun letsDoThis() {
		client.get() // get 요청을 보낸다.
			.uri("/coffees") // http://localhost:8080/coffees로 요청을 보낸다.
			.retrieve()
			.bodyToFlux(Coffee::class.java) // response를 이용해 Coffee 정보를 생성한다.
			.filter { coffee -> coffee.name.equals("kona", true) } // Coffee 정보 중에 Kona 커피를 찾는다.
			.flatMap { coffee -> client.get()
				.uri("/coffees/{id}/orders", coffee.id) // http://localhost:8080/coffees/{id}/orders로 요청을 보낸다.
				.retrieve()
				.bodyToFlux(CoffeeOrder::class.java)} // CoffeeOrder 정보를 생성한다.
			.subscribe { println("### $it") } // stream 구독을 시작하고, 전달되는 데이터를 출력한다.
	}
}

1
2
3
interface CoffeeRepository: ReactiveCrudRepository<Coffee, String> {
	fun findCoffeeByName(name: String): Mono<Coffee>
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@Service
class CoffeeService(private val repo: CoffeeRepository) {

	fun getAllCoffees(): Flux<Coffee> {
		return repo.findAll()
	}

	fun getCoffeeById(id: String): Mono<Coffee> {
		return repo.findById(id)
	}

	// 이 부분이 추가됐다.
	fun getCoffeeByName(name: String): Mono<Coffee> {
		return repo.findCoffeeByName(name) // 이름을 검색한다.
			.defaultIfEmpty(Coffee("12345", "My favorite coffee")) // 일치하는 정보가 없으면 기본값을 넘겨준다.
	}

	fun getOrdersForCoffee(coffeeId: String): Flux<CoffeeOrder> {
		return Flux.interval(Duration.ofSeconds(1)) // 1초 간격으로 숫자를 발행한다.
			.onBackpressureDrop() // 처리량이 많아서 처리를 못하는 요소는 버린다.
			.map { CoffeeOrder(coffeeId, Instant.now()) } // 주문정보를 생성한다.
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Controller
class RSController(private val service: CoffeeService) {

	@MessageMapping("coffees")
	fun supplyCoffees(): Flux<Coffee> {
		return service.getAllCoffees()
	}

	@MessageMapping("orders.{name}")
	fun orders(@DestinationVariable name: String): Flux<CoffeeOrder> {
		return service.getCoffeeByName(name)
			.flatMapMany { coffee -> service.getOrdersForCoffee(coffee.id!!) }
	}
}

1
2
3
4
5
6
7
8
@Component
class TestClient(private val client: WebClient) {

//	@PostConstruct 어노테이션을 주석처리해서 자동으로 실행되지 않도록 한다.
	fun letsDoThis() {
		// ...생략...
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
@SpringBootApplication
class CoffeeClientApplication {
	@Bean
	fun client(): WebClient {
		return WebClient.create("http://localhost:8080")
	}

	// 이 부분이 추가됐다.
	@Bean
	fun requester(builder: RSocketRequester.Builder): RSocketRequester {
		return builder.connectTcp("localhost", 8901).block()!!
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
@RestController
class ClientController(private val requester: RSocketRequester) {

	@GetMapping("/coffees")
	fun coffees(): Flux<Coffee> {
		return requester.route("coffees").retrieveFlux(Coffee::class.java)
	}

	@GetMapping("/orders/{name}", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
	fun orders(@PathVariable name: String): Flux<CoffeeOrder> {
		return requester.route("orders.$name").retrieveFlux(CoffeeOrder::class.java)
	}
}