HCOLL-185 refactored VanillaChronicleMap and ReplicatedChronicleMap; split BytesMapEventListener from MapEventListener; many bug fixes and optimizations; Stateless client queries without serialization/deserialization not working yet

src/main/java/net/openhft/chronicle/hash/serialization/internal/BytesBytesInterop.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.serialization.internal;
+
+import net.openhft.chronicle.hash.serialization.BytesInterop;
+import net.openhft.chronicle.hash.serialization.Hasher;
+import net.openhft.lang.io.Bytes;
+
+public enum BytesBytesInterop implements BytesInterop<Bytes> {
+    INSTANCE;
+
+    @Override
+    public boolean startsWith(Bytes bytes, Bytes e) {
+        return bytes.startsWith(e);
+    }
+
+    @Override
+    public long hash(Bytes e) {
+        return Hasher.hash(e);
+    }
+
+    @Override
+    public long size(Bytes e) {
+        return e.remaining();
+    }
+
+    @Override
+    public void write(Bytes bytes, Bytes e) {
+        bytes.write(e);
+    }
+}

src/main/java/net/openhft/chronicle/hash/serialization/internal/MetaProvider.java

@@ -22,7 +22,7 @@ import net.openhft.lang.threadlocal.ThreadLocalCopies;
 
 import java.io.Serializable;
 
