Skip to content

Commit ad9d79d

Browse files
artem-vartem-v
authored
Merge pull request #723 from scalecube/update-versions
Update versions
2 parents 2b29e1e + 3d1c814 commit ad9d79d

File tree

4 files changed

+40
-56
lines changed

4 files changed

+40
-56
lines changed

‎pom.xml‎

Lines changed: 10 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -22,19 +22,18 @@
2222
</scm>
2323

2424
<properties>
25-
<jackson.version>2.10.0.pr1</jackson.version>
26-
<scalecube-cluster.version>2.4.10</scalecube-cluster.version>
25+
<jackson.version>2.11.0</jackson.version>
26+
<scalecube-cluster.version>2.4.11</scalecube-cluster.version>
2727
<scalecube-commons.version>1.0.2</scalecube-commons.version>
2828
<scalecube-benchmarks.version>1.2.2</scalecube-benchmarks.version>
2929
<scalecube-config.version>0.4.3</scalecube-config.version>
30-
<reactor.version>Dysprosium-RELEASE</reactor.version>
31-
<rsocket.version>1.0.0-RC5</rsocket.version>
30+
<reactor.version>Dysprosium-SR7</reactor.version>
31+
<rsocket.version>1.0.0-RC7</rsocket.version>
3232
<metrics.version>3.1.2</metrics.version>
3333
<protostuff.version>1.6.0</protostuff.version>
34-
<netty.version>4.1.37.Final</netty.version>
35-
<reactor-netty.version>0.9.0.RELEASE</reactor-netty.version>
34+
<netty.version>4.1.48.Final</netty.version>
3635
<slf4j.version>1.7.30</slf4j.version>
37-
<log4j.version>2.11.0</log4j.version>
36+
<log4j.version>2.13.2</log4j.version>
3837
<disruptor.version>3.4.2</disruptor.version>
3938
<jsr305.version>3.0.2</jsr305.version>
4039
<jctools.version>2.1.2</jctools.version>
@@ -104,14 +103,12 @@
104103
</dependency>
105104
<dependency>
106105
<groupId>org.apache.logging.log4j</groupId>
107-
<artifactId>log4j-slf4j-impl</artifactId>
108-
<version>${log4j.version}</version>
109-
</dependency>
110-
<dependency>
111-
<groupId>org.apache.logging.log4j</groupId>
112-
<artifactId>log4j-core</artifactId>
106+
<artifactId>log4j-bom</artifactId>
113107
<version>${log4j.version}</version>
108+
<type>pom</type>
109+
<scope>import</scope>
114110
</dependency>
111+
115112
<!-- Disruptor -->
116113
<dependency>
117114
<groupId>com.lmax</groupId>

‎services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketClientTransport.java‎

Lines changed: 5 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package io.scalecube.services.transport.rsocket;
22

