HCOLL-191 - added tests and fix for reading keys and values

2021-09-13 01:26:12 +00:00 · 2014-11-06 22:45:18 +00:00
parent 6327a6ab9d
commit 912308c4d3
2 changed files with 189 additions and 9 deletions
@@ -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<Integer, CharSequence> map1;
+    ChronicleMap<Integer, CharSequence> 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<Integer, CharSequence> map1;
-        ChronicleMap<Integer, CharSequence> map2;
        final AtomicBoolean putWasCalled = new AtomicBoolean(false);

-        int udpPort = 1234;
+        final AtomicReference<Integer> keyRef = new AtomicReference<Integer>();
+
+
+        final AtomicReference<CharSequence> valueRef = new AtomicReference<CharSequence>();
+
        MapEventListener<Integer, CharSequence, ChronicleMap<Integer, CharSequence>> eventListener = new MapEventListener<Integer, CharSequence, ChronicleMap<Integer, CharSequence>>() {
            private static final long serialVersionUID = 1L;

@@ -51,6 +75,8 @@ public class EventListenerTestWithTCPSocketReplication {
            public void onPut(ChronicleMap<Integer, CharSequence> 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<Integer, CharSequence> 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<Integer, CharSequence> 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<Integer> keyRef = new AtomicReference<Integer>();
+
+
+        final AtomicReference<CharSequence> valueRef = new AtomicReference<CharSequence>();
+
+        MapEventListener<Integer, CharSequence, ChronicleMap<Integer, CharSequence>> eventListener = new MapEventListener<Integer, CharSequence, ChronicleMap<Integer, CharSequence>>() {
+            private static final long serialVersionUID = 1L;
+
+            @Override
+            public void onPut(ChronicleMap<Integer, CharSequence> 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<Integer, CharSequence> 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<Integer, CharSequence> 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<Integer> keyRef = new AtomicReference<Integer>();
+
+
+        final AtomicReference<CharSequence> valueRef = new AtomicReference<CharSequence>();
+
+        MapEventListener<Integer, CharSequence, ChronicleMap<Integer, CharSequence>> eventListener = new MapEventListener<Integer, CharSequence, ChronicleMap<Integer, CharSequence>>() {
+            private static final long serialVersionUID = 1L;
+
+            @Override
+            public void onPut(ChronicleMap<Integer, CharSequence> 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<Integer, CharSequence> 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<Integer, CharSequence> 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);
+    }
+
 }