diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListener.java b/src/main/java/net/openhft/chronicle/map/MapEventListener.java index a4dbe53f9..9c0cc1fae 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListener.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListener.java @@ -20,30 +20,28 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.lang.io.Bytes; +import org.jetbrains.annotations.Nullable; import java.io.Serializable; /** - * Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases: - * + * Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases: * - *

This is an adapter class - all methods have default implementations as no-ops. Extend this - * class and override only methods corresponding the events you are interested in. + *

This is an adapter class - all methods have default implementations as no-ops. Extend this class and + * override only methods corresponding the events you are interested in. * - *

To configure {@code MapEventListener} for {@code ChronicleMap}, use - * {@link AbstractChronicleMapBuilder#eventListener(MapEventListener)} method. + *

To configure {@code MapEventListener} for {@code ChronicleMap}, use {@link + * AbstractChronicleMapBuilder#eventListener(MapEventListener)} method. * - *

{@link MapEventListeners} uninstantiable class contains several logging implementations and - * the default {@linkplain MapEventListeners#nop() no-op implementation}. + *

{@link MapEventListeners} uninstantiable class contains several logging implementations and the default + * {@linkplain MapEventListeners#nop() no-op implementation}. * - *

All these calls are synchronous while a {@code ChronicleMap} lock is held so make them - * as quick as possible. + *

All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as quick as + * possible. * - *

The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data - * area. See {@link ChronicleHashBuilder#metaDataBytes(int)} for more information. + *

The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data area. See + * {@link ChronicleHashBuilder#metaDataBytes(int)} for more information. * * @param key type of the maps, trackable by this event listener * @param value type of the maps, trackable by this event listener @@ -56,14 +54,12 @@ public abstract class MapEventListener> private static final long serialVersionUID = 0L; /** - * This method is called if the key is found in the map during {@link ChronicleMap#get get}, - * {@link ChronicleMap#getUsing getUsing} or {@link ChronicleMap#acquireUsing acquireUsing} - * method call. + * This method is called if the key is found in the map during {@link ChronicleMap#get get}, {@link + * ChronicleMap#getUsing getUsing} or {@link ChronicleMap#acquireUsing acquireUsing} method call. * * @param map the accessed map * @param entry bytes of the entry - * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} - * for this map + * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} for this map * @param key the key looked up * @param value the value found for the key */ @@ -71,32 +67,32 @@ public abstract class MapEventListener> // do nothing } - void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V value, + void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V replacedValue, V value, long pos, SharedSegment segment) { - onPut(map, entry, metaDataBytes, added, key, value); + onPut(map, entry, metaDataBytes, added, key, value, replacedValue); } /** - * This method is called whenever a new value is put for the key in the map during calls of such - * methods as {@link ChronicleMap#put put}, {@link ChronicleMap#putIfAbsent putIfAbsent}, - * {@link ChronicleMap#replace(Object, Object, Object) replace}, etc. When a new value is - * {@linkplain AbstractChronicleMapBuilder#defaultValue(Object) default} for the map or obtained - * during {@link ChronicleMap#acquireUsing acquireUsing} call is put for the key, this method - * is called as well. + * This method is called whenever a new value is put for the key in the map during calls of such methods + * as {@link ChronicleMap#put put}, {@link ChronicleMap#putIfAbsent putIfAbsent}, {@link + * ChronicleMap#replace(Object, Object, Object) replace}, etc. When a new value is {@linkplain + * AbstractChronicleMapBuilder#defaultValue(Object) default} for the map or obtained during {@link + * ChronicleMap#acquireUsing acquireUsing} call is put for the key, this method is called as well. * *

