Skip to content

Commit ff6abf3

Browse files
artem-vartem-v
authored
Merge pull request #702 from scalecube/update/new-configuration-methods-in-sc-discovery
Added more ScalecubeServiceDiscovery config fucntions
2 parents c6454eb + 16918ad commit ff6abf3

File tree

21 files changed

+183
-185
lines changed

21 files changed

+183
-185
lines changed

‎services-benchmarks/src/main/java/io/scalecube/services/benchmarks/transport/BenchmarkServiceState.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,15 +37,15 @@ public void beforeAll() {
3737
.transport(RSocketServiceTransport::new)
3838
.startAwait();
3939

40-
Address seedAddress = seed.discovery().address();
40+
final Address seedAddress = seed.discovery().address();
4141

4242
node =
4343
Microservices.builder()
4444
.metrics(registry())
4545
.discovery(
46-
serviceEndpoint ->
47-
new ScalecubeServiceDiscovery(serviceEndpoint)
48-
.options(opts -> opts.membership(cfg -> cfg.seedMembers(seedAddress))))
46+
endpoint ->
47+
new ScalecubeServiceDiscovery(endpoint)
48+
.membership(cfg -> cfg.seedMembers(seedAddress)))
4949
.transport(RSocketServiceTransport::new)
5050
.services(services)
5151
.startAwait();

‎services-discovery/src/main/java/io/scalecube/services/discovery/ScalecubeServiceDiscovery.java‎

