[Mastering Spring 5.0] 11.2 Spring Reactive - 1

1
2

implementation ('org.reactivestreams:reactive-streams:1.0.2')
implementation ('org.reactivestreams:reactive-streams-tck:1.0.2')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// Publisher
package org.reactivestreams;

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> var1);
}

// Subscriber
package org.reactivestreams;

public interface Subscriber<T> {
    void onSubscribe(Subscription var1);

    void onNext(T var1);

    void onError(Throwable var1);

    void onComplete();
}

// Subscription
package org.reactivestreams;

public interface Subscription {
    void request(long var1);

    void cancel();
}

1
2

implementation      ('io.projectreactor:reactor-core:3.2.5.RELEASE')
testImplementation  ('io.projectreactor:reactor-test:3.2.5.RELEASE')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

plugins {
    id 'java'
    id 'war'
    id 'org.springframework.boot' version '2.1.1.RELEASE'
}

apply plugin: 'java'
apply plugin: 'io.spring.dependency-management'

group 'com.mastering.spring'
version '1.0-SNAPSHOT'

sourceCompatibility = 1.11

repositories {
    mavenCentral()
}

dependencies {

    implementation ('org.springframework.boot:spring-boot-starter')

    implementation ('org.reactivestreams:reactive-streams:1.0.2')
    implementation ('org.reactivestreams:reactive-streams-tck:1.0.2')

    implementation      ('io.projectreactor:reactor-core:3.2.5.RELEASE')
    testImplementation  ('io.projectreactor:reactor-test:3.2.5.RELEASE')

    testImplementation  ('org.springframework.boot:spring-boot-starter-test:2.1.2.RELEASE')
    testImplementation  ('junit:junit:4.12')
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

package com.mastering.spring.reactive;

import org.junit.Test;
import reactor.core.publisher.Mono;

import java.time.Duration;

public class SpringReactiveTest {

    @Test
    public void monoExample() throws InterruptedException {

        // 5초 후에 하나의 요소를 방출한다.
        Mono<String> stubMonoWithADelay = Mono.just("Ranga").delayElement(Duration.ofSeconds(5));

        // 모노 이벤트를 수신하고 콘솔에 기록한다.
        stubMonoWithADelay.subscribe(System.out::println);

        // 테스트 실행시간을 지연시킨다.
        Thread.sleep(10000);
    }

}

1
2
3
4
5
6
7
8
9

/**
 * Consumer 를 명시적으로 정의
 */
class SystemOutConsumer implements Consumer<String> {
    @Override
    public void accept(String s) {
        System.out.println("Received " + s + " at" + new Date());
    }
}

1
2
3

// 모노 이벤트를 수신하고 콘솔에 기록한다.
// stubMonoWithADelay.subscribe(System.out::println);
stubMonoWithADelay.subscribe(new SystemOutConsumer());

1
2
3
4
5
6
7

class WelcomeConsumer implements Consumer<String> {
    @Override
    public void accept(String s) {
        System.out.println("Welcome " + s);

    }
}

1
2
3
4

// 모노 이벤트를 수신하고 콘솔에 기록한다.
// stubMonoWithADelay.subscribe(System.out::println);
stubMonoWithADelay.subscribe(new SystemOutConsumer());
stubMonoWithADelay.subscribe(new WelcomeConsumer());

1
2
3
4
5
6
7

@Test
public void simpleFluxStream() {
        Flux<String> stubFluxStream = Flux.just("Jane", "Joe");
        stubFluxStream.subscribe(new SystemOutConsumer());
    }
}

1

implementation ('org.springframework.boot:spring-boot-starter-webflux')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

@RestController
public class StockPriceEventController {

    @GetMapping("/stocks/price/{stockCode}")
    Flux<String> retrieveStockPriceHardcoded(@PathVariable("stockCode") String stockCode) {
        return Flux.interval(Duration.ofSeconds(5))
                .map(l -> getCurrentDate() +" : " + getRandomNumber(100, 125)).log();

    }

    private String getCurrentDate() {
        return (new Date()).toString();
    }

    private int getRandomNumber(int min, int max) {
        return ThreadLocalRandom.current().nextInt(min, max + 1);
    }

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    <p>
        <button id="subscribe-button">Get Latest IBM Price</button>
        <ul id="display"></ul>
    </p>

    <script>
        addEvent("click", document.getElementById('subscribe-button'),
            () => registerEventSourceAndAddResponseTo("/stocks/price/IBM", "display"));

        function registerEventSourceAndAddResponseTo(uri, elementId) {
            let stringEvents = document.getElementById(elementId);
            let stringEventSource = new EventSource(uri)

            stringEventSource.onmessage = (e) => {
                let newElement = document.createElement("li");
                newElement.innerHTML = e.data;
                stringEvents.appendChild(newElement);
            }
        }

        function addEvent(event, elem, func) {
            if (typeof (EventSource) !== 'undefined') {
                elem.addEventListener(event, func, false);
            } else {
                elem[event] = func;
            }
        }

    </script>
</body>
</html>