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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
| @SpringBootApplication
@EnableAsync
@Slf4j
public class StudyApplication {
@RestController
public static class MyController {
AsyncRestTemplate rt = new AsyncRestTemplate(new Netty4ClientHttpRequestFactory(new NioEventLoopGroup(1)));
@Autowired
MyService myService;
static final String URL1 = "http://localhost:8081/service?req={req}";
static final String URL2 = "http://localhost:8081/service2?req={req}";
@GetMapping("/rest")
public DeferredResult<String> rest(int idx) {
DeferredResult<String> dr = new DeferredResult<>();
Completion
.from(rt.getForEntity(URL1, String.class, "hello" + idx))
.andApply(s -> rt.getForEntity(URL2, String.class, s.getBody()))
.andApply(s -> myService.work(s.getBody()))
.andError(e -> dr.setErrorResult(e.toString()))
.andAccept(s -> dr.setResult(s));
return dr;
}
}
public static class AcceptCompletion<S> extends Completion<S, Void> {
Consumer<S> con;
public AcceptCompletion(Consumer<S> con) {
this.con = con;
}
@Override
public void run(S value) {
con.accept(value);
}
}
public static class ErrorCompletion<T> extends Completion<T, T> {
Consumer<Throwable> econ;
public ErrorCompletion(Consumer<Throwable> econ) {
this.econ = econ;
}
@Override
public void run(T value) {
if (next != null) {
next.run(value);
}
}
@Override
public void error(Throwable e) {
econ.accept(e);
}
}
public static class AsyncCompletion<S, T> extends Completion<S, T> {
Function<S, ListenableFuture<T>> fn;
public AsyncCompletion(Function<S, ListenableFuture<T>> fn) {
this.fn = fn;
}
@Override
public void run(S value) {
ListenableFuture<T> lf = fn.apply(value);
lf.addCallback(s -> complete(s), e -> error(e));
}
}
public static class Completion<S, T> {
Completion next;
public static <S, T> Completion<S, T> from(ListenableFuture<T> lf) {
Completion<S, T> c = new Completion<>();
lf.addCallback(s -> {
c.complete(s);
}, e -> {
c.error(e);
});
return c;
}
public <V> Completion<T, V> andApply(Function<T, ListenableFuture<V>> fn) {
Completion<T, V> c = new AsyncCompletion<>(fn);
this.next = c;
return c;
}
public Completion<T, T> andError(Consumer<Throwable> econ) {
Completion<T, T> c = new ErrorCompletion<>(econ);
this.next = c;
return c;
}
public void andAccept(Consumer<T> con) {
Completion<T, Void> c = new AcceptCompletion<>(con);
this.next = c;
}
public void complete(T s) {
if (next != null) next.run(s);
}
public void run(S value) {
}
public void error(Throwable e) {
if (next != null) next.error(e);
}
}
@Service
public static class MyService {
@Async
public ListenableFuture<String> work(String req) {
return new AsyncResult<>(req + "/asyncwork");
}
}
@Bean
public ThreadPoolTaskExecutor myThreadPool() {
ThreadPoolTaskExecutor te = new ThreadPoolTaskExecutor();
te.setCorePoolSize(1);
te.setMaxPoolSize(1);
te.initialize();
return te;
}
public static void main(String[] args) {
SpringApplication.run(StudyApplication.class, args);
}
}
|