Lines changed: 47 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,11 @@
44
import io.scalecube.cluster.ClusterConfig;
55
import io.scalecube.cluster.ClusterImpl;
66
import io.scalecube.cluster.ClusterMessageHandler;
7+
import io.scalecube.cluster.fdetector.FailureDetectorConfig;
8+
import io.scalecube.cluster.gossip.GossipConfig;
9+
import io.scalecube.cluster.membership.MembershipConfig;
710
import io.scalecube.cluster.membership.MembershipEvent;
11+
import io.scalecube.cluster.transport.api.TransportConfig;
812
import io.scalecube.net.Address;
913
import io.scalecube.services.ServiceEndpoint;
1014
import io.scalecube.services.discovery.api.ServiceDiscovery;
@@ -32,10 +36,8 @@ public final class ScalecubeServiceDiscovery implements ServiceDiscovery {
3236
private static final Logger LOGGER =
3337
LoggerFactory.getLogger("io.scalecube.services.discovery.ServiceDiscovery");
3438

35-
private final ServiceEndpoint serviceEndpoint;
36-
39+
private ServiceEndpoint serviceEndpoint;
3740
private ClusterConfig clusterConfig;
38-
3941
private Cluster cluster;
4042

4143
private final DirectProcessor<ServiceDiscoveryEvent> subject = DirectProcessor.create();
@@ -63,9 +65,9 @@ private ScalecubeServiceDiscovery(ScalecubeServiceDiscovery other) {
6365
}
6466

6567
/**
66-
* Setter for {@code ClusterConfig.Builder} options.
68+
* Setter for {@code ClusterConfig} options.
6769
*
68-
* @param opts ClusterConfig options builder
70+
* @param opts options operator
6971
* @return new instance of {@code ScalecubeServiceDiscovery}
7072
*/
7173
public ScalecubeServiceDiscovery options(UnaryOperator<ClusterConfig> opts) {
@@ -74,6 +76,46 @@ public ScalecubeServiceDiscovery options(UnaryOperator<ClusterConfig> opts) {
7476
return d;
7577
}
7678

79+
/**
80+
* Setter for {@code TransportConfig} options.
81+
*
82+
* @param opts options operator
83+
* @return new instance of {@code ScalecubeServiceDiscovery}
84+
*/
85+
public ScalecubeServiceDiscovery transport(UnaryOperator<TransportConfig> opts) {
86+
return options(cfg -> cfg.transport(opts));
87+
}
88+
89+
/**
90+
* Setter for {@code MembershipConfig} options.
91+
*
92+
* @param opts options operator
93+
* @return new instance of {@code ScalecubeServiceDiscovery}
94+
*/
95+
public ScalecubeServiceDiscovery membership(UnaryOperator<MembershipConfig> opts) {
96+
return options(cfg -> cfg.membership(opts));
97+
}
98+
99+
/**
100+
* Setter for {@code GossipConfig} options.
101+
*
102+
* @param opts options operator
103+
* @return new instance of {@code ScalecubeServiceDiscovery}
104+
*/
105+
public ScalecubeServiceDiscovery gossip(UnaryOperator<GossipConfig> opts) {
106+
return options(cfg -> cfg.gossip(opts));
107+
}
108+
109+
/**
110+
* Setter for {@code FailureDetectorConfig} options.
111+
*
112+
* @param opts options operator
113+
* @return new instance of {@code ScalecubeServiceDiscovery}
114+
*/
115+
public ScalecubeServiceDiscovery failureDetector(UnaryOperator<FailureDetectorConfig> opts) {
116+
return options(cfg -> cfg.failureDetector(opts));
117+
}
118+
77119
@Override
78120
public Address address() {
79121
return cluster.address();

‎services-discovery/src/test/java/io/scalecube/services/discovery/ScalecubeServiceDiscoveryTest.java‎

Lines changed: 8 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -219,21 +219,19 @@ public static ServiceEndpoint newServiceEndpoint() {
219219
private Mono<ServiceDiscovery> newServiceDiscovery(
220220
Address seedAddress, MetadataCodec metadataCodec) {
221221
return Mono.fromCallable(
222-
() -> {
223-
ServiceEndpoint serviceEndpoint = newServiceEndpoint();
224-
return new ScalecubeServiceDiscovery(serviceEndpoint)
225-
.options(opts -> opts.metadataCodec(metadataCodec))
226-
.options(opts -> opts.gossip(cfg -> GOSSIP_CONFIG))
227-
.options(opts -> opts.membership(cfg -> MEMBERSHIP_CONFIG))
228-
.options(opts -> opts.membership(cfg -> cfg.seedMembers(seedAddress)));
229-
});
222+
() ->
223+
new ScalecubeServiceDiscovery(newServiceEndpoint())
224+
.options(opts -> opts.metadataCodec(metadataCodec))
225+
.gossip(cfg -> GOSSIP_CONFIG)
226+
.membership(cfg -> MEMBERSHIP_CONFIG)
227+
.membership(cfg -> cfg.seedMembers(seedAddress)));
230228
}
231229

232230
private Address startSeed(MetadataCodec metadataCodec) {
233231
return new ScalecubeServiceDiscovery(newServiceEndpoint())
234232
.options(opts -> opts.metadataCodec(metadataCodec))
235-
.options(opts -> opts.gossip(cfg -> GOSSIP_CONFIG))
236-
.options(opts -> opts.membership(cfg -> MEMBERSHIP_CONFIG))
233+
.gossip(cfg -> GOSSIP_CONFIG)
234+
.membership(cfg -> MEMBERSHIP_CONFIG)
237235
.start()
238236
.block()
239237
.address();

‎services-examples-parent/services-examples-runner/src/main/java/io/scalecube/services/examples/ExamplesRunner.java‎

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -32,9 +32,8 @@ public class ExamplesRunner {
3232
* Main method of gateway runner.
3333
*
3434
* @param args program arguments
35-
* @throws Exception exception thrown
3635
*/
37-
public static void main(String[] args) throws Exception {
36+
public static void main(String[] args) {
3837
ConfigRegistry configRegistry = ConfigBootstrap.configRegistry();
3938

4039
Config config =
@@ -53,7 +52,7 @@ public static void main(String[] args) throws Exception {
5352
LOGGER.info("Number of worker threads: " + numOfThreads);
5453

5554
Microservices.builder()
56-
.discovery(serviceEndpoint -> serviceDiscovery(serviceEndpoint, config))
55+
.discovery(endpoint -> serviceDiscovery(endpoint, config))
5756
.transport(
5857
() ->
5958
new RSocketServiceTransport()
@@ -77,14 +76,10 @@ public static void main(String[] args) throws Exception {
7776
.block();
7877
}
7978

80-
private static ServiceDiscovery serviceDiscovery(ServiceEndpoint serviceEndpoint, Config config) {
81-
return new ScalecubeServiceDiscovery(serviceEndpoint)
82-
.options(
83-
opts ->
84-
opts.membership(cfg1 -> cfg1.seedMembers(config.seedAddresses()))
85-
.transport(cfg1 -> cfg1.port(config.discoveryPort()))
86-
.memberHost(config.memberHost())
87-
.memberPort(config.memberPort()));
79+
private static ServiceDiscovery serviceDiscovery(ServiceEndpoint endpoint, Config config) {
80+
return new ScalecubeServiceDiscovery(endpoint)
81+
.membership(cfg -> cfg.seedMembers(config.seedAddresses()))
82+
.transport(cfg -> cfg.port(config.discoveryPort()));
8883
}
8984

9085
public static class Config {

‎services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/exceptions/ExceptionMapperExample.java‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,9 @@
11
package io.scalecube.services.examples.exceptions;
22

3+
import io.scalecube.net.Address;
34
import io.scalecube.services.Microservices;
4-
import io.scalecube.services.ServiceEndpoint;
55
import io.scalecube.services.ServiceInfo;
66
import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
7-
import io.scalecube.services.discovery.api.ServiceDiscovery;
87
import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
98
import java.util.Collections;
109

@@ -30,9 +29,14 @@ public static void main(String[] args) throws InterruptedException {
3029

3130
System.err.println("ms1 started: " + ms1.serviceAddress());
3231

32+
final Address address1 = ms1.discovery().address();
33+
3334
Microservices ms2 =
3435
Microservices.builder()
35-
.discovery(serviceEndpoint -> serviceDiscovery(serviceEndpoint, ms1))
36+
.discovery(
37+
endpoint ->
38+
new ScalecubeServiceDiscovery(endpoint)
39+
.membership(cfg -> cfg.seedMembers(address1)))
3640
.transport(RSocketServiceTransport::new)
3741
.services(
3842
call -> {
@@ -63,10 +67,4 @@ public static void main(String[] args) throws InterruptedException {
6367

6468
Thread.currentThread().join();
6569
}
66-
67-
private static ServiceDiscovery serviceDiscovery(
68-
ServiceEndpoint serviceEndpoint, Microservices ms1) {
69-
return new ScalecubeServiceDiscovery(serviceEndpoint)
70-
.options(opts -> opts.membership(cfg -> cfg.seedMembers(ms1.discovery().address())));
71-
}
7270
}

‎services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/helloworld/Example1.java‎

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

3+
import io.scalecube.net.Address;
34
import io.scalecube.services.Microservices;
45
import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
56
import io.scalecube.services.examples.helloworld.service.GreetingServiceImpl;
@@ -29,16 +30,15 @@ public static void main(String[] args) {
2930
.transport(RSocketServiceTransport::new)
3031
.startAwait();
3132

33+
final Address seedAddress = seed.discovery().address();
34+
3235
// Construct a ScaleCube node which joins the cluster hosting the Greeting Service
3336
Microservices ms =
3437
Microservices.builder()
3538
.discovery(
36-
serviceEndpoint ->
37-
new ScalecubeServiceDiscovery(serviceEndpoint)
38-
.options(
39-
opts ->
40-
opts.membership(
41-
cfg -> cfg.seedMembers(seed.discovery().address()))))
39+
endpoint ->
40+
new ScalecubeServiceDiscovery(endpoint)
41+
.membership(cfg -> cfg.seedMembers(seedAddress)))
4242
.transport(RSocketServiceTransport::new)
4343
.services(new GreetingServiceImpl())
4444
.startAwait();

‎services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/helloworld/Example2.java‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,10 @@
11
package io.scalecube.services.examples.helloworld;
22

3+
import io.scalecube.net.Address;
34
import io.scalecube.services.Microservices;
45
import io.scalecube.services.ServiceCall;
5-
import io.scalecube.services.ServiceEndpoint;
66
import io.scalecube.services.api.ServiceMessage;
77
import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
8-
import io.scalecube.services.discovery.api.ServiceDiscovery;
98
import io.scalecube.services.examples.helloworld.service.GreetingServiceImpl;
109
import io.scalecube.services.examples.helloworld.service.api.Greeting;
1110
import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
@@ -40,9 +39,14 @@ public static void main(String[] args) {
4039
.startAwait();
4140

4241
// Construct a ScaleCube node which joins the cluster hosting the Greeting Service
42+
final Address seedAddress = seed.discovery().address();
43+
4344
Microservices ms =
4445
Microservices.builder()
45-
.discovery(serviceEndpoint -> serviceDiscovery(serviceEndpoint, seed))
46+
.discovery(
47+
endpoint ->
48+
new ScalecubeServiceDiscovery(endpoint)
49+
.membership(cfg -> cfg.seedMembers(seedAddress)))
4650
.transport(RSocketServiceTransport::new)
4751
.services(new GreetingServiceImpl())
4852
.startAwait();
@@ -68,10 +72,4 @@ public static void main(String[] args) {
6872
seed.onShutdown().block();
6973
ms.onShutdown().block();
7074
}
71-
72-
private static ServiceDiscovery serviceDiscovery(
73-
ServiceEndpoint serviceEndpoint, Microservices seed) {
74-
return new ScalecubeServiceDiscovery(serviceEndpoint)
75-
.options(opts -> opts.membership(cfg -> cfg.seedMembers(seed.discovery().address())));
76-
}
7775
}

‎services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/helloworld/Example3.java‎

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

3+
import io.scalecube.net.Address;
34
import io.scalecube.services.Microservices;
45
import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
56
import io.scalecube.services.examples.helloworld.service.BidiGreetingImpl;
@@ -30,16 +31,15 @@ public static void main(String[] args) {
3031
.transport(RSocketServiceTransport::new)
3132
.startAwait();
3233

34+
final Address seedAddress = seed.discovery().address();
35+
3336
// Construct a ScaleCube node which joins the cluster hosting the Greeting Service
3437
Microservices ms =
3538
Microservices.builder()
3639
.discovery(
37-
serviceEndpoint ->
38-
new ScalecubeServiceDiscovery(serviceEndpoint)
39-
.options(
40-
opts ->
41-
opts.membership(
42-
cfg -> cfg.seedMembers(seed.discovery().address()))))
40+
endpoint ->
41+
new ScalecubeServiceDiscovery(endpoint)
42+
.membership(cfg -> cfg.seedMembers(seedAddress)))
4343
.transport(RSocketServiceTransport::new)
4444
.services(new BidiGreetingImpl())
4545
.startAwait();

‎services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/orderbook/Example1.java‎

Lines changed: 7 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,7 @@
22

33
import io.scalecube.net.Address;
44
import io.scalecube.services.Microservices;
5-
import io.scalecube.services.ServiceEndpoint;
65
import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
7-
import io.scalecube.services.discovery.api.ServiceDiscovery;
86
import io.scalecube.services.examples.orderbook.service.DefaultMarketDataService;
97
import io.scalecube.services.examples.orderbook.service.OrderBookSnapshoot;
108
import io.scalecube.services.examples.orderbook.service.OrderRequest;
@@ -40,10 +38,14 @@ public static void main(String[] args) throws InterruptedException {
4038
.transport(RSocketServiceTransport::new)
4139
.startAwait();
4240

41+
final Address gatewayAddress = gateway.discovery().address();
42+
4343
Microservices ms =
4444
Microservices.builder()
4545
.discovery(
46-
serviceEndpoint -> serviceDiscovery(serviceEndpoint, gateway.discovery().address()))
46+
endpoint ->
47+
new ScalecubeServiceDiscovery(endpoint)
48+
.membership(cfg -> cfg.seedMembers(gatewayAddress)))
4749
.transport(RSocketServiceTransport::new)
4850
.services(new DefaultMarketDataService())
4951
.startAwait();
@@ -64,7 +66,7 @@ public static void main(String[] args) throws InterruptedException {
6466
new Order(
6567
new PriceLevel(Side.BUY, RANDOM.nextInt(10) + 1), // prices
6668
System.currentTimeMillis(),
67-
Long.valueOf(RANDOM.nextInt(110) + 1 + "")), // units
69+
Long.parseLong(RANDOM.nextInt(110) + 1 + "")), // units
6870
INSTRUMENT))
6971
.block();
7072
} else {
@@ -74,7 +76,7 @@ public static void main(String[] args) throws InterruptedException {
7476
new Order(
7577
new PriceLevel(Side.SELL, RANDOM.nextInt(10) + 1), // prices
7678
System.currentTimeMillis(),
77-
Long.valueOf(RANDOM.nextInt(70) + 1 + "")), // units
79+
Long.parseLong(RANDOM.nextInt(70) + 1 + "")), // units
7880
INSTRUMENT))
7981
.block();
8082
}
@@ -89,12 +91,6 @@ public static void main(String[] args) throws InterruptedException {
8991
Thread.currentThread().join();
9092
}
9193

92-
private static ServiceDiscovery serviceDiscovery(
93-
ServiceEndpoint serviceEndpoint, Address address) {
94-
return new ScalecubeServiceDiscovery(serviceEndpoint)
95-
.options(opts -> opts.membership(cfg -> cfg.seedMembers(address)));
96-
}
97-
9894
private static void print(OrderBookSnapshoot snapshot) {
9995

10096
System.out.println("====== Asks ========");

0 commit comments

Comments
 (0)