33
import io.rsocket.RSocket;
4-
import io.rsocket.RSocketFactory;
4+
import io.rsocket.core.RSocketConnector;
55
import io.rsocket.frame.decoder.PayloadDecoder;
66
import io.rsocket.transport.netty.client.TcpClientTransport;
77
import io.scalecube.net.Address;
@@ -46,16 +46,10 @@ public ClientChannel create(Address address) {
4646

4747
private Mono<RSocket> connect(Address address, Map<Address, Mono<RSocket>> monoMap) {
4848
TcpClient tcpClient = this.tcpClient.host(address.host()).port(address.port());
49-
50-
Mono<RSocket> rsocketMono =
51-
RSocketFactory.connect()
52-
.frameDecoder(PayloadDecoder.DEFAULT)
53-
.errorConsumer(
54-
th -> LOGGER.warn("Exception occurred at rsocket client transport: " + th))
55-
.transport(() -> TcpClientTransport.create(tcpClient))
56-
.start();
57-
58-
return rsocketMono
49+
return RSocketConnector.create()
50+
.payloadDecoder(PayloadDecoder.DEFAULT)
51+
.errorConsumer(th1 -> LOGGER.warn("Exception occurred at rsocket client transport: " + th1))
52+
.connect(() -> TcpClientTransport.create(tcpClient))
5953
.doOnSuccess(
6054
rsocket -> {
6155
LOGGER.info("Connected successfully on {}", address);

‎services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServerTransport.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package io.scalecube.services.transport.rsocket;
22

33
import io.rsocket.RSocketFactory;
4+
import io.rsocket.core.RSocketServer;
45
import io.rsocket.frame.decoder.PayloadDecoder;
56
import io.rsocket.transport.netty.server.CloseableChannel;
67
import io.rsocket.transport.netty.server.TcpServerTransport;
@@ -52,14 +53,12 @@ public Mono<ServerTransport> bind(ServiceMethodRegistry methodRegistry) {
5253
connection.onDispose(
5354
() -> LOGGER.info("Connection closed on {}", connection.channel()));
5455
});
55-
56-
return RSocketFactory.receive()
57-
.frameDecoder(PayloadDecoder.DEFAULT)
56+
return RSocketServer.create()
57+
.acceptor(new RSocketServiceAcceptor(codec, methodRegistry))
58+
.payloadDecoder(PayloadDecoder.DEFAULT)
5859
.errorConsumer(
5960
th -> LOGGER.warn("Exception occurred at rsocket server transport: " + th))
60-
.acceptor(new RSocketServiceAcceptor(codec, methodRegistry))
61-
.transport(() -> TcpServerTransport.create(tcpServer))
62-
.start()
61+
.bind(TcpServerTransport.create(tcpServer))
6362
.doOnSuccess(channel -> serverChannel = channel)
6463
.thenReturn(this);
6564
});

‎services/src/test/java/io/scalecube/services/ServiceRegistryTest.java‎

Lines changed: 20 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
import io.scalecube.services.sut.GreetingServiceImpl;
1919
import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
2020
import java.time.Duration;
21-
import java.util.ArrayList;
2221
import java.util.List;
2322
import java.util.concurrent.CopyOnWriteArrayList;
2423
import java.util.function.Function;
@@ -32,7 +31,7 @@
3231

3332
public class ServiceRegistryTest extends BaseTest {
3433

35-
public static final Duration TIMEOUT = Duration.ofSeconds(6);
34+
public static final Duration TIMEOUT = Duration.ofSeconds(30);
3635

3736
private static Stream<Arguments> metadataCodecSource() {
3837
return Stream.of(
@@ -44,16 +43,15 @@ private static Stream<Arguments> metadataCodecSource() {
4443
@ParameterizedTest
4544
@MethodSource("metadataCodecSource")
4645
public void test_added_removed_registration_events(MetadataCodec metadataCodec) {
47-
48-
List<ServiceDiscoveryEvent> events = new ArrayList<>();
46+
ReplayProcessor<ServiceDiscoveryEvent> events = ReplayProcessor.create();
4947

5048
Microservices seed =
5149
Microservices.builder()
5250
.discovery(defServiceDiscovery(metadataCodec))
5351
.transport(RSocketServiceTransport::new)
5452
.startAwait();
5553

56-
seed.discovery().listenDiscovery().subscribe(events::add);
54+
seed.discovery().listenDiscovery().subscribe(events);
5755

5856
Address seedAddress = seed.discovery().address();
5957

@@ -71,17 +69,17 @@ public void test_added_removed_registration_events(MetadataCodec metadataCodec)
7169
.services(new GreetingServiceImpl())
7270
.startAwait();
7371

74-
Mono.when(ms1.shutdown(), ms2.shutdown()).then(Mono.delay(TIMEOUT)).block();
75-
76-
assertEquals(6, events.size());
77-
assertEquals(ENDPOINT_ADDED, events.get(0).type());
78-
assertEquals(ENDPOINT_ADDED, events.get(1).type());
79-
assertEquals(ENDPOINT_LEAVING, events.get(2).type());
80-
assertEquals(ENDPOINT_LEAVING, events.get(3).type());
81-
assertEquals(ENDPOINT_REMOVED, events.get(4).type());
82-
assertEquals(ENDPOINT_REMOVED, events.get(5).type());
83-
84-
seed.shutdown().block();
72+
StepVerifier.create(events)
73+
.assertNext(event -> assertEquals(ENDPOINT_ADDED, event.type()))
74+
.assertNext(event -> assertEquals(ENDPOINT_ADDED, event.type()))
75+
.then(() -> Mono.whenDelayError(ms1.shutdown(), ms2.shutdown()).block(TIMEOUT))
76+
.assertNext(event -> assertEquals(ENDPOINT_LEAVING, event.type()))
77+
.assertNext(event -> assertEquals(ENDPOINT_LEAVING, event.type()))
78+
.assertNext(event -> assertEquals(ENDPOINT_REMOVED, event.type()))
79+
.assertNext(event -> assertEquals(ENDPOINT_REMOVED, event.type()))
80+
.then(() -> seed.shutdown().block(TIMEOUT))
81+
.thenCancel()
82+
.verify(TIMEOUT);
8583
}
8684

8785
@ParameterizedTest
@@ -126,13 +124,11 @@ public void test_listen_to_discovery_events(MetadataCodec metadataCodec) {
126124
cluster.add(ms2);
127125
})
128126
.assertNext(event -> assertEquals(ENDPOINT_ADDED, event.type()))
129-
.then(() -> cluster.remove(2).shutdown().block())
127+
.then(() -> cluster.remove(2).shutdown().block(TIMEOUT))
130128
.assertNext(event -> assertEquals(ENDPOINT_LEAVING, event.type()))
131-
.thenAwait(TIMEOUT)
132129
.assertNext(event -> assertEquals(ENDPOINT_REMOVED, event.type()))
133-
.then(() -> cluster.remove(1).shutdown().block())
130+
.then(() -> cluster.remove(1).shutdown().block(TIMEOUT))
134131
.assertNext(event -> assertEquals(ENDPOINT_LEAVING, event.type()))
135-
.thenAwait(TIMEOUT)
136132
.assertNext(event -> assertEquals(ENDPOINT_REMOVED, event.type()))
137133
.thenCancel()
138134
.verify(TIMEOUT);
@@ -147,9 +143,8 @@ public void test_listen_to_discovery_events(MetadataCodec metadataCodec) {
147143
.thenCancel()
148144
.verify(TIMEOUT);
149145

150-
Mono.when(cluster.stream().map(Microservices::shutdown).toArray(Mono[]::new))
151-
.then(Mono.delay(TIMEOUT))
152-
.block();
146+
Mono.whenDelayError(cluster.stream().map(Microservices::shutdown).toArray(Mono[]::new))
147+
.block(TIMEOUT);
153148
}
154149

155150
@ParameterizedTest
@@ -203,9 +198,8 @@ public void test_delayed_listen_to_discovery_events(MetadataCodec metadataCodec)
203198
.thenCancel()
204199
.verify(TIMEOUT);
205200

206-
Mono.when(cluster.stream().map(Microservices::shutdown).toArray(Mono[]::new))
207-
.then(Mono.delay(TIMEOUT))
208-
.block();
201+
Mono.whenDelayError(cluster.stream().map(Microservices::shutdown).toArray(Mono[]::new))
202+
.block(TIMEOUT);
209203
}
210204

211205
private Function<ServiceEndpoint, ServiceDiscovery> defServiceDiscovery(

0 commit comments

Comments
 (0)