A {@code ReplicationChannel} could be used to replicate only one {@link ChronicleHash} - * instance. + *
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 - * + * This class holds all configurations of * multicontainer replication, 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 - * + * 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 * the corresponding section in ChronicleMap manual for more information. * *
Create instances of this class using this pattern:
{@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
- * [0, {@link #maxNumberOfChannels()}) 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
[0, {@link
+ * #maxNumberOfChannels()}) 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 {
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.
*
* 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.
*
*
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 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 map;
+ int maxEntrySize = 10 * 1024;
- public FindMapByName(byte identifier) throws IOException {
+ private final AtomicInteger nextFreeChannel = new AtomicInteger(1);
+ private final Map 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 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
+ * @return a new instance of ChronicleMap
+ * @throws IllegalArgumentException
+ * @throws IOException
+ * @see ChronicleMapBuilder#createPersistedTo(File file)
+ */
+ public 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 nodeDiscoveryEventListenerAtomicReference =
new AtomicReference();
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 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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, Closeable {
return null;
} else
// get the data back from the server
- return readKey(sizeLocation);
+ return readValue(sizeLocation);
}
@@ -456,7 +456,7 @@ class StatelessChronicleMap implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, 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 implements ChronicleMap, Closeable {
assert buffer.capacity() == size;
assert buffer.capacity() == bytes.capacity();
assert bytes.limit() == bytes.capacity();
-
+ bytes.position(start);
}
@@ -964,6 +960,8 @@ class StatelessChronicleMap implements ChronicleMap, 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 implements ChronicleMap, 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 {
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 builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
+ final ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
.minSegments(2)
.name("map1")
.removeReturnsNull(true);
@@ -46,11 +53,16 @@ public class MapByNameTest {
findMapByName.add(builder);
}
- ChronicleMap map = findMapByName.get("map1").create();
+ final ChronicleMap 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 serverMap = ChronicleMapBuilder.of(String.class, Map
+ .class)
+ .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) {
+
+ try (ChronicleMap 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 data = new HashMap();
+
+ for (int i = 0; i < 60; i++) {
+ data.put("" + i, new String(new char[10]));
+ }
+
+ try (ChronicleMap serverMap = ChronicleMapBuilder.of(String.class, Map
+ .class)
+ .entrySize(1024)
+ .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) {
+
+ try (ChronicleMap 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 serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class)