This method is called when put is already happened. * * @param map the accessed map * @param entry bytes of the entry (with the value already written to) - * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} - * for this map - * @param added {@code true} is the key was absent in the map before current - * {@code ChronicleMap} method call, led to putting a value and inherently - * calling this listener method + * @param metaDataBytes length of {@link net.openhft.chronicle.hash.ChronicleHashBuilder#metaDataBytes(int) + * meta data} for this map + * @param added {@code true} is the key was absent in the map before current {@code ChronicleMap} + * method call, led to putting a value and inherently calling this listener method * @param key the key the given value is put for * @param value the value which is now associated with the given key + * @param replacedValue the value which was replaced by {@code value}, is NULL if {@code added} is true */ - public void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V value) { + public void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V value, + @Nullable V replacedValue) { // do nothing } @@ -106,13 +102,12 @@ public abstract class MapEventListener> } /** - * This is called when an entry is removed. Misses, i. e. when {@code map.remove(key)} - * is called, but key is already absent in the map, are not notified. + * This is called when an entry is removed. Misses, i. e. when {@code map.remove(key)} is called, but key + * is already absent in the map, are not notified. * * @param map accessed * @param entry bytes of the entry just removed, but any data is not yet erased - * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} - * for this map + * @param metaDataBytes length of {@link ChronicleHashBuilder#metaDataBytes(int) meta data} for this map * @param key the key removed from the map * @param value the value which was associated with the given key */ diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java index 4ddb4d251..754c54f9c 100644 --- a/src/main/java/net/openhft/chronicle/map/MapEventListeners.java +++ b/src/main/java/net/openhft/chronicle/map/MapEventListeners.java @@ -68,7 +68,7 @@ public final class MapEventListeners { @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, - Object key, Object value, long pos, SharedSegment segment) { + Object key, Object replacedValue, Object value, long pos, SharedSegment segment) { logOperation(map, entry, metaDataBytes, added ? " +put " : " put "); } @@ -90,7 +90,7 @@ public final class MapEventListeners { @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, - Object key, Object value, long pos, SharedSegment segment) { + Object key, Object replacedValue, Object value, long pos, SharedSegment segment) { LOGGER.info("{} put {} => {}", map.file(), key, value); } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 46c7586cc..7dd3f8404 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -199,13 +199,20 @@ final class ReplicatedChronicleMap, * as this will trigger the modification iterators to send out another replication event. So instead with * call this method iterators * - * @return + * @return the event listener added by the user */ - MapEventListener> eventListenerWithoutModIterators() { + MapEventListener> attachedEventListener() { return eventListener; } + /** + * @return true if the user has added an event listerner + */ + boolean hasAttachedEventListener() { + return eventListener != null; + } + @Override public V put(K key, V value) { return put0(key, value, true, localIdentifier, timeProvider.currentTimeMillis()); @@ -635,7 +642,7 @@ final class ReplicatedChronicleMap, throw new IllegalStateException("ENTRY WRITE_BUFFER_SIZE TOO LARGE : Replicated " + "ChronicleMap's" + " are restricted to an " + - "entry size of " + Integer.MAX_VALUE + ", " + + "entry size of " + Integer.MAX_VALUE + ", " + "your entry size=" + result); return result; @@ -728,7 +735,8 @@ final class ReplicatedChronicleMap, void notifyRemotePut(long offset, boolean added, final long pos, long keySize, final Bytes inBytes, - final long valueSize, final long valuePosition, final long keyPosition) { + final long valueSize, final long valuePosition, + final long keyPosition, final V replacedValue) { if (eventListener() != MapEventListeners.NOP) { final MultiStoreBytes b = acquireTmpBytes(); b.storePositionAndSize(bytes, offset, entrySize); @@ -747,8 +755,8 @@ final class ReplicatedChronicleMap, inBytes.position(valuePosition); final V value = valueReaderProvider.get(copiesV, originalValueReader).read(inBytes, valueSize); - eventListener().onPut(ReplicatedChronicleMap.this, b, metaDataBytes, - added, key, value, pos, this); + attachedEventListener().onPut(ReplicatedChronicleMap.this, b, metaDataBytes, + added, key, replacedValue, value, pos, this); } } @@ -759,10 +767,19 @@ final class ReplicatedChronicleMap, private void remotePut(@NotNull final Bytes inBytes, long hash2, final byte identifier, final long timestamp, long valuePos, long valueLimit) { + + writeLock(); try { // inBytes position and limit correspond to the key final long keySize = inBytes.remaining(); + V replacedValue = null; + + ThreadLocalCopies keyCopies = keyInteropProvider.getCopies(null); + BytesReader keyReader = keyReaderProvider.get(keyCopies, originalKeyReader); + + ThreadLocalCopies valueCopies = valueReaderProvider.getCopies(keyCopies); + BytesReader valueReader = valueReaderProvider.get(valueCopies, originalValueReader); hashLookupLiveAndDeleted.startSearch(hash2); for (long pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0L; ) { @@ -775,6 +792,7 @@ final class ReplicatedChronicleMap, // key is found + entry.skip(keySize); final long timeStampPos = entry.positionAddr(); @@ -793,13 +811,17 @@ final class ReplicatedChronicleMap, entry.writeBoolean(false); long valueSizePos = entry.position(); - long valueSize = readValueSize(entry); - long entryEndAddr = entry.positionAddr() + valueSize; + long replacedValueSize = readValueSize(entry); + long entryEndAddr = entry.positionAddr() + replacedValueSize; // write the value inBytes.limit(valueLimit); inBytes.position(valuePos); + // if we have event listeners attached we have to read back the old value + if (hasAttachedEventListener()) + replacedValue = valueReader.read(entry, replacedValueSize, null); + putValue(pos, offset, entry, valueSizePos, entryEndAddr, null, null, inBytes, hashLookupLiveAndDeleted); @@ -809,10 +831,21 @@ final class ReplicatedChronicleMap, incrementSize(); } - // this is just like notifyPut, but it won't trigger the modification iterators - notifyRemotePut(offset, false, posFromOffset(offset), keySize, inBytes, valueLimit - -valuePos, - valuePos, keyPosition); + if (hasAttachedEventListener()) { + + // this is just like notifyPut, but it won't trigger the modification iterators + notifyRemotePut(offset, + false, + posFromOffset(offset), + keySize, + inBytes, + valueLimit - valuePos, + valuePos, + keyPosition, + replacedValue + ); + } + return; } @@ -848,7 +881,8 @@ final class ReplicatedChronicleMap, incrementSize(); // this is just like notifyPut, but it won't trigger the modification iterators - notifyRemotePut(offset, true, posFromOffset(offset), keySize, inBytes, valueSize, valuePos, kPosition); + notifyRemotePut(offset, true, posFromOffset(offset), keySize, inBytes, valueSize, valuePos, + kPosition, null); } finally { writeUnlock(); @@ -913,13 +947,14 @@ final class ReplicatedChronicleMap, long offset = putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value, identifier, timestamp, hashLookup); incrementSize(); - notifyPut(offset, true, key, value, posFromOffset(offset)); + notifyPut(offset, true, key, value, null, posFromOffset(offset)); return null; } finally { writeUnlock(); } } + /** * Used only with replication, its sometimes possible to receive an old ( or stale update ) from a * remote map. This method is used to determine if we should ignore such updates.

We can reject @@ -1309,12 +1344,12 @@ final class ReplicatedChronicleMap, @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, - boolean added, K key, V value, long pos, SharedSegment segment) { + boolean added, K key, V replacedValue, V value, long pos, SharedSegment segment) { assert ReplicatedChronicleMap.this == map : "ModificationIterator.onPut() is called from outside of the parent map"; try { - nextListener.onPut(map, entry, metaDataBytes, added, key, value, pos, segment); + nextListener.onPut(map, entry, metaDataBytes, added, key, replacedValue, value, pos, segment); } catch (Exception e) { LOG.error("", e); } @@ -1323,7 +1358,7 @@ final class ReplicatedChronicleMap, try { final ModificationIterator modificationIterator = modificationIterators.get((int) next); modificationIterator.onPut(map, entry, metaDataBytes, - added, key, value, pos, segment); + added, key, replacedValue, value, pos, segment); } catch (Exception e) { LOG.error("", e); } @@ -1445,7 +1480,7 @@ final class ReplicatedChronicleMap, @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, - boolean added, K key, V value, long pos, SharedSegment segment) { + boolean added, K key, V replacedValue, V value, long pos, SharedSegment segment) { assert ReplicatedChronicleMap.this == map : "ModificationIterator.onPut() is called from outside of the parent map"; diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 98a46941c..1f2b4ffef 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -19,7 +19,9 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashErrorListener; -import net.openhft.chronicle.hash.serialization.*; +import net.openhft.chronicle.hash.serialization.BytesReader; +import net.openhft.chronicle.hash.serialization.Hasher; +import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter; import net.openhft.chronicle.hash.serialization.internal.MetaProvider; @@ -32,6 +34,7 @@ import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,8 +61,7 @@ class VanillaChronicleMap, private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class); /** - * Because DirectBitSet implementations couldn't find more than 64 continuous clear or set - * bits. + * Because DirectBitSet implementations couldn't find more than 64 continuous clear or set bits. */ static final int MAX_ENTRY_OVERSIZE_FACTOR = 64; static final ThreadLocal tmpBytes = new ThreadLocal<>(); @@ -484,9 +486,9 @@ class VanillaChronicleMap, /** - * removes ( if there exists ) an entry from the map, if the {@param key} and {@param - * expectedValue} match that of a maps.entry. If the {@param expectedValue} equals null then ( - * if there exists ) an entry whose key equals {@param key} this is removed. + * removes ( if there exists ) an entry from the map, if the {@param key} and {@param expectedValue} match + * that of a maps.entry. If the {@param expectedValue} equals null then ( if there exists ) an entry whose + * key equals {@param key} this is removed. * * @param k the key of the entry to remove * @param expectedValue null if not required @@ -546,8 +548,8 @@ class VanillaChronicleMap, * replace the value in a map, only if the existing entry equals {@param existingValue} * * @param key the key into the map - * @param existingValue the expected existing value in the map ( could be null when we don't - * wish to do this check ) + * @param existingValue the expected existing value in the map ( could be null when we don't wish to do + * this check ) * @param newValue the new value you wish to store in the map * @return the value that was replaced */ @@ -680,7 +682,7 @@ class VanillaChronicleMap, /** * close for native values does not call put on close */ - final MutableLockedEntry writeUnlock = isNativeValueClass + final WriteLocked writeUnlock = isNativeValueClass ? new NativeWriteLocked() : new HeapWriteLocked(); @@ -766,7 +768,7 @@ class VanillaChronicleMap, } - public MutableLockedEntry readLock() throws IllegalStateException { + public ReadLocked readLock() throws IllegalStateException { while (true) { final boolean success = segmentHeader.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS); if (success) { @@ -784,7 +786,7 @@ class VanillaChronicleMap, } } - public MutableLockedEntry writeLock() throws IllegalStateException { + public WriteLocked writeLock() throws IllegalStateException { while (true) { final boolean success = segmentHeader.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS); if (success) return writeUnlock; @@ -926,7 +928,7 @@ class VanillaChronicleMap, entry.skip(keySizeMarshaller.sizeEncodingSize(keySize) + keySize); skipReplicationBytes(entry); V v = readValue(copies, entry, usingValue); - notifyPut(offset, true, key, v, posFromOffset(offset)); + notifyPut(offset, true, key, v, null, posFromOffset(offset)); notifyGet(offset, key, v); return v; } @@ -965,7 +967,7 @@ class VanillaChronicleMap, // key is not found long offset = putEntry(copies, metaKeyInterop, keyInterop, key, keySize, value); incrementSize(); - notifyPut(offset, true, key, value, posFromOffset(offset)); + notifyPut(offset, true, key, value, null, posFromOffset(offset)); return null; } @@ -981,7 +983,7 @@ class VanillaChronicleMap, // putValue may relocate entry and change offset offset = putValue(pos, offset, entry, valueSizePos, entryEndAddr, copies, value, null, searchedHashLookup); - notifyPut(offset, false, key, value, posFromOffset(offset)); + notifyPut(offset, false, key, value, prevValue, posFromOffset(offset)); return prevValue; } @@ -1082,7 +1084,7 @@ class VanillaChronicleMap, return readValue(copies, entry, value, readValueSize(entry)); } - V readValue(ThreadLocalCopies copies, MultiStoreBytes entry, V value, long valueSize) { + V readValue(ThreadLocalCopies copies, NativeBytes entry, V value, long valueSize) { copies = valueReaderProvider.getCopies(copies); BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); return valueReader.read(entry, valueSize, value); @@ -1097,12 +1099,12 @@ class VanillaChronicleMap, /** * Removes a key (or key-value pair) from the Segment. * - *

The entry will only be removed if {@code expectedValue} equals to {@code null} or the - * value previously corresponding to the specified key. + *

The entry will only be removed if {@code expectedValue} equals to {@code null} or the value + * previously corresponding to the specified key. * * @param hash2 a hash code related to the {@code keyBytes} - * @return the value of the entry that was removed if the entry corresponding to the {@code - * keyBytes} exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise + * @return the value of the entry that was removed if the entry corresponding to the {@code keyBytes} + * exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise */ V remove(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V expectedValue, long hash2) { @@ -1162,9 +1164,9 @@ class VanillaChronicleMap, } /** - * Replaces the specified value for the key with the given value. {@code newValue} is set - * only if the existing value corresponding to the specified key is equal to {@code - * expectedValue} or {@code expectedValue == null}. + * Replaces the specified value for the key with the given value. {@code newValue} is set only if the + * existing value corresponding to the specified key is equal to {@code expectedValue} or {@code + * expectedValue == null}. * * @param hash2 a hash code related to the {@code keyBytes} * @return the replaced value or {@code null} if the value was not replaced @@ -1205,19 +1207,20 @@ class VanillaChronicleMap, // putValue may relocate entry and change offset offset = putValue(pos, offset, entry, valueSizePos, entryEndAddr, copies, newValue, null, searchedHashLookup); - notifyPut(offset, false, key, newValue, posFromOffset(offset)); + notifyPut(offset, false, key, newValue, valueRead, posFromOffset(offset)); return valueRead; } return null; } - void notifyPut(long offset, boolean added, K key, V value, final long pos) { + void notifyPut(long offset, boolean added, K key, V value, @Nullable final V replacedValue, + final long pos) { if (eventListener() != MapEventListeners.NOP) { MultiStoreBytes b = acquireTmpBytes(); b.storePositionAndSize(bytes, offset, entrySize); eventListener().onPut(VanillaChronicleMap.this, b, metaDataBytes, - added, key, value, pos, this); + added, key, replacedValue, value, pos, this); } } @@ -1240,17 +1243,17 @@ class VanillaChronicleMap, } /** - * Replaces value in existing entry. May cause entry relocation, because there may be not - * enough space for new value in location already allocated for this entry. + * Replaces value in existing entry. May cause entry relocation, because there may be not enough space + * for new value in location already allocated for this entry. * * @param pos index of the first block occupied by the entry - * @param offset relative offset of the entry in Segment bytes (before, i. e. - * including metaData) + * @param offset relative offset of the entry in Segment bytes (before, i. e. including + * metaData) * @param entry relative pointer in Segment bytes * @param valueSizePos relative position of value size in entry * @param entryEndAddr absolute address of the entry end - * @return relative offset of the entry in Segment bytes after putting value (that may cause - * entry relocation) + * @return relative offset of the entry in Segment bytes after putting value (that may cause entry + * relocation) */ long putValue(long pos, long offset, NativeBytes entry, long valueSizePos, long entryEndAddr, ThreadLocalCopies copies, V value, Bytes valueBytes, @@ -1661,7 +1664,7 @@ class VanillaChronicleMap, } - private static abstract class ReadLocked extends MutableLockedEntry extends MutableLockedEntry implements ReadContext { boolean present; @@ -1680,7 +1683,7 @@ class VanillaChronicleMap, } } - private static abstract class WriteLocked extends MutableLockedEntry extends MutableLockedEntry implements WriteContext { } } diff --git a/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java b/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java index 21204db14..65f699a45 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMMetaDataTest.java @@ -81,7 +81,7 @@ public class CHMMetaDataTest { @Override public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, - boolean added, String key, String value, long pos, + boolean added, String key, String replacedValue, String value, long pos, SharedSegment segment) { assertEquals(8, metaDataBytes); if (added) diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java index f7782a1d0..0bf456639 100644 --- a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java +++ b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java @@ -44,6 +44,27 @@ public class EventListenerTestWithTCPSocketReplication { ChronicleMap map1; ChronicleMap map2; + final AtomicReference keyRef = new AtomicReference(); + + final AtomicReference replacedValueRef = new AtomicReference(); + final AtomicReference valueRef = new AtomicReference(); + final AtomicBoolean putWasCalled = new AtomicBoolean(false); + + final 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, CharSequence replacedValue) { + putWasCalled.getAndSet(true); + keyRef.set(key); + valueRef.set(value); + replacedValueRef.set(replacedValue); + } + }; + + @After public void tearDown() throws InterruptedException { @@ -55,7 +76,12 @@ public class EventListenerTestWithTCPSocketReplication { } } - // Thread.sleep(1600); + putWasCalled.set(false); + valueRef.set(null); + replacedValueRef.set(null); + keyRef.set(null); + + } @@ -63,60 +89,26 @@ public class EventListenerTestWithTCPSocketReplication { public void testAdded() 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 - TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); - ChronicleMapBuilder map1Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 2, tcpConfig); + ChronicleMapBuilder map1Builder = serverBuilder(); map1 = map1Builder.create(); } // ---------- SERVER2 2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); - ChronicleMapBuilder map2Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 1, tcpConfig) - .eventListener(eventListener); + ChronicleMapBuilder map2Builder = clientBuilder(); map2 = map2Builder.create(); } // 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); - } + waitTillReplicated(); Assert.assertEquals("Maps should be equal", map1, map2); Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); @@ -126,8 +118,6 @@ public class EventListenerTestWithTCPSocketReplication { Assert.assertEquals(Integer.valueOf(5), keyRef.get()); Assert.assertEquals("EXAMPLE", valueRef.get()); - keyRef.set(null); - valueRef.set(null); } @@ -135,63 +125,29 @@ public class EventListenerTestWithTCPSocketReplication { @Test 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 - TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); - ChronicleMapBuilder map1Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 2, tcpConfig); + ChronicleMapBuilder map1Builder = serverBuilder(); map1 = map1Builder.create(); + } // ---------- SERVER2 2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); - ChronicleMapBuilder map2Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 1, tcpConfig) - .eventListener(eventListener); - map2 = map2Builder.create(); + map2 = clientBuilder().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); - } + waitTillReplicated(); Assert.assertEquals("Maps should be equal", map1, map2); Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); @@ -200,74 +156,37 @@ public class EventListenerTestWithTCPSocketReplication { 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); + Assert.assertEquals("WILL_GET_OVER-WRITTEN", replacedValueRef.get()); } + @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 - TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); - ChronicleMapBuilder map1Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 2, tcpConfig); + map1 = serverBuilder().create(); - map1 = map1Builder.create(); - map1.put(5, "WILL_GET_OVERWRITEN"); } // ---------- SERVER2 2 on the same server as ---------- { - TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); - ChronicleMapBuilder map2Builder = - ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(20000L) - .replication((byte) 1, tcpConfig) - .eventListener(eventListener); - map2 = map2Builder.create(); + map2 = clientBuilder().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); - } + + waitTillReplicated(); Assert.assertEquals("Maps should be equal", map1, map2); Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); @@ -275,10 +194,35 @@ public class EventListenerTestWithTCPSocketReplication { Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); Assert.assertEquals(Integer.valueOf(5), keyRef.get()); + Assert.assertEquals("small", replacedValueRef.get()); Assert.assertEquals("EXAMPLE", valueRef.get()); - keyRef.set(null); - valueRef.set(null); + } + + private ChronicleMapBuilder clientBuilder() { + TcpConfig tcpConfig = TcpConfig.forReceivingOnlyNode(8077); + return ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replication((byte) 1, tcpConfig) + .eventListener(eventListener); + } + + + private void waitTillReplicated() throws InterruptedException { + int t = 0; + for (; t < 5000; t++) { + if (map1.equals(map2)) { + break; + } + Thread.sleep(1); + } + } + + private ChronicleMapBuilder serverBuilder() { + TcpConfig tcpConfig = forSendingNode(8076, new InetSocketAddress("localhost", 8077)); + return ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replication((byte) 2, tcpConfig); } } \ No newline at end of file