Skip to content
This repository was archived by the owner on Dec 19, 2023. It is now read-only.

Commit 7d312d9

Browse files
oliemansmoliemansm
authored
Merge pull request #346 from seongahjo/master
add reactor subscription example
2 parents c94fdf0 + 2059c4f commit 7d312d9

File tree

4 files changed

+129
-7
lines changed

4 files changed

+129
-7
lines changed

‎example-graphql-subscription/build.gradle‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ dependencies {
1616
compile "com.graphql-java-kickstart:graphql-java-tools:$LIB_GRAPHQL_JAVA_TOOLS_VER"
1717

1818
compile "io.reactivex.rxjava2:rxjava:2.1.5"
19-
19+
compile "io.projectreactor:reactor-core:3.3.2.RELEASE"
2020
compile("org.springframework.boot:spring-boot-starter-web:$LIB_SPRING_BOOT_VER")
2121

2222
testCompile "org.springframework.boot:spring-boot-starter-test:$LIB_SPRING_BOOT_VER"

‎example-graphql-subscription/src/main/java/com/oembedler/moon/graphql/boot/publishers/StockTickerReactorPublisher.java‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
package com.oembedler.moon.graphql.boot.publishers;
2+
3+
import java.math.BigDecimal;
4+
import java.math.RoundingMode;
5+
import java.time.LocalDateTime;
6+
import java.util.ArrayList;
7+
import java.util.List;
8+
import java.util.Map;
9+
import java.util.Random;
10+
import java.util.concurrent.ConcurrentHashMap;
11+
import java.util.concurrent.Executors;
12+
import java.util.concurrent.ScheduledExecutorService;
13+
import java.util.concurrent.TimeUnit;
14+
15+
import org.slf4j.Logger;
16+
import org.slf4j.LoggerFactory;
17+
import org.springframework.stereotype.Component;
18+
import com.oembedler.moon.graphql.boot.resolvers.StockPriceUpdate;
19+
import reactor.core.publisher.ConnectableFlux;
20+
import reactor.core.publisher.Flux;
21+
import reactor.core.publisher.FluxSink;
22+
23+
@Component
24+
public class StockTickerReactorPublisher {
25+
private static final Logger LOG = LoggerFactory.getLogger(StockTickerRxPublisher.class);
26+
27+
private final Flux<StockPriceUpdate> publisher;
28+
29+
public StockTickerReactorPublisher() {
30+
Flux<StockPriceUpdate> stockPriceUpdateFlux = Flux.create(emitter -> {
31+
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
32+
executorService.scheduleAtFixedRate(newStockTick(emitter), 0, 2, TimeUnit.SECONDS);
33+
}, FluxSink.OverflowStrategy.BUFFER);
34+
ConnectableFlux<StockPriceUpdate> connectableFlux = stockPriceUpdateFlux.share().publish();
35+
connectableFlux.connect();
36+
37+
publisher = Flux.from(connectableFlux);
38+
}
39+
40+
private Runnable newStockTick(FluxSink<StockPriceUpdate> emitter) {
41+
return () -> {
42+
List<StockPriceUpdate> stockPriceUpdates = getUpdates(rollDice(0, 5));
43+
if (stockPriceUpdates != null) {
44+
emitStocks(emitter, stockPriceUpdates);
45+
}
46+
};
47+
}
48+
49+
private void emitStocks(FluxSink<StockPriceUpdate> emitter, List<StockPriceUpdate> stockPriceUpdates) {
50+
for (StockPriceUpdate stockPriceUpdate : stockPriceUpdates) {
51+
try {
52+
emitter.next(stockPriceUpdate);
53+
} catch (RuntimeException e) {
54+
LOG.error("Cannot send StockUpdate", e);
55+
}
56+
}
57+
}
58+
59+
public Flux<StockPriceUpdate> getPublisher() {
60+
return publisher;
61+
}
62+
63+
public Flux<StockPriceUpdate> getPublisher(List<String> stockCodes) {
64+
if (stockCodes != null) {
65+
return publisher.filter(stockPriceUpdate -> stockCodes.contains(stockPriceUpdate.getStockCode()));
66+
}
67+
return publisher;
68+
}
69+
70+
private List<StockPriceUpdate> getUpdates(int number) {
71+
List<StockPriceUpdate> updates = new ArrayList<>();
72+
for (int i = 0; i < number; i++) {
73+
updates.add(rollUpdate());
74+
}
75+
return updates;
76+
}
77+
78+
private final static Map<String, BigDecimal> CURRENT_STOCK_PRICES = new ConcurrentHashMap<>();
79+
80+
static {
81+
CURRENT_STOCK_PRICES.put("TEAM", dollars(39, 64));
82+
CURRENT_STOCK_PRICES.put("IBM", dollars(147, 10));
83+
CURRENT_STOCK_PRICES.put("AMZN", dollars(1002, 94));
84+
CURRENT_STOCK_PRICES.put("MSFT", dollars(77, 49));
85+
CURRENT_STOCK_PRICES.put("GOOGL", dollars(1007, 87));
86+
}
87+
88+
private StockPriceUpdate rollUpdate() {
89+
ArrayList<String> STOCK_CODES = new ArrayList<>(CURRENT_STOCK_PRICES.keySet());
90+
91+
String stockCode = STOCK_CODES.get(rollDice(0, STOCK_CODES.size() - 1));
92+
BigDecimal currentPrice = CURRENT_STOCK_PRICES.get(stockCode);
93+
94+
BigDecimal incrementDollars = dollars(rollDice(0, 1), rollDice(0, 99));
95+
if (rollDice(0, 10) > 7) {
96+
// 0.3 of the time go down
97+
incrementDollars = incrementDollars.negate();
98+
}
99+
BigDecimal newPrice = currentPrice.add(incrementDollars);
100+
101+
CURRENT_STOCK_PRICES.put(stockCode, newPrice);
102+
return new StockPriceUpdate(stockCode, LocalDateTime.now(), newPrice, incrementDollars);
103+
}
104+
105+
private static BigDecimal dollars(int dollars, int cents) {
106+
return truncate("" + dollars + "." + cents);
107+
}
108+
109+
private static BigDecimal truncate(final String text) {
110+
BigDecimal bigDecimal = new BigDecimal(text);
111+
if (bigDecimal.scale() > 2)
112+
bigDecimal = new BigDecimal(text).setScale(2, RoundingMode.HALF_UP);
113+
return bigDecimal.stripTrailingZeros();
114+
}
115+
116+
private final static Random rand = new Random();
117+
118+
private static int rollDice(int min, int max) {
119+
return rand.nextInt((max - min) + 1) + min;
120+
}
121+
122+
}

‎example-graphql-subscription/src/main/java/com/oembedler/moon/graphql/boot/publishers/StockTickerPublisher.java‎ renamed to ‎example-graphql-subscription/src/main/java/com/oembedler/moon/graphql/boot/publishers/StockTickerRxPublisher.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,13 @@
2323
import java.util.concurrent.TimeUnit;
2424

2525
@Component
26-
public class StockTickerPublisher {
26+
public class StockTickerRxPublisher {
2727

28-
private static final Logger LOG = LoggerFactory.getLogger(StockTickerPublisher.class);
28+
private static final Logger LOG = LoggerFactory.getLogger(StockTickerRxPublisher.class);
2929

3030
private final Flowable<StockPriceUpdate> publisher;
3131

32-
public StockTickerPublisher() {
32+
public StockTickerRxPublisher() {
3333
Observable<StockPriceUpdate> stockPriceUpdateObservable = Observable.create(emitter -> {
3434

3535
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

‎example-graphql-subscription/src/main/java/com/oembedler/moon/graphql/boot/resolvers/Subscription.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package com.oembedler.moon.graphql.boot.resolvers;
22

3+
import com.oembedler.moon.graphql.boot.publishers.StockTickerRxPublisher;
34
import graphql.kickstart.tools.GraphQLSubscriptionResolver;
4-
import com.oembedler.moon.graphql.boot.publishers.StockTickerPublisher;
55
import org.reactivestreams.Publisher;
66
import org.springframework.stereotype.Component;
77

@@ -10,9 +10,9 @@
1010
@Component
1111
class Subscription implements GraphQLSubscriptionResolver {
1212

13-
private StockTickerPublisher stockTickerPublisher;
13+
private StockTickerRxPublisher stockTickerPublisher;
1414

15-
Subscription(StockTickerPublisher stockTickerPublisher) {
15+
Subscription(StockTickerRxPublisher stockTickerPublisher) {
1616
this.stockTickerPublisher = stockTickerPublisher;
1717
}
1818

0 commit comments

Comments
 (0)