Skip to content

Commit 74c0250

Browse files
Stephane Maldinirstoyanchev
Stephane Maldini
authored and
rstoyanchev
committed
Upgrade to Reactor 2
Issue: SPR-12599
1 parent 122d347 commit 74c0250

File tree

13 files changed

+388
-306
lines changed

13 files changed

+388
-306
lines changed

‎build.gradle‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ configure(allprojects) { project ->
5353
ext.nettyVersion = "4.0.26.Final"
5454
ext.openjpaVersion = "2.2.2" // 2.3.0 not Java 8 compatible (based on ASM 4)
5555
ext.protobufVersion = "2.6.1"
56-
ext.reactorVersion = "1.1.6.RELEASE"
56+
ext.reactorVersion = "2.0.1.BUILD-SNAPSHOT"
5757
ext.slf4jVersion = "1.7.11"
5858
ext.snakeyamlVersion = "1.15"
5959
ext.snifferVersion = "1.14"
@@ -117,6 +117,7 @@ configure(allprojects) { project ->
117117
}
118118

119119
repositories {
120+
mavenLocal()
120121
maven { url "https://repo.spring.io/libs-release" }
121122
maven { url "https://repo.spring.io/milestone" }
122123
}
@@ -486,16 +487,17 @@ project("spring-messaging") {
486487
compile(project(":spring-beans"))
487488
compile(project(":spring-core"))
488489
compile(project(":spring-context"))
489-
optional("org.projectreactor:reactor-core:${reactorVersion}")
490-
optional("org.projectreactor:reactor-net:${reactorVersion}") {
490+
optional("io.projectreactor:reactor-net:${reactorVersion}") {
491491
exclude group: "io.netty", module: "netty-all"
492492
}
493+
optional "io.netty:netty-all:$nettyVersion"
493494
optional("org.eclipse.jetty.websocket:websocket-server:${jettyVersion}") {
494495
exclude group: "javax.servlet", module: "javax.servlet-api"
495496
}
496497
optional("org.eclipse.jetty.websocket:websocket-client:${jettyVersion}")
497498
optional("com.fasterxml.jackson.core:jackson-databind:${jackson2Version}")
498499
testCompile(project(":spring-test"))
500+
testCompile('org.slf4j:slf4j-log4j12:1.7.10')
499501
testCompile("javax.inject:javax.inject-tck:1")
500502
testCompile("javax.servlet:javax.servlet-api:3.1.0")
501503
testCompile("javax.validation:validation-api:1.0.0.GA")
@@ -756,8 +758,7 @@ project("spring-websocket") {
756758
testCompile("org.apache.tomcat.embed:tomcat-embed-core:${tomcatVersion}")
757759
testCompile("org.apache.tomcat.embed:tomcat-embed-websocket:${tomcatVersion}")
758760
testCompile("org.apache.tomcat.embed:tomcat-embed-logging-juli:${tomcatVersion}")
759-
testCompile("org.projectreactor:reactor-core:${reactorVersion}")
760-
testCompile("org.projectreactor:reactor-net:${reactorVersion}")
761+
testCompile("io.projectreactor:reactor-net:${reactorVersion}")
761762
testCompile("log4j:log4j:1.2.17")
762763
testCompile("org.slf4j:slf4j-jcl:${slf4jVersion}")
763764
}

‎spring-core/src/main/java/org/springframework/core/convert/support/StreamConverter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor t
9999
}
100100

101101
private Object convertFromStream(Stream<?> source, TypeDescriptor streamType, TypeDescriptor targetType) {
102-
List<Object> content = source.collect(Collectors.toList());
102+
List<Object> content = source.collect(Collectors.<Object>toList());
103103
TypeDescriptor listType = TypeDescriptor.collection(List.class, streamType.getElementTypeDescriptor());
104104
return this.conversionService.convert(content, listType, targetType);
105105
}

‎spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor11StompCodec.java‎ renamed to ‎spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2StompCodec.java‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,13 @@
1818

1919
import java.nio.ByteBuffer;
2020

21-
import reactor.function.Consumer;
22-
import reactor.function.Function;
23-
import reactor.io.Buffer;
24-
import reactor.io.encoding.Codec;
2521

2622
import org.springframework.messaging.Message;
2723
import org.springframework.util.Assert;
24+
import reactor.fn.Consumer;
25+
import reactor.fn.Function;
26+
import reactor.io.buffer.Buffer;
27+
import reactor.io.codec.Codec;
2828

2929
/**
3030
* A Reactor TCP {@link Codec} for sending and receiving STOMP messages.
@@ -33,7 +33,7 @@
3333
* @author Rossen Stoyanchev
3434
* @since 4.0
3535
*/
36-
public class Reactor11StompCodec implements Codec<Buffer, Message<byte[]>, Message<byte[]>> {
36+
public class Reactor2StompCodec extends Codec<Buffer, Message<byte[]>, Message<byte[]>> {
3737

3838
private final StompDecoder stompDecoder;
3939

@@ -42,11 +42,11 @@ public class Reactor11StompCodec implements Codec<Buffer, Message<byte[]>, Messa
4242
private final Function<Message<byte[]>, Buffer> encodingFunction;
4343

4444

45-
public Reactor11StompCodec() {
45+
public Reactor2StompCodec() {
4646
this(new StompEncoder(), new StompDecoder());
4747
}
4848

49-
public Reactor11StompCodec(StompEncoder encoder, StompDecoder decoder) {
49+
public Reactor2StompCodec(StompEncoder encoder, StompDecoder decoder) {
5050
Assert.notNull(encoder, "'encoder' is required");
5151
Assert.notNull(decoder, "'decoder' is required");
5252
this.stompEncoder = encoder;
@@ -64,6 +64,10 @@ public Function<Message<byte[]>, Buffer> encoder() {
6464
return this.encodingFunction;
6565
}
6666

67+
@Override
68+
public Buffer apply(Message<byte[]> message) {
69+
return encodingFunction.apply(message);
70+
}
6771

6872
private static class EncodingFunction implements Function<Message<byte[]>, Buffer> {
6973

‎spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor11TcpStompClient.java‎ renamed to ‎spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/Reactor2TcpStompClient.java‎

Lines changed: 46 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -15,72 +15,88 @@
1515
*/
1616
package org.springframework.messaging.simp.stomp;
1717

18-
import java.util.Arrays;
19-
import java.util.Properties;
20-
21-
import reactor.core.Environment;
22-
import reactor.core.configuration.ConfigurationReader;
23-
import reactor.core.configuration.DispatcherConfiguration;
24-
import reactor.core.configuration.DispatcherType;
25-
import reactor.core.configuration.ReactorConfiguration;
26-
import reactor.net.netty.tcp.NettyTcpClient;
27-
import reactor.net.tcp.TcpClient;
28-
import reactor.net.tcp.spec.TcpClientSpec;
29-
3018
import org.springframework.messaging.Message;
3119
import org.springframework.messaging.tcp.TcpOperations;
32-
import org.springframework.messaging.tcp.reactor.Reactor11TcpClient;
20+
import org.springframework.messaging.tcp.reactor.Reactor2TcpClient;
3321
import org.springframework.util.concurrent.ListenableFuture;
22+
import reactor.Environment;
23+
import reactor.core.config.ConfigurationReader;
24+
import reactor.core.config.DispatcherConfiguration;
25+
import reactor.core.config.DispatcherType;
26+
import reactor.core.config.ReactorConfiguration;
27+
import reactor.fn.Function;
28+
import reactor.io.net.Spec;
29+
30+
import java.util.Arrays;
31+
import java.util.Properties;
3432

3533
/**
3634
* A STOMP over TCP client that uses
37-
* {@link org.springframework.messaging.tcp.reactor.Reactor11TcpClient
35+
* {@link Reactor2TcpClient
3836
* Reactor11TcpClient}.
3937
*
4038
* @author Rossen Stoyanchev
4139
* @since 4.2
4240
*/
43-
public class Reactor11TcpStompClient extends StompClientSupport {
41+
public class Reactor2TcpStompClient extends StompClientSupport {
4442

4543
private final TcpOperations<byte[]> tcpClient;
4644

4745

4846
/**
4947
* Create an instance with host "127.0.0.1" and port 61613.
5048
*/
51-
public Reactor11TcpStompClient() {
49+
public Reactor2TcpStompClient() {
5250
this("127.0.0.1", 61613);
5351
}
5452

5553
/**
5654
* Create an instance with the given host and port.
55+
*
5756
* @param host the host
5857
* @param port the port
5958
*/
60-
public Reactor11TcpStompClient(String host, int port) {
61-
this.tcpClient = new Reactor11TcpClient<byte[]>(createNettyTcpClient(host, port));
59+
public Reactor2TcpStompClient(final String host, final int port) {
60+
this.tcpClient = new Reactor2TcpClient<byte[]>(createNettyTcpClientFactory(host, port));
6261
}
6362

64-
private TcpClient<Message<byte[]>, Message<byte[]>> createNettyTcpClient(String host, int port) {
65-
return new TcpClientSpec<Message<byte[]>, Message<byte[]>>(NettyTcpClient.class)
66-
.env(new Environment(new StompClientDispatcherConfigReader()))
67-
.codec(new Reactor11StompCodec(new StompEncoder(), new StompDecoder()))
68-
.connect(host, port)
69-
.get();
63+
private Function<Spec.TcpClientSpec<Message<byte[]>, Message<byte[]>>,
64+
Spec.TcpClientSpec<Message<byte[]>, Message<byte[]>>> createNettyTcpClientFactory(
65+
final String host, final int port
66+
) {
67+
68+
final Environment environment = new Environment(new StompClientDispatcherConfigReader()).assignErrorJournal();
69+
70+
return new Function<Spec.TcpClientSpec<Message<byte[]>, Message<byte[]>>,
71+
Spec.TcpClientSpec<Message<byte[]>, Message<byte[]>>>() {
72+
73+
@Override
74+
public Spec.TcpClientSpec<Message<byte[]>, Message<byte[]>> apply(Spec.TcpClientSpec<Message<byte[]>,
75+
Message<byte[]>> spec) {
76+
77+
return spec
78+
.codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder()))
79+
.env(environment)
80+
.dispatcher(environment.getCachedDispatchers("StompClient").get())
81+
.connect(host, port);
82+
}
83+
};
7084
}
7185

7286
/**
7387
* Create an instance with a pre-configured TCP client.
88+
*
7489
* @param tcpClient the client to use
7590
*/
76-
public Reactor11TcpStompClient(TcpOperations<byte[]> tcpClient) {
91+
public Reactor2TcpStompClient(TcpOperations<byte[]> tcpClient) {
7792
this.tcpClient = tcpClient;
7893
}
7994

8095

8196
/**
8297
* Connect and notify the given {@link StompSessionHandler} when connected
8398
* on the STOMP level,
99+
*
84100
* @param handler the handler for the STOMP session
85101
* @return ListenableFuture for access to the session when ready for use
86102
*/
@@ -91,8 +107,9 @@ public ListenableFuture<StompSession> connect(StompSessionHandler handler) {
91107
/**
92108
* An overloaded version of {@link #connect(StompSessionHandler)} that
93109
* accepts headers to use for the STOMP CONNECT frame.
110+
*
94111
* @param connectHeaders headers to add to the CONNECT frame
95-
* @param handler the handler for the STOMP session
112+
* @param handler the handler for the STOMP session
96113
* @return ListenableFuture for access to the session when ready for use
97114
*/
98115
public ListenableFuture<StompSession> connect(StompHeaders connectHeaders, StompSessionHandler handler) {
@@ -117,9 +134,10 @@ private static class StompClientDispatcherConfigReader implements ConfigurationR
117134
@Override
118135
public ReactorConfiguration read() {
119136
String dispatcherName = "StompClient";
120-
DispatcherType dispatcherType = DispatcherType.THREAD_POOL_EXECUTOR;
137+
DispatcherType dispatcherType = DispatcherType.DISPATCHER_GROUP;
121138
DispatcherConfiguration config = new DispatcherConfiguration(dispatcherName, dispatcherType, 128, 0);
122-
return new ReactorConfiguration(Arrays.<DispatcherConfiguration>asList(config), dispatcherName, new Properties());
139+
return new ReactorConfiguration(Arrays.<DispatcherConfiguration>asList(config), dispatcherName, new Properties
140+
());
123141
}
124142
}
125143

‎spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@
4040
import org.springframework.messaging.tcp.TcpConnection;
4141
import org.springframework.messaging.tcp.TcpConnectionHandler;
4242
import org.springframework.messaging.tcp.TcpOperations;
43-
import org.springframework.messaging.tcp.reactor.Reactor11TcpClient;
43+
import org.springframework.messaging.tcp.reactor.Reactor2TcpClient;
4444
import org.springframework.util.Assert;
4545
import org.springframework.util.concurrent.ListenableFuture;
4646
import org.springframework.util.concurrent.ListenableFutureCallback;
@@ -327,7 +327,7 @@ public String getVirtualHost() {
327327

328328
/**
329329
* Configure a TCP client for managing TCP connections to the STOMP broker.
330-
* By default {@link org.springframework.messaging.tcp.reactor.Reactor11TcpClient} is used.
330+
* By default {@link Reactor2TcpClient} is used.
331331
*/
332332
public void setTcpClient(TcpOperations<byte[]> tcpClient) {
333333
this.tcpClient = tcpClient;
@@ -379,7 +379,7 @@ protected void startInternal() {
379379
if (this.tcpClient == null) {
380380
StompDecoder decoder = new StompDecoder();
381381
decoder.setHeaderInitializer(getHeaderInitializer());
382-
Reactor11StompCodec codec = new Reactor11StompCodec(new StompEncoder(), decoder);
382+
Reactor2StompCodec codec = new Reactor2StompCodec(new StompEncoder(), decoder);
383383
this.tcpClient = new StompTcpClientFactory().create(this.relayHost, this.relayPort, codec);
384384
}
385385

@@ -956,8 +956,8 @@ public ListenableFuture<Void> forward(Message<?> message, StompHeaderAccessor ac
956956

957957
private static class StompTcpClientFactory {
958958

959-
public TcpOperations<byte[]> create(String relayHost, int relayPort, Reactor11StompCodec codec) {
960-
return new Reactor11TcpClient<byte[]>(relayHost, relayPort, codec);
959+
public TcpOperations<byte[]> create(String relayHost, int relayPort, Reactor2StompCodec codec) {
960+
return new Reactor2TcpClient<byte[]>(relayHost, relayPort, codec);
961961
}
962962
}
963963

‎spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/AbstractPromiseToListenableFutureAdapter.java‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,14 @@
2020
import java.util.concurrent.TimeUnit;
2121
import java.util.concurrent.TimeoutException;
2222

23-
import reactor.core.composable.Promise;
24-
import reactor.function.Consumer;
25-
2623
import org.springframework.util.Assert;
2724
import org.springframework.util.concurrent.FailureCallback;
2825
import org.springframework.util.concurrent.ListenableFuture;
2926
import org.springframework.util.concurrent.ListenableFutureCallback;
3027
import org.springframework.util.concurrent.ListenableFutureCallbackRegistry;
3128
import org.springframework.util.concurrent.SuccessCallback;
29+
import reactor.fn.Consumer;
30+
import reactor.rx.Promise;
3231

3332
/**
3433
* Adapts a reactor {@link Promise} to {@link ListenableFuture} optionally converting
@@ -55,8 +54,7 @@ protected AbstractPromiseToListenableFutureAdapter(Promise<S> promise) {
5554
public void accept(S result) {
5655
try {
5756
registry.success(adapt(result));
58-
}
59-
catch (Throwable t) {
57+
} catch (Throwable t) {
6058
registry.failure(t);
6159
}
6260
}

‎spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/PassThroughPromiseToListenableFutureAdapter.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,8 @@
1616

1717
package org.springframework.messaging.tcp.reactor;
1818

19-
import reactor.core.composable.Promise;
19+
20+
import reactor.rx.Promise;
2021

2122
/**
2223
* A Promise-to-ListenableFutureAdapter where the source and the target from the Promise and

0 commit comments

Comments
 (0)