mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
added back in moved files
This commit is contained in:
Rob Austin
@@ -0,0 +1,436 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash;
|
||||
|
||||
|
||||
import net.openhft.chronicle.map.*;
|
||||
import net.openhft.lang.collection.DirectBitSet;
|
||||
import net.openhft.lang.collection.SingleThreadedDirectBitSet;
|
||||
import net.openhft.lang.io.ByteBufferBytes;
|
||||
import net.openhft.lang.io.Bytes;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CopyOnWriteArraySet;
|
||||
import java.util.concurrent.LinkedTransferQueue;
|
||||
import java.util.concurrent.atomic.AtomicReferenceArray;
|
||||
|
||||
import static java.lang.Math.min;
|
||||
import static java.nio.ByteBuffer.wrap;
|
||||
import static net.openhft.chronicle.map.Replica.EntryExternalizable;
|
||||
import static net.openhft.chronicle.map.Replica.ModificationIterator;
|
||||
import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public final class ChannelProvider implements Closeable {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName());
|
||||
|
||||
private static final byte BOOTSTRAP_MESSAGE = 'B';
|
||||
final EntryExternalizable asEntryExternalizable = new EntryExternalizable() {
|
||||
/**
|
||||
* writes the entry to the chronicle channel provided
|
||||
*
|
||||
* @param entry the byte location of the entry to be stored
|
||||
* @param destination a buffer the entry will be written to, the segment may reject
|
||||
* this operation and add zeroBytes, if the identifier in the entry
|
||||
* did not match the maps local
|
||||
* @param chronicleChannel used in cluster into identify the canonical map or queue
|
||||
*/
|
||||
@Override
|
||||
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
|
||||
int chronicleChannel) {
|
||||
destination.writeStopBit(chronicleChannel);
|
||||
channelEntryExternalizables[chronicleChannel]
|
||||
.writeExternalEntry(entry, destination, chronicleChannel);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readExternalEntry(@NotNull Bytes source) {
|
||||
final int chronicleId = (int) source.readStopBit();
|
||||
if (chronicleId < chronicleChannels.length)
|
||||
channelEntryExternalizables[chronicleId].readExternalEntry(source);
|
||||
else
|
||||
LOG.info("skipped entry with chronicleId=" + chronicleId + ", ");
|
||||
}
|
||||
};
|
||||
private final byte localIdentifier;
|
||||
final Replica asReplica = new Replica() {
|
||||
|
||||
@Override
|
||||
public byte identifier() {
|
||||
return localIdentifier;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ModificationIterator acquireModificationIterator(
|
||||
final short remoteIdentifier, final ModificationNotifier notifier) {
|
||||
final ModificationIterator result = modificationIterator.get(remoteIdentifier);
|
||||
if (result != null)
|
||||
return result;
|
||||
|
||||
final ModificationIterator result0 = new ModificationIterator() {
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
// old-style iteration to avoid 1) iterator object creation
|
||||
// 2) ConcurrentModificationException
|
||||
for (int i = 0, len = chronicleChannelList.size(); i < len; i++) {
|
||||
final ModificationIterator modificationIterator =
|
||||
chronicleChannelList.get(i)
|
||||
.acquireModificationIterator(remoteIdentifier, notifier);
|
||||
if (modificationIterator.hasNext())
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean nextEntry(@NotNull EntryCallback callback,
|
||||
final int na) {
|
||||
for (int i = 0, len = chronicleChannelList.size(); i < len; i++) {
|
||||
Replica chronicleChannel = chronicleChannelList.get(i);
|
||||
final ModificationIterator modificationIterator = chronicleChannel
|
||||
.acquireModificationIterator(remoteIdentifier, notifier);
|
||||
if (modificationIterator.nextEntry(callback, chronicleChannelIds.get(i)))
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void dirtyEntries(long fromTimeStamp) {
|
||||
for (int i = 0, len = chronicleChannelList.size(); i < len; i++) {
|
||||
chronicleChannelList.get(i)
|
||||
.acquireModificationIterator(remoteIdentifier, notifier)
|
||||
.dirtyEntries(fromTimeStamp);
|
||||
notifier.onChange();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
modificationIterator.set((int) remoteIdentifier, result0);
|
||||
return result0;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* gets the earliest modification time for all of the chronicles
|
||||
*/
|
||||
@Override
|
||||
public long lastModificationTime(byte remoteIdentifier) {
|
||||
long t = 0;
|
||||
// not including the SystemQueue at index 0
|
||||
for (int i = 1, len = chronicleChannelList.size(); i < len; i++) {
|
||||
t = (t == 0) ? chronicleChannels[i].lastModificationTime(remoteIdentifier) :
|
||||
min(t, chronicleChannels[i].lastModificationTime(remoteIdentifier));
|
||||
}
|
||||
return t;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
ChannelProvider.this.close();
|
||||
}
|
||||
};
|
||||
private final int maxEntrySize;
|
||||
private final Replica[] chronicleChannels;
|
||||
private final List<Replica> chronicleChannelList;
|
||||
private final List<Integer> chronicleChannelIds;
|
||||
private final EntryExternalizable[] channelEntryExternalizables;
|
||||
private final AtomicReferenceArray<PayloadProvider> systemModificationIterator =
|
||||
new AtomicReferenceArray<PayloadProvider>(128);
|
||||
private final DirectBitSet systemModificationIteratorBitSet =
|
||||
newBitSet(systemModificationIterator.length());
|
||||
private final AtomicReferenceArray<ModificationIterator> modificationIterator =
|
||||
new AtomicReferenceArray<ModificationIterator>(128);
|
||||
private final Set<Closeable> replicators = new CopyOnWriteArraySet<Closeable>();
|
||||
|
||||
ChannelProvider(ChannelProviderBuilder builder) {
|
||||
localIdentifier = builder.identifier;
|
||||
maxEntrySize = builder.maxEntrySize;
|
||||
chronicleChannels = new Replica[builder.maxNumberOfChronicles];
|
||||
channelEntryExternalizables = new EntryExternalizable[builder.maxNumberOfChronicles];
|
||||
chronicleChannelList = new ArrayList<Replica>();
|
||||
chronicleChannelIds = new ArrayList<Integer>();
|
||||
MessageHandler systemMessageHandler = new MessageHandler() {
|
||||
@Override
|
||||
public void onMessage(Bytes bytes) {
|
||||
final byte type = bytes.readByte();
|
||||
if (type == BOOTSTRAP_MESSAGE) {
|
||||
onBootstrapMessage(bytes);
|
||||
} else {
|
||||
LOG.info("message of type=" + type + " was ignored.");
|
||||
}
|
||||
}
|
||||
};
|
||||
SystemQueue systemMessageQueue = new SystemQueue(
|
||||
systemModificationIteratorBitSet, systemModificationIterator, systemMessageHandler);
|
||||
add((short) 0, systemMessageQueue.asReplica, systemMessageQueue.asEntryExternalizable);
|
||||
}
|
||||
|
||||
public int maxEntrySize() {
|
||||
return maxEntrySize;
|
||||
}
|
||||
|
||||
/**
|
||||
* creates a bit set based on a number of bits
|
||||
*
|
||||
* @param numberOfBits the number of bits the bit set should include
|
||||
* @return a new DirectBitSet backed by a byteBuffer
|
||||
*/
|
||||
private static DirectBitSet newBitSet(int numberOfBits) {
|
||||
final ByteBufferBytes bytes = new ByteBufferBytes(wrap(new byte[(numberOfBits + 7) / 8]));
|
||||
return new SingleThreadedDirectBitSet(bytes);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a replicator, dedicated to the specified channel. Channel is basically just a number, that
|
||||
* should correspond on different servers for instances of the same replicated map. Only one replicator
|
||||
* per channel could be obtained from a single {@code ReplicatingChannel}. The returned replicator could
|
||||
* be applied to a map at most once.
|
||||
*
|
||||
* @param channel number to create
|
||||
* @return a replicator, dedicated to the specified channel
|
||||
*/
|
||||
public ChronicleChannel createChannel(short channel) {
|
||||
return new ChronicleChannel(channel);
|
||||
}
|
||||
|
||||
/**
|
||||
* called whenever we receive a bootstrap message
|
||||
*/
|
||||
private void onBootstrapMessage(Bytes bytes) {
|
||||
final short remoteIdentifier = bytes.readByte();
|
||||
final int chronicleChannel = bytes.readUnsignedShort();
|
||||
final long lastModificationTime = bytes.readLong();
|
||||
|
||||
// this could be null if one node has a chronicle channel before the other
|
||||
if (chronicleChannels[chronicleChannel] != null) {
|
||||
chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP)
|
||||
.dirtyEntries(lastModificationTime);
|
||||
}
|
||||
}
|
||||
|
||||
private ByteBufferBytes toBootstrapMessage(short chronicleChannel, final long lastModificationTime) {
|
||||
final ByteBufferBytes writeBuffer = new ByteBufferBytes(ByteBuffer.allocate(1 + 1 + 2 + 8));
|
||||
writeBuffer.writeByte(BOOTSTRAP_MESSAGE);
|
||||
writeBuffer.writeByte(localIdentifier);
|
||||
writeBuffer.writeUnsignedShort(chronicleChannel);
|
||||
writeBuffer.writeLong(lastModificationTime);
|
||||
writeBuffer.flip();
|
||||
return writeBuffer;
|
||||
}
|
||||
|
||||
private void add(short chronicleChannel,
|
||||
Replica replica, EntryExternalizable entryExternalizable) {
|
||||
if (chronicleChannels[chronicleChannel] != null) {
|
||||
throw new IllegalStateException("chronicleId=" + chronicleChannel +
|
||||
" is already in use.");
|
||||
}
|
||||
chronicleChannels[chronicleChannel] = replica;
|
||||
chronicleChannelList.add(replica);
|
||||
chronicleChannelIds.add((int) chronicleChannel);
|
||||
channelEntryExternalizables[chronicleChannel] = entryExternalizable;
|
||||
|
||||
if (chronicleChannel == 0)
|
||||
return;
|
||||
|
||||
for (int i = (int) systemModificationIteratorBitSet.nextSetBit(0); i > 0;
|
||||
i = (int) systemModificationIteratorBitSet.nextSetBit(i + 1)) {
|
||||
byte remoteIdentifier = (byte) i;
|
||||
final long lastModificationTime = replica.lastModificationTime(remoteIdentifier);
|
||||
final ByteBufferBytes message =
|
||||
toBootstrapMessage(chronicleChannel, lastModificationTime);
|
||||
systemModificationIterator.get(remoteIdentifier).addPayload(message);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
for (Closeable replicator : replicators) {
|
||||
replicator.close();
|
||||
}
|
||||
// indexed reverse loop to avoid ConcurrentModificationException and remove elements from
|
||||
// the end of the loop for correct iteration
|
||||
for (int i = chronicleChannelList.size() - 1; i >= 0; i--) {
|
||||
Replica chronicleChannel = chronicleChannelList.get(i);
|
||||
chronicleChannel.close();
|
||||
}
|
||||
}
|
||||
|
||||
void add(Closeable replicator) {
|
||||
replicators.add(replicator);
|
||||
}
|
||||
|
||||
private interface MessageHandler {
|
||||
void onMessage(Bytes bytes);
|
||||
}
|
||||
|
||||
private interface PayloadProvider extends ModificationIterator {
|
||||
void addPayload(final Bytes bytes);
|
||||
}
|
||||
|
||||
/**
|
||||
* used to send system messages such as bootstrap from one remote node to another, it also can be used in
|
||||
* a broadcast context
|
||||
*/
|
||||
static class SystemQueue {
|
||||
|
||||
|
||||
final Replica asReplica = new Replica() {
|
||||
|
||||
@Override
|
||||
public byte identifier() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ModificationIterator acquireModificationIterator(
|
||||
final short remoteIdentifier, final ModificationNotifier modificationNotifier) {
|
||||
|
||||
final ModificationIterator result = systemModificationIterator.get(remoteIdentifier);
|
||||
|
||||
if (result != null)
|
||||
return result;
|
||||
|
||||
final PayloadProvider iterator = new PayloadProvider() {
|
||||
|
||||
|
||||
// todo it maybe possibel to replace this with a single atomic refernce,
|
||||
// todo this way no aditional objects will be created
|
||||
final Queue<Bytes> payloads = new LinkedTransferQueue<Bytes>();
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return payloads.peek() != null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean nextEntry(@NotNull EntryCallback callback, int na) {
|
||||
final Bytes bytes = payloads.poll();
|
||||
if (bytes == null)
|
||||
return false;
|
||||
callback.onEntry(bytes, 0);
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void dirtyEntries(long fromTimeStamp) {
|
||||
// do nothing
|
||||
}
|
||||
|
||||
@Override
|
||||
public void addPayload(Bytes bytes) {
|
||||
if (bytes.remaining() == 0)
|
||||
return;
|
||||
payloads.add(bytes);
|
||||
// notifies that a change has been made, this will nudge the OP_WRITE
|
||||
// selector to push this update out over the nio socket
|
||||
modificationNotifier.onChange();
|
||||
}
|
||||
};
|
||||
|
||||
systemModificationIterator.set(remoteIdentifier, iterator);
|
||||
systemModificationIteratorBitSet.set(remoteIdentifier);
|
||||
|
||||
return iterator;
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public long lastModificationTime(byte remoteIdentifier) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
// do nothing
|
||||
}
|
||||
};
|
||||
final EntryExternalizable asEntryExternalizable = new EntryExternalizable() {
|
||||
@Override
|
||||
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
|
||||
int na) {
|
||||
destination.write(entry);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readExternalEntry(@NotNull Bytes source) {
|
||||
messageHandler.onMessage(source);
|
||||
}
|
||||
};
|
||||
private final DirectBitSet systemModificationIteratorBitSet;
|
||||
private final AtomicReferenceArray<PayloadProvider> systemModificationIterator;
|
||||
private final MessageHandler messageHandler;
|
||||
|
||||
SystemQueue(DirectBitSet systemModificationIteratorBitSet,
|
||||
AtomicReferenceArray<PayloadProvider> systemModificationIterator,
|
||||
MessageHandler messageHandler) {
|
||||
|
||||
this.systemModificationIteratorBitSet = systemModificationIteratorBitSet;
|
||||
this.systemModificationIterator = systemModificationIterator;
|
||||
this.messageHandler = messageHandler;
|
||||
}
|
||||
}
|
||||
|
||||
public class ChronicleChannel extends Replicator implements Closeable {
|
||||
|
||||
private final short chronicleChannel;
|
||||
|
||||
private ChronicleChannel(short chronicleChannel) {
|
||||
this.chronicleChannel = chronicleChannel;
|
||||
}
|
||||
|
||||
public byte identifier() {
|
||||
return localIdentifier;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
|
||||
Replica map, EntryExternalizable entryExternalizable,
|
||||
final ChronicleMap chronicleMap) {
|
||||
add(chronicleChannel, map, entryExternalizable);
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChannelProvider provider() {
|
||||
return ChannelProvider.this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
chronicleChannelList.remove(chronicleChannels[chronicleChannel]);
|
||||
chronicleChannelIds.remove(chronicleChannel);
|
||||
chronicleChannels[chronicleChannel] = null;
|
||||
channelEntryExternalizables[chronicleChannel] = null;
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash;
|
||||
|
||||
import net.openhft.chronicle.map.TcpReplicator;
|
||||
import net.openhft.chronicle.map.UdpReplicator;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public final class ChannelProviderBuilder {
|
||||
|
||||
byte identifier;
|
||||
int maxEntrySize = 1024;
|
||||
int maxNumberOfChronicles = 128;
|
||||
private UdpReplicationConfig udpReplicationConfig = null;
|
||||
private TcpReplicationConfig tcpReplicationConfig = null;
|
||||
|
||||
|
||||
|
||||
public int maxEntrySize() {
|
||||
return maxEntrySize;
|
||||
}
|
||||
|
||||
public ChannelProviderBuilder maxEntrySize(int maxEntrySize) {
|
||||
this.maxEntrySize = maxEntrySize;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChannelProviderBuilder replicators(byte identifier, ReplicationConfig... replicationConfigs) {
|
||||
this.identifier = identifier;
|
||||
|
||||
|
||||
for (ReplicationConfig replicationConfig : replicationConfigs) {
|
||||
|
||||
if (replicationConfig instanceof TcpReplicationConfig) {
|
||||
this.tcpReplicationConfig = (TcpReplicationConfig) replicationConfig;
|
||||
} else if (replicationConfig instanceof UdpReplicationConfig) {
|
||||
this.udpReplicationConfig = (UdpReplicationConfig) replicationConfig;
|
||||
} else
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChannelProviderBuilder maxNumberOfChronicles(int maxNumberOfChronicles) {
|
||||
this.maxNumberOfChronicles = maxNumberOfChronicles;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChannelProvider create() throws IOException {
|
||||
final ChannelProvider replicatingCluster = new ChannelProvider(this);
|
||||
if (tcpReplicationConfig != null) {
|
||||
final TcpReplicator tcpReplicator = new TcpReplicator(
|
||||
replicatingCluster.asReplica,
|
||||
replicatingCluster.asEntryExternalizable,
|
||||
tcpReplicationConfig,
|
||||
maxEntrySize, null);
|
||||
replicatingCluster.add(tcpReplicator);
|
||||
}
|
||||
|
||||
if (udpReplicationConfig != null) {
|
||||
final UdpReplicator udpReplicator =
|
||||
new UdpReplicator(replicatingCluster.asReplica,
|
||||
replicatingCluster.asEntryExternalizable,
|
||||
udpReplicationConfig,
|
||||
maxEntrySize);
|
||||
replicatingCluster.add(udpReplicator);
|
||||
}
|
||||
return replicatingCluster;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public interface ReplicationConfig<B> {
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public class StatelessBuilder {
|
||||
|
||||
private long timeoutMs = TimeUnit.SECONDS.toMillis(10);
|
||||
|
||||
private InetSocketAddress remoteAddress;
|
||||
|
||||
public long timeoutMs() {
|
||||
return timeoutMs;
|
||||
}
|
||||
|
||||
public void timeout(long timeout, TimeUnit units) {
|
||||
this.timeoutMs = units.toMillis(timeout);
|
||||
}
|
||||
|
||||
private StatelessBuilder(InetSocketAddress remoteAddress) {
|
||||
this.remoteAddress = remoteAddress;
|
||||
}
|
||||
|
||||
public static StatelessBuilder remoteAddress(InetSocketAddress inetSocketAddress) {
|
||||
return new StatelessBuilder(inetSocketAddress);
|
||||
}
|
||||
|
||||
public InetSocketAddress remoteAddress() {
|
||||
return remoteAddress;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash;
|
||||
|
||||
import net.openhft.chronicle.map.IdentifierListener;
|
||||
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static java.util.Collections.unmodifiableSet;
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
|
||||
|
||||
public class TcpReplicationConfig implements ReplicationConfig {
|
||||
|
||||
private static final int DEFAULT_PACKET_SIZE = 1024 * 8;
|
||||
private static final long DEFAULT_HEART_BEAT_INTERVAL = 20;
|
||||
private static final TimeUnit DEFAULT_HEART_BEAT_INTERVAL_UNIT = SECONDS;
|
||||
|
||||
private int serverPort = DEFAULT_PACKET_SIZE;
|
||||
private Set<InetSocketAddress> endpoints;
|
||||
private int packetSize = DEFAULT_PACKET_SIZE;
|
||||
private boolean autoReconnectedUponDroppedConnection = true;
|
||||
private ThrottlingConfig throttlingConfig = ThrottlingConfig.noThrottling();
|
||||
private long heartBeatInterval = DEFAULT_HEART_BEAT_INTERVAL;
|
||||
private TimeUnit heartBeatIntervalUnit = DEFAULT_HEART_BEAT_INTERVAL_UNIT;
|
||||
public IdentifierListener identifierListener;
|
||||
|
||||
TcpReplicationConfig() {
|
||||
|
||||
}
|
||||
|
||||
public IdentifierListener identifierListener() {
|
||||
return identifierListener;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig nonUniqueIdentifierListener(@NotNull final IdentifierListener
|
||||
identifierListener) {
|
||||
this.identifierListener = identifierListener;
|
||||
return this;
|
||||
}
|
||||
|
||||
public boolean autoReconnectedUponDroppedConnection() {
|
||||
return autoReconnectedUponDroppedConnection;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) {
|
||||
this.autoReconnectedUponDroppedConnection = autoReconnectedUponDroppedConnection;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ThrottlingConfig throttlingConfig() {
|
||||
return throttlingConfig;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
|
||||
this.throttlingConfig = throttlingConfig;
|
||||
return this;
|
||||
}
|
||||
|
||||
public long heartBeatInterval(TimeUnit unit) {
|
||||
return unit.convert(heartBeatInterval, heartBeatIntervalUnit);
|
||||
}
|
||||
|
||||
public int serverPort() {
|
||||
return serverPort;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig serverPort(int serverPort) {
|
||||
this.serverPort = serverPort;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Set<InetSocketAddress> endpoints() {
|
||||
return endpoints;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig endpoints(Set<InetSocketAddress> endpoints) {
|
||||
|
||||
for (final InetSocketAddress endpoint : endpoints) {
|
||||
if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName()))
|
||||
throw new IllegalArgumentException("endpoint=" + endpoint
|
||||
+ " can not point to the same port as the server");
|
||||
}
|
||||
|
||||
this.endpoints = endpoints;
|
||||
return this;
|
||||
}
|
||||
|
||||
public int packetSize() {
|
||||
return packetSize;
|
||||
}
|
||||
|
||||
public static TcpReplicationConfig of(int serverPort, InetSocketAddress... endpoints) {
|
||||
return of(serverPort, Arrays.asList(endpoints));
|
||||
}
|
||||
|
||||
public static TcpReplicationConfig of(int serverPort, Collection<InetSocketAddress> endpoints) {
|
||||
// at least in tests, we configure "receive-only" replication without endpoints.
|
||||
// TODO decide what to do with this case
|
||||
// if (endpoints.isEmpty())
|
||||
// throw new IllegalArgumentException("There should be some endpoints");
|
||||
for (final InetSocketAddress endpoint : endpoints) {
|
||||
if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName()))
|
||||
throw new IllegalArgumentException("endpoint=" + endpoint
|
||||
+ " can not point to the same port as the server");
|
||||
}
|
||||
|
||||
TcpReplicationConfig tcpReplicationConfig = new TcpReplicationConfig();
|
||||
|
||||
tcpReplicationConfig.serverPort(serverPort);
|
||||
tcpReplicationConfig.endpoints = unmodifiableSet(new HashSet<InetSocketAddress>(endpoints));
|
||||
|
||||
return tcpReplicationConfig;
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
public TcpReplicationConfig packetSize(int packetSize) {
|
||||
if (packetSize <= 0)
|
||||
throw new IllegalArgumentException();
|
||||
this.packetSize = packetSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) {
|
||||
this.heartBeatInterval = heartBeatInterval;
|
||||
this.heartBeatIntervalUnit = heartBeatIntervalUnit;
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) return true;
|
||||
if (o == null || getClass() != o.getClass()) return false;
|
||||
|
||||
TcpReplicationConfig that = (TcpReplicationConfig) o;
|
||||
|
||||
if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection) return false;
|
||||
if (heartBeatInterval != that.heartBeatInterval) return false;
|
||||
if (packetSize != that.packetSize) return false;
|
||||
if (serverPort != that.serverPort) return false;
|
||||
if (endpoints != null ? !endpoints.equals(that.endpoints) : that.endpoints != null) return false;
|
||||
if (heartBeatIntervalUnit != that.heartBeatIntervalUnit) return false;
|
||||
if (throttlingConfig != null ? !throttlingConfig.equals(that.throttlingConfig) : that.throttlingConfig != null)
|
||||
return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int result = serverPort;
|
||||
result = 31 * result + (endpoints != null ? endpoints.hashCode() : 0);
|
||||
result = 31 * result + packetSize;
|
||||
result = 31 * result + (autoReconnectedUponDroppedConnection ? 1 : 0);
|
||||
result = 31 * result + (throttlingConfig != null ? throttlingConfig.hashCode() : 0);
|
||||
result = 31 * result + (int) (heartBeatInterval ^ (heartBeatInterval >>> 32));
|
||||
result = 31 * result + (heartBeatIntervalUnit != null ? heartBeatIntervalUnit.hashCode() : 0);
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "TcpReplicationConfig{" +
|
||||
"serverPort=" + serverPort +
|
||||
", endpoints=" + endpoints +
|
||||
", packetSize=" + packetSize +
|
||||
", autoReconnectedUponDroppedConnection=" + autoReconnectedUponDroppedConnection +
|
||||
", throttlingConfig=" + throttlingConfig +
|
||||
", heartBeatInterval=" + heartBeatInterval +
|
||||
", heartBeatIntervalUnit=" + heartBeatIntervalUnit +
|
||||
'}';
|
||||
}
|
||||
|
||||
public TcpReplicationConfig heartBeatInterval(long time, TimeUnit unit) {
|
||||
this.heartBeatInterval = time;
|
||||
this.heartBeatIntervalUnit = unit;
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash;
|
||||
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static java.util.concurrent.TimeUnit.MILLISECONDS;
|
||||
|
||||
|
||||
public class ThrottlingConfig {
|
||||
|
||||
private static final long DEFAULT_BUCKET_INTERVAL = 100L;
|
||||
private static final TimeUnit DEFAULT_BUCKET_INTERVAL_UNIT = MILLISECONDS;
|
||||
|
||||
private static final ThrottlingConfig NO_THROTTLING = create(0L, MILLISECONDS,
|
||||
DEFAULT_BUCKET_INTERVAL, DEFAULT_BUCKET_INTERVAL_UNIT);
|
||||
|
||||
private final long throttle;
|
||||
private final TimeUnit throttlePerUnit;
|
||||
private final long bucketInterval;
|
||||
private final TimeUnit bucketIntervalUnit;
|
||||
|
||||
|
||||
ThrottlingConfig(
|
||||
long throttle,
|
||||
TimeUnit throttlePerUnit,
|
||||
long bucketInterval,
|
||||
TimeUnit bucketIntervalUnit) {
|
||||
this.throttle = throttle;
|
||||
if (throttlePerUnit == null) {
|
||||
throw new NullPointerException("Null throttlePerUnit");
|
||||
}
|
||||
this.throttlePerUnit = throttlePerUnit;
|
||||
this.bucketInterval = bucketInterval;
|
||||
if (bucketIntervalUnit == null) {
|
||||
throw new NullPointerException("Null bucketIntervalUnit");
|
||||
}
|
||||
this.bucketIntervalUnit = bucketIntervalUnit;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a config which prescribe that a replicator shouldn't throttle.
|
||||
*
|
||||
* @return a config which prescribe that a replicator shouldn't throttle
|
||||
*/
|
||||
public static ThrottlingConfig noThrottling() {
|
||||
return NO_THROTTLING;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a config which throttle the specified max bits per the given time unit with default bucketing
|
||||
* interval.
|
||||
*
|
||||
* @param maxBits the preferred maximum bits
|
||||
* @param perUnit the time unit per which maximum bits specified
|
||||
* @return a config which throttle the specified max bits per the given time unit
|
||||
* @throws IllegalArgumentException if the specified {@code maxBits} is non-positive value
|
||||
*/
|
||||
public static ThrottlingConfig throttle(long maxBits, TimeUnit perUnit) {
|
||||
if (maxBits <= 0L)
|
||||
throw new IllegalArgumentException();
|
||||
return create(maxBits, perUnit, DEFAULT_BUCKET_INTERVAL, DEFAULT_BUCKET_INTERVAL_UNIT);
|
||||
}
|
||||
|
||||
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 " +
|
||||
replicator + " replicator");
|
||||
}
|
||||
}
|
||||
|
||||
static ThrottlingConfig create(
|
||||
long throttle, TimeUnit throttlePerUnit,
|
||||
long throttleBucketInterval, TimeUnit throttleBucketIntervalUnit) {
|
||||
return new ThrottlingConfig(throttle, throttlePerUnit,
|
||||
throttleBucketInterval, throttleBucketIntervalUnit);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Returns maximum bits per the given time unit, i. e. the throttling. {@code 0} (zero) designates there
|
||||
* is no throttling.
|
||||
*
|
||||
* @param perUnit maximum bits is returned per this time unit
|
||||
* @return maximum bits per the given time unit, or {@code 0} if there is no throttling
|
||||
*/
|
||||
public final long throttling(TimeUnit perUnit) {
|
||||
return throttlePerUnit.convert(throttle, perUnit);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Returns the throttle bucketing interval in the given time units. <p>Default throttle bucketing
|
||||
* interval is 100 millis.
|
||||
*
|
||||
* @param unit the time unit of the interval
|
||||
* @return the bucketing interval for throttling in the given time units
|
||||
*/
|
||||
public long bucketInterval(TimeUnit unit) {
|
||||
return unit.convert(bucketInterval, bucketIntervalUnit);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a copy of this config with the specified bucketing interval.
|
||||
*
|
||||
* @param throttleBucketInterval the bucketing interval for throttling
|
||||
* @param unit the time unit of the interval
|
||||
* @return a copy of this config with the specified bucketing interval
|
||||
* @throws IllegalArgumentException if the given bucketing interval is non-positive
|
||||
*/
|
||||
public ThrottlingConfig bucketInterval(long throttleBucketInterval, TimeUnit unit) {
|
||||
if (throttleBucketInterval <= 0L)
|
||||
throw new IllegalArgumentException();
|
||||
return create(throttle, throttlePerUnit, throttleBucketInterval, unit);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "ThrottlingConfig{"
|
||||
+ "throttle=" + throttle
|
||||
+ ", throttlePerUnit=" + throttlePerUnit
|
||||
+ ", bucketInterval=" + bucketInterval
|
||||
+ ", bucketIntervalUnit=" + bucketIntervalUnit
|
||||
+ "}";
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (o == this) {
|
||||
return true;
|
||||
}
|
||||
if (o instanceof ThrottlingConfig) {
|
||||
ThrottlingConfig that = (ThrottlingConfig) o;
|
||||
return (this.throttle == that.throttle)
|
||||
&& (this.throttlePerUnit.equals(that.throttlePerUnit))
|
||||
&& (this.bucketInterval == that.bucketInterval)
|
||||
&& (this.bucketIntervalUnit.equals(that.bucketIntervalUnit));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int h = 1;
|
||||
h *= 1000003;
|
||||
h ^= (throttle >>> 32) ^ throttle;
|
||||
h *= 1000003;
|
||||
h ^= throttlePerUnit.hashCode();
|
||||
h *= 1000003;
|
||||
h ^= (bucketInterval >>> 32) ^ bucketInterval;
|
||||
h *= 1000003;
|
||||
h ^= bucketIntervalUnit.hashCode();
|
||||
return h;
|
||||
}
|
||||
}
|
||||
@@ -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.hash;
|
||||
|
||||
import net.openhft.chronicle.map.ChronicleMap;
|
||||
import net.openhft.chronicle.set.ChronicleSet;
|
||||
|
||||
import java.io.ObjectStreamException;
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* TimeProvider was aims to possibly later provide an optimization to
|
||||
* {@link System#currentTimeMillis()} on every call to replicated {@link ChronicleMap}
|
||||
* or {@link ChronicleSet}.
|
||||
*
|
||||
* Subclasses should be immutable, because {@link ChronicleHashBuilder} doesn't make defensive
|
||||
* copies.
|
||||
*
|
||||
* @author Rob Austin.
|
||||
* @see ChronicleHashBuilder#timeProvider(TimeProvider)
|
||||
*/
|
||||
public abstract class TimeProvider implements Serializable {
|
||||
/**
|
||||
* Delegates {@link #currentTimeMillis()} to {@link System#currentTimeMillis()}.
|
||||
*/
|
||||
public static final TimeProvider SYSTEM = new System();
|
||||
private static final long serialVersionUID = 0L;
|
||||
|
||||
public abstract long currentTimeMillis();
|
||||
|
||||
private static class System extends TimeProvider {
|
||||
private static final long serialVersionUID = 0L;
|
||||
|
||||
public long currentTimeMillis() {
|
||||
return java.lang.System.currentTimeMillis();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
return o != null && o.getClass() == getClass();
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return getClass().hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return getClass().getSimpleName();
|
||||
}
|
||||
|
||||
private Object readResolve() throws ObjectStreamException {
|
||||
return SYSTEM;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash;
|
||||
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;
|
||||
private final NetworkInterface networkInterface;
|
||||
private final ThrottlingConfig throttlingConfig;
|
||||
|
||||
UdpReplicationConfigBean(
|
||||
InetAddress address,
|
||||
int port,
|
||||
NetworkInterface networkInterface,
|
||||
ThrottlingConfig throttlingConfig) {
|
||||
if (address == null) {
|
||||
throw new NullPointerException("Null address");
|
||||
}
|
||||
this.address = address;
|
||||
this.port = port;
|
||||
|
||||
this.networkInterface = networkInterface;
|
||||
if (throttlingConfig == null) {
|
||||
throw new NullPointerException("Null throttlingConfig");
|
||||
}
|
||||
this.throttlingConfig = throttlingConfig;
|
||||
}
|
||||
|
||||
@Override
|
||||
public InetAddress address() {
|
||||
return address;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int port() {
|
||||
return port;
|
||||
}
|
||||
|
||||
@Override
|
||||
public NetworkInterface networkInterface() {
|
||||
return networkInterface;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ThrottlingConfig throttlingConfig() {
|
||||
return throttlingConfig;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "UdpReplicationConfig{"
|
||||
+ "address=" + address
|
||||
+ ", port=" + port
|
||||
+ ", networkInterface=" + networkInterface
|
||||
+ ", throttlingConfig=" + throttlingConfig
|
||||
+ "}";
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (o == this) {
|
||||
return true;
|
||||
}
|
||||
if (o instanceof UdpReplicationConfig) {
|
||||
UdpReplicationConfig that = (UdpReplicationConfig) o;
|
||||
return (this.address.equals(that.address()))
|
||||
&& (this.port == that.port())
|
||||
&& (this.networkInterface.equals(that.networkInterface()))
|
||||
&& (this.throttlingConfig.equals(that.throttlingConfig()));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int h = 1;
|
||||
h *= 1000003;
|
||||
h ^= address.hashCode();
|
||||
h *= 1000003;
|
||||
h ^= port;
|
||||
h *= 1000003;
|
||||
h ^= networkInterface.hashCode();
|
||||
h *= 1000003;
|
||||
h ^= throttlingConfig.hashCode();
|
||||
return h;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash.serialization;
|
||||
|
||||
public interface AgileBytesMarshaller<E> extends BytesReader<E>, BytesInterop<E>, SizeMarshaller {
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
/*
|
||||
* Copyright 2014 Higher Frequency Trading
|
||||
*
|
||||
* http://www.higherfrequencytrading.com
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.hash.serialization;
|
||||
|
||||
import net.openhft.lang.io.Bytes;
|
||||
import net.openhft.lang.io.serialization.BytesMarshaller;
|
||||
|
||||
public final class BytesReaders {
|
||||
|
||||
public static <E> BytesReader<E> fromBytesMarshaller(BytesMarshaller<E> marshaller) {
|
||||
return new SimpleBytesReader<E>(marshaller);
|
||||
}
|
||||
|
||||
private static class SimpleBytesReader<E> implements BytesReader<E> {
|
||||
private static final long serialVersionUID = 0L;
|
||||
|
||||
private final BytesMarshaller<E> marshaller;
|
||||
|
||||
public SimpleBytesReader(BytesMarshaller<E> marshaller) {
|
||||
this.marshaller = marshaller;
|
||||
}
|
||||
|
||||
@Override
|
||||
public E read(Bytes bytes, long size) {
|
||||
return marshaller.read(bytes);
|
||||
}
|
||||
|
||||
@Override
|
||||
public E read(Bytes bytes, long size, E e) {
|
||||
return marshaller.read(bytes, e);
|
||||
}
|
||||
}
|
||||
|
||||
private BytesReaders() {}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
/*
|
||||
* 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.net.SocketAddress;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public interface IdentifierListener {
|
||||
|
||||
/**
|
||||
* checks the identifier that is unique and we haven't seen it before, unless it comes from the same port
|
||||
* and host.
|
||||
*
|
||||
* @param remoteIdentifier remoteIdentifier
|
||||
* @param remoteAddress remoteAddress
|
||||
* @return true if unique
|
||||
*/
|
||||
boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress);
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user