From 270366fdfec8d10243ef7659bbf7fde4a6a71604 Mon Sep 17 00:00:00 2001
From: Artem Vysochyn <artem-v@users.noreply.github.com>
Date: Mon, 23 Sep 2024 15:17:32 +0300
Subject: [PATCH] Get rid of address (#394)
Get rid of `Address` and `scalecube-commons`
---
cluster-api/pom.xml | 8 +-
.../java/io/scalecube/cluster/Cluster.java | 9 +-
.../java/io/scalecube/cluster/Member.java | 11 +-
.../cluster/membership/MembershipConfig.java | 9 +-
.../cluster/utils/NetworkEmulator.java | 41 +++---
.../utils/NetworkEmulatorTransport.java | 7 +-
.../cluster/utils/NetworkEmulatorTest.java | 14 +-
cluster/pom.xml | 8 +-
.../io/scalecube/cluster/ClusterImpl.java | 56 +++----
.../fdetector/FailureDetectorImpl.java | 17 +--
.../cluster/gossip/GossipProtocolImpl.java | 10 +-
.../membership/MembershipProtocol.java | 5 +-
.../membership/MembershipProtocolImpl.java | 51 ++++---
.../cluster/metadata/MetadataStoreImpl.java | 5 +-
.../io/scalecube/cluster/ClusterTest.java | 102 ++++++-------
.../fdetector/FailureDetectorTest.java | 41 +++---
.../cluster/gossip/GossipDelayTest.java | 9 +-
.../cluster/gossip/GossipProtocolTest.java | 11 +-
.../cluster/gossip/GossipRequestTest.java | 3 +-
.../membership/MembershipProtocolTest.java | 48 +++---
.../membership/MembershipRecordTest.java | 5 +-
pom.xml | 13 +-
transport-parent/transport-api/pom.xml | 11 +-
.../cluster/transport/api/DistinctErrors.java | 139 ++++++++++++++++++
.../cluster/transport/api/Message.java | 17 +--
.../cluster/transport/api/Transport.java | 64 +++++++-
.../transport/api/TransportConfig.java | 7 +-
.../io/scalecube/transport/netty/Sender.java | 3 +-
.../transport/netty/TransportImpl.java | 60 ++++----
.../transport/netty/tcp/TcpSender.java | 12 +-
.../netty/websocket/WebsocketSender.java | 12 +-
.../scalecube/transport/netty/BaseTest.java | 3 +-
.../netty/tcp/TcpTransportSendOrderTest.java | 3 +-
.../transport/netty/tcp/TcpTransportTest.java | 7 +-
.../WebsocketTransportSendOrderTest.java | 3 +-
.../websocket/WebsocketTransportTest.java | 7 +-
36 files changed, 505 insertions(+), 326 deletions(-)
create mode 100644 transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/DistinctErrors.java
diff --git a/cluster-api/pom.xml b/cluster-api/pom.xml
index c81d3a51..16ff3dcf 100644
--- a/cluster-api/pom.xml
+++ b/cluster-api/pom.xml
@@ -1,5 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -12,10 +14,6 @@
<name>ScaleCube/ClusterApi</name>
<dependencies>
- <dependency>
- <groupId>io.scalecube</groupId>
- <artifactId>scalecube-commons</artifactId>
- </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>scalecube-transport-api</artifactId>
diff --git a/cluster-api/src/main/java/io/scalecube/cluster/Cluster.java b/cluster-api/src/main/java/io/scalecube/cluster/Cluster.java
index c0df446b..6c949224 100644
--- a/cluster-api/src/main/java/io/scalecube/cluster/Cluster.java
+++ b/cluster-api/src/main/java/io/scalecube/cluster/Cluster.java
@@ -1,7 +1,6 @@
package io.scalecube.cluster;
import io.scalecube.cluster.transport.api.Message;
-import io.scalecube.net.Address;
import java.util.Collection;
import java.util.Optional;
import reactor.core.publisher.Mono;
@@ -10,11 +9,11 @@
public interface Cluster {
/**
- * Returns {@link Address} of this cluster instance.
+ * Returns address of this cluster instance.
*
* @return cluster address
*/
- Address address();
+ String address();
/**
* Spreads given message between cluster members using gossiping protocol.
@@ -52,7 +51,7 @@ public interface Cluster {
*
* @return member by id
*/
- Optional<Member> member(String id);
+ Optional<Member> memberById(String id);
/**
* Returns cluster member by given address or null if no member with such address exists at joined
@@ -60,7 +59,7 @@ public interface Cluster {
*
* @return member by address
*/
- Optional<Member> member(Address address);
+ Optional<Member> memberByAddress(String address);
/**
* Returns list of all members of the joined cluster. This will include all cluster members
diff --git a/cluster-api/src/main/java/io/scalecube/cluster/Member.java b/cluster-api/src/main/java/io/scalecube/cluster/Member.java
index 65dc7835..8963930f 100644
--- a/cluster-api/src/main/java/io/scalecube/cluster/Member.java
+++ b/cluster-api/src/main/java/io/scalecube/cluster/Member.java
@@ -1,7 +1,6 @@
package io.scalecube.cluster;
import io.scalecube.cluster.membership.MembershipConfig;
-import io.scalecube.net.Address;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
@@ -20,7 +19,7 @@ public final class Member implements Externalizable {
private String id;
private String alias;
- private Address address;
+ private String address;
private String namespace;
public Member() {}
@@ -33,7 +32,7 @@ public Member() {}
* @param address member address; not null
* @param namespace namespace; not null
*/
- public Member(String id, String alias, Address address, String namespace) {
+ public Member(String id, String alias, String address, String namespace) {
this.id = Objects.requireNonNull(id, "member id");
this.alias = alias; // optional
this.address = Objects.requireNonNull(address, "member address");
@@ -76,7 +75,7 @@ public String namespace() {
* @see io.scalecube.cluster.transport.api.TransportConfig#port(int)
* @return member address
*/
- public Address address() {
+ public String address() {
return address;
}
@@ -110,7 +109,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
out.writeUTF(alias);
}
// address
- out.writeUTF(address.toString());
+ out.writeUTF(address);
// namespace
out.writeUTF(namespace);
}
@@ -125,7 +124,7 @@ public void readExternal(ObjectInput in) throws IOException {
alias = in.readUTF();
}
// address
- address = Address.from(in.readUTF());
+ address = in.readUTF();
// namespace
this.namespace = in.readUTF();
}
diff --git a/cluster-api/src/main/java/io/scalecube/cluster/membership/MembershipConfig.java b/cluster-api/src/main/java/io/scalecube/cluster/membership/MembershipConfig.java
index 01b8114c..630210e3 100644
--- a/cluster-api/src/main/java/io/scalecube/cluster/membership/MembershipConfig.java
+++ b/cluster-api/src/main/java/io/scalecube/cluster/membership/MembershipConfig.java
@@ -1,6 +1,5 @@
package io.scalecube.cluster.membership;
-import io.scalecube.net.Address;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -24,7 +23,7 @@ public final class MembershipConfig implements Cloneable {
public static final int DEFAULT_LOCAL_SUSPICION_MULT = 3;
public static final int DEFAULT_LOCAL_SYNC_INTERVAL = 15_000;
- private List<Address> seedMembers = Collections.emptyList();
+ private List<String> seedMembers = Collections.emptyList();
private int syncInterval = DEFAULT_SYNC_INTERVAL;
private int syncTimeout = DEFAULT_SYNC_TIMEOUT;
private int suspicionMult = DEFAULT_SUSPICION_MULT;
@@ -67,7 +66,7 @@ public static MembershipConfig defaultLocalConfig() {
.syncInterval(DEFAULT_LOCAL_SYNC_INTERVAL);
}
- public List<Address> seedMembers() {
+ public List<String> seedMembers() {
return seedMembers;
}
@@ -77,7 +76,7 @@ public List<Address> seedMembers() {
* @param seedMembers seed members
* @return new {@code MembershipConfig} instance
*/
- public MembershipConfig seedMembers(Address... seedMembers) {
+ public MembershipConfig seedMembers(String... seedMembers) {
return seedMembers(Arrays.asList(seedMembers));
}
@@ -87,7 +86,7 @@ public MembershipConfig seedMembers(Address... seedMembers) {
* @param seedMembers seed members
* @return new {@code MembershipConfig} instance
*/
- public MembershipConfig seedMembers(List<Address> seedMembers) {
+ public MembershipConfig seedMembers(List<String> seedMembers) {
MembershipConfig m = clone();
m.seedMembers = Collections.unmodifiableList(new ArrayList<>(seedMembers));
return m;
diff --git a/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulator.java b/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulator.java
index d51a7526..552c10bc 100644
--- a/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulator.java
+++ b/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulator.java
@@ -1,7 +1,6 @@
package io.scalecube.cluster.utils;
import io.scalecube.cluster.transport.api.Message;
-import io.scalecube.net.Address;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
@@ -30,21 +29,21 @@ public final class NetworkEmulator {
private volatile OutboundSettings defaultOutboundSettings = new OutboundSettings(0, 0);
private volatile InboundSettings defaultInboundSettings = new InboundSettings(true);
- private final Map<Address, OutboundSettings> outboundSettings = new ConcurrentHashMap<>();
- private final Map<Address, InboundSettings> inboundSettings = new ConcurrentHashMap<>();
+ private final Map<String, OutboundSettings> outboundSettings = new ConcurrentHashMap<>();
+ private final Map<String, InboundSettings> inboundSettings = new ConcurrentHashMap<>();
private final AtomicLong totalMessageSentCount = new AtomicLong();
private final AtomicLong totalOutboundMessageLostCount = new AtomicLong();
private final AtomicLong totalInboundMessageLostCount = new AtomicLong();
- private final Address address;
+ private final String address;
/**
* Creates new instance of network emulator.
*
* @param address local address
*/
- NetworkEmulator(Address address) {
+ NetworkEmulator(String address) {
this.address = address;
}
@@ -56,7 +55,7 @@ public final class NetworkEmulator {
* @param destination address of target endpoint
* @return network outbound settings
*/
- public OutboundSettings outboundSettings(Address destination) {
+ public OutboundSettings outboundSettings(String destination) {
return outboundSettings.getOrDefault(destination, defaultOutboundSettings);
}
@@ -67,7 +66,7 @@ public OutboundSettings outboundSettings(Address destination) {
* @param lossPercent loss in percents
* @param meanDelay mean delay
*/
- public void outboundSettings(Address destination, int lossPercent, int meanDelay) {
+ public void outboundSettings(String destination, int lossPercent, int meanDelay) {
OutboundSettings settings = new OutboundSettings(lossPercent, meanDelay);
outboundSettings.put(destination, settings);
LOGGER.debug("[{}] Set outbound settings {} to {}", address, settings, destination);
@@ -103,7 +102,7 @@ public void unblockAllOutbound() {
*
* @param destinations collection of target endpoints where to apply
*/
- public void blockOutbound(Address... destinations) {
+ public void blockOutbound(String... destinations) {
blockOutbound(Arrays.asList(destinations));
}
@@ -112,8 +111,8 @@ public void blockOutbound(Address... destinations) {
*
* @param destinations collection of target endpoints where to apply
*/
- public void blockOutbound(Collection<Address> destinations) {
- for (Address destination : destinations) {
+ public void blockOutbound(Collection<String> destinations) {
+ for (String destination : destinations) {
outboundSettings.put(destination, new OutboundSettings(100, 0));
}
LOGGER.debug("[{}] Blocked outbound to {}", address, destinations);
@@ -124,7 +123,7 @@ public void blockOutbound(Collection<Address> destinations) {
*
* @param destinations collection of target endpoints where to apply
*/
- public void unblockOutbound(Address... destinations) {
+ public void unblockOutbound(String... destinations) {
unblockOutbound(Arrays.asList(destinations));
}
@@ -133,7 +132,7 @@ public void unblockOutbound(Address... destinations) {
*
* @param destinations collection of target endpoints where to apply
*/
- public void unblockOutbound(Collection<Address> destinations) {
+ public void unblockOutbound(Collection<String> destinations) {
destinations.forEach(outboundSettings::remove);
LOGGER.debug("[{}] Unblocked outbound {}", address, destinations);
}
@@ -164,7 +163,7 @@ public long totalOutboundMessageLostCount() {
* @param address target address
* @return mono message
*/
- public Mono<Message> tryFailOutbound(Message msg, Address address) {
+ public Mono<Message> tryFailOutbound(Message msg, String address) {
return Mono.defer(
() -> {
totalMessageSentCount.incrementAndGet();
@@ -187,7 +186,7 @@ public Mono<Message> tryFailOutbound(Message msg, Address address) {
* @param address target address
* @return mono message
*/
- public Mono<Message> tryDelayOutbound(Message msg, Address address) {
+ public Mono<Message> tryDelayOutbound(Message msg, String address) {
return Mono.defer(
() -> {
totalMessageSentCount.incrementAndGet();
@@ -209,7 +208,7 @@ public Mono<Message> tryDelayOutbound(Message msg, Address address) {
* @param destination address of target endpoint
* @return network inbound settings
*/
- public InboundSettings inboundSettings(Address destination) {
+ public InboundSettings inboundSettings(String destination) {
return inboundSettings.getOrDefault(destination, defaultInboundSettings);
}
@@ -218,7 +217,7 @@ public InboundSettings inboundSettings(Address destination) {
*
* @param shallPass shallPass inbound flag
*/
- public void inboundSettings(Address destination, boolean shallPass) {
+ public void inboundSettings(String destination, boolean shallPass) {
InboundSettings settings = new InboundSettings(shallPass);
inboundSettings.put(destination, settings);
LOGGER.debug("[{}] Set inbound settings {} to {}", address, settings, destination);
@@ -253,7 +252,7 @@ public void unblockAllInbound() {
*
* @param destinations collection of target endpoints where to apply
*/
- public void blockInbound(Address... destinations) {
+ public void blockInbound(String... destinations) {
blockInbound(Arrays.asList(destinations));
}
@@ -262,8 +261,8 @@ public void blockInbound(Address... destinations) {
*
* @param destinations collection of target endpoints where to apply
*/
- public void blockInbound(Collection<Address> destinations) {
- for (Address destination : destinations) {
+ public void blockInbound(Collection<String> destinations) {
+ for (String destination : destinations) {
inboundSettings.put(destination, new InboundSettings(false));
}
LOGGER.debug("[{}] Blocked inbound from {}", address, destinations);
@@ -274,7 +273,7 @@ public void blockInbound(Collection<Address> destinations) {
*
* @param destinations collection of target endpoints where to apply
*/
- public void unblockInbound(Address... destinations) {
+ public void unblockInbound(String... destinations) {
unblockInbound(Arrays.asList(destinations));
}
@@ -283,7 +282,7 @@ public void unblockInbound(Address... destinations) {
*
* @param destinations collection of target endpoints where to apply
*/
- public void unblockInbound(Collection<Address> destinations) {
+ public void unblockInbound(Collection<String> destinations) {
destinations.forEach(inboundSettings::remove);
LOGGER.debug("[{}] Unblocked inbound from {}", address, destinations);
}
diff --git a/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulatorTransport.java b/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulatorTransport.java
index 381042c5..dcfbf677 100644
--- a/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulatorTransport.java
+++ b/cluster-testlib/src/main/java/io/scalecube/cluster/utils/NetworkEmulatorTransport.java
@@ -2,7 +2,6 @@
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -26,7 +25,7 @@ public NetworkEmulator networkEmulator() {
}
@Override
- public Address address() {
+ public String address() {
return transport.address();
}
@@ -46,7 +45,7 @@ public boolean isStopped() {
}
@Override
- public Mono<Void> send(Address address, Message message) {
+ public Mono<Void> send(String address, Message message) {
return Mono.defer(
() ->
Mono.just(enhanceWithSender(message))
@@ -56,7 +55,7 @@ public Mono<Void> send(Address address, Message message) {
}
@Override
- public Mono<Message> requestResponse(Address address, Message request) {
+ public Mono<Message> requestResponse(String address, Message request) {
return Mono.defer(
() ->
Mono.just(enhanceWithSender(request))
diff --git a/cluster-testlib/src/test/java/io/scalecube/cluster/utils/NetworkEmulatorTest.java b/cluster-testlib/src/test/java/io/scalecube/cluster/utils/NetworkEmulatorTest.java
index 1919bc6f..b65137db 100644
--- a/cluster-testlib/src/test/java/io/scalecube/cluster/utils/NetworkEmulatorTest.java
+++ b/cluster-testlib/src/test/java/io/scalecube/cluster/utils/NetworkEmulatorTest.java
@@ -1,7 +1,6 @@
package io.scalecube.cluster.utils;
import io.scalecube.cluster.utils.NetworkEmulator.OutboundSettings;
-import io.scalecube.net.Address;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -10,24 +9,23 @@ public class NetworkEmulatorTest extends BaseTest {
@Test
public void testResolveLinkSettingsBySocketAddress() {
// Init network emulator
- Address address = Address.from("localhost:1234");
- NetworkEmulator networkEmulator = new NetworkEmulator(address);
- networkEmulator.outboundSettings(Address.create("localhost", 5678), 25, 10);
- networkEmulator.outboundSettings(Address.create("192.168.0.1", 8765), 10, 20);
+ NetworkEmulator networkEmulator = new NetworkEmulator("localhost:1234");
+ networkEmulator.outboundSettings("localhost:" + 5678, 25, 10);
+ networkEmulator.outboundSettings("192.168.0.1:" + 8765, 10, 20);
networkEmulator.setDefaultOutboundSettings(0, 2);
// Check resolve by hostname:port
- OutboundSettings link1 = networkEmulator.outboundSettings(Address.create("localhost", 5678));
+ OutboundSettings link1 = networkEmulator.outboundSettings("localhost:" + 5678);
Assertions.assertEquals(25, link1.lossPercent());
Assertions.assertEquals(10, link1.meanDelay());
// Check resolve by ipaddr:port
- OutboundSettings link2 = networkEmulator.outboundSettings(Address.create("192.168.0.1", 8765));
+ OutboundSettings link2 = networkEmulator.outboundSettings("192.168.0.1:" + 8765);
Assertions.assertEquals(10, link2.lossPercent());
Assertions.assertEquals(20, link2.meanDelay());
// Check default link settings
- OutboundSettings link3 = networkEmulator.outboundSettings(Address.create("localhost", 8765));
+ OutboundSettings link3 = networkEmulator.outboundSettings("localhost:" + 8765);
Assertions.assertEquals(0, link3.lossPercent());
Assertions.assertEquals(2, link3.meanDelay());
}
diff --git a/cluster/pom.xml b/cluster/pom.xml
index 1fd9ae60..8b9287ed 100644
--- a/cluster/pom.xml
+++ b/cluster/pom.xml
@@ -1,5 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -12,10 +14,6 @@
<name>ScaleCube/Cluster</name>
<dependencies>
- <dependency>
- <groupId>io.scalecube</groupId>
- <artifactId>scalecube-commons</artifactId>
- </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>scalecube-transport-api</artifactId>
diff --git a/cluster/src/main/java/io/scalecube/cluster/ClusterImpl.java b/cluster/src/main/java/io/scalecube/cluster/ClusterImpl.java
index 77942be8..27309586 100644
--- a/cluster/src/main/java/io/scalecube/cluster/ClusterImpl.java
+++ b/cluster/src/main/java/io/scalecube/cluster/ClusterImpl.java
@@ -1,6 +1,8 @@
package io.scalecube.cluster;
-import static io.scalecube.reactor.RetryNonSerializedEmitFailureHandler.RETRY_NON_SERIALIZED;
+import static io.scalecube.cluster.transport.api.Transport.parseHost;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.busyLooping;
import io.scalecube.cluster.fdetector.FailureDetectorConfig;
import io.scalecube.cluster.fdetector.FailureDetectorImpl;
@@ -16,9 +18,9 @@
import io.scalecube.cluster.transport.api.Transport;
import io.scalecube.cluster.transport.api.TransportConfig;
import io.scalecube.cluster.transport.api.TransportFactory;
-import io.scalecube.net.Address;
import java.io.Serializable;
import java.nio.ByteBuffer;
+import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.Objects;
@@ -111,14 +113,14 @@ private void initLifecycle() {
start
.asMono()
.then(doStart())
- .doOnSuccess(avoid -> onStart.emitEmpty(RETRY_NON_SERIALIZED))
- .doOnError(th -> onStart.emitError(th, RETRY_NON_SERIALIZED))
+ .doOnSuccess(avoid -> onStart.emitEmpty(busyLooping(Duration.ofSeconds(3))))
+ .doOnError(th -> onStart.emitError(th, busyLooping(Duration.ofSeconds(3))))
.subscribe(null, th -> LOGGER.error("[{}][doStart] Exception occurred:", localMember, th));
shutdown
.asMono()
.then(doShutdown())
- .doFinally(s -> onShutdown.emitEmpty(RETRY_NON_SERIALIZED))
+ .doFinally(s -> onShutdown.emitEmpty(busyLooping(Duration.ofSeconds(3))))
.subscribe(
null,
th ->
@@ -224,7 +226,7 @@ public ClusterImpl handler(Function<Cluster, ClusterMessageHandler> handler) {
public Mono<Cluster> start() {
return Mono.defer(
() -> {
- start.emitEmpty(RETRY_NON_SERIALIZED);
+ start.emitEmpty(busyLooping(Duration.ofSeconds(3)));
return onStart.asMono().thenReturn(this);
});
}
@@ -244,7 +246,7 @@ private Mono<Cluster> doStart0() {
localMember = createLocalMember(boundTransport.address());
transport = new SenderAwareTransport(boundTransport, localMember.address());
- scheduler = Schedulers.newSingle("sc-cluster-" + localMember.address().port(), true);
+ scheduler = Schedulers.newSingle("sc-cluster-" + localMember.address(), true);
failureDetector =
new FailureDetectorImpl(
@@ -283,9 +285,10 @@ private Mono<Cluster> doStart0() {
/*.publishOn(scheduler)*/
// Dont uncomment, already beign executed inside scalecube-cluster thread
.subscribe(
- event -> membershipSink.emitNext(event, RETRY_NON_SERIALIZED),
+ event ->
+ membershipSink.emitNext(event, busyLooping(Duration.ofSeconds(3))),
ex -> LOGGER.error("[{}][membership][error] cause:", localMember, ex),
- () -> membershipSink.emitComplete(RETRY_NON_SERIALIZED)));
+ () -> membershipSink.emitComplete(busyLooping(Duration.ofSeconds(3)))));
return Mono.fromRunnable(() -> failureDetector.start())
.then(Mono.fromRunnable(() -> gossip.start()))
@@ -371,14 +374,13 @@ private Flux<MembershipEvent> listenMembership() {
* @param address transport address
* @return local cluster member with cluster address and cluster member id
*/
- private Member createLocalMember(Address address) {
- int port = Optional.ofNullable(config.externalPort()).orElse(address.port());
+ private Member createLocalMember(String address) {
+ final String finalHost =
+ Optional.ofNullable(config.externalHost()).orElseGet(() -> parseHost(address));
+ final int finalPort =
+ Optional.ofNullable(config.externalPort()).orElseGet(() -> parsePort(address));
- // calculate local member cluster address
- Address memberAddress =
- Optional.ofNullable(config.externalHost())
- .map(host -> Address.create(host, port))
- .orElseGet(() -> Address.create(address.host(), port));
+ final String memberAddress = finalHost + ":" + finalPort;
return new Member(
config.memberId() != null ? config.memberId() : UUID.randomUUID().toString(),
@@ -388,7 +390,7 @@ private Member createLocalMember(Address address) {
}
@Override
- public Address address() {
+ public String address() {
return member().address();
}
@@ -431,13 +433,13 @@ public Member member() {
}
@Override
- public Optional<Member> member(String id) {
- return membership.member(id);
+ public Optional<Member> memberById(String id) {
+ return membership.memberById(id);
}
@Override
- public Optional<Member> member(Address address) {
- return membership.member(address);
+ public Optional<Member> memberByAddress(String address) {
+ return membership.memberByAddress(address);
}
@Override
@@ -449,7 +451,7 @@ public <T> Mono<Void> updateMetadata(T metadata) {
@Override
public void shutdown() {
- shutdown.emitEmpty(RETRY_NON_SERIALIZED);
+ shutdown.emitEmpty(busyLooping(Duration.ofSeconds(3)));
}
private Mono<Void> doShutdown() {
@@ -498,15 +500,15 @@ public Mono<Void> onShutdown() {
private static class SenderAwareTransport implements Transport {
private final Transport transport;
- private final Address address;
+ private final String address;
- private SenderAwareTransport(Transport transport, Address address) {
+ private SenderAwareTransport(Transport transport, String address) {
this.transport = Objects.requireNonNull(transport);
this.address = Objects.requireNonNull(address);
}
@Override
- public Address address() {
+ public String address() {
return transport.address();
}
@@ -526,12 +528,12 @@ public boolean isStopped() {
}
@Override
- public Mono<Void> send(Address address, Message message) {
+ public Mono<Void> send(String address, Message message) {
return Mono.defer(() -> transport.send(address, enhanceWithSender(message)));
}
@Override
- public Mono<Message> requestResponse(Address address, Message request) {
+ public Mono<Message> requestResponse(String address, Message request) {
return Mono.defer(() -> transport.requestResponse(address, enhanceWithSender(request)));
}
diff --git a/cluster/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorImpl.java b/cluster/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorImpl.java
index c7540bc0..274f5a08 100644
--- a/cluster/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorImpl.java
+++ b/cluster/src/main/java/io/scalecube/cluster/fdetector/FailureDetectorImpl.java
@@ -1,6 +1,6 @@
package io.scalecube.cluster.fdetector;
-import static io.scalecube.reactor.RetryNonSerializedEmitFailureHandler.RETRY_NON_SERIALIZED;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.busyLooping;
import io.scalecube.cluster.Member;
import io.scalecube.cluster.fdetector.PingData.AckType;
@@ -8,7 +8,6 @@
import io.scalecube.cluster.membership.MembershipEvent;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -117,7 +116,7 @@ public void stop() {
actionsDisposables.dispose();
// Stop publishing events
- sink.emitComplete(RETRY_NON_SERIALIZED);
+ sink.emitComplete(busyLooping(Duration.ofSeconds(3)));
}
@Override
@@ -145,7 +144,7 @@ private void doPing() {
Message pingMsg = Message.withData(pingData).qualifier(PING).correlationId(cid).build();
LOGGER.debug("[{}][{}] Send Ping to {}", localMember, period, pingMember);
- Address address = pingMember.address();
+ String address = pingMember.address();
transport
.requestResponse(address, pingMsg)
.timeout(Duration.ofMillis(config.pingTimeout()), scheduler)
@@ -232,7 +231,7 @@ private void onMessage(Message message) {
/** Listens to PING message and answers with ACK. */
private void onPing(Message message) {
long period = this.currentPeriod;
- Address sender = message.sender();
+ String sender = message.sender();
LOGGER.debug("[{}][{}] Received Ping from {}", localMember, period, sender);
PingData data = message.data();
data = data.withAckType(AckType.DEST_OK);
@@ -249,7 +248,7 @@ private void onPing(Message message) {
String correlationId = message.correlationId();
Message ackMessage =
Message.withData(data).qualifier(PING_ACK).correlationId(correlationId).build();
- Address address = data.getFrom().address();
+ String address = data.getFrom().address();
LOGGER.debug("[{}][{}] Send PingAck to {}", localMember, period, address);
transport
.send(address, ackMessage)
@@ -275,7 +274,7 @@ private void onPingReq(Message message) {
PingData pingReqData = new PingData(localMember, target, originalIssuer);
Message pingMessage =
Message.withData(pingReqData).qualifier(PING).correlationId(correlationId).build();
- Address address = target.address();
+ String address = target.address();
LOGGER.debug("[{}][{}] Send transit Ping to {}", localMember, period, address);
transport
.send(address, pingMessage)
@@ -305,7 +304,7 @@ private void onTransitPingAck(Message message) {
PingData originalAckData = new PingData(target, data.getTo()).withAckType(ackType);
Message originalAckMessage =
Message.withData(originalAckData).qualifier(PING_ACK).correlationId(correlationId).build();
- Address address = target.address();
+ String address = target.address();
LOGGER.debug("[{}][{}] Resend transit PingAck to {}", localMember, period, address);
transport
.send(address, originalAckMessage)
@@ -378,7 +377,7 @@ private List<Member> selectPingReqMembers(Member pingMember) {
private void publishPingResult(long period, Member member, MemberStatus status) {
LOGGER.debug("[{}][{}] Member {} detected as {}", localMember, period, member, status);
- sink.emitNext(new FailureDetectorEvent(member, status), RETRY_NON_SERIALIZED);
+ sink.emitNext(new FailureDetectorEvent(member, status), busyLooping(Duration.ofSeconds(3)));
}
private MemberStatus computeMemberStatus(Message message, long period) {
diff --git a/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java b/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java
index 1acfd060..3657c06b 100644
--- a/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java
+++ b/cluster/src/main/java/io/scalecube/cluster/gossip/GossipProtocolImpl.java
@@ -1,13 +1,13 @@
package io.scalecube.cluster.gossip;
-import static io.scalecube.reactor.RetryNonSerializedEmitFailureHandler.RETRY_NON_SERIALIZED;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.busyLooping;
import io.scalecube.cluster.ClusterMath;
import io.scalecube.cluster.Member;
import io.scalecube.cluster.membership.MembershipEvent;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -120,7 +120,7 @@ public void stop() {
actionsDisposables.dispose();
// Stop publishing events
- sink.emitComplete(RETRY_NON_SERIALIZED);
+ sink.emitComplete(busyLooping(Duration.ofSeconds(3)));
}
@Override
@@ -208,7 +208,7 @@ private void onGossipRequest(Message message) {
if (gossipState == null) { // new gossip
gossipState = new GossipState(gossip, period);
gossips.put(gossip.gossipId(), gossipState);
- sink.emitNext(gossip.message(), RETRY_NON_SERIALIZED);
+ sink.emitNext(gossip.message(), busyLooping(Duration.ofSeconds(3)));
}
}
if (gossipState != null) {
@@ -287,7 +287,7 @@ private void spreadGossipsTo(long period, Member member) {
}
// Send gossip request
- Address address = member.address();
+ String address = member.address();
gossips.stream()
.map(this::buildGossipRequestMessage)
diff --git a/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocol.java b/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocol.java
index 08907e21..d127a097 100644
--- a/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocol.java
+++ b/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocol.java
@@ -1,7 +1,6 @@
package io.scalecube.cluster.membership;
import io.scalecube.cluster.Member;
-import io.scalecube.net.Address;
import java.util.Collection;
import java.util.Optional;
import reactor.core.publisher.Flux;
@@ -53,7 +52,7 @@ public interface MembershipProtocol {
*
* @return member by id
*/
- Optional<Member> member(String id);
+ Optional<Member> memberById(String id);
/**
* Returns cluster member by given address or null if no member with such address exists at joined
@@ -61,5 +60,5 @@ public interface MembershipProtocol {
*
* @return member by address
*/
- Optional<Member> member(Address address);
+ Optional<Member> memberByAddress(String address);
}
diff --git a/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java b/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
index fad3bca3..5c73af81 100644
--- a/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
+++ b/cluster/src/main/java/io/scalecube/cluster/membership/MembershipProtocolImpl.java
@@ -3,7 +3,8 @@
import static io.scalecube.cluster.membership.MemberStatus.ALIVE;
import static io.scalecube.cluster.membership.MemberStatus.DEAD;
import static io.scalecube.cluster.membership.MemberStatus.LEAVING;
-import static io.scalecube.reactor.RetryNonSerializedEmitFailureHandler.RETRY_NON_SERIALIZED;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.busyLooping;
import io.scalecube.cluster.ClusterConfig;
import io.scalecube.cluster.ClusterMath;
@@ -15,8 +16,8 @@
import io.scalecube.cluster.metadata.MetadataStore;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
import java.net.InetAddress;
+import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -74,7 +75,7 @@ private enum MembershipUpdateReason {
private final Transport transport;
private final MembershipConfig membershipConfig;
private final FailureDetectorConfig failureDetectorConfig;
- private final List<Address> seedMembers;
+ private final List<String> seedMembers;
private final FailureDetector failureDetector;
private final GossipProtocol gossipProtocol;
private final MetadataStore metadataStore;
@@ -160,18 +161,18 @@ public MembershipProtocolImpl(
}
// Remove duplicates and local address(es)
- private List<Address> cleanUpSeedMembers(Collection<Address> seedMembers) {
- InetAddress localIpAddress = Address.getLocalIpAddress();
+ private List<String> cleanUpSeedMembers(Collection<String> seedMembers) {
+ InetAddress localIpAddress = getLocalIpAddress();
String hostAddress = localIpAddress.getHostAddress();
String hostName = localIpAddress.getHostName();
- Address memberAddr = localMember.address();
- Address transportAddr = transport.address();
- Address memberAddrByHostAddress = Address.create(hostAddress, memberAddr.port());
- Address transportAddrByHostAddress = Address.create(hostAddress, transportAddr.port());
- Address memberAddByHostName = Address.create(hostName, memberAddr.port());
- Address transportAddrByHostName = Address.create(hostName, transportAddr.port());
+ String memberAddr = localMember.address();
+ String transportAddr = transport.address();
+ String memberAddrByHostAddress = hostAddress + ":" + parsePort(memberAddr);
+ String transportAddrByHostAddress = hostAddress + ":" + parsePort(transportAddr);
+ String memberAddByHostName = hostName + ":" + parsePort(memberAddr);
+ String transportAddrByHostName = hostName + ":" + parsePort(transportAddr);
return new LinkedHashSet<>(seedMembers)
.stream()
@@ -184,7 +185,15 @@ private List<Address> cleanUpSeedMembers(Collection<Address> seedMembers) {
.collect(Collectors.toList());
}
- private boolean checkAddressesNotEqual(Address address0, Address address1) {
+ private static InetAddress getLocalIpAddress() {
+ try {
+ return InetAddress.getLocalHost();
+ } catch (UnknownHostException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private boolean checkAddressesNotEqual(String address0, String address1) {
if (!address0.equals(address1)) {
return true;
} else {
@@ -300,7 +309,7 @@ public void stop() {
suspicionTimeoutTasks.clear();
// Stop publishing events
- sink.emitComplete(RETRY_NON_SERIALIZED);
+ sink.emitComplete(busyLooping(Duration.ofSeconds(3)));
}
@Override
@@ -320,18 +329,18 @@ public Member member() {
}
@Override
- public Optional<Member> member(String id) {
+ public Optional<Member> memberById(String id) {
return Optional.ofNullable(members.get(id));
}
@Override
- public Optional<Member> member(Address address) {
+ public Optional<Member> memberByAddress(String address) {
return new ArrayList<>(members.values())
.stream().filter(member -> member.address().equals(address)).findFirst();
}
private void doSync() {
- Address address = selectSyncAddress().orElse(null);
+ String address = selectSyncAddress().orElse(null);
if (address == null) {
return;
}
@@ -390,7 +399,7 @@ private Mono<Void> onSyncAck(Message syncAckMsg, boolean onStart) {
private Mono<Void> onSync(Message syncMsg) {
return Mono.defer(
() -> {
- final Address sender = syncMsg.sender();
+ final String sender = syncMsg.sender();
LOGGER.debug("[{}] Received Sync from {}", localMember, sender);
return syncMembership(syncMsg.data(), false)
.doOnSuccess(
@@ -425,7 +434,7 @@ private void onFailureDetectorEvent(FailureDetectorEvent fdEvent) {
// Alive won't override SUSPECT so issue instead extra sync with member to force it spread
// alive with inc + 1
Message syncMsg = prepareSyncDataMsg(SYNC, null);
- Address address = fdEvent.member().address();
+ String address = fdEvent.member().address();
transport
.send(address, syncMsg)
.subscribe(
@@ -464,8 +473,8 @@ private void onMembershipGossip(Message message) {
}
}
- private Optional<Address> selectSyncAddress() {
- List<Address> addresses =
+ private Optional<String> selectSyncAddress() {
+ List<String> addresses =
Stream.concat(seedMembers.stream(), otherMembers().stream().map(Member::address))
.collect(Collectors.collectingAndThen(Collectors.toSet(), ArrayList::new));
Collections.shuffle(addresses);
@@ -726,7 +735,7 @@ private Mono<Void> onLeavingDetected(MembershipRecord r0, MembershipRecord r1) {
private void publishEvent(MembershipEvent event) {
LOGGER.info("[{}][publishEvent] {}", localMember, event);
- sink.emitNext(event, RETRY_NON_SERIALIZED);
+ sink.emitNext(event, busyLooping(Duration.ofSeconds(3)));
}
private Mono<Void> onDeadMemberDetected(MembershipRecord r1) {
diff --git a/cluster/src/main/java/io/scalecube/cluster/metadata/MetadataStoreImpl.java b/cluster/src/main/java/io/scalecube/cluster/metadata/MetadataStoreImpl.java
index 35ba5328..f86680cf 100644
--- a/cluster/src/main/java/io/scalecube/cluster/metadata/MetadataStoreImpl.java
+++ b/cluster/src/main/java/io/scalecube/cluster/metadata/MetadataStoreImpl.java
@@ -4,7 +4,6 @@
import io.scalecube.cluster.Member;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.HashMap;
@@ -148,7 +147,7 @@ public Mono<ByteBuffer> fetchMetadata(Member member) {
return Mono.defer(
() -> {
final String cid = UUID.randomUUID().toString();
- final Address targetAddress = member.address();
+ final String targetAddress = member.address();
LOGGER.debug("[{}][{}] Getting metadata for member {}", localMember, cid, member);
@@ -196,7 +195,7 @@ private void onMessage(Message message) {
}
private void onMetadataRequest(Message message) {
- final Address sender = message.sender();
+ final String sender = message.sender();
LOGGER.debug("[{}] Received GetMetadataReq from {}", localMember, sender);
GetMetadataRequest reqData = message.data();
diff --git a/cluster/src/test/java/io/scalecube/cluster/ClusterTest.java b/cluster/src/test/java/io/scalecube/cluster/ClusterTest.java
index dd104ade..f8ac8eda 100644
--- a/cluster/src/test/java/io/scalecube/cluster/ClusterTest.java
+++ b/cluster/src/test/java/io/scalecube/cluster/ClusterTest.java
@@ -1,5 +1,6 @@
package io.scalecube.cluster;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -8,9 +9,9 @@
import io.scalecube.cluster.membership.MembershipEvent;
import io.scalecube.cluster.membership.MembershipEvent.Type;
import io.scalecube.cluster.metadata.MetadataCodec;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.tcp.TcpTransportFactory;
import java.net.InetAddress;
+import java.net.UnknownHostException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -37,7 +38,7 @@ public class ClusterTest extends BaseTest {
@Test
public void testStartStopRepeatedly() throws Exception {
- Address address = Address.from("localhost:4848");
+ String address = "localhost:4848";
// Start seed node
Cluster seedNode =
@@ -45,7 +46,7 @@ public void testStartStopRepeatedly() throws Exception {
.gossip(opts -> opts.gossipInterval(100))
.failureDetector(opts -> opts.pingInterval(100))
.membership(opts -> opts.syncInterval(100))
- .transport(opts -> opts.port(address.port()))
+ .transport(opts -> opts.port(parsePort(address)))
.transport(opts -> opts.connectTimeout(CONNECT_TIMEOUT))
.transportFactory(TcpTransportFactory::new)
.startAwait();
@@ -72,7 +73,7 @@ public void testStartStopRepeatedly() throws Exception {
.gossip(opts -> opts.gossipInterval(100))
.failureDetector(opts -> opts.pingInterval(100))
.membership(opts -> opts.syncInterval(100))
- .transport(opts -> opts.port(address.port()))
+ .transport(opts -> opts.port(parsePort(address)))
.transport(opts -> opts.connectTimeout(CONNECT_TIMEOUT))
.transportFactory(TcpTransportFactory::new)
.startAwait();
@@ -101,8 +102,8 @@ public void testMembersAccessFromScheduler() {
// Members by address
- Optional<Member> otherNodeOnSeedNode = seedNode.member(otherNode.address());
- Optional<Member> seedNodeOnOtherNode = otherNode.member(seedNode.address());
+ Optional<Member> otherNodeOnSeedNode = seedNode.memberByAddress(otherNode.address());
+ Optional<Member> seedNodeOnOtherNode = otherNode.memberByAddress(seedNode.address());
assertEquals(otherNode.member(), otherNodeOnSeedNode.orElse(null));
assertEquals(seedNode.member(), seedNodeOnOtherNode.orElse(null));
@@ -112,15 +113,11 @@ public void testMembersAccessFromScheduler() {
@Test
public void testJoinLocalhostIgnored() throws InterruptedException {
- InetAddress localIpAddress = Address.getLocalIpAddress();
- Address localAddressByHostname = Address.create(localIpAddress.getHostName(), 4801);
- Address localAddressByIp = Address.create(localIpAddress.getHostAddress(), 4801);
- Address[] addresses = {
- Address.from("localhost:4801"),
- Address.from("127.0.0.1:4801"),
- Address.from("127.0.1.1:4801"),
- localAddressByHostname,
- localAddressByIp
+ InetAddress localIpAddress = getLocalIpAddress();
+ String localAddressByHostname = localIpAddress.getHostName() + ":" + 4801;
+ String localAddressByIp = localIpAddress.getHostAddress() + ":" + 4801;
+ String[] addresses = {
+ "localhost:4801", "127.0.0.1:4801", "127.0.1.1:4801", localAddressByHostname, localAddressByIp
};
// Start seed node
@@ -141,15 +138,11 @@ public void testJoinLocalhostIgnored() throws InterruptedException {
@Test
public void testJoinLocalhostIgnoredWithOverride() throws InterruptedException {
- InetAddress localIpAddress = Address.getLocalIpAddress();
- Address localAddressByHostname = Address.create(localIpAddress.getHostName(), 7878);
- Address localAddressByIp = Address.create(localIpAddress.getHostAddress(), 7878);
- Address[] addresses = {
- Address.from("localhost:7878"),
- Address.from("127.0.0.1:7878"),
- Address.from("127.0.1.1:7878"),
- localAddressByHostname,
- localAddressByIp
+ InetAddress localIpAddress = getLocalIpAddress();
+ String localAddressByHostname = localIpAddress.getHostName() + ":" + 7878;
+ String localAddressByIp = localIpAddress.getHostAddress() + ":" + 7878;
+ String[] addresses = {
+ "localhost:7878", "127.0.0.1:7878", "127.0.1.1:7878", localAddressByHostname, localAddressByIp
};
// Start seed node
@@ -242,7 +235,7 @@ public void onMembershipEvent(MembershipEvent event) {
// Check all test members know valid metadata
for (Cluster node : otherNodes) {
- Optional<Member> memberOptional = node.member(metadataNode.member().id());
+ Optional<Member> memberOptional = node.memberById(metadataNode.member().id());
assertTrue(memberOptional.isPresent());
Member member = memberOptional.get();
assertEquals(metadata, node.metadata(member).orElse(null));
@@ -254,7 +247,7 @@ public void onMembershipEvent(MembershipEvent event) {
// Check all nodes had updated metadata member
for (Cluster node : otherNodes) {
- Optional<Member> memberOptional = node.member(metadataNode.member().id());
+ Optional<Member> memberOptional = node.memberById(metadataNode.member().id());
assertTrue(memberOptional.isPresent());
Member member = memberOptional.get();
assertEquals(updatedMetadata, node.metadata(member).orElse(null));
@@ -315,7 +308,7 @@ public void onMembershipEvent(MembershipEvent event) {
// Check all test members know valid metadata
for (Cluster node : otherNodes) {
- Optional<Member> memberOptional = node.member(metadataNode.member().id());
+ Optional<Member> memberOptional = node.memberById(metadataNode.member().id());
assertTrue(memberOptional.isPresent());
Member member = memberOptional.get();
assertEquals(metadata, node.metadata(member).orElse(null));
@@ -328,7 +321,7 @@ public void onMembershipEvent(MembershipEvent event) {
// Check all nodes had updated metadata member
for (Cluster node : otherNodes) {
- Optional<Member> memberOptional = node.member(metadataNode.member().id());
+ Optional<Member> memberOptional = node.memberById(metadataNode.member().id());
assertTrue(memberOptional.isPresent());
Member member = memberOptional.get();
//noinspection unchecked,OptionalGetWithoutIsPresent
@@ -393,7 +386,7 @@ public void onMembershipEvent(MembershipEvent event) {
// Check all test members know valid metadata
for (Cluster node : otherNodes) {
- Optional<Member> memberOptional = node.member(metadataNode.member().id());
+ Optional<Member> memberOptional = node.memberById(metadataNode.member().id());
assertTrue(memberOptional.isPresent());
Member member = memberOptional.get();
assertEquals(metadata, node.metadata(member).orElse(null));
@@ -407,7 +400,7 @@ public void onMembershipEvent(MembershipEvent event) {
// Check all nodes had updated metadata member
for (Cluster node : otherNodes) {
- Optional<Member> memberOptional = node.member(metadataNode.member().id());
+ Optional<Member> memberOptional = node.memberById(metadataNode.member().id());
assertTrue(memberOptional.isPresent());
Member member = memberOptional.get();
//noinspection unchecked,OptionalGetWithoutIsPresent
@@ -559,9 +552,9 @@ public void testJoinSeedClusterWithNoExistingSeedMember() {
// Start seed node
Cluster seedNode = new ClusterImpl().transportFactory(TcpTransportFactory::new).startAwait();
- Address nonExistingSeed1 = Address.from("localhost:1234");
- Address nonExistingSeed2 = Address.from("localhost:5678");
- Address[] seeds = new Address[] {nonExistingSeed1, nonExistingSeed2, seedNode.address()};
+ String nonExistingSeed1 = "localhost:1234";
+ String nonExistingSeed2 = "localhost:5678";
+ String[] seeds = new String[] {nonExistingSeed1, nonExistingSeed2, seedNode.address()};
Cluster otherNode =
new ClusterImpl()
@@ -575,29 +568,15 @@ public void testJoinSeedClusterWithNoExistingSeedMember() {
shutdown(Arrays.asList(seedNode, otherNode));
}
- private void shutdown(List<Cluster> nodes) {
- try {
- Mono.whenDelayError(
- nodes.stream()
- .peek(Cluster::shutdown)
- .map(Cluster::onShutdown)
- .collect(Collectors.toList()))
- .block(TIMEOUT);
- } catch (Exception ex) {
- LOGGER.error("Exception on cluster shutdown", ex);
- }
- }
-
@Test
public void testExplicitLocalMemberId() {
- ClusterConfig config = ClusterConfig.defaultConfig()
- .memberId("test-member");
+ ClusterConfig config = ClusterConfig.defaultConfig().memberId("test-member");
ClusterImpl cluster = null;
try {
- cluster = (ClusterImpl) new ClusterImpl(config)
- .transportFactory(TcpTransportFactory::new)
- .startAwait();
+ cluster =
+ (ClusterImpl)
+ new ClusterImpl(config).transportFactory(TcpTransportFactory::new).startAwait();
assertEquals("test-member", cluster.member().id());
} finally {
@@ -606,4 +585,25 @@ public void testExplicitLocalMemberId() {
}
}
}
+
+ private static InetAddress getLocalIpAddress() {
+ try {
+ return InetAddress.getLocalHost();
+ } catch (UnknownHostException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void shutdown(List<Cluster> nodes) {
+ try {
+ Mono.whenDelayError(
+ nodes.stream()
+ .peek(Cluster::shutdown)
+ .map(Cluster::onShutdown)
+ .collect(Collectors.toList()))
+ .block(TIMEOUT);
+ } catch (Exception ex) {
+ LOGGER.error("Exception on cluster shutdown", ex);
+ }
+ }
}
diff --git a/cluster/src/test/java/io/scalecube/cluster/fdetector/FailureDetectorTest.java b/cluster/src/test/java/io/scalecube/cluster/fdetector/FailureDetectorTest.java
index 3c2241d8..bcb89765 100644
--- a/cluster/src/test/java/io/scalecube/cluster/fdetector/FailureDetectorTest.java
+++ b/cluster/src/test/java/io/scalecube/cluster/fdetector/FailureDetectorTest.java
@@ -2,6 +2,7 @@
import static io.scalecube.cluster.membership.MemberStatus.ALIVE;
import static io.scalecube.cluster.membership.MemberStatus.SUSPECT;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -12,7 +13,6 @@
import io.scalecube.cluster.transport.api.Transport;
import io.scalecube.cluster.transport.api.TransportConfig;
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
-import io.scalecube.net.Address;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -53,7 +53,7 @@ public void testTrusted() {
Transport a = createTransport();
Transport b = createTransport();
Transport c = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), c.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -82,7 +82,7 @@ public void testSuspected() {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), c.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -119,7 +119,7 @@ public void testTrustedDespiteBadNetwork() {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), c.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -151,7 +151,7 @@ public void testTrustedDespiteDifferentPingTimings() {
Transport a = createTransport();
Transport b = createTransport();
Transport c = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), c.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -183,7 +183,7 @@ public void testSuspectedMemberWithBadNetworkGetsPartitioned() throws Exception
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
NetworkEmulatorTransport d = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), c.address(), d.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), c.address(), d.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -242,7 +242,7 @@ public void testSuspectedMemberWithNormalNetworkGetsPartitioned() throws Excepti
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
NetworkEmulatorTransport d = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), c.address(), d.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), c.address(), d.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -303,7 +303,7 @@ public void testMemberStatusChangeAfterNetworkRecovery() throws Exception {
// Create transports
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address());
+ List<String> members = Arrays.asList(a.address(), b.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -346,7 +346,7 @@ public void testStatusChangeAfterMemberRestart() throws Exception {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport x = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), x.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), x.address());
// Create failure detectors
FailureDetectorImpl fdA = createFd(a, members);
@@ -374,7 +374,7 @@ public void testStatusChangeAfterMemberRestart() throws Exception {
TimeUnit.SECONDS.sleep(2);
// restart node X as XX
- xx = createTransport(new TransportConfig().port(x.address().port()));
+ xx = createTransport(new TransportConfig().port(parsePort(x.address())));
assertEquals(x.address(), xx.address());
fdetectors = Arrays.asList(fdA, fdB, fdXx = createFd(xx, members));
@@ -397,7 +397,7 @@ public void testStatusChangeAfterMemberRestart() throws Exception {
}
}
- private FailureDetectorImpl createFd(Transport transport, List<Address> members) {
+ private FailureDetectorImpl createFd(Transport transport, List<String> members) {
FailureDetectorConfig failureDetectorConfig =
FailureDetectorConfig.defaultLocalConfig() // faster config for local testing
.pingTimeout(100)
@@ -407,15 +407,16 @@ private FailureDetectorImpl createFd(Transport transport, List<Address> members)
}
private FailureDetectorImpl createFd(
- Transport transport, List<Address> addresses, FailureDetectorConfig config) {
+ Transport transport, List<String> addresses, FailureDetectorConfig config) {
Member localMember =
- new Member("member-" + transport.address().port(), null, transport.address(), NAMESPACE);
+ new Member(
+ "member-" + parsePort(transport.address()), null, transport.address(), NAMESPACE);
Flux<MembershipEvent> membershipFlux =
Flux.fromIterable(addresses)
.filter(address -> !transport.address().equals(address))
- .map(address -> new Member("member-" + address.port(), null, address, NAMESPACE))
+ .map(address -> new Member("member-" + parsePort(address), null, address, NAMESPACE))
.map(member -> MembershipEvent.createAdded(member, null, 0));
return new FailureDetectorImpl(localMember, transport, membershipFlux, config, scheduler);
@@ -437,11 +438,11 @@ private static void stop(List<FailureDetectorImpl> fdetectors) {
}
private static void assertStatus(
- Address address,
+ String address,
MemberStatus status,
Collection<FailureDetectorEvent> events,
- Address... expected) {
- List<Address> actual =
+ String... expected) {
+ List<String> actual =
events.stream()
.filter(event -> event.status() == status)
.map(FailureDetectorEvent::member)
@@ -454,7 +455,7 @@ private static void assertStatus(
address, expected.length, status, Arrays.toString(expected), events);
assertEquals(expected.length, actual.size(), msg1);
- for (Address member : expected) {
+ for (String member : expected) {
String msg2 =
String.format("Node %s expected as %s %s, but was: %s", address, status, member, events);
assertTrue(actual.contains(member), msg2);
@@ -462,7 +463,7 @@ private static void assertStatus(
}
private static Future<List<FailureDetectorEvent>> listenNextEventFor(
- FailureDetectorImpl fd, List<Address> addresses) {
+ FailureDetectorImpl fd, List<String> addresses) {
final Transport transport = BaseTest.getField(fd, "transport");
addresses = new ArrayList<>(addresses);
addresses.remove(transport.address()); // exclude self
@@ -471,7 +472,7 @@ private static Future<List<FailureDetectorEvent>> listenNextEventFor(
}
List<CompletableFuture<FailureDetectorEvent>> resultFuture = new ArrayList<>();
- for (final Address member : addresses) {
+ for (final String member : addresses) {
final CompletableFuture<FailureDetectorEvent> future = new CompletableFuture<>();
fd.listen().filter(event -> event.member().address() == member).subscribe(future::complete);
resultFuture.add(future);
diff --git a/cluster/src/test/java/io/scalecube/cluster/gossip/GossipDelayTest.java b/cluster/src/test/java/io/scalecube/cluster/gossip/GossipDelayTest.java
index 3857d1e6..1ea1e583 100644
--- a/cluster/src/test/java/io/scalecube/cluster/gossip/GossipDelayTest.java
+++ b/cluster/src/test/java/io/scalecube/cluster/gossip/GossipDelayTest.java
@@ -1,5 +1,6 @@
package io.scalecube.cluster.gossip;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
import static org.junit.jupiter.api.Assertions.assertEquals;
import io.scalecube.cluster.BaseTest;
@@ -9,7 +10,6 @@
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
-import io.scalecube.net.Address;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -74,7 +74,7 @@ private NetworkEmulatorTransport getNetworkEmulatorTransport(int lostPercent, in
return transport;
}
- private GossipProtocolImpl initGossipProtocol(Transport transport, List<Address> members) {
+ private GossipProtocolImpl initGossipProtocol(Transport transport, List<String> members) {
GossipConfig gossipConfig =
new GossipConfig()
.gossipFanout(gossipFanout)
@@ -82,12 +82,13 @@ private GossipProtocolImpl initGossipProtocol(Transport transport, List<Address>
.gossipRepeatMult(gossipRepeatMultiplier);
Member localMember =
- new Member("member-" + transport.address().port(), null, transport.address(), NAMESPACE);
+ new Member(
+ "member-" + parsePort(transport.address()), null, transport.address(), NAMESPACE);
Flux<MembershipEvent> membershipFlux =
Flux.fromIterable(members)
.filter(address -> !transport.address().equals(address))
- .map(address -> new Member("member-" + address.port(), null, address, NAMESPACE))
+ .map(address -> new Member("member-" + parsePort(address), null, address, NAMESPACE))
.map(member -> MembershipEvent.createAdded(member, null, 0));
GossipProtocolImpl gossipProtocol =
diff --git a/cluster/src/test/java/io/scalecube/cluster/gossip/GossipProtocolTest.java b/cluster/src/test/java/io/scalecube/cluster/gossip/GossipProtocolTest.java
index ae82bca2..9842c07c 100644
--- a/cluster/src/test/java/io/scalecube/cluster/gossip/GossipProtocolTest.java
+++ b/cluster/src/test/java/io/scalecube/cluster/gossip/GossipProtocolTest.java
@@ -4,6 +4,7 @@
import static io.scalecube.cluster.ClusterMath.gossipDisseminationTime;
import static io.scalecube.cluster.ClusterMath.maxMessagesPerGossipPerNode;
import static io.scalecube.cluster.ClusterMath.maxMessagesPerGossipTotal;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -15,7 +16,6 @@
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
-import io.scalecube.net.Address;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -230,7 +230,7 @@ private LongSummaryStatistics computeMessageLostStats(List<GossipProtocolImpl> g
private List<GossipProtocolImpl> initGossipProtocols(int count, int lostPercent, int meanDelay) {
final List<Transport> transports = initTransports(count, lostPercent, meanDelay);
- List<Address> members = new ArrayList<>();
+ List<String> members = new ArrayList<>();
for (Transport transport : transports) {
members.add(transport.address());
}
@@ -251,7 +251,7 @@ private List<Transport> initTransports(int count, int lostPercent, int meanDelay
return transports;
}
- private GossipProtocolImpl initGossipProtocol(Transport transport, List<Address> members) {
+ private GossipProtocolImpl initGossipProtocol(Transport transport, List<String> members) {
GossipConfig gossipConfig =
new GossipConfig()
.gossipFanout(gossipFanout)
@@ -259,12 +259,13 @@ private GossipProtocolImpl initGossipProtocol(Transport transport, List<Address>
.gossipRepeatMult(gossipRepeatMultiplier);
Member localMember =
- new Member("member-" + transport.address().port(), null, transport.address(), NAMESPACE);
+ new Member(
+ "member-" + parsePort(transport.address()), null, transport.address(), NAMESPACE);
Flux<MembershipEvent> membershipFlux =
Flux.fromIterable(members)
.filter(address -> !transport.address().equals(address))
- .map(address -> new Member("member-" + address.port(), null, address, NAMESPACE))
+ .map(address -> new Member("member-" + parsePort(address), null, address, NAMESPACE))
.map(member -> MembershipEvent.createAdded(member, null, 0));
GossipProtocolImpl gossipProtocol =
diff --git a/cluster/src/test/java/io/scalecube/cluster/gossip/GossipRequestTest.java b/cluster/src/test/java/io/scalecube/cluster/gossip/GossipRequestTest.java
index e2dfa1fc..37a3e86d 100644
--- a/cluster/src/test/java/io/scalecube/cluster/gossip/GossipRequestTest.java
+++ b/cluster/src/test/java/io/scalecube/cluster/gossip/GossipRequestTest.java
@@ -8,7 +8,6 @@
import io.scalecube.cluster.Member;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.MessageCodec;
-import io.scalecube.net.Address;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.Serializable;
@@ -39,7 +38,7 @@ public void init() {
@Test
public void testSerializationAndDeserialization() throws Exception {
- Member from = new Member("0", null, Address.from("localhost:1234"), NAMESPACE);
+ Member from = new Member("0", null, "localhost:1234", NAMESPACE);
List<Gossip> gossips = getGossips();
Message message =
Message.withData(new GossipRequest(gossips, from.id())).correlationId("CORR_ID").build();
diff --git a/cluster/src/test/java/io/scalecube/cluster/membership/MembershipProtocolTest.java b/cluster/src/test/java/io/scalecube/cluster/membership/MembershipProtocolTest.java
index 4fced8e2..310e947f 100644
--- a/cluster/src/test/java/io/scalecube/cluster/membership/MembershipProtocolTest.java
+++ b/cluster/src/test/java/io/scalecube/cluster/membership/MembershipProtocolTest.java
@@ -1,5 +1,6 @@
package io.scalecube.cluster.membership;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -18,7 +19,6 @@
import io.scalecube.cluster.transport.api.TransportConfig;
import io.scalecube.cluster.utils.NetworkEmulator;
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
-import io.scalecube.net.Address;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
@@ -76,7 +76,7 @@ public void testLeaveCluster() {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -109,9 +109,8 @@ public void testLeaveCluster() {
public void testLeaveClusterCameBeforeAlive() {
final NetworkEmulatorTransport a = createTransport();
final NetworkEmulatorTransport b = createTransport();
- final Member anotherMember =
- new Member("leavingNodeId-1", null, Address.from("localhost:9236"), NAMESPACE);
- final List<Address> addresses = Arrays.asList(a.address(), b.address());
+ final Member anotherMember = new Member("leavingNodeId-1", null, "localhost:9236", NAMESPACE);
+ final List<String> addresses = Arrays.asList(a.address(), b.address());
final MembershipProtocolImpl cmA = createMembership(a, addresses);
final MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -154,9 +153,8 @@ public void testLeaveClusterCameBeforeAlive() {
public void testLeaveClusterOnly() {
final NetworkEmulatorTransport a = createTransport();
final NetworkEmulatorTransport b = createTransport();
- final Member anotherMember =
- new Member("leavingNodeId-1", null, Address.from("localhost:9236"), NAMESPACE);
- final List<Address> addresses = Arrays.asList(a.address(), b.address());
+ final Member anotherMember = new Member("leavingNodeId-1", null, "localhost:9236", NAMESPACE);
+ final List<String> addresses = Arrays.asList(a.address(), b.address());
final MembershipProtocolImpl cmA = createMembership(a, addresses);
final MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -188,9 +186,8 @@ public void testLeaveClusterOnly() {
public void testLeaveClusterOnSuspectedNode() {
final NetworkEmulatorTransport a = createTransport();
final NetworkEmulatorTransport b = createTransport();
- final Member anotherMember =
- new Member("leavingNodeId-1", null, Address.from("localhost:9236"), NAMESPACE);
- final List<Address> addresses = Arrays.asList(a.address(), b.address());
+ final Member anotherMember = new Member("leavingNodeId-1", null, "localhost:9236", NAMESPACE);
+ final List<String> addresses = Arrays.asList(a.address(), b.address());
final MembershipProtocolImpl cmA = createMembership(a, addresses);
final MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -234,7 +231,7 @@ public void testLeaveClusterOnSuspectedNode() {
public void testLeaveClusterOnAliveAndSuspectedNode() {
final NetworkEmulatorTransport a = createTransport();
final NetworkEmulatorTransport b = createTransport();
- final List<Address> addresses = Arrays.asList(a.address(), b.address());
+ final List<String> addresses = Arrays.asList(a.address(), b.address());
final MembershipProtocolImpl cmA = createMembership(a, addresses);
final MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -270,7 +267,7 @@ public void testInitialPhaseOk() {
Transport a = createTransport();
Transport b = createTransport();
Transport c = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -295,7 +292,7 @@ public void testNetworkPartitionDueNoOutboundThenRecover() {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -341,7 +338,7 @@ public void testMemberLostNetworkDueNoOutboundThenRecover() {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
- List<Address> members = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> members = Arrays.asList(a.address(), b.address(), c.address());
MembershipProtocolImpl cmA = createMembership(a, members);
MembershipProtocolImpl cmB = createMembership(b, members);
@@ -397,7 +394,7 @@ public void testNetworkPartitionTwiceDueNoOutboundThenRecover() {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -467,7 +464,7 @@ public void testNetworkLostOnAllNodesDueNoOutboundThenRecover() {
NetworkEmulatorTransport a = createTransport();
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -523,7 +520,7 @@ public void testLongNetworkPartitionDueNoOutboundThenRemoved() {
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
NetworkEmulatorTransport d = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -576,7 +573,7 @@ public void testRestartStoppedMembers() {
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
NetworkEmulatorTransport d = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -656,7 +653,7 @@ public void testRestartStoppedMembersOnSameAddresses() {
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
NetworkEmulatorTransport d = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -691,8 +688,8 @@ public void testRestartStoppedMembersOnSameAddresses() {
assertSuspected(cmB, cmC.member(), cmD.member());
// Restart C and D on same ports
- c_Restarted = createTransport(new TransportConfig().port(c.address().port()));
- d_Restarted = createTransport(new TransportConfig().port(d.address().port()));
+ c_Restarted = createTransport(new TransportConfig().port(parsePort(c.address())));
+ d_Restarted = createTransport(new TransportConfig().port(parsePort(d.address())));
cmC_Restarted = createMembership(c_Restarted, addresses);
cmD_Restarted = createMembership(d_Restarted, addresses);
@@ -1047,7 +1044,7 @@ public void testNetworkPartitionManyDueNoInboundThenRemovedThenRecover() {
NetworkEmulatorTransport b = createTransport();
NetworkEmulatorTransport c = createTransport();
NetworkEmulatorTransport d = createTransport();
- List<Address> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
+ List<String> addresses = Arrays.asList(a.address(), b.address(), c.address(), d.address());
MembershipProtocolImpl cmA = createMembership(a, addresses);
MembershipProtocolImpl cmB = createMembership(b, addresses);
@@ -1117,7 +1114,7 @@ public void testNetworkPartitionManyDueNoInboundThenRemovedThenRecover() {
}
}
- private static ClusterConfig testConfig(List<Address> seedAddresses) {
+ private static ClusterConfig testConfig(List<String> seedAddresses) {
// Create faster config for local testing
return new ClusterConfig()
.membership(opts -> opts.seedMembers(seedAddresses))
@@ -1129,8 +1126,7 @@ private static ClusterConfig testConfig(List<Address> seedAddresses) {
.metadataTimeout(100);
}
- private MembershipProtocolImpl createMembership(
- Transport transport, List<Address> seedAddresses) {
+ private MembershipProtocolImpl createMembership(Transport transport, List<String> seedAddresses) {
return createMembership(transport, testConfig(seedAddresses));
}
diff --git a/cluster/src/test/java/io/scalecube/cluster/membership/MembershipRecordTest.java b/cluster/src/test/java/io/scalecube/cluster/membership/MembershipRecordTest.java
index 0a5028cb..54a9c869 100644
--- a/cluster/src/test/java/io/scalecube/cluster/membership/MembershipRecordTest.java
+++ b/cluster/src/test/java/io/scalecube/cluster/membership/MembershipRecordTest.java
@@ -9,13 +9,12 @@
import io.scalecube.cluster.BaseTest;
import io.scalecube.cluster.Member;
-import io.scalecube.net.Address;
import org.junit.jupiter.api.Test;
public class MembershipRecordTest extends BaseTest {
- private final Member member = new Member("0", "0", Address.from("localhost:1234"), "ns-0");
- private final Member anotherMember = new Member("1", "1", Address.from("localhost:4567"), "ns-1");
+ private final Member member = new Member("0", "0", "localhost:1234", "ns-0");
+ private final Member anotherMember = new Member("1", "1", "localhost:4567", "ns-1");
private final MembershipRecord r0Null = null;
diff --git a/pom.xml b/pom.xml
index e21c8b82..584671d8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1,5 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -33,8 +35,6 @@
</scm>
<properties>
- <scalecube-commons.version>1.0.24</scalecube-commons.version>
-
<slf4j.version>1.7.36</slf4j.version>
<log4j.version>2.17.2</log4j.version>
<reactor.version>2020.0.32</reactor.version>
@@ -117,13 +117,6 @@
<scope>import</scope>
</dependency>
- <!-- Scalecube Commons -->
- <dependency>
- <groupId>io.scalecube</groupId>
- <artifactId>scalecube-commons</artifactId>
- <version>${scalecube-commons.version}</version>
- </dependency>
-
<!-- Jackson -->
<dependency>
<groupId>com.fasterxml.jackson</groupId>
diff --git a/transport-parent/transport-api/pom.xml b/transport-parent/transport-api/pom.xml
index 9a1f4c38..b0a2d71a 100644
--- a/transport-parent/transport-api/pom.xml
+++ b/transport-parent/transport-api/pom.xml
@@ -1,5 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>scalecube-transport-parent</artifactId>
<groupId>io.scalecube</groupId>
@@ -10,11 +12,4 @@
<artifactId>scalecube-transport-api</artifactId>
<name>ScaleCube/ClusterTransportApi</name>
- <dependencies>
- <dependency>
- <groupId>io.scalecube</groupId>
- <artifactId>scalecube-commons</artifactId>
- </dependency>
- </dependencies>
-
</project>
diff --git a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/DistinctErrors.java b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/DistinctErrors.java
new file mode 100644
index 00000000..a421de0a
--- /dev/null
+++ b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/DistinctErrors.java
@@ -0,0 +1,139 @@
+package io.scalecube.cluster.transport.api;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+public class DistinctErrors {
+
+ private final List<DistinctObservation> distinctObservations = new ArrayList<>();
+ private final long evictionInterval;
+
+ /** Constructor. */
+ public DistinctErrors() {
+ this(null);
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param evictionInterval optional, how long consider incoming observation as unique.
+ */
+ public DistinctErrors(Duration evictionInterval) {
+ this.evictionInterval =
+ evictionInterval != null && evictionInterval.toMillis() > 0
+ ? evictionInterval.toMillis()
+ : Long.MAX_VALUE;
+ }
+
+ /**
+ * Return true if there is an observation (or at least in the eviction time window) of this error
+ * type for a stack trace. Otherwise a new entry will be created and kept.
+ *
+ * @param observation an error observation
+ * @return true if such observation exists.
+ */
+ public boolean contains(Throwable observation) {
+ synchronized (this) {
+ final long now = System.currentTimeMillis();
+ DistinctObservation distinctObservation = find(now, distinctObservations, observation);
+
+ if (distinctObservation == null) {
+ distinctObservations.add(new DistinctObservation(observation, now + evictionInterval));
+ return false;
+ }
+
+ if (distinctObservation.deadline > now) {
+ distinctObservation.resetDeadline(now + evictionInterval);
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private static DistinctObservation find(
+ long now, List<DistinctObservation> existingObservations, Throwable observation) {
+ DistinctObservation existingObservation = null;
+
+ for (int lastIndex = existingObservations.size() - 1, i = lastIndex; i >= 0; i--) {
+ final DistinctObservation o = existingObservations.get(lastIndex);
+
+ if (equals(o.throwable, observation)) {
+ existingObservation = o;
+ break;
+ }
+
+ if (o.deadline > now) {
+ if (i == lastIndex) {
+ existingObservations.remove(i);
+ } else {
+ existingObservations.set(i, existingObservations.remove(lastIndex));
+ }
+ lastIndex--;
+ }
+ }
+
+ return existingObservation;
+ }
+
+ private static boolean equals(Throwable lhs, Throwable rhs) {
+ while (true) {
+ if (lhs == rhs) {
+ return true;
+ }
+
+ if (lhs.getClass() == rhs.getClass()
+ && Objects.equals(lhs.getMessage(), rhs.getMessage())
+ && equals(lhs.getStackTrace(), rhs.getStackTrace())) {
+ lhs = lhs.getCause();
+ rhs = rhs.getCause();
+
+ if (null == lhs && null == rhs) {
+ return true;
+ } else if (null != lhs && null != rhs) {
+ continue;
+ }
+ }
+
+ return false;
+ }
+ }
+
+ private static boolean equals(
+ StackTraceElement[] lhsStackTrace, StackTraceElement[] rhsStackTrace) {
+ if (lhsStackTrace.length != rhsStackTrace.length) {
+ return false;
+ }
+
+ for (int i = 0, length = lhsStackTrace.length; i < length; i++) {
+ final StackTraceElement lhs = lhsStackTrace[i];
+ final StackTraceElement rhs = rhsStackTrace[i];
+
+ if (lhs.getLineNumber() != rhs.getLineNumber()
+ || !lhs.getClassName().equals(rhs.getClassName())
+ || !Objects.equals(lhs.getMethodName(), rhs.getMethodName())
+ || !Objects.equals(lhs.getFileName(), rhs.getFileName())) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private static final class DistinctObservation {
+
+ private final Throwable throwable;
+ private long deadline;
+
+ DistinctObservation(Throwable throwable, long deadline) {
+ this.throwable = throwable;
+ this.deadline = deadline;
+ }
+
+ void resetDeadline(long deadline) {
+ this.deadline = deadline;
+ }
+ }
+}
diff --git a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Message.java b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Message.java
index b5e3e879..96f5105f 100644
--- a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Message.java
+++ b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Message.java
@@ -1,6 +1,5 @@
package io.scalecube.cluster.transport.api;
-import io.scalecube.net.Address;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
@@ -33,8 +32,8 @@ public final class Message implements Externalizable {
public static final String HEADER_CORRELATION_ID = "cid";
/**
- * This header represents sender address of type {@link Address}. It's an address of message
- * originator. This header is optional.
+ * This header represents sender address. It is an address of message originator. This header is
+ * optional.
*/
public static final String HEADER_SENDER = "sender";
@@ -186,12 +185,12 @@ public <T> T data() {
}
/**
- * Returns {@link Address} of the sender of this message.
+ * Returns address of the sender of this message.
*
- * @return address
+ * @return address, or null
*/
- public Address sender() {
- return Optional.ofNullable(header(HEADER_SENDER)).map(Address::from).orElse(null);
+ public String sender() {
+ return Optional.ofNullable(header(HEADER_SENDER)).orElse(null);
}
@Override
@@ -281,8 +280,8 @@ public Builder correlationId(String correlationId) {
return header(HEADER_CORRELATION_ID, correlationId);
}
- public Builder sender(Address sender) {
- return header(HEADER_SENDER, sender.toString());
+ public Builder sender(String sender) {
+ return header(HEADER_SENDER, sender);
}
public Message build() {
diff --git a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Transport.java b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Transport.java
index 2cd0b025..f8ad2d18 100644
--- a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Transport.java
+++ b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/Transport.java
@@ -1,7 +1,8 @@
package io.scalecube.cluster.transport.api;
-import io.scalecube.net.Address;
import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -12,13 +13,14 @@
*/
public interface Transport {
+ Pattern ADDRESS_FORMAT = Pattern.compile("(?<host>^.*):(?<port>\\d+$)");
+
/**
- * Returns local {@link Address} on which current instance of transport listens for incoming
- * messages.
+ * Returns local address on which current instance of transport listens for incoming messages.
*
* @return address
*/
- Address address();
+ String address();
/**
* Start transport. After this call method {@link #address()} shall be eligible for calling.
@@ -50,7 +52,7 @@ public interface Transport {
* @return promise which will be completed with result of sending (void or exception)
* @throws IllegalArgumentException if {@code message} or {@code address} is null
*/
- Mono<Void> send(Address address, Message message);
+ Mono<Void> send(String address, Message message);
/**
* Sends message to the given address. It will issue connect in case if no transport channel by
@@ -62,7 +64,7 @@ public interface Transport {
* @return promise which will be completed with result of sending (message or exception)
* @throws IllegalArgumentException if {@code message} or {@code address} is null
*/
- Mono<Message> requestResponse(Address address, Message request);
+ Mono<Message> requestResponse(String address, Message request);
/**
* Returns stream of received messages. For each observers subscribed to the returned observable:
@@ -124,4 +126,54 @@ static Mono<Transport> bind(TransportConfig config) {
Objects.requireNonNull(config.transportFactory(), "[bind] transportFactory");
return config.transportFactory().createTransport(config).start();
}
+
+ /**
+ * Parses string in format {@code host:port} and returns host part.
+ *
+ * @param address address, must be string in format {@code host:port}
+ * @return address host, or throwing exception
+ */
+ static String parseHost(String address) {
+ if (address == null || address.isEmpty()) {
+ throw new IllegalArgumentException("Cannot parse address host from: " + address);
+ }
+
+ Matcher matcher = ADDRESS_FORMAT.matcher(address);
+ if (!matcher.find()) {
+ throw new IllegalArgumentException("Cannot parse address host from: " + address);
+ }
+
+ String host = matcher.group(1);
+ if (host == null || host.isEmpty()) {
+ throw new IllegalArgumentException("Cannot parse address host from: " + address);
+ }
+
+ return host;
+ }
+
+ /**
+ * Parses string in format {@code host:port} and returns port part.
+ *
+ * @param address address, must be string in format {@code host:port}
+ * @return address port, or throwing exception
+ */
+ static int parsePort(String address) {
+ if (address == null || address.isEmpty()) {
+ throw new IllegalArgumentException("Cannot parse address port from: " + address);
+ }
+
+ Matcher matcher = ADDRESS_FORMAT.matcher(address);
+ if (!matcher.find()) {
+ throw new IllegalArgumentException("Cannot parse address port from: " + address);
+ }
+
+ int port;
+ try {
+ port = Integer.parseInt(matcher.group(2));
+ } catch (NumberFormatException ex) {
+ throw new IllegalArgumentException("Cannot parse address port from: " + address, ex);
+ }
+
+ return port;
+ }
}
diff --git a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/TransportConfig.java b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/TransportConfig.java
index 06129f83..b6837f9e 100644
--- a/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/TransportConfig.java
+++ b/transport-parent/transport-api/src/main/java/io/scalecube/cluster/transport/api/TransportConfig.java
@@ -1,6 +1,5 @@
package io.scalecube.cluster.transport.api;
-import io.scalecube.net.Address;
import java.util.StringJoiner;
import java.util.function.Function;
import reactor.core.Exceptions;
@@ -22,7 +21,7 @@ public final class TransportConfig implements Cloneable {
private MessageCodec messageCodec = MessageCodec.INSTANCE;
private int maxFrameLength = 2 * 1024 * 1024; // 2 MB
private TransportFactory transportFactory;
- private Function<Address, Address> addressMapper = Function.identity();
+ private Function<String, String> addressMapper = Function.identity();
public TransportConfig() {}
@@ -143,13 +142,13 @@ public TransportConfig maxFrameLength(int maxFrameLength) {
* @param addressMapper address mapper
* @return new {@code TransportConfig} instance
*/
- public TransportConfig addressMapper(Function<Address, Address> addressMapper) {
+ public TransportConfig addressMapper(Function<String, String> addressMapper) {
TransportConfig t = clone();
t.addressMapper = addressMapper;
return t;
}
- public Function<Address, Address> addressMapper() {
+ public Function<String, String> addressMapper() {
return addressMapper;
}
diff --git a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/Sender.java b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/Sender.java
index 5c0cbdd6..3d5ef35c 100644
--- a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/Sender.java
+++ b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/Sender.java
@@ -1,13 +1,12 @@
package io.scalecube.transport.netty;
import io.scalecube.cluster.transport.api.Message;
-import io.scalecube.net.Address;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
public interface Sender {
- Mono<Connection> connect(Address address);
+ Mono<Connection> connect(String address);
Mono<Void> send(Message message);
}
diff --git a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/TransportImpl.java b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/TransportImpl.java
index 9838c68d..bcb93ad8 100644
--- a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/TransportImpl.java
+++ b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/TransportImpl.java
@@ -1,6 +1,6 @@
package io.scalecube.transport.netty;
-import static io.scalecube.reactor.RetryNonSerializedEmitFailureHandler.RETRY_NON_SERIALIZED;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.busyLooping;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
@@ -10,13 +10,13 @@
import io.netty.handler.codec.DecoderException;
import io.netty.handler.codec.EncoderException;
import io.netty.util.ReferenceCountUtil;
+import io.scalecube.cluster.transport.api.DistinctErrors;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.MessageCodec;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.errors.DistinctErrors;
-import io.scalecube.net.Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
import java.time.Duration;
import java.util.Map;
import java.util.Objects;
@@ -36,6 +36,7 @@
public final class TransportImpl implements Transport {
private static final Logger LOGGER = LoggerFactory.getLogger(Transport.class);
+
private static final DistinctErrors DISTINCT_ERRORS = new DistinctErrors(Duration.ofMinutes(1));
private final MessageCodec messageCodec;
@@ -48,15 +49,15 @@ public final class TransportImpl implements Transport {
private final Sinks.One<Void> onStop = Sinks.one();
// Server
- private Address address;
+ private String address;
private DisposableServer server;
- private final Map<Address, Mono<? extends Connection>> connections = new ConcurrentHashMap<>();
+ private final Map<String, Mono<? extends Connection>> connections = new ConcurrentHashMap<>();
private final LoopResources loopResources = LoopResources.create("sc-cluster-io", 1, true);
// Transport factory
private final Receiver receiver;
private final Sender sender;
- private final Function<Address, Address> addressMapper;
+ private final Function<String, String> addressMapper;
/**
* Constructor with config as parameter.
@@ -72,21 +73,30 @@ public TransportImpl(
MessageCodec messageCodec,
Receiver receiver,
Sender sender,
- Function<Address, Address> addressMapper) {
+ Function<String, String> addressMapper) {
this.messageCodec = messageCodec;
this.receiver = receiver;
this.sender = sender;
this.addressMapper = addressMapper;
}
- private static Address prepareAddress(DisposableServer server) {
+ private static String prepareAddress(DisposableServer server) {
final InetSocketAddress serverAddress = (InetSocketAddress) server.address();
- InetAddress inetAddress = serverAddress.getAddress();
- int port = serverAddress.getPort();
+ final InetAddress inetAddress = serverAddress.getAddress();
+ final int port = serverAddress.getPort();
+
if (inetAddress.isAnyLocalAddress()) {
- return Address.create(Address.getLocalIpAddress().getHostAddress(), port);
+ return getLocalHostAddress() + ":" + port;
} else {
- return Address.create(inetAddress.getHostAddress(), port);
+ return inetAddress.getHostAddress() + ":" + port;
+ }
+ }
+
+ private static String getLocalHostAddress() {
+ try {
+ return InetAddress.getLocalHost().getHostAddress();
+ } catch (UnknownHostException e) {
+ throw new RuntimeException(e);
}
}
@@ -96,7 +106,7 @@ private void init(DisposableServer server) {
// Setup cleanup
stop.asMono()
.then(doStop())
- .doFinally(s -> onStop.emitEmpty(RETRY_NON_SERIALIZED))
+ .doFinally(s -> onStop.emitEmpty(busyLooping(Duration.ofSeconds(3))))
.subscribe(
null, ex -> LOGGER.warn("[{}][doStop] Exception occurred: {}", address, ex.toString()));
}
@@ -122,7 +132,7 @@ public Mono<Transport> start() {
}
@Override
- public Address address() {
+ public String address() {
return address;
}
@@ -132,10 +142,10 @@ public boolean isStopped() {
}
@Override
- public final Mono<Void> stop() {
+ public Mono<Void> stop() {
return Mono.defer(
() -> {
- stop.emitEmpty(RETRY_NON_SERIALIZED);
+ stop.emitEmpty(busyLooping(Duration.ofSeconds(3)));
return onStop.asMono();
});
}
@@ -145,7 +155,7 @@ private Mono<Void> doStop() {
() -> {
LOGGER.info("[{}][doStop] Stopping", address);
// Complete incoming messages observable
- sink.emitComplete(RETRY_NON_SERIALIZED);
+ sink.emitComplete(busyLooping(Duration.ofSeconds(3)));
return Flux.concatDelayError(closeServer(), shutdownLoopResources())
.then()
.doFinally(s -> connections.clear())
@@ -154,12 +164,12 @@ private Mono<Void> doStop() {
}
@Override
- public final Flux<Message> listen() {
+ public Flux<Message> listen() {
return sink.asFlux().onBackpressureBuffer();
}
@Override
- public Mono<Void> send(Address address, Message message) {
+ public Mono<Void> send(String address, Message message) {
return Mono.deferContextual(context -> connections.computeIfAbsent(address, this::connect))
.flatMap(
connection ->
@@ -172,7 +182,7 @@ public Mono<Void> send(Address address, Message message) {
}
@Override
- public Mono<Message> requestResponse(Address address, final Message request) {
+ public Mono<Message> requestResponse(String address, final Message request) {
return Mono.create(
sink -> {
Objects.requireNonNull(request, "request must be not null");
@@ -224,8 +234,8 @@ private ByteBuf encodeMessage(Message message) {
return byteBuf;
}
- private Mono<? extends Connection> connect(Address remoteAddress) {
- final Address mappedAddr = addressMapper.apply(remoteAddress);
+ private Mono<? extends Connection> connect(String remoteAddress) {
+ final String mappedAddr = addressMapper.apply(remoteAddress);
return sender
.connect(mappedAddr)
.doOnSuccess(
@@ -277,13 +287,13 @@ private Mono<Void> shutdownLoopResources() {
public static final class ReceiverContext {
- private final Address address;
+ private final String address;
private final Sinks.Many<Message> sink;
private final LoopResources loopResources;
private final Function<ByteBuf, Message> messageDecoder;
private ReceiverContext(
- Address address,
+ String address,
Sinks.Many<Message> sink,
LoopResources loopResources,
Function<ByteBuf, Message> messageDecoder) {
@@ -313,7 +323,7 @@ public void onMessage(ByteBuf byteBuf) {
return;
}
final Message message = messageDecoder.apply(byteBuf);
- sink.emitNext(message, RETRY_NON_SERIALIZED);
+ sink.emitNext(message, busyLooping(Duration.ofSeconds(3)));
} catch (Exception e) {
LOGGER.error("[{}][onMessage] Exception occurred:", address, e);
}
diff --git a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/tcp/TcpSender.java b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/tcp/TcpSender.java
index 708d1a95..80450f31 100644
--- a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/tcp/TcpSender.java
+++ b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/tcp/TcpSender.java
@@ -1,9 +1,11 @@
package io.scalecube.transport.netty.tcp;
+import static io.scalecube.cluster.transport.api.Transport.parseHost;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
+
import io.netty.channel.ChannelOption;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.TransportConfig;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.Sender;
import io.scalecube.transport.netty.TransportImpl.SenderContext;
import reactor.core.publisher.Mono;
@@ -19,7 +21,7 @@ public final class TcpSender implements Sender {
}
@Override
- public Mono<Connection> connect(Address address) {
+ public Mono<Connection> connect(String address) {
return Mono.deferContextual(context -> Mono.just(context.get(SenderContext.class)))
.map(context -> newTcpClient(context, address))
.flatMap(TcpClient::connect);
@@ -38,12 +40,12 @@ public Mono<Void> send(Message message) {
});
}
- private TcpClient newTcpClient(SenderContext context, Address address) {
+ private TcpClient newTcpClient(SenderContext context, String address) {
TcpClient tcpClient =
TcpClient.newConnection()
.runOn(context.loopResources())
- .host(address.host())
- .port(address.port())
+ .host(parseHost(address))
+ .port(parsePort(address))
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.SO_REUSEADDR, true)
diff --git a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/websocket/WebsocketSender.java b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/websocket/WebsocketSender.java
index 704a7f3d..34762917 100644
--- a/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/websocket/WebsocketSender.java
+++ b/transport-parent/transport-netty/src/main/java/io/scalecube/transport/netty/websocket/WebsocketSender.java
@@ -1,10 +1,12 @@
package io.scalecube.transport.netty.websocket;
+import static io.scalecube.cluster.transport.api.Transport.parseHost;
+import static io.scalecube.cluster.transport.api.Transport.parsePort;
+
import io.netty.channel.ChannelOption;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.TransportConfig;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.Sender;
import io.scalecube.transport.netty.TransportImpl.SenderContext;
import reactor.core.publisher.Mono;
@@ -21,7 +23,7 @@ public WebsocketSender(TransportConfig config) {
}
@Override
- public Mono<Connection> connect(Address address) {
+ public Mono<Connection> connect(String address) {
return Mono.deferContextual(context -> Mono.just(context.get(SenderContext.class)))
.map(context -> newWebsocketSender(context, address))
.flatMap(sender -> sender.uri("/").connect());
@@ -44,12 +46,12 @@ public Mono<Void> send(Message message) {
});
}
- private HttpClient.WebsocketSender newWebsocketSender(SenderContext context, Address address) {
+ private HttpClient.WebsocketSender newWebsocketSender(SenderContext context, String address) {
HttpClient httpClient =
HttpClient.newConnection()
.runOn(context.loopResources())
- .host(address.host())
- .port(address.port())
+ .host(parseHost(address))
+ .port(parsePort(address))
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.SO_REUSEADDR, true)
diff --git a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/BaseTest.java b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/BaseTest.java
index 3d07e84d..a8638bf8 100644
--- a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/BaseTest.java
+++ b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/BaseTest.java
@@ -4,7 +4,6 @@
import io.scalecube.cluster.transport.api.Transport;
import io.scalecube.cluster.transport.api.TransportConfig;
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.tcp.TcpTransportFactory;
import io.scalecube.transport.netty.websocket.WebsocketTransportFactory;
import java.time.Duration;
@@ -37,7 +36,7 @@ public final void baseTearDown(TestInfo testInfo) {
* @param to destination
* @param msg request
*/
- protected Mono<Void> send(Transport transport, Address to, Message msg) {
+ protected Mono<Void> send(Transport transport, String to, Message msg) {
return transport
.send(to, msg)
.doOnError(
diff --git a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportSendOrderTest.java b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportSendOrderTest.java
index 753622d7..417c6f0b 100644
--- a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportSendOrderTest.java
+++ b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportSendOrderTest.java
@@ -4,7 +4,6 @@
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.BaseTest;
import java.time.Duration;
import java.util.ArrayList;
@@ -220,7 +219,7 @@ private void assertSendOrder(int total, List<Message> received) {
}
}
- private Callable<Void> sender(int id, Transport client, Address address, int total) {
+ private Callable<Void> sender(int id, Transport client, String address, int total) {
return () -> {
for (int j = 0; j < total; j++) {
String correlationId = id + "/" + j;
diff --git a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportTest.java b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportTest.java
index f8bb8daa..9c5ed4b9 100644
--- a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportTest.java
+++ b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/tcp/TcpTransportTest.java
@@ -8,7 +8,6 @@
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.BaseTest;
import java.io.IOException;
import java.net.UnknownHostException;
@@ -45,7 +44,7 @@ public void testUnresolvedHostConnection() {
client = createTcpTransport();
// create transport with wrong host
try {
- Address address = Address.from("wronghost:49255");
+ String address = "wronghost:49255";
Message message = Message.withData("q").build();
client.send(address, message).block(Duration.ofSeconds(20));
fail("fail");
@@ -57,7 +56,7 @@ public void testUnresolvedHostConnection() {
@Test
public void testInteractWithNoConnection(TestInfo testInfo) {
- Address serverAddress = Address.from("localhost:49255");
+ String serverAddress = "localhost:49255";
for (int i = 0; i < 10; i++) {
LOGGER.debug("####### {} : iteration = {}", testInfo.getDisplayName(), i);
@@ -94,7 +93,7 @@ public void testPingPongClientTfListenAndServerTfListen() throws Exception {
.listen()
.subscribe(
message -> {
- Address address = message.sender();
+ String address = message.sender();
assertEquals(client.address(), address, "Expected clientAddress");
send(server, address, Message.fromQualifier("hi client")).subscribe();
});
diff --git a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportSendOrderTest.java b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportSendOrderTest.java
index ab17c5b1..8ed11235 100644
--- a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportSendOrderTest.java
+++ b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportSendOrderTest.java
@@ -4,7 +4,6 @@
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.transport.api.Transport;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.BaseTest;
import java.time.Duration;
import java.util.ArrayList;
@@ -220,7 +219,7 @@ private void assertSendOrder(int total, List<Message> received) {
}
}
- private Callable<Void> sender(int id, Transport client, Address address, int total) {
+ private Callable<Void> sender(int id, Transport client, String address, int total) {
return () -> {
for (int j = 0; j < total; j++) {
String correlationId = id + "/" + j;
diff --git a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportTest.java b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportTest.java
index 050474af..9d5867c4 100644
--- a/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportTest.java
+++ b/transport-parent/transport-netty/src/test/java/io/scalecube/transport/netty/websocket/WebsocketTransportTest.java
@@ -8,7 +8,6 @@
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.cluster.utils.NetworkEmulatorTransport;
-import io.scalecube.net.Address;
import io.scalecube.transport.netty.BaseTest;
import java.io.IOException;
import java.net.UnknownHostException;
@@ -45,7 +44,7 @@ public void testUnresolvedHostConnection() {
client = createWebsocketTransport();
// create transport with wrong host
try {
- Address address = Address.from("wronghost:49255");
+ String address = "wronghost:49255";
Message message = Message.withData("q").build();
client.send(address, message).block(Duration.ofSeconds(20));
fail("fail");
@@ -57,7 +56,7 @@ public void testUnresolvedHostConnection() {
@Test
public void testInteractWithNoConnection(TestInfo testInfo) {
- Address serverAddress = Address.from("localhost:49255");
+ String serverAddress = "localhost:49255";
for (int i = 0; i < 10; i++) {
LOGGER.debug("####### {} : iteration = {}", testInfo.getDisplayName(), i);
@@ -94,7 +93,7 @@ public void testPingPongClientTfListenAndServerTfListen() throws Exception {
.listen()
.subscribe(
message -> {
- Address address = message.sender();
+ String address = message.sender();
assertEquals(client.address(), address, "Expected clientAddress");
send(server, address, Message.fromQualifier("hi client")).subscribe();
});