-public interface MetaProvider<E, W, MW extends MetaBytesWriter<E, W>> extends Serializable {
+public interface MetaProvider<E, W, MW extends MetaBytesWriter<E, ? super W>> extends Serializable {
 
     MW get(ThreadLocalCopies copies, MW originalMetaWriter, W writer, E e);
 

src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java

@@ -22,6 +22,7 @@ import net.openhft.chronicle.hash.replication.ThrottlingConfig;
 import net.openhft.lang.io.ByteBufferBytes;
 import net.openhft.lang.io.Bytes;
 import net.openhft.lang.thread.NamedThreadFactory;
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
@@ -69,6 +70,9 @@ abstract class AbstractChannelReplicator implements Closeable {
     volatile boolean isClosed = false;
 
 
+    ThreadLocalCopies copies;
+    VanillaChronicleMap.SegmentState segmentState;
+
     AbstractChannelReplicator(String name, ThrottlingConfig throttlingConfig,
                               int maxEntrySizeBytes)
             throws IOException {
@@ -211,6 +215,8 @@ abstract class AbstractChannelReplicator implements Closeable {
                     @Override
                     public void run() {
                         try {
+                            copies = VanillaChronicleMap.SegmentState.getCopies(null);
+                            segmentState = VanillaChronicleMap.SegmentState.get(copies);
                             processEvent();
                         } catch (Exception e) {
                             LOG.error("", e);
@@ -227,7 +233,7 @@ abstract class AbstractChannelReplicator implements Closeable {
         executorService.shutdown();
         closeables.closeQuietly();
         executorService.shutdownNow();
-
+        segmentState.close();
 
         try {
             // we HAVE to be sure we have terminated before calling close() on the

src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java

@@ -136,15 +136,15 @@ abstract class AbstractChronicleMapBuilder<K, V,
     private TimeProvider timeProvider = TimeProvider.SYSTEM;
     private BytesMarshallerFactory bytesMarshallerFactory;
     private ObjectSerializer objectSerializer;
-    private MapEventListener<K, V, ChronicleMap<K, V>> eventListener =
-            MapEventListeners.nop();
+    private MapEventListener<K, V> eventListener = null;
+    private BytesMapEventListener bytesEventListener = null;
     private V defaultValue = null;
     private DefaultValueProvider<K, V> defaultValueProvider = null;
     private PrepareValueBytes<K, V> prepareValueBytes = null;
 
     private SingleChronicleHashReplication singleHashReplication = null;
     private InetSocketAddress[] pushToAddresses;
-    public boolean bootstapOnlyLocalEntries = false;
+    public boolean bootstrapOnlyLocalEntries = false;
 
     AbstractChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
         keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
@@ -680,15 +680,24 @@ abstract class AbstractChronicleMapBuilder<K, V,
         return self();
     }
 
-    public B eventListener(MapEventListener<K, V, ChronicleMap<K, V>> eventListener) {
+    public B eventListener(MapEventListener<K, V> eventListener) {
         this.eventListener = eventListener;
         return self();
     }
 
-    MapEventListener<K, V, ChronicleMap<K, V>> eventListener() {
+    MapEventListener<K, V> eventListener() {
         return eventListener;
     }
 
+    public B bytesEventListener(BytesMapEventListener eventListener) {
+        this.bytesEventListener = eventListener;
+        return self();
+    }
+
+    BytesMapEventListener bytesEventListener() {
+        return bytesEventListener;
+    }
+
     /**
      * Specifies the value to be put for each key queried in {@link ChronicleMap#acquireUsing acquireUsing()}
      * method, if the key is absent in the map, created by this builder.
@@ -859,7 +868,7 @@ abstract class AbstractChronicleMapBuilder<K, V,
         preMapConstruction(true);
         VanillaChronicleMap<K, ?, ?, V, ?, ?> map =
                 new ReplicatedChronicleMap<K, Object, MetaBytesInterop<K, Object>,
-                        V, Object, MetaBytesWriter<V, Object>>(this, identifier);
+                        V, Object, MetaBytesInterop<V, Object>>(this, identifier);
         BytesStore bytesStore = new DirectStore(JDKZObjectSerializer.INSTANCE,
                 map.sizeInBytes(), true);
         map.createMappedStoreAndSegments(bytesStore);
@@ -895,7 +904,7 @@ abstract class AbstractChronicleMapBuilder<K, V,
             for (int i = 0; i < pushToAddresses.length; i++) {
                 statelessClients[i] = cmb.statelessClient(pushToAddresses[i]).create();
             }
-            eventListener = (MapEventListener<K, V, ChronicleMap<K, V>>) constructor.newInstance((Object) statelessClients);
+            eventListener = (MapEventListener<K, V>) constructor.newInstance((Object) statelessClients);
         } catch (ClassNotFoundException e) {
             LoggerFactory.getLogger(getClass().getName()).warn("Chronicle Enterprise not found in the class path");
         } catch (Exception e) {
@@ -924,10 +933,10 @@ abstract class AbstractChronicleMapBuilder<K, V,
                 identifier = channel.hub().identifier();
             }
             return new ReplicatedChronicleMap<K, Object, MetaBytesInterop<K, Object>,
-                    V, Object, MetaBytesWriter<V, Object>>(this, identifier);
+                    V, Object, MetaBytesInterop<V, Object>>(this, identifier);
         } else {
             return new VanillaChronicleMap<K, Object, MetaBytesInterop<K, Object>,
-                    V, Object, MetaBytesWriter<V, Object>>(this);
+                    V, Object, MetaBytesInterop<V, Object>>(this);
         }
     }
 
@@ -998,8 +1007,8 @@ abstract class AbstractChronicleMapBuilder<K, V,
 
 
     @Override
-    public ChronicleMapBuilderI<K, V> bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) {
-        this.bootstapOnlyLocalEntries = bootstapOnlyLocalEntries;
+    public ChronicleMapBuilderI<K, V> bootstrapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) {
+        this.bootstrapOnlyLocalEntries = bootstapOnlyLocalEntries;
         return this;
     }
 

src/main/java/net/openhft/chronicle/map/Asserts.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import org.jetbrains.annotations.NotNull;
+
+final class Asserts {
+
+    public static void assertNotNull(@NotNull Object obj) {
+        assert obj != null;
+    }
+
+    private Asserts() {}
+}

src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.hash.serialization.SizeMarshaller;
+import net.openhft.lang.io.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+
+/**
+ * <ul>
+ *     <li>Methods must not update {@code entry} state ({@linkplain Bytes#position(long) position},
+ *     {@linkplain Bytes#limit() limit}).</li>
+ *     <li>Methods must not update entry bytes, except in metadata area</li>
+ *     <li>From {@code metaDataPos} offset metadata area starts in the given {@code entry}, listener
+ *     should be itself aware of it's length (see {@link ChronicleMapBuilder#metaDataBytes()}).</li>
+ *     <li>From {@code keyPos} offset key area in the given {@code entry}, serialized key size
+ *     (using {@linkplain ChronicleMapBuilder#keySizeMarshaller(SizeMarshaller)}), directly followed
+ *     by the serialized key itself.</li>
+ *     <li>From {@code valuePos} offset value area in the given {@code entry}, serialized value size
+ *     (using {@linkplain ChronicleMapBuilder#keySizeMarshaller(SizeMarshaller)}), then (optionally)
+ *     alignment (see {@link OffHeapUpdatableChronicleMapBuilder#entryAndValueAlignment(Alignment)}
+ *     ), followed by the serialized value itself.</li>
+ * </ul>
+ */
+public abstract class BytesMapEventListener implements Serializable {
+    private static final long serialVersionUID = 0L;
+
+    public void onGetFound(Bytes entry, long metaDataPos, long keyPos, long valuePos) {
+        // do nothing
+    }
+
+    public void onPut(Bytes entry, long metaDataPos, long keyPos, long valuePos, boolean added) {
+        // do nothing
+    }
+
+    public void onRemove(Bytes entry, long metaDataPos, long keyPos, long valuePos) {
+        // do nothing
+    }
+}

src/main/java/net/openhft/chronicle/map/ChannelProvider.java

@@ -26,6 +26,7 @@ import net.openhft.lang.collection.DirectBitSet;
 import net.openhft.lang.collection.SingleThreadedDirectBitSet;
 import net.openhft.lang.io.ByteBufferBytes;
 import net.openhft.lang.io.Bytes;
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -108,7 +109,9 @@ final class ChannelProvider implements Closeable {
         }
 
         @Override
-        public void readExternalEntry(@NotNull Bytes source) {
+        public void readExternalEntry(
+                @NotNull ThreadLocalCopies copies,
+                @NotNull VanillaChronicleMap.SegmentState segmentState, @NotNull Bytes source) {
             channelDataLock.readLock().lock();
             try {
                 final int chronicleId = (int) source.readStopBit();
@@ -116,7 +119,8 @@ final class ChannelProvider implements Closeable {
 
                     // this channel is has not currently been created so it updates will be ignored
                    if (channelEntryExternalizables[chronicleId] != null)
-                        channelEntryExternalizables[chronicleId].readExternalEntry(source);
+                        channelEntryExternalizables[chronicleId]
+                                .readExternalEntry(copies, segmentState, source);
                 } else
                     LOG.info("skipped entry with chronicleId=" + chronicleId + ", ");
             } finally {
@@ -484,7 +488,9 @@ final class ChannelProvider implements Closeable {
             }
 
             @Override
-            public void readExternalEntry(@NotNull Bytes source) {
+            public void readExternalEntry(
+                    @NotNull ThreadLocalCopies copies,
+                    @NotNull VanillaChronicleMap.SegmentState segmentState, @NotNull Bytes source) {
                 messageHandler.onMessage(source);
             }
         };

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java

@@ -492,27 +492,34 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
     }
 
     @Override
-    public ChronicleMapBuilder<K, V> eventListener(MapEventListener<K, V, ChronicleMap<K, V>> eventListener) {
+    public ChronicleMapBuilder<K, V> eventListener(MapEventListener<K, V> eventListener) {
         delegate.eventListener(eventListener);
         return this;
     }
 
+    @Override
+    public ChronicleMapBuilderI<K, V> bytesEventListener(BytesMapEventListener eventListener) {
+        delegate.bytesEventListener(eventListener);
+        return this;
+    }
+
+
     @Override
     public ChronicleMap<K, V> createPersistedTo(File file) throws IOException {
         return delegate.createPersistedTo(file);
     }
 
     /**
-     * @param bootstapOnlyLocalEntries if set to true - when a new node joins a TCP replication grid, the new
-     *                                 node will be populated with data, only for the nodes that created that
-     *                                 data. Otherwise, all the nodes will publish all the data they have (
-     *                                 potentially swamping the new node with duplicates ) however this does
-     *                                 guarantee that all the data is replicated over to the new node, and is
-     *                                 useful especially in the case that the originating node is not
-     *                                 currently running.
+     * @param bootstrapOnlyLocalEntries if set to true - when a new node joins a TCP replication grid, the new
+     *                                  node will be populated with data, only for the nodes that created that
+     *                                  data. Otherwise, all the nodes will publish all the data they have (
+     *                                  potentially swamping the new node with duplicates ) however this does
+     *                                  guarantee that all the data is replicated over to the new node, and is
+     *                                  useful especially in the case that the originating node is not
+     *                                  currently running.
      */
-    public ChronicleMapBuilder<K, V> bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) {
-        delegate.bootstapOnlyLocalEntries(bootstapOnlyLocalEntries);
+    public ChronicleMapBuilder<K, V> bootstrapOnlyLocalEntries(boolean bootstrapOnlyLocalEntries) {
+        delegate.bootstrapOnlyLocalEntries(bootstrapOnlyLocalEntries);
         return this;
     }
 

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java

@@ -290,7 +290,9 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
 
     ChronicleMapBuilderI<K, V> clone();
 
-    ChronicleMapBuilderI<K, V> eventListener(MapEventListener<K, V, ChronicleMap<K, V>> eventListener);
+    ChronicleMapBuilderI<K, V> eventListener(MapEventListener<K, V> eventListener);
+
+    ChronicleMapBuilderI<K, V> bytesEventListener(BytesMapEventListener eventListener);
 
     ChronicleMapBuilderI<K, V> putReturnsNull(boolean b);
 
@@ -307,7 +309,7 @@ public interface ChronicleMapBuilderI<K, V> extends Serializable {
      *                                 useful especially in the case that the originating node is not
      *                                 currently running.
      */
-    ChronicleMapBuilderI<K, V> bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries);
+    ChronicleMapBuilderI<K, V> bootstrapOnlyLocalEntries(boolean bootstapOnlyLocalEntries);
 
     String name();
 

src/main/java/net/openhft/chronicle/map/GetValueInterops.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
+import org.jetbrains.annotations.NotNull;
+
+interface GetValueInterops<VB, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>> {
+    MVBI getMetaValueInterop(@NotNull ThreadLocalCopies copies, VBI valueInterop, VB value);
+
+    VBI getValueInterop(@NotNull ThreadLocalCopies copies);
+}

src/main/java/net/openhft/chronicle/map/InstanceOrBytesToInstance.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
+import org.jetbrains.annotations.NotNull;
+
+interface InstanceOrBytesToInstance<TB, T> {
+    T toInstance(@NotNull ThreadLocalCopies copies, TB instanceOrBytes, long size);
+}

src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java

@@ -120,6 +120,12 @@ class Int24Int24MultiMap implements MultiMap {
         return (pos -= ENTRY_SIZE) >= 0L ? pos : pos + capacityInBytes;
     }
 
+    @Override
+    public void putPosition(long value) {
+        checkValueForPut(value);
+        positions.set(value);
+    }
+
     @Override
     public void put(long key, long value) {
         key = maskUnsetKey(key);
@@ -134,6 +140,12 @@ class Int24Int24MultiMap implements MultiMap {
         }
     }
 
+    @Override
+    public void removePosition(long value) {
+        checkValueForRemove(value);
+        positions.clear(value);
+    }
+
     @Override
     public void remove(long key, long value) {
         key = maskUnsetKey(key);

src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java

@@ -121,6 +121,12 @@ class IntIntMultiMap implements MultiMap {
         return (pos - ENTRY_SIZE) & capacityMask2;
     }
 
+    @Override
+    public void putPosition(long value) {
+        checkValueForPut(value);
+        positions.set(value);
+    }
+
     @Override
     public void put(long key, long value) {
         key = maskUnsetKey(key);
@@ -135,6 +141,12 @@ class IntIntMultiMap implements MultiMap {
         }
     }
 
+    @Override
+    public void removePosition(long value) {
+        checkValueForRemove(value);
+        positions.clear(value);
+    }
+
     @Override
     public void remove(long key, long value) {
         key = maskUnsetKey(key);

src/main/java/net/openhft/chronicle/map/MapEventListener.java

@@ -19,8 +19,9 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.chronicle.hash.ChronicleHashBuilder;
-import net.openhft.lang.io.Bytes;
 import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
 
@@ -34,8 +35,7 @@ import java.io.Serializable;
  * <p>To configure {@code MapEventListener} for {@code ChronicleMap}, use {@link
  * AbstractChronicleMapBuilder#eventListener(MapEventListener)} method.
  *
- * <p>{@link MapEventListeners} uninstantiable class contains several logging implementations and the default
- * {@linkplain MapEventListeners#nop() no-op implementation}.
+ * <p>See {@link #logging(String)} implementation.
  *
  * <p>All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as quick as
  * possible.
@@ -45,33 +45,61 @@ import java.io.Serializable;
  *
  * @param <K> key type of the maps, trackable by this event listener
  * @param <V> value type of the maps, trackable by this event listener
- * @param <M> {@code ChronicleMap} subtype, trackable by this event listener
  * @see AbstractChronicleMapBuilder#eventListener(MapEventListener)
- * @see MapEventListeners
  */
-public abstract class MapEventListener<K, V, M extends ChronicleMap<K, V>>
-        implements Serializable {
+public abstract class MapEventListener<K, V> implements Serializable {
     private static final long serialVersionUID = 0L;
 
+    /**
+     * Returns the map event listener, which logs strings like
+     * "{@linkplain ChronicleMap#file() map file} opType key value", where opType is either "get",
+     * "put" or "remove", to the logger provided by SLF4J.
+     *
+     * @param <K> the map key type
+     * @param <V> the map value type
+     * @return the logging event listener
+     */
+    public static <K, V> MapEventListener<K, V> logging(String prefix) {
+        return new LoggingMapEventListener(prefix);
+    }
+
+    private static class LoggingMapEventListener extends MapEventListener {
+        private static final long serialVersionUID = 0L;
+        public final static Logger LOGGER = LoggerFactory.getLogger(LoggingMapEventListener.class);
+
+        private final String prefix;
+
+        private LoggingMapEventListener(String prefix) {
+            this.prefix = prefix;
+        }
+
+        @Override
+        public void onGetFound(Object key, Object value) {
+            LOGGER.info("get {} => {}", prefix, key, value);
+        }
+
+        @Override
+        public void onPut(Object key, Object value, Object replacedValue) {
+            LOGGER.info("{} put {} => {}", prefix, key, value);
+        }
+
+        @Override
+        public void onRemove(Object key, Object value) {
+            LOGGER.info("{} remove {} was {}", prefix, key, value);
+        }
+    }
+
     /**
      * 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 key           the key looked up
-     * @param value         the value found for the key
+     * @param foundValue    the value found for the key
      */
-    public void onGetFound(M map, Bytes entry, int metaDataBytes, K key, V value) {
+    public void onGetFound(K key, V foundValue) {
         // do nothing
     }
 
-    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, 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
@@ -81,41 +109,23 @@ public abstract class MapEventListener<K, V, M extends ChronicleMap<K, V>>
      *
      * <p>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 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
+     * @param newValue      the value which is now associated with the given key
+     * @param replacedValue the value which was replaced by {@code newValue}, {@code null}
+     *                      if the key was absent in the map before current {@code ChronicleMap}
      */
-    public void onPut(M map, Bytes entry, int metaDataBytes, boolean added, K key, V value,
-                      @Nullable V replacedValue) {
+    public void onPut(K key, V newValue, @Nullable V replacedValue) {
         // do nothing
     }
 
-    void onRemove(M map, Bytes entry, int metaDataBytes, K key, V value,
-                  long pos, SharedSegment segment) {
-        onRemove(map, entry, metaDataBytes, key, value);
-    }
-
     /**
-     * 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 key           the key removed from the map
      * @param value         the value which was associated with the given key
      */
-    public void onRemove(M map, Bytes entry, int metaDataBytes, K key, V value) {
-        // do nothing
-    }
-
-    void onRelocation(long pos, SharedSegment segment) {
+    public void onRemove(K key, V value) {
         // do nothing
     }
 }

src/main/java/net/openhft/chronicle/map/MapEventListeners.java

@@ -1,153 +0,0 @@
-/*
- * Copyright 2014 Higher Frequency Trading
- *
- * http://www.higherfrequencytrading.com
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package net.openhft.chronicle.map;
-
-
-import net.openhft.lang.io.Bytes;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Methods returning useful implementations of {@link MapEventListener}.
- */
-public final class MapEventListeners {
-
-    /**
-     * Factories a more flexible than public static instances.
-     * We can add some configuration-on-the-first-call in the future.
-     */
-
-    static final MapEventListener NOP = new MapEventListener() {
-        private static final long serialVersionUID = 0L;
-    };
-    private static final MapEventListener BYTES_LOGGING = new MapEventListener() {
-        private static final long serialVersionUID = 0L;
-        public final Logger LOGGER = LoggerFactory.getLogger(getClass());
-
-        @Override
-        public void onGetFound(ChronicleMap map, Bytes entry, int metaDataBytes,
-                               Object key, Object value) {
-            logOperation(map, entry, metaDataBytes, " get ");
-        }
-
-        private void logOperation(ChronicleMap map, Bytes entry, int metaDataBytes, String oper) {
-            StringBuilder sb = new StringBuilder();
-            sb.append(map.file()).append(oper);
-            if (metaDataBytes > 0) {
-                sb.append("Meta: ");
-                entry.toString(sb, 0L, 0L, metaDataBytes);
-                sb.append(" | ");
-            }
-            Bytes slice = entry.slice(metaDataBytes, entry.limit() - metaDataBytes);
-            VanillaChronicleMap vanillaMap = (VanillaChronicleMap) map;
-            long keySize = vanillaMap.keySizeMarshaller.readSize(slice);
-            slice.toString(sb, slice.position(), 0L, slice.position() + keySize);
-            slice.position(slice.position() + keySize);
-            long valueSize = vanillaMap.valueSizeMarshaller.readSize(slice);
-            vanillaMap.alignment.alignPositionAddr(slice);
-            sb.append(" = ");
-            slice.toString(sb, slice.position(), 0L, slice.position() + valueSize);
-            LOGGER.info(sb.toString());
-        }
-
-        @Override
-        public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added,
-                          Object key, Object replacedValue, Object value, long pos, SharedSegment segment) {
-            logOperation(map, entry, metaDataBytes, added ? " +put " : " put ");
-        }
-
-        @Override
-        public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes,
-                             Object key, Object value, long pos, SharedSegment segment) {
-            logOperation(map, entry, metaDataBytes, " remove ");
-        }
-    };
-    private static final MapEventListener KEY_VALUE_LOGGING = new MapEventListener() {
-        private static final long serialVersionUID = 0L;
-        public final Logger LOGGER = LoggerFactory.getLogger(getClass());
-
-        @Override
-        public void onGetFound(ChronicleMap map, Bytes entry, int metaDataBytes,
-                               Object key, Object value) {
-            LOGGER.info("{} get {} => {}", map.file(), key, value);
-        }
-
-        @Override
-        public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added,
-                          Object key, Object replacedValue, Object value, long pos, SharedSegment segment) {
-            LOGGER.info("{} put {} => {}", map.file(), key, value);
-        }
-
-        @Override
-        public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes,
-                             Object key, Object value, long pos, SharedSegment segment) {
-            LOGGER.info("{} remove {} was {}", map.file(), key, value);
-        }
-    };
-
-    private MapEventListeners() {
-    }
-
-    /**
-     * Returns the no-op event listener, i. e. all it's method implementations have empty bodies.
-     *
-     * @param <K> the map key type
-     * @param <V> the map value type
-     * @param <M> the {@code ChronicleMap} subclass, or {@code ChronicleMap} type itself.
-     *           Typically this type should be just {@code ChronicleMap}.
-     * @return the no-op event listener
-     */
-    @SuppressWarnings("unchecked")
-    public static <K, V, M extends ChronicleMap<K, V>> MapEventListener<K, V, M> nop() {
-        return NOP;
-    }
-
-    /**
-     * Returns the map event listener, which logs strings like
-     * "Meta: [meta bytes] | [key bytes] = [value bytes]", where bytes are meta, key and value
-     * areas from {@link Bytes} passed to {@code MapEventListener} methods, printed in ASCII
-     * coding.
-     *
-     * @param <K> the map key type
-     * @param <V> the map value type
-     * @param <M> the {@code ChronicleMap} subclass, or {@code ChronicleMap} type itself.
-     *           Typically this type should be just {@code ChronicleMap}.
-     * @return the bytes logging event listener
-     */
-    @SuppressWarnings("unchecked")
-    public static <K, V, M extends ChronicleMap<K, V>> MapEventListener<K, V, M> bytesLogging() {
-        return BYTES_LOGGING;
-    }
-
-    /**
-     * Returns the map event listener, which logs strings like
-     * "{@linkplain ChronicleMap#file() map file} opType key value", where opType is either "get",
-     * "put" or "remove", to the logger provided by SLF4J.
-     *
-     * @param <K> the map key type
-     * @param <V> the map value type
-     * @param <M> the {@code ChronicleMap} subclass, or {@code ChronicleMap} type itself.
-     *           Typically this type should be just {@code ChronicleMap}.
-     * @return the logging event listener
-     */
-    @SuppressWarnings("unchecked")
-    public static <K, V, M extends ChronicleMap<K, V>> MapEventListener<K, V, M> keyValueLogging() {
-        return KEY_VALUE_LOGGING;
-    }
-}

src/main/java/net/openhft/chronicle/map/MultiMap.java

@@ -36,6 +36,8 @@ interface MultiMap {
      */
     void put(long key, long value);
 
+    void putPosition(long value);
+
     /**
      * Remove a key/value pair.
      *
@@ -45,6 +47,8 @@ interface MultiMap {
      */
     void remove(long key, long value);
 
+    void removePosition(long value);
+
     void replace(long key, long oldValue, long newValue);
 
     /**

src/main/java/net/openhft/chronicle/map/ReadValue.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.io.Bytes;
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * @param <RV> result type, map value type or Bytes
+ */
+interface ReadValue<RV> {
+    RV readValue(@NotNull ThreadLocalCopies copies, Bytes entry, RV usingValue, long valueSize);
+    RV readNull();
+}

src/main/java/net/openhft/chronicle/map/Replica.java

@@ -19,6 +19,7 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.lang.io.Bytes;
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.Closeable;
@@ -140,9 +141,13 @@ interface Replica extends Closeable {
          * is typically called when we receive a remote replication event, this event could originate from
          * either a remote {@code put(K key, V value)} or {@code remove(Object key)}
          *
+         * @param copies
+         * @param segmentState
          * @param source bytes to read an entry from
          */
-        void readExternalEntry(@NotNull Bytes source);
+        void readExternalEntry(
+                @NotNull ThreadLocalCopies copies,
+                @NotNull VanillaChronicleMap.SegmentState segmentState, @NotNull Bytes source);
 
     }
 

src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java

@@ -64,17 +64,15 @@ class ReplicationHubFindByName<K> implements FindByName {
         this.replicationHub = replicationHub;
         ReplicationChannel channel = replicationHub.createChannel((short) MAP_BY_NAME_CHANNEL);
 
-        final MapEventListener<CharSequence, ChronicleMapBuilderWithChannelId, ChronicleMap<CharSequence, ChronicleMapBuilderWithChannelId>> listener = new
-                MapEventListener<CharSequence, ChronicleMapBuilderWithChannelId, ChronicleMap<CharSequence, ChronicleMapBuilderWithChannelId>>() {
+        final MapEventListener<CharSequence, ChronicleMapBuilderWithChannelId> listener = new
+                MapEventListener<CharSequence, ChronicleMapBuilderWithChannelId>() {
 
                     // creates a map based on the details that are sent to the map of builders
                     @Override
-                    void onPut(ChronicleMap<CharSequence, ChronicleMapBuilderWithChannelId> map, Bytes
-                            entry, int metaDataBytes, boolean added, CharSequence key,
-                               ChronicleMapBuilderWithChannelId replaced, ChronicleMapBuilderWithChannelId
-                            value, long pos, SharedSegment segment) {
-                        super.onPut(map, entry, metaDataBytes, added, key, replaced, value, pos, segment);
-
+                    public void onPut(CharSequence key, ChronicleMapBuilderWithChannelId value,
+                               ChronicleMapBuilderWithChannelId replacedValue) {
+                        super.onPut(key, value, replacedValue);
+                        boolean added = replacedValue == null;
                         if (!added || value == null)
                             return;
 

src/main/java/net/openhft/chronicle/map/SharedSegment.java

@@ -19,6 +19,8 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.lang.io.NativeBytes;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 
 import java.util.Map;
 
@@ -28,18 +30,19 @@ import java.util.Map;
  */
 interface SharedSegment<K, V> {
 
-    AutoCloseable readLock();
+    /**
+     * if passed segmentState is null, null is returned instead of readLock
+     */
+    AutoCloseable readLock(@Nullable VanillaChronicleMap.SegmentState segmentState);
 
-    AutoCloseable writeLock();
+    AutoCloseable writeLock(@Nullable VanillaChronicleMap.SegmentState segmentState);
 
-    Map.Entry<K, V> getEntry(long pos);
+    Map.Entry<K, V> getEntry(@NotNull VanillaChronicleMap.SegmentState segmentState, long pos);
 
     void readUnlock();
 
     void writeUnlock();
 
-    NativeBytes entry(long pos);
-
     int getIndex();
 
     long offsetFromPos(long pos);

src/main/java/net/openhft/chronicle/map/ShortShortMultiMap.java

@@ -120,6 +120,12 @@ class ShortShortMultiMap implements MultiMap {
         return (long) (entry & 0xFFFF);
     }
 
+    @Override
+    public void putPosition(long value) {
+        checkValueForPut(value);
+        positions.set(value);
+    }
+
     @Override
     public void put(long key, long value) {
         key = maskUnsetKey(key);
@@ -134,6 +140,12 @@ class ShortShortMultiMap implements MultiMap {
         }
     }
 
+    @Override
+    public void removePosition(long value) {
+        checkValueForRemove(value);
+        positions.clear(value);
+    }
+
     @Override
     public void remove(long key, long value) {
         key = maskUnsetKey(key);
@@ -240,6 +252,7 @@ class ShortShortMultiMap implements MultiMap {
         bytes.writeInt(searchPos, entry(searchHash, value));
     }
 
+    @Override
     public long getSearchHash() {
         return searchHash;
     }

src/main/java/net/openhft/chronicle/map/TcpReplicator.java

@@ -1050,7 +1050,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
          *
          * @param socketChannel the socket to publish the buffer to
          * @param approxTime    an approximation of the current time in millis
-         * @param key           the selectors key
          * @throws IOException
          */
         private int writeBufferToSocket(@NotNull final SocketChannel socketChannel,
@@ -1282,7 +1281,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
                     }
 
                 } else
-                    externalizable.readExternalEntry(out);
+                    externalizable.readExternalEntry(copies, segmentState, out);
 
                 out.limit(limit);
 
@@ -1506,41 +1505,35 @@ class StatelessServerConnector<K, V> {
     }
 
     private Work removeWithValue(Bytes reader, Bytes writer, final long sizeLocation) {
-        if (MAP_SUPPORTS_BYTES) {
-            map.removeWithValue(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K key = readKey(reader, local);
-            final V readValue = readValue(reader, local);
-
-            final boolean result;
-            try {
-                result = map.remove(key, readValue);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                writer.writeBoolean(map.removeWithValue(reader));
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K key = readKey(reader, local);
+                final V readValue = readValue(reader, local);
+                writer.writeBoolean(map.remove(key, readValue));
             }
-            writer.writeBoolean(result);
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
     }
 
     private Work replaceWithOldAndNew(Bytes reader, Bytes writer, final long sizeLocation) {
-        if (MAP_SUPPORTS_BYTES) {
-            map.replaceWithOldAndNew(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K key = readKey(reader, local);
-            final V oldValue = readValue(reader, local);
-            final V newValue = readValue(reader, local);
-            boolean replaced;
-
-            try {
-                replaced = map.replace(key, oldValue, newValue);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                map.replaceWithOldAndNew(reader, writer);
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K key = readKey(reader, local);
+                final V oldValue = readValue(reader, local);
+                final V newValue = readValue(reader, local);
+                writer.writeBoolean(map.replace(key, oldValue, newValue));
             }
-            writer.writeBoolean(replaced);
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
@@ -1621,52 +1614,46 @@ class StatelessServerConnector<K, V> {
     }
 
     private Work containsKey(Bytes reader, Bytes writer, final long sizeLocation) {
-        if (MAP_SUPPORTS_BYTES) {
-            map.containsKey(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K k = readKey(reader, local);
-
-            try {
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                writer.writeBoolean(map.containsKey(reader));
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K k = readKey(reader, local);
                 writer.writeBoolean(map.containsKey(k));
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
             }
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
     }
 
     private Work containsValue(Bytes reader, Bytes writer, final long sizeLocation) {
+        // todo optimize -- eliminate
+        final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+        final V v = readValue(reader, local);
 
-        if (MAP_SUPPORTS_BYTES) {
-            map.containsValue(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final V v = readValue(reader, local);
-
-            try {
-                writer.writeBoolean(map.containsValue(v));
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
-            }
+        try {
+            writer.writeBoolean(map.containsValue(v));
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
     }
 
     private Work get(Bytes reader, Bytes writer, final long sizeLocation) {
-        if (MAP_SUPPORTS_BYTES) {
-            map.get(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K k = readKey(reader, local);
-
-            try {
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                map.get(reader, writer);
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K k = readKey(reader, local);
                 writeValue(map.get(k), writer, local);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
             }
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
@@ -1688,18 +1675,17 @@ class StatelessServerConnector<K, V> {
     }
 
     private Work put(Bytes reader, Bytes writer, final long sizeLocation) {
-        if (MAP_SUPPORTS_BYTES) {
-            map.put(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K k = readKey(reader, local);
-            final V v = readValue(reader, local);
-
-            try {
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                map.put(reader, writer);
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K k = readKey(reader, local);
+                final V v = readValue(reader, local);
                 writeValue(map.put(k, v), writer, local);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
             }
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
@@ -1707,7 +1693,7 @@ class StatelessServerConnector<K, V> {
 
     private Work remove(Bytes reader) {
         if (MAP_SUPPORTS_BYTES) {
-            map.removeWithBytes(reader);
+            map.removeKeyAsBytes(reader);
         } else {
             final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
             final K key = readKey(reader, local);
@@ -1719,40 +1705,34 @@ class StatelessServerConnector<K, V> {
 
 
     private Work remove(Bytes reader, Bytes writer, final long sizeLocation) {
-        if (MAP_SUPPORTS_BYTES) {
-            map.remove(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K key = readKey(reader, local);
-
-            try {
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                map.remove(reader, writer);
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K key = readKey(reader, local);
                 if (LOG.isDebugEnabled())
                     LOG.debug("removing entry key=" + key);
                 writeValue(map.remove(key), writer, local);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
             }
-
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
-
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
     }
 
     private Work putAll(Bytes reader, Bytes writer, final long sizeLocation) {
-
-        if (MAP_SUPPORTS_BYTES) {
-            map.putAll(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final Map<K, V> m = readEntries(reader, local);
-
-            try {
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                map.putAll(reader);
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final Map<K, V> m = readEntries(reader, local);
                 map.putAll(m);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
             }
-
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
@@ -1924,23 +1904,17 @@ class StatelessServerConnector<K, V> {
 
 
     private Work putIfAbsent(Bytes reader, Bytes writer, final long sizeLocation) {
-
-        if (MAP_SUPPORTS_BYTES) {
-            map.putIfAbsent(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K key = readKey(reader, local);
-            final V v = readValue(reader, local);
-
-            final V result;
-
-            try {
-                result = map.putIfAbsent(key, v);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                map.putIfAbsent(reader, writer);
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K key = readKey(reader, local);
+                final V v = readValue(reader, local);
+                writeValue(map.putIfAbsent(key, v), writer, local);
             }
-
-            writeValue(result, writer, local);
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
@@ -1948,23 +1922,18 @@ class StatelessServerConnector<K, V> {
 
 
     private Work replace(Bytes reader, Bytes writer, final long sizeLocation) {
-
-        if (MAP_SUPPORTS_BYTES) {
-            map.replaceKV(reader, writer);
-        } else {
-            final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
-            final K k = readKey(reader, local);
-            final V v = readValue(reader, local);
-            final V replaced;
-
-            try {
-                replaced = map.replace(k, v);
-            } catch (Throwable e) {
-                return sendException(writer, sizeLocation, e);
+        try {
+            if (MAP_SUPPORTS_BYTES) {
+                map.replaceKV(reader, writer);
+            } else {
+                final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
+                final K k = readKey(reader, local);
+                final V v = readValue(reader, local);
+                writeValue(map.replace(k, v), writer, local);
             }
-            writeValue(replaced, writer, local);
+        } catch (Throwable e) {
+            return sendException(writer, sizeLocation, e);
         }
-
         writeSizeAndFlags(sizeLocation, false, writer);
         return null;
     }

src/main/java/net/openhft/chronicle/map/UdpReplicator.java

@@ -60,7 +60,7 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
         start();
     }
 
-    private static class UdpSocketChannelEntryReader implements EntryReader {
+    private class UdpSocketChannelEntryReader implements EntryReader {
 
         private final Replica.EntryExternalizable externalizable;
         private final ByteBuffer in;
@@ -113,7 +113,7 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
             if (out.remaining() != size)
                 return;
 
-            externalizable.readExternalEntry(out);
+            externalizable.readExternalEntry(copies, segmentState, out);
         }
 
     }