From 912308c4d31665b42c44d587c76b1514efe1d3d0 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 6 Nov 2014 22:45:18 +0000 Subject: [PATCH] HCOLL-191 - added tests and fix for reading keys and values --- .../chronicle/map/ReplicatedChronicleMap.java | 3 +- ...tListenerTestWithTCPSocketReplication.java | 195 +++++++++++++++++- 2 files changed, 189 insertions(+), 9 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index d4d54fc67..9f4a2e0bc 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -811,7 +811,8 @@ class ReplicatedChronicleMap, } // this is just like notifyPut, but it won't trigger the modification iterators - notifyRemotePut(offset, false, posFromOffset(offset), keySize, inBytes, valueSize, + notifyRemotePut(offset, false, posFromOffset(offset), keySize, inBytes, valueLimit + -valuePos, valuePos, keyPosition); return; } diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java index 9dcf2ff26..d48b76535 100644 --- a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java +++ b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java @@ -20,12 +20,15 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.TcpReplicationConfig; import net.openhft.lang.io.Bytes; +import org.junit.After; import org.junit.Assert; import org.junit.Test; +import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; /** * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket @@ -36,14 +39,35 @@ import java.util.concurrent.atomic.AtomicBoolean; public class EventListenerTestWithTCPSocketReplication { + ChronicleMap map1; + ChronicleMap map2; + + @After + public void tearDown() throws InterruptedException { + + for (final Closeable closeable : new Closeable[]{map1, map2}) { + try { + closeable.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + // Thread.sleep(1600); + } + + @Test - public void test() throws IOException, InterruptedException { + public void testAdded() throws IOException, InterruptedException { + - ChronicleMap map1; - ChronicleMap map2; final AtomicBoolean putWasCalled = new AtomicBoolean(false); - int udpPort = 1234; + final AtomicReference keyRef = new AtomicReference(); + + + final AtomicReference valueRef = new AtomicReference(); + MapEventListener> eventListener = new MapEventListener>() { private static final long serialVersionUID = 1L; @@ -51,6 +75,8 @@ public class EventListenerTestWithTCPSocketReplication { public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, CharSequence value) { putWasCalled.getAndSet(true); + keyRef.set(key); + valueRef.set(value); } }; @@ -63,7 +89,7 @@ public class EventListenerTestWithTCPSocketReplication { ChronicleMapBuilder map1Builder = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) - .replicators((byte) 1, tcpConfig); + .replicators((byte) 2, tcpConfig); map1 = map1Builder.create(); } @@ -74,7 +100,7 @@ public class EventListenerTestWithTCPSocketReplication { ChronicleMapBuilder map2Builder = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) - .replicators((byte) 2, tcpConfig) + .replicators((byte) 1, tcpConfig) .eventListener(eventListener); map2 = map2Builder.create(); @@ -93,10 +119,163 @@ public class EventListenerTestWithTCPSocketReplication { Assert.assertEquals("Maps should be equal", map1, map2); Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); + Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); + Assert.assertEquals(Integer.valueOf(5), keyRef.get()); + Assert.assertEquals("EXAMPLE", valueRef.get()); + keyRef.set(null); + valueRef.set(null); - map1.close(); - map2.close(); } + + + public void testAddedOriginalValueToBeOverWritten() throws IOException, InterruptedException { + + + final AtomicBoolean putWasCalled = new AtomicBoolean(false); + + final AtomicReference keyRef = new AtomicReference(); + + + final AtomicReference valueRef = new AtomicReference(); + + MapEventListener> eventListener = new MapEventListener>() { + private static final long serialVersionUID = 1L; + + @Override + public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, + CharSequence value) { + putWasCalled.getAndSet(true); + keyRef.set(key); + valueRef.set(value); + } + }; + + // ---------- SERVER1 1 ---------- + { + + // we connect the maps via a TCP socket connection on port 8077 + + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8076, new InetSocketAddress("localhost", 8077)); + ChronicleMapBuilder map1Builder = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 2, tcpConfig); + + map1 = map1Builder.create(); + + } + // ---------- SERVER2 2 on the same server as ---------- + + { + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8077); + ChronicleMapBuilder map2Builder = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 1, tcpConfig) + .eventListener(eventListener); + + map2 = map2Builder.create(); + map2.put(5, "WILL_GET_OVER-WRITTEN"); + } + + // we will stores some data into one map here + map1.put(5, "EXAMPLE"); + int t = 0; + for (; t < 5000; t++) { + if (map1.equals(map2)) { + break; + } + Thread.sleep(1); + } + + Assert.assertEquals("Maps should be equal", map1, map2); + Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); + Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); + + Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); + Assert.assertEquals(Integer.valueOf(5), keyRef.get()); + Assert.assertEquals("EXAMPLE", valueRef.get()); + + keyRef.set(null); + valueRef.set(null); + + } + + + @Test + public void testAlreadyAddedWithSmallInitialValue() throws IOException, InterruptedException { + + + final AtomicBoolean putWasCalled = new AtomicBoolean(false); + + final AtomicReference keyRef = new AtomicReference(); + + + final AtomicReference valueRef = new AtomicReference(); + + MapEventListener> eventListener = new MapEventListener>() { + private static final long serialVersionUID = 1L; + + @Override + public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, + CharSequence value) { + putWasCalled.getAndSet(true); + keyRef.set(key); + valueRef.set(value); + } + }; + + // ---------- SERVER1 1 ---------- + { + + // we connect the maps via a TCP socket connection on port 8077 + + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8076, new InetSocketAddress("localhost", 8077)); + ChronicleMapBuilder map1Builder = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 2, tcpConfig); + + map1 = map1Builder.create(); + map1.put(5, "WILL_GET_OVERWRITEN"); + } + // ---------- SERVER2 2 on the same server as ---------- + + { + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8077); + ChronicleMapBuilder map2Builder = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 1, tcpConfig) + .eventListener(eventListener); + + map2 = map2Builder.create(); + map2.put(5, "small"); + } + + + // we will stores some data into one map here + map1.put(5, "EXAMPLE"); + int t = 0; + for (; t < 5000; t++) { + if (map1.equals(map2)) { + break; + } + Thread.sleep(1); + } + + Assert.assertEquals("Maps should be equal", map1, map2); + Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); + Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); + + Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); + Assert.assertEquals(Integer.valueOf(5), keyRef.get()); + Assert.assertEquals("EXAMPLE", valueRef.get()); + + keyRef.set(null); + valueRef.set(null); + } + } \ No newline at end of file