sites=new HashMap<>();
+
+ /**
+ * Adds a set of sites to the cache. Returns a set of sites for which notifications should be emitted. For each
+ * site S, the following happens:
+ *
+ * - S is not present: add a new entry with the given status for S and add S to the return value
+ * - S is present: if S != status: change the status and add S to the return value, else no-op
+ *
+ * @param sites
+ * @param status
+ * @return
+ */
+ public synchronized Set add(Set sites, Status status) {
+ Set retval=new HashSet<>();
+ for(String site: sites) {
+ Status s=this.sites.get(site);
+ if(s == null) {
+ this.sites.put(site, status);
+ retval.add(site);
+ }
+ else {
+ if(s != status) {
+ this.sites.put(site, status);
+ retval.add(site);
+ }
+ }
+ }
+ return retval;
+ }
+
+ public synchronized Status get(String site) {
+ return this.sites.get(site);
+ }
+
+ public synchronized SiteStatus clear() {
+ sites.clear();
+ return this;
+ }
+
+ public String toString() {
+ return sites.entrySet().stream()
+ .map(e -> String.format("%s: %s", e.getKey(), e.getValue())).collect(Collectors.joining("\n"));
+ }
+}
diff --git a/src/org/jgroups/protocols/relay/Topology.java b/src/org/jgroups/protocols/relay/Topology.java
index 1e5ab8957d..e1ccd997a0 100644
--- a/src/org/jgroups/protocols/relay/Topology.java
+++ b/src/org/jgroups/protocols/relay/Topology.java
@@ -27,7 +27,7 @@
public class Topology {
protected final RELAY2 relay;
protected final Map> cache=new ConcurrentHashMap<>(); // cache of sites and members
- protected BiConsumer rsp_handler;
+ protected BiConsumer rsp_handler;
public Topology(RELAY2 relay) {
@@ -38,9 +38,10 @@ public Topology(RELAY2 relay) {
/**
* Sets a response handler
- * @param c The response handler. Arguments are the site and MemberInfo of the member from which we received the rsp
+ * @param c The response handler. Arguments are the site and {@link Members}
+ * ({@link MemberInfo} of joined and left members)
*/
- public Topology setResponseHandler(BiConsumer c) {rsp_handler=c; return this;}
+ public Topology setResponseHandler(BiConsumer c) {rsp_handler=c; return this;}
@ManagedOperation(description="Fetches information (site, address, IP address) from all members")
@@ -118,20 +119,14 @@ protected void handleResponse(Members rsp) {
Set infos=cache.computeIfAbsent(site,s -> new ConcurrentSkipListSet<>());
if(rsp.joined != null) {
infos.addAll(rsp.joined);
- if(rsp_handler != null)
- for(MemberInfo mi: rsp.joined)
- rsp_handler.accept(site, mi);
}
-
if(rsp.left != null) {
// todo: implement
}
+ if(rsp_handler != null)
+ rsp_handler.accept(site, rsp);
}
- @FunctionalInterface
- public interface ResponseHandler {
- void handle(String site, MemberInfo rsp);
- }
/** Contains information about joined and left members for a given site */
public static class Members implements SizeStreamable {
diff --git a/tests/junit-functional/org/jgroups/tests/Relay2RpcDispatcherTest.java b/tests/junit-functional/org/jgroups/tests/Relay2RpcDispatcherTest.java
index 89873f3550..a9c2da4b35 100644
--- a/tests/junit-functional/org/jgroups/tests/Relay2RpcDispatcherTest.java
+++ b/tests/junit-functional/org/jgroups/tests/Relay2RpcDispatcherTest.java
@@ -191,7 +191,7 @@ public static long sleep(long timeout) {
protected RELAY2 createRELAY2(String site_name) {
- RELAY2 relay=new RELAY2().site(site_name).enableAddressTagging(false).asyncRelayCreation(true);
+ RELAY2 relay=new RELAY2().site(site_name).asyncRelayCreation(true);
RelayConfig.SiteConfig lon_cfg=new RelayConfig.SiteConfig(LON),
sfo_cfg=new RelayConfig.SiteConfig(SFO);
diff --git a/tests/junit-functional/org/jgroups/tests/Relay2Test.java b/tests/junit-functional/org/jgroups/tests/Relay2Test.java
index 76d29af750..bd06b21a8a 100644
--- a/tests/junit-functional/org/jgroups/tests/Relay2Test.java
+++ b/tests/junit-functional/org/jgroups/tests/Relay2Test.java
@@ -385,7 +385,11 @@ public void testSitesUp() throws Exception {
JChannel _g=createNode(SFO, "G", BRIDGE_CLUSTER, LON, NYC, SFO);
JChannel _h=createNode(SFO, "H", BRIDGE_CLUSTER, LON, NYC, SFO);
- JChannel _i=createNode(SFO, "I", BRIDGE_CLUSTER, LON, NYC, SFO);) {
+ JChannel _i=createNode(SFO, "I", BRIDGE_CLUSTER, LON, NYC, SFO)) {
+
+ Util.waitUntilAllChannelsHaveSameView(5000, 100, _a,_b,_c);
+ Util.waitUntilAllChannelsHaveSameView(5000, 100, _d,_e,_f);
+ Util.waitUntilAllChannelsHaveSameView(5000, 100, _g,_h,_i);
waitUntilRoute(NYC, true, 5000, 500, _a);
waitUntilRoute(SFO, true, 5000, 500, _a);
@@ -399,19 +403,17 @@ public void testSitesUp() throws Exception {
assert Stream.of(_a,_d,_g).map(ch -> (RELAY2)ch.getProtocolStack().findProtocol(RELAY2.class))
.allMatch(RELAY2::isSiteMaster);
- Stream.of(_b,_c,_d,_e,_f,_g,_h,_i)
+ Stream.of(_d,_e,_f,_g,_h,_i)
.map(ch -> (RELAY2)ch.getProtocolStack().findProtocol(RELAY2.class))
- .forEach(r -> r.setRouteStatusListener(new MyRouteStatusListener(r.getAddress()).verbose(true)));
+ .forEach(r -> r.setRouteStatusListener(new MyRouteStatusListener(r.getAddress()).verbose(false)));
// now stop A; B will become new site master and we should get a site-down(NYC), then site-up(NYC)
Util.close(_a);
-
- Util.waitUntil(5000, 500, () -> Stream.of(_b,_c,_d, _e, _f, _g, _h, _i)
+ Util.waitUntil(5000, 500, () -> Stream.of(_d,_e,_f,_g,_h,_i)
.map(ch -> (RELAY2)ch.getProtocolStack().findProtocol(RELAY2.class))
.peek(r -> System.out.printf("%s: %s\n", r.getAddress(), r.getRouteStatusListener()))
.map(r -> (MyRouteStatusListener)r.getRouteStatusListener())
- .allMatch(l -> l.down().size() == 1 && l.down().contains(NYC)
- && l.up().size() == 1 && l.up().contains(NYC)));
+ .allMatch(l -> l.down().contains(LON) && l.up().contains(LON)));
}
}
diff --git a/tests/junit-functional/org/jgroups/tests/SiteStatusTest.java b/tests/junit-functional/org/jgroups/tests/SiteStatusTest.java
new file mode 100644
index 0000000000..d8fa6d4143
--- /dev/null
+++ b/tests/junit-functional/org/jgroups/tests/SiteStatusTest.java
@@ -0,0 +1,46 @@
+package org.jgroups.tests;
+
+import org.jgroups.Global;
+import org.jgroups.protocols.relay.SiteStatus;
+import org.jgroups.protocols.relay.SiteStatus.Status;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.Set;
+
+/**
+ * Tests {@link org.jgroups.protocols.relay.SiteStatus}
+ * @author Bela Ban
+ * @since 5.2.17
+ */
+@Test(groups= Global.FUNCTIONAL)
+public class SiteStatusTest {
+ public void testSiteStatus() {
+ SiteStatus s=new SiteStatus();
+ Set res=s.add(Set.of("net1"), Status.up);
+ assert res.size() == 1 && res.contains("net1");
+ Status status=s.get("net1");
+ assert status == Status.up;
+ res=s.add(Set.of("net1"), Status.up);
+ assert res.isEmpty();
+ res=s.add(Set.of("net1"), Status.down);
+ assert res.size() == 1 && res.contains("net1");
+ status=s.get("net1");
+ assert status == Status.down;
+
+ status=s.get("hf");
+ assert status == null;
+ }
+
+ public void testSiteStatus2() {
+ SiteStatus s=new SiteStatus();
+ Set retval=s.add(Set.of("hf","net1","net2"),Status.down);
+ assert retval.size() == 3;
+ for(String st: Arrays.asList("hf", "net1", "net2"))
+ assert s.get(st) == Status.down;
+ retval=s.add(Set.of("net1"), Status.up);
+ assert retval.size() == 1;
+ retval=s.add(Set.of("net1", "net2"), Status.up);
+ assert retval.size() == 1 && retval.contains("net2");
+ }
+}