Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Rob Austin
2015-01-02 12:39:55 +00:00
parent 0b7860d554 96a0b57c66
commit f1083c617a
29 changed files with 526 additions and 529 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+4 -8
View File
@@ -428,7 +428,7 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
*
* @param replication the replication config
* @return this builder back
* @see ChronicleHashInstanceConfig#replicated(SingleChronicleHashReplication)
* @see ChronicleHashInstanceBuilder#replicated(SingleChronicleHashReplication)
* @see #replication(byte, TcpTransportAndNetworkConfig)
*/
B replication(SingleChronicleHashReplication replication);
@@ -441,13 +441,13 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* @param tcpTransportAndNetwork configuration of tcp connection and network
* @return this builder back
* @see #replication(SingleChronicleHashReplication)
* @see ChronicleHashInstanceConfig#replicated(byte, TcpTransportAndNetworkConfig)
* @see ChronicleHashInstanceBuilder#replicated(byte, TcpTransportAndNetworkConfig)
*/
B replication(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork);
B replication(byte identifier);
ChronicleHashInstanceConfig<C> instance();
ChronicleHashInstanceBuilder<C> instance();
/**
* Creates a new hash container, storing it's data in off-heap memory, not mapped to any file.
@@ -488,11 +488,7 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* @see ChronicleHash#file()
* @see ChronicleHash#close()
* @see #create()
* @see ChronicleHashInstanceConfig#persistedTo(File)
* @see ChronicleHashInstanceBuilder#persistedTo(File)
*/
C createPersistedTo(File file) throws IOException;
StatelessClientConfig<C> statelessClient(InetSocketAddress remoteAddress);
C createStatelessClient(InetSocketAddress serverAddress) throws IOException;
}
src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceConfig.java → src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java
+6 -6
View File
@@ -23,11 +23,11 @@ import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import java.io.File;
import java.io.IOException;
public interface ChronicleHashInstanceConfig<C extends ChronicleHash> {
public interface ChronicleHashInstanceBuilder<C extends ChronicleHash> {
ChronicleHashInstanceConfig<C> replicated(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork);
ChronicleHashInstanceBuilder<C> replicated(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork);
ChronicleHashInstanceConfig<C> replicated(SingleChronicleHashReplication replication);
ChronicleHashInstanceBuilder<C> replicated(SingleChronicleHashReplication replication);
/**
* Configures replication of the hash containers, created by this builder, via so called
@@ -49,11 +49,11 @@ public interface ChronicleHashInstanceConfig<C extends ChronicleHash> {
* @return this builder object back
* @see #replicated(SingleChronicleHashReplication)
*/
ChronicleHashInstanceConfig<C> replicatedViaChannel(ReplicationChannel channel);
ChronicleHashInstanceBuilder<C> replicatedViaChannel(ReplicationChannel channel);
ChronicleHashInstanceConfig<C> persistedTo(File file);
ChronicleHashInstanceBuilder<C> persistedTo(File file);
ChronicleHashInstanceConfig<C> name(String name);
ChronicleHashInstanceBuilder<C> name(String name);
C create() throws IOException;
}
src/main/java/net/openhft/chronicle/hash/StatelessClientConfig.java → src/main/java/net/openhft/chronicle/hash/ChronicleHashStatelessClientBuilder.java
+8 -5
View File
@@ -35,13 +35,16 @@ import java.util.concurrent.TimeUnit;
* consider to be a ClientProxy to {@link ChronicleMap} or {@link ChronicleSet} running on
* another host
*
* @param <C> the type of {@code ChronicleHash} accessed remotely, {@code ChronicleMap} or
* @param <H> the type of {@code ChronicleHash} accessed remotely, {@code ChronicleMap} or
* {@code ChronicleSet}
*/
public interface StatelessClientConfig<C extends ChronicleHash> {
StatelessClientConfig<C> timeout(long timeout, TimeUnit units);
public interface ChronicleHashStatelessClientBuilder<
C extends ChronicleHashStatelessClientBuilder<C, H>, H extends ChronicleHash> {
C timeout(long timeout, TimeUnit units);
StatelessClientConfig<C> name(String name);
C name(String name);
C create() throws IOException;
C tcpBufferSize(int tcpBufferSize);
H create() throws IOException;
}
src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java
+2 -2
View File
@@ -17,14 +17,14 @@
package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import org.jetbrains.annotations.NotNull;
import java.io.Serializable;
/**
* A token which should be created via {@link ReplicationHub#createChannel(int)} call and passed to {@link
* ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)} method, to establish a channel
* ChronicleHashInstanceBuilder#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.
*
src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
+2 -2
View File
@@ -18,7 +18,7 @@ package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.set.ChronicleSet;
import org.jetbrains.annotations.NotNull;
@@ -46,7 +46,7 @@ import java.util.Arrays;
* .persistedTo(myMapFile) // optional, for this example
* .create();}</pre>
*
* @see ChronicleHashInstanceConfig#replicatedViaChannel(ReplicationChannel)
* @see ChronicleHashInstanceBuilder#replicatedViaChannel(ReplicationChannel)
* @see ReplicationChannel
*/
public final class ReplicationHub extends AbstractReplication {
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+9 -90
View File
@@ -46,9 +46,7 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -99,7 +97,8 @@ import static net.openhft.lang.model.DataValueGenerator.firstPrimitiveFieldType;
* @see ChronicleSetBuilder
*/
public final class ChronicleMapBuilder<K, V> implements Cloneable,
ChronicleHashBuilder<K, ChronicleMap<K, V>, ChronicleMapBuilder<K, V>>, Serializable {
ChronicleHashBuilder<K, ChronicleMap<K, V>, ChronicleMapBuilder<K, V>>,
MapBuilder<ChronicleMapBuilder<K, V>>, Serializable {
static final byte UDP_REPLICATION_MODIFICATION_ITERATOR_ID = (byte) 128;
private static final int DEFAULT_KEY_OR_VALUE_SIZE = 120;
@@ -108,7 +107,6 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
private static final int MAX_SEGMENTS_TO_CHAISE_COMPACT_MULTI_MAPS = (1 << 20);
private static final Logger LOG =
LoggerFactory.getLogger(ChronicleMapBuilder.class.getName());
static final long DEFAULT_STATELESS_CLIENT_TIMEOUT = TimeUnit.SECONDS.toMillis(10);
private static final StringBuilder EMTRY_STRING_BUILDER = new StringBuilder();
@@ -117,9 +115,6 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
private static final int XML_SERIALIZATION = 1;
private static final int BINARY_SERIALIZATION = 2;
private List jsonConverters = new ArrayList();
private static boolean isDefined(double config) {
return !Double.isNaN(config);
}
@@ -288,30 +283,6 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
return this;
}
/**
* Used to a add a custom xStream converter, this converter may be used while chronicle maps
* serialises/deserialized and object(s) to and from JSON. You should add a custom converter if
* you wish to customise the JSON format, or support object(s) that can not be serialised into
* JSON.
*
* The method signature purposely has a loose coupling to xStream, it exposes {@code Object}
* parameter, rather than the com.thoughtworks.xstream.jsonConverters.Converter this is so that
* the the {@code ChronicleMapBuilder} class does not require xStream artifact/jars, however the
* parameter pass must be of type com.thoughtworks.xstream.jsonConverters.Converter as no other
* type is currently accepted.
*
* @param converter an xStream converter of type com.thoughtworks.xstream.jsonConverters.Converter
* @return this builder back
* @see ChronicleMap#putAll(java.io.File)
* @see ChronicleMap#putAll(java.io.File)
*/
public ChronicleMapBuilder<K, V> addJsonConverter(Object converter) {
this.jsonConverters.add(converter);
return this;
}
private double averageKeySize() {
return averageKeyOrValueSize(averageKeySize, keyBuilder);
}
@@ -417,10 +388,6 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
return this;
}
List jsonConverters() {
return this.jsonConverters;
}
static class EntrySizeInfo {
final double averageEntrySize;
final int worstAlignment;
@@ -858,22 +825,7 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
return errorListener;
}
/**
* Configures if the maps created by this builder should return {@code null} instead of previous
* mapped values on {@link ChronicleMap#put(Object, Object) ChornicleMap.put(key, value)} calls.
* <p> <p>{@link Map#put(Object, Object) Map.put()} returns the previous value, functionality
* which is rarely used but fairly cheap for {@link HashMap}. In the case, for an off heap
* collection, it has to create a new object and deserialize the data from off-heap memory. It's
* expensive for something you probably don't use. <p> <p>By default, of cause, {@code
* ChronicleMap} conforms the general {@code Map} contract and returns the previous mapped value
* on {@code put()} calls.
*
* @param putReturnsNull {@code true} if you want {@link ChronicleMap#put(Object, Object)
* ChronicleMap.put()} to not return the value that was replaced but
* instead return {@code null}
* @return this builder back
* @see #removeReturnsNull(boolean)
*/
@Override
public ChronicleMapBuilder<K, V> putReturnsNull(boolean putReturnsNull) {
this.putReturnsNull = putReturnsNull;
return this;
@@ -883,21 +835,7 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
return putReturnsNull;
}
/**
* Configures if the maps created by this builder should return {@code null} instead of the last
* mapped value on {@link ChronicleMap#remove(Object) ChronicleMap.remove(key)} calls. <p>
* <p>{@link Map#remove(Object) Map.remove()} returns the previous value, functionality which is
* rarely used but fairly cheap for {@link HashMap}. In the case, for an off heap collection, it
* has to create a new object and deserialize the data from off-heap memory. It's expensive for
* something you probably don't use. <p> <p>By default, of cause, {@code ChronicleMap} conforms
* the general {@code Map} contract and returns the mapped value on {@code remove()} calls.
*
* @param removeReturnsNull {@code true} if you want {@link ChronicleMap#remove(Object)
* ChronicleMap.remove()} to not return the value of the removed entry
* but instead return {@code null}
* @return this builder back
* @see #putReturnsNull(boolean)
*/
@Override
public ChronicleMapBuilder<K, V> removeReturnsNull(boolean removeReturnsNull) {
this.removeReturnsNull = removeReturnsNull;
return this;
@@ -1274,24 +1212,11 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
}
@Override
public ChronicleHashInstanceConfig<ChronicleMap<K, V>> instance() {
return new MapInstanceConfig<>(this.clone(), singleHashReplication, null, null, null,
public ChronicleHashInstanceBuilder<ChronicleMap<K, V>> instance() {
return new MapInstanceBuilder<>(this.clone(), singleHashReplication, null, null, null,
new AtomicBoolean(false));
}
@Override
public ChronicleMap<K, V> createStatelessClient(InetSocketAddress serverAddress)
throws IOException {
return statelessClient(serverAddress).create();
}
ChronicleMap<K, V> createStatelessClient(StatelessMapConfig<K, V> statelessClientConfig)
throws IOException {
pushingToMapEventListener();
preMapConstruction(false);
return new StatelessChronicleMap<>(statelessClientConfig, this);
}
@Override
public ChronicleMap<K, V> createPersistedTo(File file) throws IOException {
// clone() to make this builder instance thread-safe, because createWithFile() method
@@ -1300,13 +1225,6 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
return clone().createWithFile(file, singleHashReplication, null);
}
@Override
public StatelessClientConfig<ChronicleMap<K, V>> statelessClient(
InetSocketAddress remoteAddress) {
return new StatelessMapConfig<>(clone(), remoteAddress, DEFAULT_STATELESS_CLIENT_TIMEOUT,
null, new AtomicBoolean(false));
}
@Override
public ChronicleMap<K, V> create() {
// clone() to make this builder instance thread-safe, because createWithoutFile() method
@@ -1315,7 +1233,7 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
return clone().createWithoutFile(singleHashReplication, null);
}
ChronicleMap<K, V> create(MapInstanceConfig<K, V> ib) throws IOException {
ChronicleMap<K, V> create(MapInstanceBuilder<K, V> ib) throws IOException {
if (ib.file != null) {
return createWithFile(ib.file, ib.singleHashReplication, ib.channel);
} else {
@@ -1450,7 +1368,8 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
ChronicleMapBuilder<K, V> cmb = clone();
cmb.pushTo((InetSocketAddress[]) null);
for (int i = 0; i < pushToAddresses.length; i++) {
statelessClients[i] = cmb.createStatelessClient(pushToAddresses[i]);
statelessClients[i] = ChronicleMapStatelessClientBuilder.of(pushToAddresses[i])
.create();
}
eventListener =
(MapEventListener<K, V>) constructor.newInstance((Object) statelessClients);
src/main/java/net/openhft/chronicle/map/ChronicleMapStatelessClientBuilder.java
+141
View File
@@ -0,0 +1,141 @@
/*
* 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.ChronicleHashStatelessClientBuilder;
import net.openhft.lang.MemoryUnit;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* @author Rob Austin.
*/
public final class ChronicleMapStatelessClientBuilder<K, V> implements
ChronicleHashStatelessClientBuilder<ChronicleMapStatelessClientBuilder<K, V>,
ChronicleMap<K, V>>,
MapBuilder<ChronicleMapStatelessClientBuilder<K, V>> {
public static <K, V> ChronicleMapStatelessClientBuilder<K, V> of(
InetSocketAddress serverAddress) {
return new ChronicleMapStatelessClientBuilder<>(serverAddress);
}
// This method is meaningful, despite you could call of().create(), because Java 7 doesn't
// infer type parameters if case of chained calls, like this; and then you couldn't omit class
// name. Then it is
// ChronicleMapStatelessClientBuilder.<Integer, Integer>.of(addr).create()
// vs
// createClientOf(addr) // static import
// -- the second is really shorter.
/**
* Equivalent of {@code ChronicleMapStatelessClientBuilder.<K, V>of(serverAddress).create()}.
*
* @param serverAddress address of the server map
* @param <K> key type of the map
* @param <V> value type of the map
* @return stateless client of the ChronicleMap of server
*/
public static <K, V> ChronicleMap<K, V> createClientOf(InetSocketAddress serverAddress)
throws IOException {
return ChronicleMapStatelessClientBuilder.<K, V>of(serverAddress).create();
}
private final InetSocketAddress remoteAddress;
private boolean putReturnsNull = false;
private boolean removeReturnsNull = false;
private long timeoutMs = TimeUnit.SECONDS.toMillis(10);
private String name;
private int tcpBufferSize = (int) MemoryUnit.KILOBYTES.toBytes(64);
private final AtomicBoolean used = new AtomicBoolean(false);
ChronicleMapStatelessClientBuilder(InetSocketAddress remoteAddress) {
this.remoteAddress = remoteAddress;
}
InetSocketAddress remoteAddress() {
return remoteAddress;
}
@Override
public ChronicleMapStatelessClientBuilder<K, V> putReturnsNull(boolean putReturnsNull) {
this.putReturnsNull = putReturnsNull;
return this;
}
boolean putReturnsNull() {
return putReturnsNull;
}
@Override
public ChronicleMapStatelessClientBuilder<K, V> removeReturnsNull(boolean removeReturnsNull) {
this.removeReturnsNull = removeReturnsNull;
return this;
}
boolean removeReturnsNull() {
return removeReturnsNull;
}
@Override
public ChronicleMapStatelessClientBuilder<K, V> timeout(long timeout, TimeUnit units) {
this.timeoutMs = units.toMillis(timeout);
return this;
}
long timeoutMs() {
return timeoutMs;
}
@Override
public ChronicleMapStatelessClientBuilder<K, V> name(String name) {
this.name = name;
return this;
}
String name() {
return name;
}
@Override
public ChronicleMapStatelessClientBuilder<K, V> tcpBufferSize(int tcpBufferSize) {
this.tcpBufferSize = tcpBufferSize;
return this;
}
int tcpBufferSize() {
return tcpBufferSize;
}
@Override
public ChronicleMap<K, V> create() throws IOException {
if (!used.getAndSet(true)) {
return new StatelessChronicleMap<>(this);
} else {
throw new IllegalStateException(
"A stateless client has already been created using this builder. " +
"Create a new ChronicleMapStatelessClientBuilder " +
"to create a new stateless client");
}
}
}
src/main/java/net/openhft/chronicle/map/MapBuilder.java
+74
View File
@@ -0,0 +1,74 @@
/*
* 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 java.util.HashMap;
import java.util.Map;
/**
* Common configurations of {@link ChronicleMapBuilder} and
* {@link ChronicleMapStatelessClientBuilder}.
*
* @param <C> concrete builder class
*/
public interface MapBuilder<C extends MapBuilder<C>> {
/**
* Configures if the maps created by this {@code MapBuilder} should return {@code null}
* instead of previous mapped values on {@link ChronicleMap#put(Object, Object)
* ChornicleMap.put(key, value)} calls.
*
* <p>{@link Map#put(Object, Object) Map.put()} returns the previous value, functionality
* which is rarely used but fairly cheap for simple in-process, on-heap implementations like
* {@link HashMap}. But an off-heap collection has to create a new object and deserialize
* the data from off-heap memory. A collection hiding remote queries over the network should
* send the value back in addition to that. It's expensive for something you probably don't use.
*
* <p>By default, of cause, {@code ChronicleMap} conforms the general {@code Map} contract and
* returns the previous mapped value on {@code put()} calls.
*
* @param putReturnsNull {@code true} if you want {@link ChronicleMap#put(Object, Object)
* ChronicleMap.put()} to not return the value that was replaced but
* instead return {@code null}
* @return {@code MapBuilder} with this configuration applied
* @see #removeReturnsNull(boolean)
*/
C putReturnsNull(boolean putReturnsNull);
/**
* Configures if the maps created by this {@code MapBuilder} should return {@code null}
* instead of the last mapped value on {@link ChronicleMap#remove(Object)
* ChronicleMap.remove(key)} calls.
*
* <p>{@link Map#remove(Object) Map.remove()} returns the previous value, functionality which is
* rarely used but fairly cheap for simple in-process, on-heap implementations like {@link
* HashMap}. But an off-heap collection has to create a new object and deserialize the data
* from off-heap memory. A collection hiding remote queries over the network should send
* the value back in addition to that. It's expensive for something you probably don't use.
*
* <p>By default, of cause, {@code ChronicleMap} conforms the general {@code Map} contract and
* returns the mapped value on {@code remove()} calls.
*
* @param removeReturnsNull {@code true} if you want {@link ChronicleMap#remove(Object)
* ChronicleMap.remove()} to not return the value of the removed entry
* but instead return {@code null}
* @return {@code MapBuilder} with this configuration applied
* @see #putReturnsNull(boolean)
*/
C removeReturnsNull(boolean removeReturnsNull);
}
src/main/java/net/openhft/chronicle/map/MapInstanceConfig.java → src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java
+26 -20
View File
@@ -16,7 +16,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import net.openhft.chronicle.hash.replication.ReplicationChannel;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
@@ -26,21 +26,21 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.concurrent.atomic.AtomicBoolean;
final class MapInstanceConfig<K, V>
implements ChronicleHashInstanceConfig<ChronicleMap<K, V>>, Serializable{
final class MapInstanceBuilder<K, V>
implements ChronicleHashInstanceBuilder<ChronicleMap<K, V>>, Serializable{
final ChronicleMapBuilder<K, V> mapBuilder;
final SingleChronicleHashReplication singleHashReplication;
final ReplicationChannel channel;
final File file;
final String name;
SingleChronicleHashReplication singleHashReplication;
ReplicationChannel channel;
File file;
String name;
final AtomicBoolean used;
MapInstanceConfig(ChronicleMapBuilder<K, V> mapBuilder,
SingleChronicleHashReplication singleHashReplication,
ReplicationChannel channel,
File file, String name, AtomicBoolean used) {
MapInstanceBuilder(ChronicleMapBuilder<K, V> mapBuilder,
SingleChronicleHashReplication singleHashReplication,
ReplicationChannel channel,
File file, String name, AtomicBoolean used) {
this.mapBuilder = mapBuilder;
this.singleHashReplication = singleHashReplication;
this.channel = channel;
@@ -50,30 +50,36 @@ final class MapInstanceConfig<K, V>
}
@Override
public MapInstanceConfig<K, V> replicated(
public MapInstanceBuilder<K, V> replicated(
byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) {
return replicated(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpTransportAndNetwork).createWithId(identifier));
}
@Override
public MapInstanceConfig<K, V> replicated(SingleChronicleHashReplication replication) {
return new MapInstanceConfig<>(mapBuilder, replication, null, file, name, used);
public MapInstanceBuilder<K, V> replicated(SingleChronicleHashReplication replication) {
singleHashReplication = replication;
channel = null;
return this;
}
@Override
public MapInstanceConfig<K, V> replicatedViaChannel(ReplicationChannel channel) {
return new MapInstanceConfig<>(mapBuilder, null, channel, file, name, used);
public MapInstanceBuilder<K, V> replicatedViaChannel(ReplicationChannel channel) {
singleHashReplication = null;
this.channel = channel;
return this;
}
@Override
public MapInstanceConfig<K, V> persistedTo(File file) {
return new MapInstanceConfig<>(mapBuilder, singleHashReplication, channel, file, name, used);
public MapInstanceBuilder<K, V> persistedTo(File file) {
this.file = file;
return this;
}
@Override
public MapInstanceConfig<K, V> name(String name) {
return new MapInstanceConfig<>(mapBuilder, singleHashReplication, channel, file, name, used);
public MapInstanceBuilder<K, V> name(String name) {
this.name = name;
return this;
}
@Override
src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java
+14 -14
View File
@@ -18,7 +18,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import net.openhft.chronicle.hash.replication.ReplicationChannel;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import org.slf4j.Logger;
@@ -39,7 +39,7 @@ class ReplicationHubFindByName<K> implements FindByName {
public static final int MAP_BY_NAME_CHANNEL = 1;
private final AtomicInteger nextFreeChannel = new AtomicInteger(2);
private final Map<String, MapInstanceConfig> map;
private final Map<String, MapInstanceBuilder> map;
private final ReplicationHub replicationHub;
/**
@@ -52,12 +52,12 @@ class ReplicationHubFindByName<K> implements FindByName {
this.replicationHub = replicationHub;
ReplicationChannel channel = replicationHub.createChannel((short) MAP_BY_NAME_CHANNEL);
final MapEventListener<CharSequence, MapInstanceConfig> listener =
new MapEventListener<CharSequence, MapInstanceConfig>() {
final MapEventListener<CharSequence, MapInstanceBuilder> listener =
new MapEventListener<CharSequence, MapInstanceBuilder>() {
// creates a map based on the details that are sent to the map of builders
@Override
public void onPut(CharSequence key, MapInstanceConfig value,
MapInstanceConfig replacedValue) {
public void onPut(CharSequence key, MapInstanceBuilder value,
MapInstanceBuilder replacedValue) {
super.onPut(key, value, replacedValue);
boolean added = replacedValue == null;
if (!added || value == null)
@@ -80,7 +80,7 @@ class ReplicationHubFindByName<K> implements FindByName {
};
this.map = (Map) ChronicleMapBuilder
.of(CharSequence.class, MapInstanceConfig.class)
.of(CharSequence.class, MapInstanceBuilder.class)
.averageKeySize(10)
.averageValueSize(4000)
.entries(128)
@@ -93,7 +93,7 @@ class ReplicationHubFindByName<K> implements FindByName {
}
public <T extends ChronicleHash> T create(
MapInstanceConfig<CharSequence, CharSequence> config)
MapInstanceBuilder<CharSequence, CharSequence> config)
throws IOException, TimeoutException, InterruptedException {
int withChannelId = nextFreeChannel.incrementAndGet();
@@ -109,11 +109,11 @@ class ReplicationHubFindByName<K> implements FindByName {
return (T) get(name).create();
}
MapInstanceConfig get(String name) throws IllegalArgumentException,
MapInstanceBuilder get(String name) throws IllegalArgumentException,
TimeoutException,
InterruptedException {
MapInstanceConfig config = waitTillEntryReceived(5000, name);
MapInstanceBuilder config = waitTillEntryReceived(5000, name);
if (config == null)
throw new IllegalArgumentException("A map name=" + name + " can not be found.");
@@ -126,11 +126,11 @@ class ReplicationHubFindByName<K> implements FindByName {
* @param timeOutMs timeout in milliseconds
* @throws InterruptedException
*/
private MapInstanceConfig waitTillEntryReceived(final int timeOutMs, String name)
private MapInstanceBuilder waitTillEntryReceived(final int timeOutMs, String name)
throws TimeoutException, InterruptedException {
int t = 0;
for (; t < timeOutMs; t++) {
MapInstanceConfig config = map.get(name);
MapInstanceBuilder config = map.get(name);
if (config != null)
return config;
@@ -156,7 +156,7 @@ class ReplicationHubFindByName<K> implements FindByName {
return (T) get(name).persistedTo(file).create();
}
private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleMapBuilder builder) {
private ChronicleHashInstanceBuilder 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 " +
@@ -164,7 +164,7 @@ class ReplicationHubFindByName<K> implements FindByName {
return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) channelId));
}
private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleHashBuilder<K, ?, ?> builder, int
private ChronicleHashInstanceBuilder toReplicatedViaChannel(ChronicleHashBuilder<K, ?, ?> builder, int
withChannelId) {
return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) withChannelId));
}
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+46 -50
View File
@@ -43,6 +43,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import static java.nio.ByteBuffer.allocateDirect;
import static java.util.Collections.emptyList;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID;
import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*;
@@ -59,7 +60,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private final ByteBuffer connectionOutBuffer = ByteBuffer.wrap(connectionByte);
private final String name;
private final List jsonConverters;
private ByteBuffer outBuffer;
private ByteBufferBytes outBytes;
@@ -78,11 +78,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
@Nullable
private CloseablesManager closeables;
@NotNull
private final StatelessMapConfig config;
private final Class<K> kClass;
private final Class<V> vClass;
private final InetSocketAddress remoteAddress;
private final long timeoutMs;
private final int tcpBufferSize;
private Class<K> kClass;
private Class<V> vClass;
private boolean putReturnsNull;
private boolean removeReturnsNull;
@@ -144,12 +145,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
@NotNull
private final AtomicLong transactionID = new AtomicLong(0);
StatelessChronicleMap(@NotNull final StatelessMapConfig config,
@NotNull final ChronicleMapBuilder chronicleMapBuilder) {
this.config = config;
this.jsonConverters = chronicleMapBuilder.jsonConverters();
StatelessChronicleMap(@NotNull final ChronicleMapStatelessClientBuilder config) {
this.remoteAddress = config.remoteAddress();
this.timeoutMs = config.timeoutMs();
this.tcpBufferSize = config.tcpBufferSize();
this.name = config.name();
this.putReturnsNull = config.putReturnsNull();
this.removeReturnsNull = config.removeReturnsNull();
outBuffer = allocateDirect(128).order(ByteOrder.nativeOrder());
outBytes = new ByteBufferBytes(outBuffer.slice());
@@ -157,22 +159,19 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
inBuffer = allocateDirect(128).order(ByteOrder.nativeOrder());
inBytes = new ByteBufferBytes(inBuffer.slice());
this.vClass = chronicleMapBuilder.valueBuilder.eClass;
this.kClass = chronicleMapBuilder.keyBuilder.eClass;
this.name = config.name();
attemptConnect(config.remoteAddress());
attemptConnect(remoteAddress);
checkVersion();
loadKeyValueSerializers();
this.putReturnsNull = chronicleMapBuilder.putReturnsNull();
this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull();
}
private void loadKeyValueSerializers() {
final SerializationBuilder keyBuilder = fetchObject(SerializationBuilder.class, KEY_BUILDER);
final SerializationBuilder valueBuilder = fetchObject(SerializationBuilder.class, VALUE_BUILDER);
final SerializationBuilder keyBuilder =
fetchObject(SerializationBuilder.class, KEY_BUILDER);
kClass = keyBuilder.eClass;
final SerializationBuilder valueBuilder =
fetchObject(SerializationBuilder.class, VALUE_BUILDER);
vClass = valueBuilder.eClass;
keyReaderWithSize = new ReaderWithSize(keyBuilder);
keyWriterWithSize = new WriterWithSize(keyBuilder, outBufferResizer);
@@ -198,12 +197,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
@Override
public void getAll(File toFile) throws IOException {
JsonSerializer.getAll(toFile, this, jsonConverters);
JsonSerializer.getAll(toFile, this, emptyList());
}
@Override
public void putAll(File fromFile) throws IOException {
JsonSerializer.putAll(fromFile, this, jsonConverters);
JsonSerializer.putAll(fromFile, this, emptyList());
}
@Override
@@ -255,7 +254,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
try {
result = AbstractChannelReplicator.openSocketChannel(closeables);
if (!result.connect(config.remoteAddress())) {
if (!result.connect(remoteAddress)) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
@@ -361,7 +360,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
connectionByte[0] = STATELESS_CLIENT_IDENTIFIER;
this.connectionOutBuffer.clear();
long timeoutTime = System.currentTimeMillis() + config.timeoutMs();
long timeoutTime = System.currentTimeMillis() + timeoutMs;
// write a single byte
while (connectionOutBuffer.hasRemaining()) {
@@ -619,8 +618,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private Bytes resizeBufferOutBuffer(int newCapacity, long start) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
@@ -720,7 +717,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long sizeLocation = writeEventAnSkip(VALUES);
final long startTime = System.currentTimeMillis();
timeoutTime = System.currentTimeMillis() + config.timeoutMs();
timeoutTime = System.currentTimeMillis() + timeoutMs;
transactionId = send(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
@@ -764,7 +761,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
try {
final long sizeLocation = writeEventAnSkip(ENTRY_SET);
final long startTime = System.currentTimeMillis();
timeoutTime = System.currentTimeMillis() + config.timeoutMs();
timeoutTime = System.currentTimeMillis() + timeoutMs;
transactionId = send(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
@@ -818,7 +815,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
try {
final long sizeLocation = writeEventAnSkip(ENTRY_SET);
final long startTime = System.currentTimeMillis();
timeoutTime = System.currentTimeMillis() + config.timeoutMs();
timeoutTime = System.currentTimeMillis() + timeoutMs;
transactionId = send(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
@@ -871,7 +868,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
outBytesLock.unlock();
}
final long startTime = System.currentTimeMillis();
final long timeoutTime = startTime + config.timeoutMs();
final long timeoutTime = startTime + timeoutMs;
final long transactionId;
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
@@ -940,7 +937,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
try {
final long sizeLocation = writeEventAnSkip(KEY_SET);
final long startTime = System.currentTimeMillis();
timeoutTime = startTime + config.timeoutMs();
timeoutTime = startTime + timeoutMs;
transactionId = send(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
@@ -980,7 +977,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
assert !outBytesLock.isHeldByCurrentThread();
final long timeoutTime = startTime + this.config.timeoutMs();
final long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock.lock();
@@ -1094,12 +1091,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
assert !inBytesLock.isHeldByCurrentThread();
long transactionId = nextUniqueTransaction(startTime);
final long timeoutTime = startTime + this.config.timeoutMs();
final long timeoutTime = startTime + this.timeoutMs;
try {
for (; ; ) {
if (clientChannel == null) {
lazyConnect(config.timeoutMs(), config.remoteAddress());
lazyConnect(timeoutMs, remoteAddress);
}
try {
@@ -1115,7 +1112,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
} catch (@NotNull java.nio.channels.ClosedChannelException | ClosedConnectionException e) {
checkTimeout(timeoutTime);
lazyConnect(config.timeoutMs(), config.remoteAddress());
lazyConnect(timeoutMs, remoteAddress);
}
}
} catch (IOException e) {
@@ -1187,7 +1184,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) {
stes.remove(i);
}
InetSocketAddress address = config.remoteAddress();
InetSocketAddress address = remoteAddress;
stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort()));
StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace();
//noinspection ManualArrayToCollectionCopy
@@ -1336,7 +1333,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
long lagestEntrySoFar = 0;
long largestEntrySoFar = 0;
private long limitOfLast = 0;
private void writeBytesToSocket(long timeoutTime) throws IOException {
@@ -1347,7 +1344,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// if we have other threads waiting to send and the buffer is not full, let the other threads
// write to the buffer
if (outBytesLock.hasQueuedThreads() &&
outBytes.position() + lagestEntrySoFar <= config.tcpBufferSize()) {
outBytes.position() + largestEntrySoFar <= tcpBufferSize) {
return;
}
@@ -1355,8 +1352,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
outBuffer.position(0);
int sizeOfThisMessage = (int) (outBuffer.limit() - limitOfLast);
if (lagestEntrySoFar < sizeOfThisMessage)
lagestEntrySoFar = sizeOfThisMessage;
if (largestEntrySoFar < sizeOfThisMessage)
largestEntrySoFar = sizeOfThisMessage;
limitOfLast = outBuffer.limit();
@@ -1371,8 +1368,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// if we have queued threads then we don't have to write all the bytes as the other
// threads will write the remains bytes.
if (outBuffer.remaining() > 0 && outBytesLock.hasQueuedThreads() &&
outBuffer.remaining() + lagestEntrySoFar <= config.tcpBufferSize()) {
outBuffer.remaining() + largestEntrySoFar <= tcpBufferSize) {
LOG.debug("continuing - without all the data being written to the buffer as " +
"it will be written by the next thread");
@@ -1475,7 +1471,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private V readValue(long transactionId, long startTime, ThreadLocalCopies copies, V usingValue) {
assert !outBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock.lock();
@@ -1495,7 +1491,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private <O> O readObject(long transactionId, long startTime) {
assert !outBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock.lock();
@@ -1509,7 +1505,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private boolean readBoolean(long transactionId, long startTime) {
assert !outBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock.lock();
@@ -1524,7 +1520,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private int readInt(long transactionId, long startTime) {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock.lock();
@@ -1704,7 +1700,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void readVoid(long transactionId, long startTime) {
long timeoutTime = startTime + this.config.timeoutMs();
long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock.lock();
@@ -1729,7 +1725,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
outBytesLock.unlock();
}
long timeoutTime = startTime + this.config.timeoutMs();
long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock.lock();
src/main/java/net/openhft/chronicle/map/StatelessMapConfig.java
-96
View File
@@ -1,96 +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.chronicle.hash.StatelessClientConfig;
import net.openhft.lang.MemoryUnit;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* @author Rob Austin.
*/
class StatelessMapConfig<K, V> implements StatelessClientConfig<ChronicleMap<K, V>> {
private final ChronicleMapBuilder<K, V> mapBuilder;
private final InetSocketAddress remoteAddress;
private final long timeoutMs;
private final String name;
private int tcpBufferSize = (int) MemoryUnit.KILOBYTES.toBytes(64);
private final AtomicBoolean used;
StatelessMapConfig(ChronicleMapBuilder<K, V> mapBuilder,
InetSocketAddress remoteAddress, long timeoutMs,
String name, AtomicBoolean used) {
this.mapBuilder = mapBuilder;
this.remoteAddress = remoteAddress;
this.timeoutMs = timeoutMs;
this.name = name;
this.used = used;
}
public long timeoutMs() {
return timeoutMs;
}
public InetSocketAddress remoteAddress() {
return remoteAddress;
}
@Override
public StatelessMapConfig<K, V> timeout(long timeout, TimeUnit units) {
return new StatelessMapConfig<>(mapBuilder, remoteAddress, units.toMillis(timeout),
name, used);
}
@Override
public StatelessClientConfig<ChronicleMap<K, V>> name(String name) {
return new StatelessMapConfig<>(mapBuilder, remoteAddress, timeoutMs, name, used);
}
String name() {
return name;
}
@Override
public ChronicleMap<K, V> create() throws IOException {
if (!used.getAndSet(true)) {
return mapBuilder.createStatelessClient(this);
} else {
throw new IllegalStateException(
"A stateless client has already been created using this config chain. " +
"Create a new StatelessClientConfig (builder.statelessClient()) " +
"to create a new stateless client");
}
}
public int tcpBufferSize() {
return tcpBufferSize;
}
public StatelessMapConfig tcpBufferSize(int packetSize) {
this.tcpBufferSize = packetSize;
return this;
}
}
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+4 -4
View File
@@ -51,6 +51,7 @@ import java.util.concurrent.TimeUnit;
import static java.lang.Long.numberOfTrailingZeros;
import static java.lang.Math.max;
import static java.lang.Thread.currentThread;
import static java.util.Collections.emptyList;
import static net.openhft.chronicle.map.Asserts.assertNotNull;
import static net.openhft.chronicle.map.ChronicleMapBuilder.greatestCommonDivisor;
import static net.openhft.lang.MemoryUnit.*;
@@ -101,7 +102,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
final Class nativeValueClass;
final MultiMapFactory multiMapFactory;
final int maxChunksPerEntry;
private final List jsonConverters;
transient Provider<BytesReader<K>> keyReaderProvider;
transient Provider<KI> keyInteropProvider;
transient Provider<BytesReader<V>> valueReaderProvider;
@@ -119,7 +120,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
keySizeMarshaller = keyBuilder.sizeMarshaller();
originalKeyReader = keyBuilder.reader();
dataFileVersion = BuildVersion.version();
jsonConverters = builder.jsonConverters();
originalKeyInterop = (KI) keyBuilder.interop();
@@ -672,12 +672,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
@Override
public synchronized void getAll(File toFile) throws IOException {
JsonSerializer.getAll(toFile, this, jsonConverters);
JsonSerializer.getAll(toFile, this, emptyList());
}
@Override
public synchronized void putAll(File fromFile) throws IOException {
JsonSerializer.putAll(fromFile, this, jsonConverters);
JsonSerializer.putAll(fromFile, this, emptyList());
}
@Override
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+3 -17
View File
@@ -20,8 +20,7 @@ package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashErrorListener;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.StatelessClientConfig;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.TimeProvider;
@@ -38,7 +37,6 @@ import org.jetbrains.annotations.NotNull;
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
/**
@@ -289,20 +287,8 @@ public final class ChronicleSetBuilder<E>
}
@Override
public StatelessClientConfig<ChronicleSet<E>> statelessClient(
InetSocketAddress remoteAddress) {
return new StatelessSetConfig<>(chronicleMapBuilder.statelessClient(remoteAddress));
}
@Override
public ChronicleSet<E> createStatelessClient(InetSocketAddress serverAddress)
throws IOException {
return statelessClient(serverAddress).create();
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> instance() {
return new SetInstanceConfig<>(chronicleMapBuilder.instance());
public ChronicleHashInstanceBuilder<ChronicleSet<E>> instance() {
return new SetInstanceBuilder<>(chronicleMapBuilder.instance());
}
@Override
src/main/java/net/openhft/chronicle/set/ChronicleSetStatelessClientBuilder.java
+62
View File
@@ -0,0 +1,62 @@
/*
* 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.set;
import net.openhft.chronicle.hash.ChronicleHashStatelessClientBuilder;
import net.openhft.chronicle.map.ChronicleMapStatelessClientBuilder;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
public final class ChronicleSetStatelessClientBuilder<E> implements
ChronicleHashStatelessClientBuilder<
ChronicleSetStatelessClientBuilder<E>, ChronicleSet<E>> {
public static <E> ChronicleSetStatelessClientBuilder<E> of(InetSocketAddress serverAddress) {
return new ChronicleSetStatelessClientBuilder<>(serverAddress);
}
private final ChronicleMapStatelessClientBuilder<E, DummyValue> mapClientBuilder;
ChronicleSetStatelessClientBuilder(InetSocketAddress serverAddress) {
this.mapClientBuilder = ChronicleMapStatelessClientBuilder.of(serverAddress);
}
@Override
public ChronicleSetStatelessClientBuilder<E> timeout(long timeout, TimeUnit units) {
mapClientBuilder.timeout(timeout, units);
return this;
}
@Override
public ChronicleSetStatelessClientBuilder<E> name(String name) {
mapClientBuilder.name(name);
return this;
}
@Override
public ChronicleSetStatelessClientBuilder<E> tcpBufferSize(int tcpBufferSize) {
mapClientBuilder.tcpBufferSize(tcpBufferSize);
return this;
}
@Override
public ChronicleSet<E> create() throws IOException {
return new SetFromMap<>(mapClientBuilder.create());
}
}
src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java
+1 -1
View File
@@ -31,7 +31,7 @@ enum DummyValueMarshaller
@Override
public boolean startsWith(Bytes bytes, DummyValue dummyValue) {
throw new UnsupportedOperationException();
return true;
}
@Override
src/main/java/net/openhft/chronicle/set/SetFromMap.java
+3 -1
View File
@@ -60,7 +60,7 @@ class SetFromMap<E> extends AbstractSet<E>
}
public boolean remove(Object o) {
return m.remove(o) == DUMMY_VALUE;
return m.remove(o, DUMMY_VALUE);
}
public boolean add(E e) {
@@ -91,6 +91,8 @@ class SetFromMap<E> extends AbstractSet<E>
return o == this || s.equals(o);
}
// TODO optimize in case of stateless clients -- because bulk ops are optimized on maps
// but on key set they delegate to individual element queries => individual remote calls
public boolean containsAll(Collection<?> c) {
return s.containsAll(c);
}
src/main/java/net/openhft/chronicle/set/SetInstanceConfig.java → src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java
+21 -16
View File
@@ -16,7 +16,7 @@
package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import net.openhft.chronicle.hash.replication.ReplicationChannel;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
@@ -25,46 +25,51 @@ import net.openhft.chronicle.map.ChronicleMap;
import java.io.File;
import java.io.IOException;
final class SetInstanceConfig<E> implements ChronicleHashInstanceConfig<ChronicleSet<E>> {
final class SetInstanceBuilder<E> implements ChronicleHashInstanceBuilder<ChronicleSet<E>> {
private final ChronicleHashInstanceConfig<ChronicleMap<E, DummyValue>> mapConfig;
private final ChronicleHashInstanceBuilder<ChronicleMap<E, DummyValue>> mapBuilder;
SetInstanceConfig(ChronicleHashInstanceConfig<ChronicleMap<E, DummyValue>> mapConfig) {
this.mapConfig = mapConfig;
SetInstanceBuilder(ChronicleHashInstanceBuilder<ChronicleMap<E, DummyValue>> mapBuilder) {
this.mapBuilder = mapBuilder;
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> replicated(
public ChronicleHashInstanceBuilder<ChronicleSet<E>> replicated(
byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) {
return new SetInstanceConfig<>(mapConfig.replicated(identifier, tcpTransportAndNetwork));
mapBuilder.replicated(identifier, tcpTransportAndNetwork);
return this;
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> replicated(
public ChronicleHashInstanceBuilder<ChronicleSet<E>> replicated(
SingleChronicleHashReplication replication) {
return new SetInstanceConfig<>(mapConfig.replicated(replication));
mapBuilder.replicated(replication);
return this;
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> replicatedViaChannel(
public ChronicleHashInstanceBuilder<ChronicleSet<E>> replicatedViaChannel(
ReplicationChannel channel) {
return new SetInstanceConfig<>(mapConfig.replicatedViaChannel(channel));
mapBuilder.replicatedViaChannel(channel);
return this;
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> persistedTo(File file) {
return new SetInstanceConfig<>(mapConfig.persistedTo(file));
public ChronicleHashInstanceBuilder<ChronicleSet<E>> persistedTo(File file) {
mapBuilder.persistedTo(file);
return this;
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> name(String name) {
return new SetInstanceConfig<>(mapConfig.name(name));
public ChronicleHashInstanceBuilder<ChronicleSet<E>> name(String name) {
mapBuilder.name(name);
return this;
}
@Override
public synchronized ChronicleSet<E> create() throws IOException {
try {
return new SetFromMap<>(mapConfig.create());
return new SetFromMap<>(mapBuilder.create());
} catch (IllegalStateException e) {
if (e.getMessage() != null && e.getMessage().startsWith("A ChronicleMap")) {
throw new IllegalStateException(
src/main/java/net/openhft/chronicle/set/StatelessSetConfig.java
-47
View File
@@ -1,47 +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.set;
import net.openhft.chronicle.hash.StatelessClientConfig;
import net.openhft.chronicle.map.ChronicleMap;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
final class StatelessSetConfig<E> implements StatelessClientConfig<ChronicleSet<E>> {
private final StatelessClientConfig<ChronicleMap<E, DummyValue>> mapClientBuilder;
StatelessSetConfig(StatelessClientConfig<ChronicleMap<E, DummyValue>> mapClientBuilder) {
this.mapClientBuilder = mapClientBuilder;
}
@Override
public StatelessClientConfig<ChronicleSet<E>> timeout(long timeout, TimeUnit units) {
return new StatelessSetConfig<>(mapClientBuilder.timeout(timeout, units));
}
@Override
public StatelessClientConfig<ChronicleSet<E>> name(String name) {
return new StatelessSetConfig<>(mapClientBuilder.name(name));
}
@Override
public ChronicleSet<E> create() throws IOException {
return new SetFromMap<>(mapClientBuilder.create());
}
}