rdev/gossip-revisited
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

GOSSIP-73 gcounter to example app

Browse Source
This commit is contained in:
Edward Capriolo
2017-03-10 18:57:51 -05:00
parent 2622248e7b
commit 4eafd58ecc
4 changed files with 29 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/apache/gossip/crdt/GrowOnlyCounter.java
View File

@ -111,7 +111,7 @@ public class GrowOnlyCounter implements CrdtCounter<Long, GrowOnlyCounter> {
counter = 0L; counter = 0L;
} }
public GrowOnlyCounter.Builder increment(Integer count) { public GrowOnlyCounter.Builder increment(Long count) {
counter += count; counter += count;
return this; return this;
} }

24
src/main/java/org/apache/gossip/examples/StandAloneNodeCrdtOrSet.java
View File

@ -24,6 +24,7 @@ import java.net.URI;
import java.util.Arrays; import java.util.Arrays;
import org.apache.gossip.GossipSettings; import org.apache.gossip.GossipSettings;
import org.apache.gossip.RemoteMember; import org.apache.gossip.RemoteMember;
import org.apache.gossip.crdt.GrowOnlyCounter;
import org.apache.gossip.crdt.OrSet; import org.apache.gossip.crdt.OrSet;
import org.apache.gossip.manager.GossipManager; import org.apache.gossip.manager.GossipManager;
import org.apache.gossip.manager.GossipManagerBuilder; import org.apache.gossip.manager.GossipManagerBuilder;
@ -51,6 +52,9 @@ public class StandAloneNodeCrdtOrSet {
System.out.println("---------- " + (gossipService.findCrdt("abc") == null ? "": System.out.println("---------- " + (gossipService.findCrdt("abc") == null ? "":
gossipService.findCrdt("abc").value())); gossipService.findCrdt("abc").value()));
System.out.println("********** " + gossipService.findCrdt("abc")); System.out.println("********** " + gossipService.findCrdt("abc"));
System.out.println("^^^^^^^^^^ " + (gossipService.findCrdt("def") == null ? "":
gossipService.findCrdt("def").value()));
System.out.println("$$$$$$$$$$ " + gossipService.findCrdt("def"));
try { try {
Thread.sleep(2000); Thread.sleep(2000);
} catch (Exception e) {} } catch (Exception e) {}
@ -65,13 +69,31 @@ public class StandAloneNodeCrdtOrSet {
String val = line.substring(2); String val = line.substring(2);
if (op == 'a'){ if (op == 'a'){
addData(val, gossipService); addData(val, gossipService);
} else { } else if (op == 'r') {
removeData(val, gossipService); removeData(val, gossipService);
} else if (op == 'g'){
gcount(val, gossipService);
} }
} }
} }
} }
private static void gcount(String val, GossipManager gossipManager){
GrowOnlyCounter c = (GrowOnlyCounter) gossipManager.findCrdt("def");
Long l = Long.valueOf(val);
if (c == null){
c = new GrowOnlyCounter(new GrowOnlyCounter.Builder(gossipManager).increment((l)));
} else {
c = new GrowOnlyCounter(c, new GrowOnlyCounter.Builder(gossipManager).increment((l)));
}
SharedDataMessage m = new SharedDataMessage();
m.setExpireAt(Long.MAX_VALUE);
m.setKey("def");
m.setPayload(c);
m.setTimestamp(System.currentTimeMillis());
gossipManager.merge(m);
}
private static void removeData(String val, GossipManager gossipService){ private static void removeData(String val, GossipManager gossipService){
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
OrSet<String> s = (OrSet<String>) gossipService.findCrdt("abc"); OrSet<String> s = (OrSet<String>) gossipService.findCrdt("abc");

8
src/test/java/org/apache/gossip/DataTest.java
View File

@ -115,7 +115,7 @@ public class DataTest extends AbstractIntegrationBase {
{ {
SharedDataMessage d = new SharedDataMessage(); SharedDataMessage d = new SharedDataMessage();
d.setKey(gCounterKey); d.setKey(gCounterKey);
d.setPayload(new GrowOnlyCounter(new GrowOnlyCounter.Builder(clients.get(0)).increment(1))); d.setPayload(new GrowOnlyCounter(new GrowOnlyCounter.Builder(clients.get(0)).increment(1L)));
d.setExpireAt(Long.MAX_VALUE); d.setExpireAt(Long.MAX_VALUE);
d.setTimestamp(System.currentTimeMillis()); d.setTimestamp(System.currentTimeMillis());
clients.get(0).merge(d); clients.get(0).merge(d);
@ -123,7 +123,7 @@ public class DataTest extends AbstractIntegrationBase {
{ {
SharedDataMessage d = new SharedDataMessage(); SharedDataMessage d = new SharedDataMessage();
d.setKey(gCounterKey); d.setKey(gCounterKey);
d.setPayload(new GrowOnlyCounter(new GrowOnlyCounter.Builder(clients.get(1)).increment(2))); d.setPayload(new GrowOnlyCounter(new GrowOnlyCounter.Builder(clients.get(1)).increment(2L)));
d.setExpireAt(Long.MAX_VALUE); d.setExpireAt(Long.MAX_VALUE);
d.setTimestamp(System.currentTimeMillis()); d.setTimestamp(System.currentTimeMillis());
clients.get(1).merge(d); clients.get(1).merge(d);
@ -133,7 +133,7 @@ public class DataTest extends AbstractIntegrationBase {
private void givenIncreaseOther(final List<GossipManager> clients) { private void givenIncreaseOther(final List<GossipManager> clients) {
GrowOnlyCounter gc = (GrowOnlyCounter) clients.get(1).findCrdt(gCounterKey); GrowOnlyCounter gc = (GrowOnlyCounter) clients.get(1).findCrdt(gCounterKey);
GrowOnlyCounter gc2 = new GrowOnlyCounter(gc, GrowOnlyCounter gc2 = new GrowOnlyCounter(gc,
new GrowOnlyCounter.Builder(clients.get(1)).increment(4)); new GrowOnlyCounter.Builder(clients.get(1)).increment(4L));
SharedDataMessage d = new SharedDataMessage(); SharedDataMessage d = new SharedDataMessage();
d.setKey(gCounterKey); d.setKey(gCounterKey);
@ -195,7 +195,7 @@ public class DataTest extends AbstractIntegrationBase {
} }
private void assertThatCountIsUpdated(final List<GossipManager> clients, int finalCount) { private void assertThatCountIsUpdated(final List<GossipManager> clients, long finalCount) {
TUnit.assertThat(() -> { TUnit.assertThat(() -> {
return clients.get(0).findCrdt(gCounterKey); return clients.get(0).findCrdt(gCounterKey);
}).afterWaitingAtMost(10, TimeUnit.SECONDS).isEqualTo(new GrowOnlyCounter( }).afterWaitingAtMost(10, TimeUnit.SECONDS).isEqualTo(new GrowOnlyCounter(