HCOLL-167 moved classes to hash as they are used for set as well.

This commit is contained in:
Rob Austin
2014-10-29 12:59:18 +00:00
parent 2f73cf1f39
commit 70fdbb1fc6
27 changed files with 139 additions and 121 deletions
@@ -16,9 +16,10 @@
* limitations under the License.
*/
package net.openhft.chronicle.map;
package net.openhft.chronicle.hash;
import net.openhft.chronicle.map.*;
import net.openhft.lang.collection.DirectBitSet;
import net.openhft.lang.collection.SingleThreadedDirectBitSet;
import net.openhft.lang.io.ByteBufferBytes;
@@ -411,7 +412,7 @@ public final class ChannelProvider implements Closeable {
@Override
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, EntryExternalizable entryExternalizable,
final VanillaChronicleMap chronicleMap) {
final ChronicleMap chronicleMap) {
if (builder.entrySize() > maxEntrySize) {
throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " +
"maxEntrySize=" + maxEntrySize + " was specified, but map with " +
@@ -429,5 +430,6 @@ public final class ChannelProvider implements Closeable {
chronicleChannels[chronicleChannel] = null;
channelEntryExternalizables[chronicleChannel] = null;
}
}
}
@@ -16,7 +16,10 @@
* limitations under the License.
*/
package net.openhft.chronicle.map;
package net.openhft.chronicle.hash;
import net.openhft.chronicle.map.TcpReplicator;
import net.openhft.chronicle.map.UdpReplicator;
import java.io.IOException;
@@ -16,7 +16,7 @@
* limitations under the License.
*/
package net.openhft.chronicle.map;
package net.openhft.chronicle.hash;
/**
* @author Rob Austin.
@@ -16,8 +16,10 @@
* limitations under the License.
*/
package net.openhft.chronicle.map;
package net.openhft.chronicle.hash;
import net.openhft.chronicle.map.IdentifierListener;
import net.openhft.chronicle.map.ThrottlingConfig;
import org.jetbrains.annotations.NotNull;
import java.net.InetSocketAddress;
@@ -70,7 +72,7 @@ public class TcpReplicationConfig implements ReplicationConfig {
return this;
}
public ThrottlingConfig throttlingConfig() {
public ThrottlingConfig throttlingConfig() {
return throttlingConfig;
}
@@ -16,12 +16,64 @@
* limitations under the License.
*/
package net.openhft.chronicle.map;
package net.openhft.chronicle.hash;
import net.openhft.chronicle.map.ThrottlingConfig;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.net.InetAddress;
import java.net.NetworkInterface;
public abstract class UdpReplicationConfig implements ReplicationConfig {
/**
* Package-private constructor forbids subclassing from outside of the package
*/
UdpReplicationConfig() {
// nothing to do
}
public static UdpReplicationConfig simple(@NotNull InetAddress address, int port) {
if (address.isMulticastAddress())
throw new IllegalArgumentException();
return create(address, port, null, ThrottlingConfig.noThrottling());
}
public static UdpReplicationConfig multiCast(@NotNull InetAddress address, int port,
@NotNull NetworkInterface networkInterface) {
if (!address.isMulticastAddress() || networkInterface == null)
throw new IllegalArgumentException();
return create(address, port, networkInterface, ThrottlingConfig.noThrottling());
}
static UdpReplicationConfig create(InetAddress address, int port,
NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
return new UdpReplicationConfigBean(address, port, networkInterface,
throttlingConfig);
}
public abstract InetAddress address();
public abstract int port();
@Nullable
public abstract NetworkInterface networkInterface();
public abstract ThrottlingConfig throttlingConfig();
public UdpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP");
return create(address(), port(), networkInterface(), throttlingConfig);
}
}
final class UdpReplicationConfigBean extends UdpReplicationConfig {
private final InetAddress address;
private final int port;
@@ -42,7 +42,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
/**
* @author Rob Austin.
*/
abstract class AbstractChannelReplicator implements Closeable {
public abstract class AbstractChannelReplicator implements Closeable {
public static final int BITS_IN_A_BYTE = 8;
@@ -272,7 +272,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
return self();
}
int entrySize() {
public int entrySize() {
if (entrySize > 0)
return entryAndValueAlignment().alignSize(entrySize);
int size = metaDataBytes;
@@ -18,6 +18,8 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.TcpReplicationConfig;
import net.openhft.chronicle.hash.UdpReplicationConfig;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.collection.DirectBitSet;
import net.openhft.lang.io.ByteBufferBytes;
@@ -39,11 +41,11 @@ import java.util.concurrent.atomic.AtomicReference;
import static java.nio.ByteBuffer.allocateDirect;
import static java.util.concurrent.TimeUnit.SECONDS;
import static net.openhft.chronicle.hash.UdpReplicationConfig.simple;
import static net.openhft.chronicle.map.ConcurrentExpiryMap.getDefaultAddress;
import static net.openhft.chronicle.map.DiscoveryNodeBytesMarshallable.ProposedNodes;
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES;
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
import static net.openhft.chronicle.map.UdpReplicationConfig.simple;
/**
* @author Rob Austin.
@@ -259,7 +261,7 @@ public class NodeDiscovery {
* bitwise OR's the two bit sets or put another way, merges the source bitset into the
* destination bitset and returns the destination
*
* @param source to or
* @param source to or
* @param destination to or
* @return result.
*/
@@ -357,7 +359,7 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
* @throws java.io.IOException
*/
NodeDiscoveryBroadcaster(
@NotNull final UdpReplicationConfig replicationConfig,
final UdpReplicationConfig replicationConfig,
final int serializedEntrySize,
final BytesMarshallable externalizable)
throws IOException {
@@ -19,7 +19,7 @@ package net.openhft.chronicle.map;
import java.io.Closeable;
import java.io.IOException;
abstract class Replicator {
public abstract class Replicator {
/**
* A constructor for use in subclasses.
@@ -29,8 +29,8 @@ abstract class Replicator {
/**
* Applies the replicator to the map instance and returns a Closeable token to manage resources,
* associated with the replication. <p>This method isn't intended to be called from the
* client code.
* associated with the replication. <p>This method isn't intended to be called from the client
* code.
*
* @param builder the builder from which the map was constructed. The replicator may
* obtain some map configurations, not accessible via the map
@@ -49,6 +49,5 @@ abstract class Replicator {
*/
protected abstract Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, Replica.EntryExternalizable entryExternalizable,
final VanillaChronicleMap chronicleMap)
throws IOException;
final ChronicleMap chronicleMap) throws IOException;
}
@@ -16,6 +16,10 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.TcpReplicationConfig;
import net.openhft.chronicle.hash.UdpReplicationConfig;
import org.jetbrains.annotations.NotNull;
import java.io.Closeable;
import java.io.IOException;
@@ -28,15 +32,17 @@ final class Replicators {
return new Replicator() {
@Override
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica replica, Replica.EntryExternalizable entryExternalizable,
final VanillaChronicleMap chronicleMap) throws IOException {
protected Closeable applyTo(@NotNull final AbstractChronicleMapBuilder builder,
@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap)
throws IOException {
final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder
.keyBuilder, builder.valueBuilder);
StatelessServerConnector statelessServer = new StatelessServerConnector
(keyValueSerializer, chronicleMap, builder.entrySize());
(keyValueSerializer, (VanillaChronicleMap) chronicleMap, builder.entrySize());
return new TcpReplicator(replica, entryExternalizable, replicationConfig,
builder.entrySize(), statelessServer);
@@ -49,8 +55,10 @@ final class Replicators {
return new Replicator() {
@Override
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, Replica.EntryExternalizable entryExternalizable, final VanillaChronicleMap chronicleMap)
protected Closeable applyTo(@NotNull final AbstractChronicleMapBuilder builder,
@NotNull final Replica map,
@NotNull final Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap)
throws IOException {
return new UdpReplicator(map, entryExternalizable, replicationConfig,
builder.entrySize());
@@ -18,6 +18,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.TcpReplicationConfig;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import org.jetbrains.annotations.NotNull;
@@ -46,7 +47,7 @@ import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT;
*
* @author Rob Austin.
*/
class TcpReplicator extends AbstractChannelReplicator implements Closeable {
public class TcpReplicator extends AbstractChannelReplicator implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName());
@@ -79,11 +80,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
* @param statelessServerConnector set to NULL if not required
* @throws IOException
*/
TcpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final TcpReplicationConfig replicationConfig,
final int maxEntrySizeBytes,
@Nullable StatelessServerConnector statelessServerConnector) throws IOException {
public TcpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final TcpReplicationConfig replicationConfig,
final int maxEntrySizeBytes,
@Nullable StatelessServerConnector statelessServerConnector) throws IOException {
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(),
maxEntrySizeBytes);
@@ -79,7 +79,7 @@ public class ThrottlingConfig {
return create(maxBits, perUnit, DEFAULT_BUCKET_INTERVAL, DEFAULT_BUCKET_INTERVAL_UNIT);
}
static void checkMillisecondBucketInterval(ThrottlingConfig config, String replicator) {
public static void checkMillisecondBucketInterval(ThrottlingConfig config, String replicator) {
if (config.bucketInterval(MILLISECONDS) < 1) {
throw new IllegalArgumentException(
"Minimum throttle bucketing interval of 1 millisecond is supported by " +
@@ -18,6 +18,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.UdpReplicationConfig;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,10 +41,10 @@ import static java.nio.channels.SelectionKey.OP_WRITE;
/**
* The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), typically, you
* should use the UdpReplicator if you have a large number of nodes, and you wish to receive the data before
* it becomes available on TCP/IP. In order to not miss data, UdpReplicator should be used in conjunction with
* the TCP Replicator.
* The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ),
* typically, you should use the UdpReplicator if you have a large number of nodes, and you wish to
* receive the data before it becomes available on TCP/IP. In order to not miss data, UdpReplicator
* should be used in conjunction with the TCP Replicator.
*/
class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.ModificationNotifier, Closeable {
@@ -69,7 +70,8 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
* @throws IOException
*/
UdpChannelReplicator(@NotNull final UdpReplicationConfig replicationConfig,
final int maxPayloadSize, final byte localIdentifier)
final int maxPayloadSize,
final byte localIdentifier)
throws IOException {
super("UdpReplicator-" + localIdentifier, replicationConfig.throttlingConfig(),
@@ -1,72 +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 org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.net.InetAddress;
import java.net.NetworkInterface;
public abstract class UdpReplicationConfig implements ReplicationConfig {
/**
* Package-private constructor forbids subclassing from outside of the package
*/
UdpReplicationConfig() {
// nothing to do
}
public static UdpReplicationConfig simple(@NotNull InetAddress address, int port) {
if (address.isMulticastAddress())
throw new IllegalArgumentException();
return create(address, port, null, ThrottlingConfig.noThrottling());
}
public static UdpReplicationConfig multiCast(@NotNull InetAddress address, int port,
@NotNull NetworkInterface networkInterface) {
if (!address.isMulticastAddress() || networkInterface == null)
throw new IllegalArgumentException();
return create(address, port, networkInterface, ThrottlingConfig.noThrottling());
}
static UdpReplicationConfig create(InetAddress address, int port,
NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
return new UdpReplicationConfigBean(address, port, networkInterface,
throttlingConfig);
}
public abstract InetAddress address();
public abstract int port();
@Nullable
public abstract NetworkInterface networkInterface();
public abstract ThrottlingConfig throttlingConfig();
public UdpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP");
return create(address(), port(), networkInterface(), throttlingConfig);
}
}
@@ -18,6 +18,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.UdpReplicationConfig;
import net.openhft.lang.io.ByteBufferBytes;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
@@ -35,15 +36,15 @@ import java.nio.channels.DatagramChannel;
* it becomes available on TCP/IP. In order to not miss data, UdpReplicator should be used in conjunction with
* the TCP Replicator.
*/
class UdpReplicator extends UdpChannelReplicator implements Replica.ModificationNotifier, Closeable {
public class UdpReplicator extends UdpChannelReplicator implements Replica.ModificationNotifier, Closeable {
private static final Logger LOG =
LoggerFactory.getLogger(UdpReplicator.class.getName());
UdpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable entryExternalizable,
@NotNull final UdpReplicationConfig replicationConfig,
final int serializedEntrySize)
public UdpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable entryExternalizable,
@NotNull final UdpReplicationConfig replicationConfig,
final int serializedEntrySize)
throws IOException {
super(replicationConfig, serializedEntrySize, replica.identifier());
@@ -18,11 +18,10 @@
package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashErrorListener;
import net.openhft.chronicle.hash.StatelessBuilder;
import net.openhft.chronicle.hash.TimeProvider;
import net.openhft.chronicle.map.*;
import net.openhft.chronicle.hash.*;
import net.openhft.chronicle.map.Alignment;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.lang.io.serialization.BytesMarshaller;
import net.openhft.lang.io.serialization.BytesMarshallerFactory;
import net.openhft.lang.io.serialization.ObjectFactory;