From f854e963616bf24d266de66f1461577ddeea0876 Mon Sep 17 00:00:00 2001
From: Rob Austin <rob.austin@boundedbuffer>
Date: Mon, 17 Nov 2014 19:31:56 +0000
Subject: [PATCH] HCOLL-213 - started to integrate MapByName and improved the
 stateless clients handling of large values

---
 .../hash/replication/ReplicationChannel.java  |  28 +++--
 .../hash/replication/ReplicationHub.java      |  78 +++++++------
 .../chronicle/map/ChannelProvider.java        |  11 +-
 .../openhft/chronicle/map/FindMapByName.java  |  81 +++++++++----
 .../openhft/chronicle/map/NodeDiscovery.java  |  40 +++++--
 .../chronicle/map/StatelessChronicleMap.java  | 107 ++++++++++++------
 .../openhft/chronicle/map/TcpReplicator.java  |   3 +-
 .../openhft/chronicle/map/MapByNameTest.java  |  20 +++-
 .../chronicle/map/StatelessClientTest.java    |  61 +++++++++-
 9 files changed, 288 insertions(+), 141 deletions(-)

diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java
index 7e0e48d44..7c1efd0af 100644
--- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java
+++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java
@@ -22,33 +22,31 @@ import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
 import org.jetbrains.annotations.NotNull;
 
 /**
- * A token which should be created via {@link ReplicationHub#createChannel(short)} call and passed
- * to {@link ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)} method, to
- * establish a channel replication. See
- * <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
+ * A token which should be created via {@link ReplicationHub#createChannel(int)} call and passed to {@link
+ * ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)} method, to establish a channel
+ * replication. See <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
  * the corresponding section in ChronicleMap manual</a> for more information.
  *
- * <p>A {@code ReplicationChannel} could be used to replicate only one {@link ChronicleHash}
- * instance.
+ * <p>A {@code ReplicationChannel} could be used to replicate only one {@link ChronicleHash} instance.
  *
  * @see ReplicationHub
- * @see ReplicationHub#createChannel(short)
+ * @see ReplicationHub#createChannel(int)
  */
 public final class ReplicationChannel {
     private ReplicationHub hub;
-    private final short channelId;
+    private final int channelId;
 
-    ReplicationChannel(ReplicationHub hub, short channelId) {
+    ReplicationChannel(ReplicationHub hub, int channelId) {
         this.hub = hub;
         this.channelId = channelId;
     }
 
     /**
      * Returns the {@link ReplicationHub} on which this {@code ReplicationChannel} was {@linkplain
-     * ReplicationHub#createChannel(short) created}.
+     * ReplicationHub#createChannel(int) created}.
      *
      * @return the {@code ReplicationHub} to which this {@code ReplicationChannel} belongs
-     * @see ReplicationHub#createChannel(short)
+     * @see ReplicationHub#createChannel(int)
      */
     @NotNull
     public ReplicationHub hub() {
@@ -56,13 +54,13 @@ public final class ReplicationChannel {
     }
 
     /**
-     * Returns the identifier of this channel, with which was {@linkplain
-     * ReplicationHub#createChannel(short) created}.
+     * Returns the identifier of this channel, with which was {@linkplain ReplicationHub#createChannel(int)
+     * created}.
      *
      * @return the identifier of this channel
-     * @see ReplicationHub#createChannel(short)
+     * @see ReplicationHub#createChannel(int)
      */
-    public short channelId() {
+    public int channelId() {
         return channelId;
     }
 }
diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
index 18aaac2b9..fb2a1c5db 100644
--- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
+++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
@@ -24,14 +24,12 @@ import net.openhft.chronicle.set.ChronicleSet;
 import org.jetbrains.annotations.NotNull;
 
 /**
- * This class holds all configurations of
- * <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
+ * This class holds all configurations of <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
  * multicontainer replication</a>, which is usable, when you want to replicate several {@linkplain
- * ChronicleHash ChronicleHashes} ({@linkplain ChronicleMap maps}, {@linkplain ChronicleSet sets},
- * etc.) between same servers. {@code ReplicationHub} allows to share TCP/UDP connection, sockets,
- * buffers, worker threads for that, considerably reducing resources usage and increasing cumulative
- * (for all replicated containers) replication throughput. See
- * <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
+ * ChronicleHash ChronicleHashes} ({@linkplain ChronicleMap maps}, {@linkplain ChronicleSet sets}, etc.)
+ * between same servers. {@code ReplicationHub} allows to share TCP/UDP connection, sockets, buffers, worker
+ * threads for that, considerably reducing resources usage and increasing cumulative (for all replicated
+ * containers) replication throughput. See <a href="https://github.com/OpenHFT/Chronicle-Map#multiple-chronicle-maps---network-distributed">
  * the corresponding section in ChronicleMap manual</a> for more information.
  *
  * <p>Create instances of this class using this pattern: <pre>{@code
@@ -67,14 +65,14 @@ public final class ReplicationHub extends AbstractReplication {
     private ReplicationHub(byte localIdentifier, Builder builder) {
         super(localIdentifier, builder);
         maxEntrySize = builder.maxEntrySize;
-        channels = new ReplicationChannel[builder.maxNumberOfChronicles];
+        channels = new ReplicationChannel[builder.maxNumberOfChannels];
     }
 
     /**
      * Returns the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of {@link
-     * ChronicleHash}es, replicated by this {@code ReplicationHub}, could have. {@code
-     * RuntimeException} is thrown on attempt to replicate with this {@code ReplicationHub}
-     * a {@code ChronicleHash} with larger entry size.
+     * ChronicleHash}es, replicated by this {@code ReplicationHub}, could have. {@code RuntimeException} is
+     * thrown on attempt to replicate with this {@code ReplicationHub} a {@code ChronicleHash} with larger
+     * entry size.
      *
      * @return the maximum entry size
      * @see Builder#maxEntrySize(int)
@@ -84,35 +82,34 @@ public final class ReplicationHub extends AbstractReplication {
     }
 
     /**
-     * Returns the maximum number of channels could be {@linkplain #createChannel(short) created}
-     * for this {@code ReplicationHub}.
+     * Returns the maximum number of channels could be {@linkplain #createChannel(int) created} for this
+     * {@code ReplicationHub}.
      *
      * @return the maximum number of channels
      * @see Builder#maxNumberOfChannels()
-     * @see #createChannel(short)
+     * @see #createChannel(int)
      */
     public int maxNumberOfChannels() {
         return channels.length;
     }
 
     /**
-     * Creates a new {@link ReplicationChannel} in this {@code ReplicationHub} with the given
-     * identifier. Identifier shouldn't be lesser than zero and greater or equal to
-     * {@code maxNumberOfChannels() - 1}. On a {@code ReplicationHub} instance,
-     * {@code ReplicationChannel} could be created only once for each possible {@code channelId}
-     * value.
+     * Creates a new {@link ReplicationChannel} in this {@code ReplicationHub} with the given identifier.
+     * Identifier shouldn't be lesser than zero and greater or equal to {@code maxNumberOfChannels() - 1}. On
+     * a {@code ReplicationHub} instance, {@code ReplicationChannel} could be created only once for each
+     * possible {@code channelId} value.
      *
-     * @param channelId the identifier of the channel. Should be equal for replicated containers
-     *                  on different nodes (servers)
-     * @return a new {@code ReplicationChannel} instance, in fact just incapsulating the given
-     *         {@code channelId} for this {@code ReplicationHub}
-     * @throws IllegalArgumentException if the specified {@code channelId} is out of<br>
-     *         <code>[0, {@link #maxNumberOfChannels()})</code> range
-     * @throws IllegalStateException if {@code ReplicationChannel} with the specified
-     *         {@code channelId} has already been acquired on this {@code ReplicationHub}
+     * @param channelId the identifier of the channel. Should be equal for replicated containers on different
+     *                  nodes (servers)
+     * @return a new {@code ReplicationChannel} instance, in fact just incapsulating the given {@code
+     * channelId} for this {@code ReplicationHub}
+     * @throws IllegalArgumentException if the specified {@code channelId} is out of<br> <code>[0, {@link
+     *                                  #maxNumberOfChannels()})</code> range
+     * @throws IllegalStateException    if {@code ReplicationChannel} with the specified {@code channelId} has
+     *                                  already been acquired on this {@code ReplicationHub}
      * @see ReplicationChannel
      */
-    public synchronized ReplicationChannel createChannel(short channelId) {
+    public synchronized ReplicationChannel createChannel(int channelId) {
         if (channelId < 0)
             throw new IllegalArgumentException("channelId should be positive");
         if (channelId >= maxNumberOfChannels())
@@ -131,41 +128,42 @@ public final class ReplicationHub extends AbstractReplication {
      */
     public static final class Builder extends AbstractReplication.Builder<ReplicationHub, Builder> {
         private int maxEntrySize = 1024;
-        private int maxNumberOfChronicles = 128;
+        private int maxNumberOfChannels = 128;
 
-        private Builder() {}
+        private Builder() {
+        }
 
         /**
-         * Configures the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of
-         * {@link ChronicleHash}es, replicated via {@code ReplicationHub}s, created by this builder.
+         * Configures the maximum {@linkplain ChronicleHashBuilder#entrySize(int) entry size} of {@link
+         * ChronicleHash}es, replicated via {@code ReplicationHub}s, created by this builder.
          *
          * <p>Default value is {@code 1024}.
          *
-         * @param maxEntrySize {@link ReplicationHub#maxEntrySize()} of {@link ReplicationHub}s,
-         *                     created by this builder
+         * @param maxEntrySize {@link ReplicationHub#maxEntrySize()} of {@link ReplicationHub}s, created by
+         *                     this builder
          * @return this builder object back, for chaining
          * @see ReplicationHub#maxEntrySize()
          */
         @NotNull
         public Builder maxEntrySize(int maxEntrySize) {
-            this.maxEntrySize = maxEntrySize;
+            this.maxEntrySize = maxEntrySize * 64;
             return this;
         }
 
         /**
-         * Configures the maximum number of channels could be {@linkplain #createChannel(short)}
-         * created for {@code ReplicationHub}s, created by this builder.
+         * Configures the maximum number of channels could be {@linkplain #createChannel(int)} created for
+         * {@code ReplicationHub}s, created by this builder.
          *
          * <p>Default value is {@code 128}.
          *
-         * @param maxNumberOfChannels {@link ReplicationHub#maxNumberOfChannels()} of
-         *                            {@link ReplicationHub}s, created by this builder
+         * @param maxNumberOfChannels {@link ReplicationHub#maxNumberOfChannels()} of {@link ReplicationHub}s,
+         *                            created by this builder
          * @return this builder object back, for chaining
          * @see ReplicationHub#maxNumberOfChannels()
          */
         @NotNull
         public Builder maxNumberOfChannels(int maxNumberOfChannels) {
-            this.maxNumberOfChronicles = maxNumberOfChannels;
+            this.maxNumberOfChannels = maxNumberOfChannels;
             return this;
         }
 
diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
index 684dff7d3..dabc9d800 100644
--- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
@@ -53,6 +53,7 @@ final class ChannelProvider implements Closeable {
     private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName());
 
     static final Map<ReplicationHub, ChannelProvider> implMapping = new IdentityHashMap<>();
+
     static synchronized ChannelProvider getProvider(ReplicationHub hub) throws IOException {
         ChannelProvider channelProvider = implMapping.get(hub);
         if (channelProvider != null)
@@ -287,7 +288,7 @@ final class ChannelProvider implements Closeable {
         return new SingleThreadedDirectBitSet(bytes);
     }
 
-    public ChronicleChannel createChannel(short channel) {
+    public ChronicleChannel createChannel(int channel) {
         return new ChronicleChannel(channel);
     }
 
@@ -306,7 +307,7 @@ final class ChannelProvider implements Closeable {
         }
     }
 
-    private ByteBufferBytes toBootstrapMessage(short chronicleChannel, final long lastModificationTime) {
+    private ByteBufferBytes toBootstrapMessage(int chronicleChannel, final long lastModificationTime) {
         final ByteBufferBytes writeBuffer = new ByteBufferBytes(ByteBuffer.allocate(1 + 1 + 2 + 8));
         writeBuffer.writeByte(BOOTSTRAP_MESSAGE);
         writeBuffer.writeByte(localIdentifier);
@@ -316,7 +317,7 @@ final class ChannelProvider implements Closeable {
         return writeBuffer;
     }
 
-    private void add(short chronicleChannel,
+    private void add(int chronicleChannel,
                      Replica replica, EntryExternalizable entryExternalizable) {
         channelDataLock.writeLock().lock();
         try {
@@ -481,9 +482,9 @@ final class ChannelProvider implements Closeable {
 
     public class ChronicleChannel extends Replicator implements Closeable {
 
-        private final short chronicleChannel;
+        private final int chronicleChannel;
 
-        private ChronicleChannel(short chronicleChannel) {
+        private ChronicleChannel(int chronicleChannel) {
             this.chronicleChannel = chronicleChannel;
         }
 
diff --git a/src/main/java/net/openhft/chronicle/map/FindMapByName.java b/src/main/java/net/openhft/chronicle/map/FindMapByName.java
index 74778866a..f9d6a1068 100644
--- a/src/main/java/net/openhft/chronicle/map/FindMapByName.java
+++ b/src/main/java/net/openhft/chronicle/map/FindMapByName.java
@@ -16,49 +16,84 @@
 
 package net.openhft.chronicle.map;
 
+import net.openhft.chronicle.hash.ChronicleHash;
+import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
 import net.openhft.chronicle.hash.replication.ReplicationHub;
-import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
 
 /**
  * @author Rob Austin.
  */
 class FindMapByName {
 
-    int maxEntrySize = 10*1024;
-
-
+    public static final int MAP_BY_NAME_CHANNEL = 1;
     private final ReplicationHub replicationHub;
-    private Map<String,ChronicleMapBuilder> map;
+    int maxEntrySize = 10 * 1024;
 
-    public FindMapByName(byte identifier) throws IOException {
+    private final AtomicInteger nextFreeChannel = new AtomicInteger(1);
+    private final Map<String, ChronicleMapBuilder> map;
 
-
-        final TcpTransportAndNetworkConfig tcpConfig =
-                TcpTransportAndNetworkConfig.of(8076)
-                        .heartBeatInterval(1, TimeUnit.SECONDS);
-
-        replicationHub = ReplicationHub.builder()
-                .maxEntrySize(maxEntrySize)
-                .tcpTransportAndNetwork(tcpConfig)
-                .createWithId(identifier);
-
-        map = (Map)ChronicleMapBuilder.of(CharSequence.class, ChronicleMapBuilder.class)
+    /**
+     * @throws IOException
+     */
+    public FindMapByName(ReplicationHub replicationHub) throws IOException {
+        this.replicationHub = replicationHub;
+        this.map = (Map) of(CharSequence.class, ChronicleMapBuilder.class)
                 .entrySize(300)
                 .instance()
-                .replicatedViaChannel(replicationHub.createChannel((short) 1)).create();
+                .replicatedViaChannel(replicationHub.createChannel((short) MAP_BY_NAME_CHANNEL)).create();
     }
 
-    public ChronicleMapBuilder get(String name) throws IOException {
-        return map.get(name);
+    ChronicleMapBuilder get(String name) throws IllegalArgumentException {
+
+        ChronicleMapBuilder chronicleMapBuilder = map.get(name);
+        if (chronicleMapBuilder == null)
+            throw new IllegalArgumentException("A map of this name can not be found name=" + name);
+
+        return chronicleMapBuilder;
     }
 
 
-    public void add(ChronicleMapBuilder builder) throws IOException {
-        builder.name();
+    public <T extends ChronicleHash> T create(String name) throws IllegalArgumentException,
+            IOException {
+        return (T) replicatedViaChannel(name).create();
+    }
+
+    private ChronicleHashInstanceConfig replicatedViaChannel(String name) {
+        return toReplicatedViaChannel(get(name));
+    }
+
+    /**
+     * similar to {@link  ChronicleMapBuilder#createPersistedTo(File file)  }
+     *
+     * @param name the name of the map to be created
+     * @param file the file which will be used as shared off heap storage
+     * @param <T>
+     * @return a new instance of ChronicleMap
+     * @throws IllegalArgumentException
+     * @throws IOException
+     * @see ChronicleMapBuilder#createPersistedTo(File file)
+     */
+    public <T extends ChronicleHash> T createPersistedTo(String name, File file) throws IllegalArgumentException,
+            IOException {
+        return (T) replicatedViaChannel(name).persistedTo(file).create();
+    }
+
+    private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleMapBuilder builder) {
+        int channelId = nextFreeChannel.incrementAndGet();
+        if (channelId > replicationHub.maxNumberOfChannels())
+            throw new IllegalStateException("There are no more free channels, you can increase the number " +
+                    "of changes in the replicationHub by calling replicationHub.maxNumberOfChannels(..);");
+        return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) channelId));
+    }
+
+    public void add(ChronicleMapBuilder builder) {
         this.map.put(builder.name(), builder);
     }
 }
diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
index be5755ed2..ea5b7f2e3 100644
--- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
@@ -19,7 +19,7 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
-import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
+import net.openhft.chronicle.hash.replication.ReplicationHub;
 import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
 import net.openhft.chronicle.hash.replication.UdpTransportConfig;
 import net.openhft.lang.collection.ATSDirectBitSet;
@@ -56,15 +56,33 @@ import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
  */
 public class NodeDiscovery {
 
+    public static final int DEFAULT_UDP_BROADCAST_PORT = 8124;
+    public static final short DEFAULT_TCP_PORT = (short) 8123;
+    public static final int SERIALIZED_ENTRY_SIZE = 1024;
     private final AddressAndPort ourAddressAndPort;
     private final UdpTransportConfig udpConfig;
+
+    public static final InetAddress DEFAULT_UDP_MASK;
+
+    static {
+        InetAddress mask = null;
+        try {
+            mask = Inet4Address.getByName("255.255.255.255");
+        } catch (Exception e) {
+            mask = null;
+        }
+        DEFAULT_UDP_MASK = mask;
+
+    }
+
+
     private final DiscoveryNodeBytesMarshallable discoveryNodeBytesMarshallable;
     private final AtomicReference<NodeDiscoveryEventListener> nodeDiscoveryEventListenerAtomicReference =
             new AtomicReference<NodeDiscoveryEventListener>();
     private KnownNodes knownNodes;
 
     public NodeDiscovery() throws IOException {
-        this((short) 8124, (short) 8123, getDefaultAddress(), Inet4Address.getByName("255.255.255.255"));
+        this((short) DEFAULT_UDP_BROADCAST_PORT, DEFAULT_TCP_PORT, getDefaultAddress(), DEFAULT_UDP_MASK);
     }
 
     public NodeDiscovery(short udpBroadcastPort,
@@ -79,13 +97,13 @@ public class NodeDiscovery {
                 (knownNodes, nodeDiscoveryEventListenerAtomicReference, ourAddressAndPort);
 
         final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster
-                = new NodeDiscoveryBroadcaster(udpConfig, 1024, discoveryNodeBytesMarshallable);
+                = new NodeDiscoveryBroadcaster(udpConfig, SERIALIZED_ENTRY_SIZE, discoveryNodeBytesMarshallable);
 
         discoveryNodeBytesMarshallable.setModificationNotifier(nodeDiscoveryBroadcaster);
     }
 
 
-    public synchronized ChronicleMap<Integer, CharSequence> discoverMap() throws
+    public synchronized FindMapByName discoverMap() throws
             IOException, InterruptedException {
 
         final AtomicInteger ourProposedIdentifier = new AtomicInteger();
@@ -229,14 +247,12 @@ public class NodeDiscovery {
         LOG.info("Using Remote identifier=" + identifier);
         nodeDiscoveryEventListenerAtomicReference.set(null);
 
-        return ChronicleMapBuilder.of(Integer.class,
-                CharSequence.class)
-                .entries(20000L)
-                .replication(SingleChronicleHashReplication.builder()
-                        .tcpTransportAndNetwork(tcpConfig)
-                        .remoteNodeValidator(remoteNodeValidator)
-                        .createWithId(identifier))
-                .create();
+        final ReplicationHub replicationHub = ReplicationHub.builder()
+                .maxEntrySize(10 * 1024)
+                .tcpTransportAndNetwork(tcpConfig)
+                .createWithId(identifier);
+
+        return new FindMapByName(replicationHub);
 
     }
 
diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
index 0fd25b71e..c70480efe 100644
--- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
@@ -267,7 +267,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
         final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
         writeKey(key, local);
         writeValue(value, local);
-        return readKey(sizeLocation, local);
+        return readValue(sizeLocation, local);
     }
 
 
@@ -311,7 +311,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
         writeValue(value, local);
 
         // get the data back from the server
-        return readKey(sizeLocation, local);
+        return readValue(sizeLocation, local);
     }
 
 
@@ -394,7 +394,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
         writeKey((K) key, local);
 
         // get the data back from the server
-        return readKey(sizeLocation, local);
+        return readValue(sizeLocation, local);
 
     }
 
@@ -437,7 +437,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
             return null;
         } else
             // get the data back from the server
-            return readKey(sizeLocation);
+            return readValue(sizeLocation);
     }
 
 
@@ -456,7 +456,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
             return null;
         } else
             // get the data back from the server
-            return readKey(sizeLocation, local);
+            return readValue(sizeLocation, local);
 
     }
 
@@ -499,9 +499,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
 
                 if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) {
                     LOG.debug("resizing buffer");
-                    resizeBuffer(buffer.capacity() + 1024);
-                    bytes.limit(bytes.capacity());
-                    bytes.position(start);
+                    resizeBuffer(buffer.capacity() + maxEntrySize, start);
 
                 } else
                     throw e;
@@ -562,12 +560,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
 
             numberOfEntriesReadSoFar++;
 
+            long start = bytes.position();
+
             // putAll if a bit more complicated than the others
             // as the volume of data could cause us to have to resize our buffers
-            resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar);
+            resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start);
 
-
-            final long start = bytes.position();
             final K key = e.getKey();
 
             final Class<?> keyClass = key.getClass();
@@ -597,15 +595,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
         }
     }
 
-    private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar) {
+    private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) {
 
         final long remaining = bytes.remaining();
 
         if (remaining < maxEntrySize) {
-
-            // todo check this again as I  think it may have a bug
             long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar);
-            resizeBuffer(estimatedRequiredSize + maxEntrySize);
+            resizeBuffer((int) (estimatedRequiredSize + maxEntrySize), start);
         }
     }
 
@@ -622,7 +618,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
     }
 
 
-    void resizeBuffer(long size) {
+    private void resizeBuffer(int size, long start) {
         if (size < buffer.capacity())
             throw new IllegalStateException("it not possible to resize the buffer smaller");
 
@@ -663,7 +659,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
         assert buffer.capacity() == size;
         assert buffer.capacity() == bytes.capacity();
         assert bytes.limit() == bytes.capacity();
-
+        bytes.position(start);
     }
 
 
@@ -964,6 +960,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
             bytes = new ByteBufferBytes(allocateDirect(requiredSize));
         }
 
+        bytes.limit(bytes.capacity());
+
         // block until we have received all the byte in this chunk
         receive(size, timeoutTime);
 
@@ -1054,45 +1052,82 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
 
 
     private void writeKey(K key) {
-        try {
-            keyValueSerializer.writeKey(key, bytes);
-        } catch (IllegalArgumentException e) {
-            resizeBuffer(bytes.capacity() + maxEntrySize);
-
+        long start = bytes.position();
+        for (; ; ) {
+            try {
+                keyValueSerializer.writeKey(key, bytes);
+                return;
+            } catch (IllegalArgumentException e) {
+                resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
+            } catch (IllegalStateException e) {
+                if (e.getMessage().contains("Not enough available space"))
+                    resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
+                else
+                    throw e;
+            }
         }
     }
 
 
     private void writeKey(K key, ThreadLocalCopies local) {
-        try {
-            keyValueSerializer.writeKey(key, bytes, local);
-        } catch (IllegalArgumentException e) {
-            resizeBuffer(bytes.capacity() + maxEntrySize);
+        long start = bytes.position();
+        for (; ; ) {
+            try {
+                keyValueSerializer.writeKey(key, bytes, local);
+                return;
+            } catch (IllegalArgumentException e) {
+                resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
+            } catch (IllegalStateException e) {
+                if (e.getMessage().contains("Not enough available space"))
+                    resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
+                else
+                    throw e;
+            }
         }
     }
 
-    private V readKey(final long sizeLocation) {
+    private V readValue(final long sizeLocation) {
         return keyValueSerializer.readValue(blockingFetch(sizeLocation), null);
     }
 
-    private V readKey(final long sizeLocation, ThreadLocalCopies local) {
+    private V readValue(final long sizeLocation, ThreadLocalCopies local) {
         return keyValueSerializer.readValue(blockingFetch(sizeLocation), local);
     }
 
     private void writeValue(V value) {
-        try {
-            keyValueSerializer.writeValue(value, bytes, null);
-        } catch (IllegalArgumentException e) {
-            resizeBuffer(bytes.capacity() + maxEntrySize);
+        long start = bytes.position();
+        for (; ; ) {
+            try {
+                keyValueSerializer.writeValue(value, bytes, null);
+                return;
+            } catch (IllegalArgumentException e) {
+                resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
+            }
         }
     }
 
     private void writeValue(V value, ThreadLocalCopies local) {
-        try {
-            keyValueSerializer.writeValue(value, bytes, local);
-        } catch (IllegalArgumentException e) {
-            resizeBuffer(bytes.capacity() + maxEntrySize);
+
+
+        long start = bytes.position();
+        for (; ; ) {
+            try {
+
+                keyValueSerializer.writeValue(value, bytes, null);
+
+                return;
+            } catch (IllegalArgumentException e) {
+                resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
+            } catch (IllegalStateException e) {
+                if (e.getMessage().contains("Not enough available space"))
+                    resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
+                else
+                    throw e;
+
+            }
         }
+
+
     }
 
 }
diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
index ebdabbe1b..a1e48da95 100644
--- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
@@ -2022,9 +2022,8 @@ class KeyValueSerializer<K, V> {
         assert writer.limit() == writer.capacity();
         writer.writeBoolean(value == null);
 
-        if (value != null) {
+        if (value != null)
             valueSerializer.writeMarshallable(value, writer, threadLocal);
-        }
 
     }
 
diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java
index d0866a02a..c05827626 100644
--- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java
+++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java
@@ -17,6 +17,8 @@
 package net.openhft.chronicle.map;
 
 import junit.framework.Assert;
+import net.openhft.chronicle.hash.replication.ReplicationHub;
+import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -32,13 +34,18 @@ public class MapByNameTest {
 
     @Before
     public void setUp() throws IOException {
-        findMapByName = new FindMapByName((byte) 2);
+        final ReplicationHub replicationHub = ReplicationHub.builder()
+                .maxEntrySize(10 * 1024)
+                .tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(8243))
+                .createWithId((byte)1);
+
+        findMapByName = new FindMapByName(replicationHub);
     }
 
     @Test
-    public void testSerializingBuilder() throws IOException, InterruptedException {
+    public void testSerializingBuilder() {
         {
-            ChronicleMapBuilder<CharSequence, CharSequence> builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
+            final ChronicleMapBuilder<CharSequence, CharSequence> builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
                     .minSegments(2)
                     .name("map1")
                     .removeReturnsNull(true);
@@ -46,11 +53,16 @@ public class MapByNameTest {
             findMapByName.add(builder);
         }
 
-        ChronicleMap<CharSequence, CharSequence> map = findMapByName.get("map1").create();
+        final ChronicleMap<CharSequence, CharSequence> map = findMapByName.get("map1").create();
         map.put("hello", "world");
 
         Assert.assertEquals(map.get("hello"), "world");
     }
 
+    @Test(expected = IllegalArgumentException.class)
+    public void testSerializingBuilderUnknownMap() {
+        findMapByName.get("hello");
+    }
+
 
 }
diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
index 7cf725d54..8ba5aa015 100644
--- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
+++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
@@ -24,10 +24,7 @@ import org.junit.Test;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 
 
 /**
@@ -235,6 +232,62 @@ public class StatelessClientTest {
     }
 
 
+    @Test
+    public void testStringKeyMap() throws IOException, InterruptedException {
+
+        try (ChronicleMap<String, Map> serverMap = ChronicleMapBuilder.of(String.class, Map
+                .class)
+                .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) {
+
+            try (ChronicleMap<String, Map> statelessMap = ChronicleMapBuilder.of(String.class, Map
+                    .class)
+                    .statelessClient(new InetSocketAddress("localhost", 8056)).create()) {
+
+
+                serverMap.put("hello", Collections.singletonMap("hello", "world"));
+
+                Assert.assertEquals(Collections.singletonMap("hello", "world"), statelessMap.get("hello"));
+                Assert.assertEquals(1, statelessMap.size());
+
+            }
+
+
+        }
+
+
+    }
+
+
+    @Test
+    public void testStringKeyMapPutIntoStatelesMap() throws IOException, InterruptedException {
+
+        final Map<String, String> data = new HashMap<String, String>();
+
+        for (int i = 0; i < 60; i++) {
+            data.put("" + i, new String(new char[10]));
+        }
+
+        try (ChronicleMap<String, Map> serverMap = ChronicleMapBuilder.of(String.class, Map
+                .class)
+                .entrySize(1024)
+                .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) {
+
+            try (ChronicleMap<String, Map> statelessMap = ChronicleMapBuilder.of(String.class, Map
+                    .class)
+                    .statelessClient(new InetSocketAddress("localhost", 8056)).create()) {
+
+                statelessMap.put("hello", data);
+
+                Assert.assertEquals(data, serverMap.get("hello"));
+                Assert.assertEquals(1, statelessMap.size());
+
+            }
+
+        }
+
+    }
+
+
     @Test
     public void testBufferOverFlowPutAll() throws IOException, InterruptedException {
         try (ChronicleMap<Integer, CharSequence> serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class)