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

CE-14 moved the stateswire connector for map into engine

Browse Source
This commit is contained in:
Rob Austin
2015-03-12 09:54:32 +00:00
parent cd076f8b3c
commit 894e4ccde6
10 changed files with 41 additions and 3509 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+4 -7
View File
@@ -54,8 +54,7 @@ final class ChannelProvider implements Closeable {
static final Map<ReplicationHub, ChannelProvider> implMapping = new IdentityHashMap<>();
static synchronized ChannelProvider getProvider(ReplicationHub hub,
Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException {
static synchronized ChannelProvider getProvider(ReplicationHub hub ) throws IOException {
ChannelProvider channelProvider = implMapping.get(hub);
if (channelProvider != null)
return channelProvider;
@@ -71,9 +70,7 @@ final class ChannelProvider implements Closeable {
tcpConfig,
hub.remoteNodeValidator(),
null,
hub.name(),
channelProvider.chronicleChannelList(),
statelessWiredConnectorSupplier);
hub.name());
channelProvider.add(tcpReplicator);
}
@@ -571,8 +568,8 @@ final class ChannelProvider implements Closeable {
protected Closeable applyTo(ChronicleMapBuilder builder,
Replica map,
EntryExternalizable entryExternalizable,
ChronicleMap chronicleMap,
Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException {
ChronicleMap chronicleMap
) throws IOException {
add(chronicleChannel, map, entryExternalizable);
return this;
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+2 -6
View File
@@ -1461,19 +1461,15 @@ public final class ChronicleMapBuilder<K, V> implements
} else {
final ReplicationHub hub = channel.hub();
Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier =
() -> new StatelessWiredConnector(instance(), hub);
ChannelProvider provider = ChannelProvider.getProvider(hub, statelessWiredConnectorSupplier);
ChannelProvider provider = ChannelProvider.getProvider(hub);
ChannelProvider.ChronicleChannel ch = provider.createChannel(channel.channelId());
replicators.add(ch);
}
for (Replicator replicator : replicators) {
Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier =
() -> new StatelessWiredConnector(instance(), channel.hub());
Closeable token = replicator.applyTo(this, result, result, map, statelessWiredConnectorSupplier);
Closeable token = replicator.applyTo(this, result, result, map );
if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) {
LOG.warn(Replicators.ONLY_UDP_WARN_MESSAGE);
}
src/main/java/net/openhft/chronicle/map/Replicator.java
+1 -3
View File
@@ -18,7 +18,6 @@ package net.openhft.chronicle.map;
import java.io.Closeable;
import java.io.IOException;
import java.util.function.Supplier;
abstract class Replicator {
@@ -50,6 +49,5 @@ abstract class Replicator {
*/
protected abstract Closeable applyTo(ChronicleMapBuilder builder,
Replica map, Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap,
Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException;
final ChronicleMap chronicleMap) throws IOException;
}
src/main/java/net/openhft/chronicle/map/Replicators.java
+14 -20
View File
@@ -27,7 +27,6 @@ import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.function.Supplier;
final class Replicators {
@@ -112,28 +111,24 @@ final class Replicators {
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
Replica map,
Replica.EntryExternalizable entryExternalizable,
ChronicleMap chronicleMap,
Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException {
protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap) throws IOException {
TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork();
TcpReplicator.StatelessClientParameters statelessClientParameters =
new TcpReplicator.StatelessClientParameters(
(VanillaChronicleMap) chronicleMap,
builder.keyBuilder,
builder.valueBuilder);
(VanillaChronicleMap) chronicleMap,
builder.keyBuilder,
builder.valueBuilder);
return new TcpReplicator(map, entryExternalizable,
return new TcpReplicator(replica, entryExternalizable,
tcpConfig,
replication.remoteNodeValidator(), statelessClientParameters,
replication.name(),
null, statelessWiredConnectorSupplier);
replication.name());
}
};
}
@@ -141,15 +136,14 @@ final class Replicators {
final UdpTransportConfig replicationConfig) {
return new Replicator() {
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
Replica map,
Replica.EntryExternalizable entryExternalizable,
ChronicleMap chronicleMap,
Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException {
protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
@NotNull final Replica map,
@NotNull final Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap)
throws IOException {
return new UdpReplicator(map, entryExternalizable, replicationConfig
);
}
};
}
}
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
-1093
View File
File diff suppressed because it is too large Load Diff
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+20 -108
View File
@@ -19,13 +19,13 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.function.SerializableFunction;
import net.openhft.chronicle.hash.impl.util.BuildVersion;
import net.openhft.chronicle.hash.serialization.internal.ReaderWithSize;
import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.internal.ReaderWithSize;
import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.hash.impl.util.BuildVersion;
import net.openhft.lang.io.AbstractBytes;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
@@ -45,13 +45,12 @@ import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import static java.nio.channels.SelectionKey.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static net.openhft.chronicle.hash.impl.util.BuildVersion.version;
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.hash.impl.util.BuildVersion.version;
import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT;
interface Work {
@@ -73,7 +72,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
public static final long TIMESTAMP_FACTOR = 10000;
private static final int STATELESS_CLIENT = -127;
public static final int WIRED_CONNECTION = -126;
private static final byte NOT_SET = (byte) HEARTBEAT.ordinal();
private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName());
private static final int BUFFER_SIZE = 0x100000; // 1MB
@@ -86,8 +84,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
new KeyInterestUpdater(OP_WRITE, selectionKeysStore);
private final BitSet activeKeys = new BitSet(selectionKeysStore.length);
private final long heartBeatIntervalMillis;
private long largestEntrySoFar = 128;
@NotNull
@@ -109,9 +105,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
StatelessClientParameters<K, V> statelessClientParameters;
private Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier;
private List<Replica> channelList;
static class StatelessClientParameters<K, V> {
public StatelessClientParameters(VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
SerializationBuilder<K> keySerializationBuilder,
@@ -134,17 +127,13 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
@NotNull final TcpTransportAndNetworkConfig replicationConfig,
@Nullable final RemoteNodeValidator remoteNodeValidator,
@Nullable final StatelessClientParameters statelessClientParameters,
String name,
@Nullable final List<Replica> channelList,
@Nullable Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier)
String name)
throws IOException {
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig());
this.statelessClientParameters = statelessClientParameters;
this.channelList = channelList;
this.statelessWiredConnectorSupplier = statelessWiredConnectorSupplier;
final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig();
long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS);
@@ -160,8 +149,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
this.remoteNodeValidator = remoteNodeValidator;
this.name = name;
start();
}
@@ -562,7 +549,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
final String localVersion = BuildVersion.version();
final String remoteVersion = attached.serverVersion;
if (!remoteVersion.equals(localVersion)) {
byte remoteIdentifier = attached.remoteIdentifier;
LOG.warn("DIFFERENT CHRONICLE-MAP VERSIONS : " +
@@ -600,11 +586,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
if (attached.remoteIdentifier == Byte.MIN_VALUE) {
final byte remoteIdentifier = reader.identifierFromBuffer();
if (remoteIdentifier == WIRED_CONNECTION) {
attached.handShakingComplete = true;
return;
}
if (remoteIdentifier == STATELESS_CLIENT) {
attached.handShakingComplete = true;
attached.hasRemoteHeartbeatInterval = false;
@@ -628,6 +609,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
// this can occur sometimes when if 2 or more remote node attempt to use node discovery
// at the same time
final SocketAddress remoteAddress = socketChannel.getRemoteAddress();
if ((remoteNodeValidator != null &&
@@ -725,12 +707,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
return;
}
if (attached.useWire) {
attached.statelessWiredConnector.onWrite(socketChannel, key);
return;
}
TcpSocketChannelEntryWriter entryWriter = attached.entryWriter;
if (entryWriter == null) throw
@@ -782,67 +758,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
return;
}
boolean forceRead = false;
if (!attached.isHandShakingComplete()) {
byte[] data = new byte[1];
// sniff just a single byte from the socket and write it to attached.entryReader.in
// this way we can switch to StatelessWiredConnector without reading any other data
int len;
for (; ; ) {
len = socketChannel.read(ByteBuffer.wrap(data));
if (len != 0)
break;
}
if (len == -1) {
socketChannel.register(selector, 0);
if (replicationConfig.autoReconnectedUponDroppedConnection()) {
AbstractConnector connector = attached.connector;
if (connector != null)
connector.connectLater();
} else
socketChannel.close();
return;
}
byte b = data[0];
attached.entryReader.in.put(b);
attached.entryReader.out.limit(1);
if (b == WIRED_CONNECTION)
attached.useWire = true;
forceRead = true;
attached.statelessWiredConnector = statelessWiredConnectorSupplier.get();
if (attached.statelessWiredConnector != null) {
attached.statelessWiredConnector.localIdentifier(localIdentifier);
attached.statelessWiredConnector.setChannelList(channelList);
} else {
LOG.error("", new RuntimeException("statelessWiredConnector not set up"));
}
attached.handShakingComplete = true;
}
if (attached.useWire) {
int len = attached.statelessWiredConnector.onRead(socketChannel, key);
if (len == -1) {
socketChannel.register(selector, 0);
if (replicationConfig.autoReconnectedUponDroppedConnection()) {
AbstractConnector connector = attached.connector;
if (connector != null)
connector.connectLater();
} else
socketChannel.close();
}
return;
}
try {
int len = attached.entryReader.readSocketToBuffer(socketChannel);
@@ -857,7 +772,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
return;
}
if (len == 0 && !forceRead)
if (len == 0)
return;
if (attached.entryWriter.isWorkIncomplete())
@@ -1081,22 +996,20 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
*/
class Attached<K, V> implements Replica.ModificationNotifier {
TcpSocketChannelEntryReader entryReader;
TcpSocketChannelEntryWriter entryWriter;
TcpSocketChannelEntryReader entryReader;
TcpSocketChannelEntryWriter entryWriter;
@Nullable
Replica.ModificationIterator remoteModificationIterator;
AbstractConnector connector;
long remoteBootstrapTimestamp = Long.MIN_VALUE;
byte remoteIdentifier = Byte.MIN_VALUE;
boolean hasRemoteHeartbeatInterval;
Replica.ModificationIterator remoteModificationIterator;
AbstractConnector connector;
long remoteBootstrapTimestamp = Long.MIN_VALUE;
byte remoteIdentifier = Byte.MIN_VALUE;
boolean hasRemoteHeartbeatInterval;
// true if its socket is a ServerSocket
boolean isServer; // the frequency the remote node will send a heartbeat
boolean handShakingComplete;
String serverVersion;
long remoteHeartbeatInterval = heartBeatIntervalMillis;
boolean useWire;
StatelessWiredConnector statelessWiredConnector;
boolean isServer; // the frequency the remote node will send a heartbeat
boolean handShakingComplete;
String serverVersion;
long remoteHeartbeatInterval = heartBeatIntervalMillis;
boolean isHandShakingComplete() {
return handShakingComplete;
@@ -1117,6 +1030,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
*/
public class TcpSocketChannelEntryWriter implements MapIOBuffer {
@NotNull
private final EntryCallback entryCallback;
// if uncompletedWork is set ( not null ) , this must be completed before any further work
@@ -1420,13 +1334,11 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
/**
* reads entries from the buffer till empty
*
* @param attached * @param key the selection key
* channel
* @param attached
* @param key
* @throws InterruptedException
*/
void entriesFromBuffer(@NotNull Attached attached, @NotNull SelectionKey key) {
int entriesRead = 0;
try {
for (; ; entriesRead++) {
src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java
-166
View File
@@ -1,166 +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.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 WiredChronicleMapStatelessClientBuilder<K, V> implements
ChronicleHashStatelessClientBuilder<WiredChronicleMapStatelessClientBuilder<K, V>,
ChronicleMap<K, V>>,
MapBuilder<WiredChronicleMapStatelessClientBuilder<K, V>> {
WiredStatelessClientTcpConnectionHub hub;
private Class keyClass;
private Class valueClass;
private byte localIdentifier;
private short channelID;
public WiredChronicleMapStatelessClientBuilder(InetSocketAddress remoteAddress, Class keyClass, Class valueClass, short channelID) {
this.keyClass = keyClass;
this.valueClass = valueClass;
this.remoteAddress = remoteAddress;
this.channelID = channelID;
}
public WiredChronicleMapStatelessClientBuilder(WiredStatelessClientTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) {
this.keyClass = keyClass;
this.valueClass = valueClass;
this.channelID = channelID;
}
// 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.
private 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);
InetSocketAddress remoteAddress() {
return remoteAddress;
}
@Override
public WiredChronicleMapStatelessClientBuilder<K, V> putReturnsNull(boolean putReturnsNull) {
this.putReturnsNull = putReturnsNull;
return this;
}
boolean putReturnsNull() {
return putReturnsNull;
}
@Override
public WiredChronicleMapStatelessClientBuilder<K, V> removeReturnsNull(boolean removeReturnsNull) {
this.removeReturnsNull = removeReturnsNull;
return this;
}
boolean removeReturnsNull() {
return removeReturnsNull;
}
@Override
public WiredChronicleMapStatelessClientBuilder<K, V> timeout(long timeout, TimeUnit units) {
this.timeoutMs = units.toMillis(timeout);
return this;
}
long timeoutMs() {
return timeoutMs;
}
@Override
public WiredChronicleMapStatelessClientBuilder<K, V> name(String name) {
this.name = name;
return this;
}
String name() {
return name;
}
@Override
public WiredChronicleMapStatelessClientBuilder<K, V> tcpBufferSize(int tcpBufferSize) {
this.tcpBufferSize = tcpBufferSize;
return this;
}
int tcpBufferSize() {
return tcpBufferSize;
}
@Override
public ChronicleMap<K, V> create() throws IOException {
// todo clean this up
if (hub == null)
hub = new WiredStatelessClientTcpConnectionHub(this, localIdentifier);
if (!used.getAndSet(true)) {
return new WiredStatelessChronicleMap<K, V>(this, keyClass, valueClass, channelID);
} else {
throw new IllegalStateException(
"A stateless client has already been created using this builder. " +
"Create a new ChronicleMapStatelessClientBuilder " +
"to create a new stateless client");
}
}
public void identifier(byte identifier) {
this.localIdentifier = identifier;
}
public short channelID() {
return channelID;
}
public WiredStatelessClientTcpConnectionHub hub() {
return hub;
}
public WiredChronicleMapStatelessClientBuilder<K, V> hub(WiredStatelessClientTcpConnectionHub hub) {
this.hub = hub;
return this;
}
}
src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java
-911
View File
@@ -1,911 +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.function.SerializableFunction;
import net.openhft.chronicle.wire.Marshallable;
import net.openhft.chronicle.wire.Wire;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Predicate;
import static java.util.Collections.emptyList;
import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.TO_STRING;
import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.*;
/**
* @author Rob Austin.
*/
public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable {
private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class);
private final WiredStatelessClientTcpConnectionHub hub;
protected Class<K> kClass;
protected Class<V> vClass;
private boolean putReturnsNull;
private boolean removeReturnsNull;
private short channelID;
public WiredStatelessChronicleMap(@NotNull final WiredChronicleMapStatelessClientBuilder config,
@NotNull final Class kClass,
@NotNull final Class vClass,
short channelID) {
this.channelID = channelID;
hub = config.hub;
this.putReturnsNull = config.putReturnsNull();
this.removeReturnsNull = config.removeReturnsNull();
this.kClass = kClass;
this.vClass = vClass;
}
@SuppressWarnings("UnusedDeclaration")
void identifier(int localIdentifier) {
hub.localIdentifier = localIdentifier;
}
@Override
public void getAll(File toFile) throws IOException {
JsonSerializer.getAll(toFile, this, emptyList());
}
@Override
public void putAll(File fromFile) throws IOException {
JsonSerializer.putAll(fromFile, this, emptyList());
}
@Override
public V newValueInstance() {
return VanillaChronicleMap.newInstance(vClass, false);
}
@Override
public K newKeyInstance() {
return VanillaChronicleMap.newInstance(kClass, true);
}
@Override
public Class<K> keyClass() {
return kClass;
}
@Override
public boolean forEachEntryWhile(Predicate<? super MapKeyContext<K, V>> predicate) {
// TODO implement!
throw new UnsupportedOperationException();
}
@Override
public void forEachEntry(Consumer<? super MapKeyContext<K, V>> action) {
// TODO implement!
throw new UnsupportedOperationException();
}
public String serverApplicationVersion() {
return hub.serverApplicationVersion(channelID);
}
@Override
public void close() {
// todo add ref count
}
@Override
public Class<V> valueClass() {
return vClass;
}
@NotNull
public File file() {
throw new UnsupportedOperationException();
}
@SuppressWarnings("NullableProblems")
public V putIfAbsent(K key, V value) {
if (key == null || value == null)
throw new NullPointerException();
return proxyReturnObject(PUT_IF_ABSENT.toString(), key, value, vClass);
}
@SuppressWarnings("NullableProblems")
public boolean remove(Object key, Object value) {
if (key == null)
throw new NullPointerException();
return value != null && proxyReturnBoolean(REMOVE_WITH_VALUE.toString(), (K) key, (V) value);
}
@SuppressWarnings("NullableProblems")
public boolean replace(K key, V oldValue, V newValue) {
if (key == null || oldValue == null || newValue == null)
throw new NullPointerException();
return proxyReturnBoolean(REPLACE_WITH_OLD_AND_NEW_VALUE.toString(), key, oldValue, newValue);
}
@SuppressWarnings("NullableProblems")
public V replace(K key, V value) {
if (key == null || value == null)
throw new NullPointerException();
return proxyReturnObject(REPLACE.toString(), key, value, vClass);
}
public int size() {
return (int) longSize();
}
/**
* calling this method should be avoided at all cost, as the entire {@code object} is
* serialized. This equals can be used to compare map that extends ChronicleMap. So two
* Chronicle Maps that contain the same data are considered equal, even if the instances of the
* chronicle maps were of different types
*
* @param object the object that you are comparing against
* @return true if the contain the same data
*/
@Override
public boolean equals(@Nullable Object object) {
if (this == object) return true;
if (object == null || object.getClass().isAssignableFrom(Map.class))
return false;
final Map<? extends K, ? extends V> that = (Map<? extends K, ? extends V>) object;
final int size = size();
if (that.size() != size)
return false;
final Set<Map.Entry<K, V>> entries = entrySet();
return that.entrySet().equals(entries);
}
@Override
public int hashCode() {
return proxyReturnInt(HASH_CODE.toString());
}
@NotNull
public String toString() {
return hub.proxyReturnString(TO_STRING.toString(), channelID);
}
@NotNull
public String serverPersistedDataVersion() {
return hub.proxyReturnString(PERSISTED_DATA_VERSION.toString(), channelID);
}
public boolean isEmpty() {
return proxyReturnBoolean(IS_EMPTY.toString());
}
public boolean containsKey(Object key) {
return proxyReturnBooleanK(CONTAINS_KEY.toString(), (K) key);
}
@NotNull
private NullPointerException keyNotNullNPE() {
return new NullPointerException("key can not be null");
}
public boolean containsValue(Object value) {
return proxyReturnBooleanV(CONTAINS_VALUE.toString(), (V) value);
}
public long longSize() {
return proxyReturnLong(LONG_SIZE.toString());
}
@Override
public MapKeyContext<K, V> context(K key) {
throw new UnsupportedOperationException("Contexts are not supported by stateless clients");
}
public V get(Object key) {
return proxyReturnObject(vClass, GET.toString(), (K) key);
}
@Nullable
public V getUsing(K key, V usingValue) {
throw new UnsupportedOperationException();
}
@NotNull
public V acquireUsing(@NotNull K key, V usingValue) {
throw new UnsupportedOperationException(
"acquireUsing() is not supported for stateless clients");
}
@NotNull
@Override
public MapKeyContext<K, V> acquireContext(@NotNull K key, @NotNull V usingValue) {
throw new UnsupportedOperationException("Contexts are not supported by stateless clients");
}
public V remove(Object key) {
if (key == null)
throw keyNotNullNPE();
return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key);
}
public void createChannel(short channelID) {
proxyReturnVoid(
() -> {
writeField(StatelessWiredConnector.Fields.METHOD_NAME, "CREATE_CHANNEL");
hub.outWire().write(StatelessWiredConnector.Fields.ARG_1).int16(channelID);
});
}
public void putAll(@NotNull Map<? extends K, ? extends V> map) {
final long startTime = System.currentTimeMillis();
long transactionId = hub.nextUniqueTransaction(startTime);
Set<? extends Map.Entry<? extends K, ? extends V>> entries = map.entrySet();
Iterator<? extends Map.Entry<? extends K, ? extends V>> iterator = entries.iterator();
boolean hasNext;
do {
hasNext = iterator.hasNext();
hub.outBytesLock().lock();
try {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
Wire wire = hub.outWire();
assert hub.outBytesLock().isHeldByCurrentThread();
hub.markSize(wire);
hub.startTime(startTime);
wire.write(StatelessWiredConnector.Fields.TRANSACTION_ID).int64(transactionId);
wire.write(StatelessWiredConnector.Fields.TIME_STAMP).int64(startTime);
wire.write(StatelessWiredConnector.Fields.CHANNEL_ID).int16(channelID);
writeField(StatelessWiredConnector.Fields.METHOD_NAME, "PUT_ALL");
hub.outWire().write(StatelessWiredConnector.Fields.HAS_NEXT).bool(hasNext);
if (hasNext) {
Map.Entry<? extends K, ? extends V> e = iterator.next();
writeField(StatelessWiredConnector.Fields.ARG_1, e.getKey());
writeField(StatelessWiredConnector.Fields.ARG_2, e.getValue());
}
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
} while (hasNext);
// wait for the transaction id to be received, this will only be received once the
// last chunk has been processed
readVoid(transactionId, startTime);
}
public V put(K key, V value) {
if (key == null || value == null)
throw new NullPointerException();
return proxyReturnObject(putReturnsNull ? PUT_WITHOUT_ACC.toString() : PUT.toString(), key, value, vClass);
}
@Nullable
public <R> R getMapped(@Nullable K key, @NotNull SerializableFunction<? super V, R> function) {
throw new UnsupportedOperationException();
}
@Nullable
@Override
public V putMapped(@Nullable K key, @NotNull UnaryOperator<V> unaryOperator) {
throw new UnsupportedOperationException();
}
public void clear() {
proxyReturnVoid(() -> writeField(StatelessWiredConnector.Fields.METHOD_NAME, CLEAR.toString()));
}
public Set<K> keySet() {
final long startTime = System.currentTimeMillis();
final Set<K> result = new HashSet<K>();
// send
final long transactionId = proxySend("KEY_SET", startTime);
assert !hub.outBytesLock().isHeldByCurrentThread();
final long timeoutTime = startTime + hub.timeoutMs;
for (; ; ) {
// receive
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) {
K k = readK(StatelessWiredConnector.Fields.RESULT, wireIn, null);
result.add(k);
} else
break;
} finally {
hub.inBytesLock().unlock();
}
}
return result;
}
public Collection<V> values() {
final long startTime = System.currentTimeMillis();
final Set<V> result = new HashSet<V>();
// send
final long transactionId = proxySend("VALUES", startTime);
assert !hub.outBytesLock().isHeldByCurrentThread();
final long timeoutTime = startTime + hub.timeoutMs;
for (; ; ) {
// receive
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) {
V v = readV(StatelessWiredConnector.Fields.RESULT, wireIn, null);
result.add(v);
} else
break;
} finally {
hub.inBytesLock().unlock();
}
}
return result;
}
@NotNull
public Set<Map.Entry<K, V>> entrySet() {
final long startTime = System.currentTimeMillis();
final Map<K, V> result = new HashMap<K, V>();
// send
final long transactionId = proxySend("ENTRY_SET", startTime);
assert !hub.outBytesLock().isHeldByCurrentThread();
final long timeoutTime = startTime + hub.timeoutMs;
for (; ; ) {
// receive
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) {
result.put(
readK(StatelessWiredConnector.Fields.RESULT_KEY, wireIn, null),
readV(StatelessWiredConnector.Fields.RESULT_VALUE, wireIn, null));
} else
break;
} finally {
hub.inBytesLock().unlock();
}
}
return result.entrySet();
}
/**
* @param callback each entry is passed to the callback
*/
void entrySet(@NotNull MapEntryCallback<K, V> callback) {
throw new UnsupportedOperationException();
}
private long readLong(long transactionId, long startTime) {
assert !hub.outBytesLock().isHeldByCurrentThread();
final long timeoutTime = startTime + hub.timeoutMs;
// receive
hub.inBytesLock().lock();
try {
return hub.proxyReply(timeoutTime, transactionId).read(StatelessWiredConnector.Fields.RESULT).int64();
} finally {
hub.inBytesLock().unlock();
}
}
private void writeField(StatelessWiredConnector.Fields fieldName, Object value) {
writeField(fieldName, value, hub.outWire());
}
private void writeField(StatelessWiredConnector.Fields fieldName, Object value, Wire wire) {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
if (value instanceof Byte)
wire.write(fieldName).int8((Byte) value);
else if (value instanceof Character)
wire.write(fieldName).text(((Character) value).toString());
else if (value instanceof Short)
wire.write(fieldName).int16((Short) value);
else if (value instanceof Integer)
wire.write(fieldName).int32((Integer) value);
else if (value instanceof Long)
wire.write(fieldName).int64((Long) value);
else if (value instanceof CharSequence) {
wire.write(fieldName).text((CharSequence) value);
} else if (value instanceof Marshallable) {
wire.write(fieldName).marshallable((Marshallable) value);
} else {
throw new IllegalStateException("type=" + value.getClass() + " is unsupported, it must either be of type Marshallable or CharSequence");
}
}
private V readValue(StatelessWiredConnector.Fields argName, long transactionId, long startTime, final V usingValue) {
assert !hub.outBytesLock().isHeldByCurrentThread();
long timeoutTime = startTime + hub.timeoutMs;
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(StatelessWiredConnector.Fields.RESULT_IS_NULL).bool())
return null;
return readV(argName, wireIn, usingValue);
} finally {
hub.inBytesLock().unlock();
}
}
private V readV(StatelessWiredConnector.Fields argName, Wire wireIn, V usingValue) {
if (StringBuilder.class.isAssignableFrom(vClass)) {
wireIn.read(argName).text((StringBuilder) usingValue);
return usingValue;
} else if (Marshallable.class.isAssignableFrom(vClass)) {
if (usingValue == null)
try {
V v = vClass.newInstance();
wireIn.read(argName).marshallable((Marshallable) v);
return v;
} catch (Exception e) {
throw new RuntimeException(e);
}
wireIn.read(argName).marshallable((Marshallable) usingValue);
return usingValue;
} else if (String.class.isAssignableFrom(vClass)) {
//noinspection unchecked
return (V) wireIn.read(argName).text();
} else {
throw new IllegalStateException("unsupported type");
}
}
private K readK(StatelessWiredConnector.Fields argName, Wire wireIn, K usingValue) {
if (StringBuilder.class.isAssignableFrom(kClass)) {
wireIn.read(argName).text((StringBuilder) usingValue);
return usingValue;
} else if (Marshallable.class.isAssignableFrom(kClass)) {
if (usingValue == null)
try {
K v = kClass.newInstance();
wireIn.read(argName).marshallable((Marshallable) v);
return v;
} catch (Exception e) {
throw new RuntimeException(e);
}
wireIn.read(argName).marshallable((Marshallable) usingValue);
return usingValue;
} else if (String.class.isAssignableFrom(vClass)) {
//noinspection unchecked
return (K) wireIn.read(argName).text();
} else {
throw new IllegalStateException("unsupported type");
}
}
private boolean readBoolean(long transactionId, long startTime) {
assert !hub.outBytesLock().isHeldByCurrentThread();
long timeoutTime = startTime + hub.timeoutMs;
// receive
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
return wireIn.read(StatelessWiredConnector.Fields.RESULT).bool();
} finally {
hub.inBytesLock().unlock();
}
}
private int readInt(long transactionId, long startTime) {
assert !hub.outBytesLock().isHeldByCurrentThread();
long timeoutTime = startTime + hub.timeoutMs;
// receive
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
return wireIn.read(StatelessWiredConnector.Fields.RESULT).int32();
} finally {
hub.inBytesLock().unlock();
}
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value) {
final long startTime = System.currentTimeMillis();
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, key);
writeField(StatelessWiredConnector.Fields.ARG_2, value);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readBoolean(transactionId, startTime);
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value1, V value2) {
final long startTime = System.currentTimeMillis();
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, value1);
writeField(StatelessWiredConnector.Fields.ARG_2, value2);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readBoolean(transactionId, startTime);
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBooleanV(@NotNull final String methodName, V value) {
final long startTime = System.currentTimeMillis();
return readBoolean(proxySend(methodName, startTime, value), startTime);
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBooleanK(@NotNull final String methodName, K key) {
final long startTime = System.currentTimeMillis();
return readBoolean(proxySend(methodName, startTime, key), startTime);
}
@SuppressWarnings("SameParameterValue")
private long proxyReturnLong(@NotNull final String methodName) {
final long startTime = System.currentTimeMillis();
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readLong(transactionId, startTime);
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final String methodName) {
final long startTime = System.currentTimeMillis();
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readBoolean(transactionId, startTime);
}
@SuppressWarnings("SameParameterValue")
private void proxyReturnVoid(Runnable r) {
final long startTime = System.currentTimeMillis();
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
r.run();
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
readVoid(transactionId, startTime);
}
@SuppressWarnings("SameParameterValue")
private int proxyReturnInt(@NotNull final String methodName) {
final long startTime = System.currentTimeMillis();
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readInt(transactionId, startTime);
}
@Nullable
private <R> R proxyReturnObject(@NotNull final String methodName, K key, V value, Class<V> resultType) {
final long startTime = System.currentTimeMillis();
long transactionId;
hub.outBytesLock().lock();
try {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, key);
writeField(StatelessWiredConnector.Fields.ARG_2, value);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
if (eventReturnsNull(methodName))
return null;
if (resultType == vClass)
return (R) readValue(StatelessWiredConnector.Fields.RESULT, transactionId, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + resultType + " is not " +
"supported");
}
@Nullable
private <R> R proxyReturnObject(Class<R> rClass, @NotNull final String methodName, Object key) {
final long startTime = System.currentTimeMillis();
long transactionId;
transactionId = proxySend(methodName, startTime, key);
if (eventReturnsNull(methodName))
return null;
if (rClass == vClass)
return (R) readValue(StatelessWiredConnector.Fields.RESULT, transactionId, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + rClass + " is not " +
"supported");
}
private long proxySend(String methodName, long startTime, Object arg1) {
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
writeField(StatelessWiredConnector.Fields.ARG_1, arg1);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return transactionId;
}
private long proxySend(String methodName, long startTime) {
long transactionId;
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return transactionId;
}
private boolean eventReturnsNull(@NotNull String methodName) {
switch (methodName) {
case "PUT_ALL_WITHOUT_ACC":
case "PUT_WITHOUT_ACC":
case "REMOVE_WITHOUT_ACC":
return true;
default:
return false;
}
}
private void readVoid(long transactionId, long startTime) {
long timeoutTime = startTime + hub.timeoutMs;
// receive
hub.inBytesLock().lock();
try {
hub.proxyReply(timeoutTime, transactionId);
} finally {
hub.inBytesLock().unlock();
}
}
static enum EventId {
HEARTBEAT,
STATEFUL_UPDATE,
LONG_SIZE,
SIZE,
IS_EMPTY,
CONTAINS_KEY,
CONTAINS_VALUE,
GET,
PUT,
PUT_WITHOUT_ACC,
REMOVE,
REMOVE_WITHOUT_ACC,
CLEAR,
KEY_SET,
VALUES,
ENTRY_SET,
REPLACE,
REPLACE_WITH_OLD_AND_NEW_VALUE,
PUT_IF_ABSENT,
REMOVE_WITH_VALUE,
TO_STRING,
APPLICATION_VERSION,
PERSISTED_DATA_VERSION,
PUT_ALL,
PUT_ALL_WITHOUT_ACC,
HASH_CODE,
MAP_FOR_KEY,
PUT_MAPPED,
KEY_BUILDER,
VALUE_BUILDER
}
class Entry implements Map.Entry<K, V> {
final K key;
final V value;
/**
* Creates new entry.
*/
Entry(K k1, V v) {
value = v;
key = k1;
}
public final K getKey() {
return key;
}
public final V getValue() {
return value;
}
public final V setValue(V newValue) {
V oldValue = value;
WiredStatelessChronicleMap.this.put(getKey(), newValue);
return oldValue;
}
public final boolean equals(Object o) {
if (!(o instanceof Map.Entry))
return false;
final Map.Entry e = (Map.Entry) o;
final Object k1 = getKey();
final Object k2 = e.getKey();
if (k1 == k2 || (k1 != null && k1.equals(k2))) {
Object v1 = getValue();
Object v2 = e.getValue();
if (v1 == v2 || (v1 != null && v1.equals(v2)))
return true;
}
return false;
}
public final int hashCode() {
return (key == null ? 0 : key.hashCode()) ^
(value == null ? 0 : value.hashCode());
}
@NotNull
public final String toString() {
return getKey() + "=" + getValue();
}
}
}
src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
-667
View File
@@ -1,667 +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 com.sun.jdi.connect.spi.ClosedConnectionException;
import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.hash.RemoteCallTimeoutException;
import net.openhft.chronicle.hash.impl.util.BuildVersion;
import net.openhft.chronicle.hash.impl.util.CloseablesManager;
import net.openhft.chronicle.wire.TextWire;
import net.openhft.chronicle.wire.Wire;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import static java.nio.ByteBuffer.wrap;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
import static net.openhft.chronicle.map.TcpReplicator.WIRED_CONNECTION;
import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.APPLICATION_VERSION;
/**
* Created by Rob Austin
*/
public class WiredStatelessClientTcpConnectionHub {
private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMap.class);
private static final ByteBuffer WIRED_FLAG = wrap(new byte[]{WIRED_CONNECTION});
protected final String name;
protected final InetSocketAddress remoteAddress;
protected final long timeoutMs;
protected final int tcpBufferSize;
private final ReentrantLock inBytesLock = new ReentrantLock(true);
private final ReentrantLock outBytesLock = new ReentrantLock();
@NotNull
private final AtomicLong transactionID = new AtomicLong(0);
private final WiredChronicleMapStatelessClientBuilder config;
@Nullable
protected CloseablesManager closeables;
private final Wire outWire = new TextWire(Bytes.elasticByteBuffer());
long largestChunkSoFar = 0;
private final Wire intWire = new TextWire(Bytes.elasticByteBuffer());
// used by the enterprise version
protected int localIdentifier;
private SocketChannel clientChannel;
// this is a transaction id and size that has been read by another thread.
private volatile long parkedTransactionId;
private volatile long parkedTransactionTimeStamp;
private long limitOfLast = 0;
// set up in the header
private long startTime;
public WiredStatelessClientTcpConnectionHub(WiredChronicleMapStatelessClientBuilder config, byte localIdentifier) {
this.localIdentifier = localIdentifier;
this.remoteAddress = config.remoteAddress();
this.tcpBufferSize = config.tcpBufferSize();
this.config = config;
this.name = config.name();
this.timeoutMs = config.timeoutMs();
attemptConnect(remoteAddress);
checkVersion(config.channelID());
}
private synchronized void attemptConnect(final InetSocketAddress remoteAddress) {
// ensures that the excising connection are closed
closeExisting();
try {
SocketChannel socketChannel = AbstractChannelReplicator.openSocketChannel(closeables);
if (socketChannel.connect(remoteAddress)) {
doHandShaking(socketChannel);
clientChannel = socketChannel;
}
} catch (IOException e) {
if (closeables != null) closeables.closeQuietly();
clientChannel = null;
}
}
ReentrantLock inBytesLock() {
return inBytesLock;
}
ReentrantLock outBytesLock() {
return outBytesLock;
}
protected void checkVersion(short channelID) {
final String serverVersion = serverApplicationVersion(channelID);
final String clientVersion = clientVersion();
if (!serverVersion.equals(clientVersion)) {
LOG.warn("DIFFERENT CHRONICLE-MAP VERSIONS: The Chronicle-Map-Server and " +
"Stateless-Client are on different " +
"versions, " +
" we suggest that you use the same version, server=" + serverApplicationVersion(channelID) + ", " +
"client=" + clientVersion);
}
}
private void checkTimeout(long timeoutTime) {
if (timeoutTime < System.currentTimeMillis())
throw new RemoteCallTimeoutException();
}
protected synchronized void lazyConnect(final long timeoutMs,
final InetSocketAddress remoteAddress) {
if (clientChannel != null)
return;
if (LOG.isDebugEnabled())
LOG.debug("attempting to connect to " + remoteAddress + " ,name=" + name);
SocketChannel result;
long timeoutAt = System.currentTimeMillis() + timeoutMs;
for (; ; ) {
checkTimeout(timeoutAt);
// ensures that the excising connection are closed
closeExisting();
try {
result = AbstractChannelReplicator.openSocketChannel(closeables);
if (!result.connect(remoteAddress)) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
continue;
}
result.socket().setTcpNoDelay(true);
doHandShaking(result);
break;
} catch (IOException e) {
if (closeables != null) closeables.closeQuietly();
} catch (Exception e) {
if (closeables != null) closeables.closeQuietly();
throw e;
}
}
clientChannel = result;
checkVersion(config.channelID());
}
/**
* closes the existing connections and establishes a new closeables
*/
protected void closeExisting() {
// ensure that any excising connection are first closed
if (closeables != null)
closeables.closeQuietly();
closeables = new CloseablesManager();
}
/**
* initiates a very simple level of handshaking with the remote server, we send a special ID of
* -127 ( when the server receives this it knows its dealing with a stateless client, receive
* back an identifier from the server
*
* @param clientChannel clientChannel
* @throws java.io.IOException
*/
protected synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException {
long timeoutTime = System.currentTimeMillis() + timeoutMs;
WIRED_FLAG.clear();
// write a single byte
while (WIRED_FLAG.hasRemaining()) {
int write = clientChannel.write(WIRED_FLAG);
if (write == -1)
throw new SocketException();
checkTimeout(timeoutTime);
}
}
public synchronized void close() {
if (closeables != null)
closeables.closeQuietly();
closeables = null;
clientChannel = null;
}
/**
* the transaction id are generated as unique timestamps
*
* @param time in milliseconds
* @return a unique transactionId
*/
long nextUniqueTransaction(long time) {
long id = time * TcpReplicator.TIMESTAMP_FACTOR;
for (; ; ) {
long old = transactionID.get();
if (old >= id) id = old + 1;
if (transactionID.compareAndSet(old, id))
break;
}
return id;
}
@NotNull
public String serverApplicationVersion(short channelID) {
TextWire wire = new TextWire(Bytes.elasticByteBuffer());
String result = proxyReturnString(APPLICATION_VERSION.toString(), channelID, wire);
return (result == null) ? "" : result;
}
@SuppressWarnings("WeakerAccess")
@NotNull
String clientVersion() {
return BuildVersion.version();
}
/**
* sends data to the server via TCP/IP
*
* @param wire the {@code wire} containing the outbound data
*/
void writeSocket(@NotNull final Wire wire) {
assert outBytesLock().isHeldByCurrentThread();
assert !inBytesLock().isHeldByCurrentThread();
final long timeoutTime = startTime + this.timeoutMs;
try {
for (; ; ) {
if (clientChannel == null)
lazyConnect(timeoutMs, remoteAddress);
try {
writeLength(wire);
// send out all the bytes
writeSocket(wire, timeoutTime);
break;
} catch (@NotNull java.nio.channels.ClosedChannelException | ClosedConnectionException e) {
checkTimeout(timeoutTime);
lazyConnect(timeoutMs, remoteAddress);
}
}
} catch (IOException e) {
close();
throw new IORuntimeException(e);
} catch (Exception e) {
close();
throw e;
}
}
private void writeLength(Wire outWire) {
assert outBytesLock().isHeldByCurrentThread();
long position = outWire.bytes().position();
if (position > Integer.MAX_VALUE || position < Integer.MIN_VALUE)
throw new IllegalStateException("message too large");
long pos = outWire.bytes().position();
try {
outWire.bytes().reset();
outWire.bytes().writeUnsignedShort((int) position);
} finally {
outWire.bytes().position(pos);
}
}
protected Wire proxyReply(long timeoutTime, final long transactionId) {
assert inBytesLock().isHeldByCurrentThread();
try {
final Wire wire = proxyReplyThrowable(timeoutTime, transactionId);
// handle an exception if the message contains the IS_EXCEPTION field
if (wire.read(() -> "IS_EXCEPTION").bool()) {
final String text = wire.read(() -> "EXCEPTION").text();
throw new RuntimeException(text);
}
return wire;
} catch (IOException e) {
close();
throw new IORuntimeException(e);
} catch (RuntimeException e) {
close();
throw e;
} catch (Exception e) {
close();
throw new RuntimeException(e);
} catch (AssertionError e) {
LOG.error("name=" + name, e);
throw e;
}
}
private Wire proxyReplyThrowable(long timeoutTime, long transactionId) throws IOException {
assert inBytesLock().isHeldByCurrentThread();
for (; ; ) {
// read the next item from the socket
if (parkedTransactionId == 0) {
assert parkedTransactionTimeStamp == 0;
// if we have processed all the bytes that we have read in
if (inWireByteBuffer().position() == intWire.bytes().position())
inWireClear();
// todo change the size to include the meta data bit
// reads just the size
readSocket(SIZE_OF_SIZE, timeoutTime);
final int messageSize = intWire.bytes().readUnsignedShort(intWire.bytes().position());
assert messageSize > 0 : "Invalid message size " + messageSize;
assert messageSize < 16 << 20 : "Invalid message size " + messageSize;
final int remainingBytes0 = messageSize - SIZE_OF_SIZE;
readSocket(remainingBytes0, timeoutTime);
intWire.bytes().skip(4);
intWire.bytes().limit(messageSize);
System.out.println("\n--------------------------------\n" +
"\nclient read:\n" + Bytes.toDebugString(intWire.bytes()));
long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64();
// check the transaction id is reasonable
/* assert transactionId0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR :
"TransactionId too small " + transactionId0 + " messageSize " + messageSize;
assert transactionId0 < 2100000000000L * TcpReplicator.TIMESTAMP_FACTOR :
"TransactionId too large " + transactionId0 + " messageSize " + messageSize;
*/
// if the transaction id is for this thread process it
if (transactionId0 == transactionId) {
clearParked();
return intWire;
} else {
// if the transaction id is not for this thread, park it
// and allow another thread to pick it up
parkedTransactionTimeStamp = System.currentTimeMillis();
parkedTransactionId = transactionId0;
pause();
continue;
}
}
// the transaction id was read by another thread, but is for this thread, process it
if (parkedTransactionId == transactionId) {
clearParked();
return intWire;
}
// time out the old transaction id
if (System.currentTimeMillis() - timeoutTime >
parkedTransactionTimeStamp) {
LOG.error("name=" + name, new IllegalStateException("Skipped Message with " +
"transaction-id=" +
parkedTransactionTimeStamp +
", this can occur when you have another thread which has called the " +
"stateless client and terminated abruptly before the message has been " +
"returned from the server and hence consumed by the other thread."));
// read the the next message
clearParked();
pause();
}
}
}
/**
* clears the wire and its underlying byte buffer
*/
private void inWireClear() {
inWireByteBuffer().clear();
intWire.bytes().clear();
}
private void clearParked() {
assert inBytesLock().isHeldByCurrentThread();
parkedTransactionId = 0;
parkedTransactionTimeStamp = 0;
}
private void pause() {
assert !outBytesLock().isHeldByCurrentThread();
assert inBytesLock().isHeldByCurrentThread();
/// don't call inBytesLock.isHeldByCurrentThread() as it not atomic
inBytesLock().unlock();
// allows another thread to enter hear
inBytesLock().lock();
}
/**
* reads up to the number of byte in {@code requiredNumberOfBytes} from the socket
*
* @param requiredNumberOfBytes the number of bytes to read
* @param timeoutTime timeout in milliseconds
* @return bytes read from the TCP/IP socket
* @throws java.io.IOException socket failed to read data
*/
@SuppressWarnings("UnusedReturnValue")
private void readSocket(int requiredNumberOfBytes, long timeoutTime) throws IOException {
assert inBytesLock().isHeldByCurrentThread();
ByteBuffer buffer = inWireByteBuffer();
buffer.limit(buffer.position() + requiredNumberOfBytes);
long start = buffer.position();
while (buffer.position() - start < requiredNumberOfBytes) {
int len = clientChannel.read(buffer);
if (len == -1)
throw new IORuntimeException("Disconnection to server");
checkTimeout(timeoutTime);
}
}
private ByteBuffer inWireByteBuffer() {
return (ByteBuffer) intWire.bytes().underlyingObject();
}
/**
* writes the bytes to the socket
*
* @param outWire the data that you wish to write
* @param timeoutTime how long before a we timeout
* @throws IOException
*/
private void writeSocket(Wire outWire, long timeoutTime) throws IOException {
assert outBytesLock().isHeldByCurrentThread();
assert !inBytesLock().isHeldByCurrentThread();
long outBytesPosition = outWire.bytes().position();
// 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() &&
outBytesPosition + largestChunkSoFar <= tcpBufferSize) {
return;
}
final ByteBuffer outBuffer = (ByteBuffer) outWire.bytes().underlyingObject();
outBuffer.limit((int) outWire.bytes().position());
outBuffer.position(4);
System.out.println("\n--------------------------------------------\n" +
"client wrote:\n\n" + Bytes.toDebugString(outBuffer));
outBuffer.position(0);
upateLargestChunkSoFarSize(outBuffer);
while (outBuffer.remaining() > 0) {
int len = clientChannel.write(outBuffer);
if (len == -1)
throw new IORuntimeException("Disconnection to server");
// 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() + largestChunkSoFar <= tcpBufferSize) {
if (LOG.isDebugEnabled())
LOG.debug("continuing - without all the data being written to the buffer as " +
"it will be written by the next thread");
outBuffer.compact();
outWire.bytes().limit(outBuffer.limit());
outWire.bytes().position(outBuffer.position());
return;
}
checkTimeout(timeoutTime);
}
outBuffer.clear();
outWire.bytes().clear();
}
/**
* calculates the size of each chunk
*
* @param outBuffer
*/
private void upateLargestChunkSoFarSize(ByteBuffer outBuffer) {
int sizeOfThisChunk = (int) (outBuffer.limit() - limitOfLast);
if (largestChunkSoFar < sizeOfThisChunk)
largestChunkSoFar = sizeOfThisChunk;
limitOfLast = outBuffer.limit();
}
private long proxySend(@NotNull final String methodName,
final long startTime,
short channelID,
@NotNull final Wire wire) {
assert outBytesLock().isHeldByCurrentThread();
assert !inBytesLock().isHeldByCurrentThread();
// send
outBytesLock().lock();
try {
long transactionId = writeHeader(startTime, channelID, wire);
wire.write(() -> "METHOD_NAME").text(methodName);
writeSocket(wire);
return transactionId;
} finally {
outBytesLock().unlock();
}
}
@SuppressWarnings("SameParameterValue")
@Nullable
String proxyReturnString(@NotNull final String messageId, short channelID) {
return proxyReturnString(messageId, channelID, outWire);
}
@SuppressWarnings("SameParameterValue")
@Nullable
String proxyReturnString(@NotNull final String messageId, short channelID, Wire outWire) {
final long startTime = System.currentTimeMillis();
long transactionId;
outBytesLock().lock();
try {
transactionId = proxySend(messageId, startTime, channelID, outWire);
} finally {
outBytesLock().unlock();
}
long timeoutTime = startTime + this.timeoutMs;
// receive
inBytesLock().lock();
try {
return proxyReply(timeoutTime, transactionId).read(() -> "RESULT").text();
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
inBytesLock().unlock();
}
}
Wire outWire() {
assert outBytesLock().isHeldByCurrentThread();
return outWire;
}
long writeHeader(long startTime, short channelID, Wire wire) {
assert outBytesLock().isHeldByCurrentThread();
markSize(wire);
startTime(startTime);
long transactionId = nextUniqueTransaction(startTime);
wire.write(() -> "TRANSACTION_ID").int64(transactionId);
wire.write(() -> "TIME_STAMP").int64(startTime);
wire.write(() -> "CHANNEL_ID").int16(channelID);
return transactionId;
}
/**
* mark the location of the outWire size
*
* @param outWire
*/
void markSize(Wire outWire) {
assert outBytesLock().isHeldByCurrentThread();
// this is where the size will go
outWire.bytes().mark();
// skip the 2 bytes for the size
outWire.bytes().skip(2);
}
void startTime(long startTime) {
this.startTime = startTime;
}
}
src/test/java/net/openhft/chronicle/map/TestWire.java
-528
View File
@@ -1,528 +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.replication.ReplicationHub;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.wire.Marshallable;
import net.openhft.chronicle.wire.WireIn;
import net.openhft.chronicle.wire.WireOut;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.*;
import static java.util.concurrent.TimeUnit.SECONDS;
import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
import static org.junit.Assert.*;
/**
* Created by Rob Austin
*/
public class TestWire {
public static int SERVER_PORT = 8086;
private ChronicleMap<byte[], byte[]> map1a;
private ChronicleMap<byte[], byte[]> map2a;
private ChronicleMap<byte[], byte[]> map3a;
private ReplicationHub hubA;
@Test
public void testUsingDetails() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, Details> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
Details.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, Details> detailsMap = builder.create();
Details details = detailsMap.get("Colours");
// todo add some sort of cross node locking
if (details == null) {
short channelID = (short) 2;
((WiredStatelessChronicleMap) detailsMap).createChannel(channelID);
details = new Details(String.class, String.class, channelID);
detailsMap.put("Colours", details);
}
ChronicleMap<String, String> colours = new WiredChronicleMapStatelessClientBuilder<String, String>(
builder.hub(),
details.keyClass,
details.valueClass,
details.channelID).hub(builder.hub()).create();
colours.put("Rob", "Blue");
assertEquals(1, colours.size());
Details result = detailsMap.get("Colours");
Assert.assertEquals(2, result.channelID);
map1a.close();
}
}
@Test
public void testCreateChannel() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, String> statelessMap = builder.create();
((WiredStatelessChronicleMap) statelessMap).createChannel((short) 2);
WiredChronicleMapStatelessClientBuilder<String, String> localhost = new WiredChronicleMapStatelessClientBuilder<String, String>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
(short) 2).hub(builder.hub());
ChronicleMap<String, String> map2 = localhost.create();
map2.put("hello", "world");
assertEquals(1, map2.size());
map1a.close();
}
}
@Test
public void testSimplePutStringWithChannels() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
short channelID = (short) 1;
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
channelID);
builder.identifier(identifier);
builder.putReturnsNull(true);
try (ChronicleMap<String, String> statelessMap = builder.create()) {
statelessMap.put("Hello", "World");
assertEquals("World", statelessMap.get("Hello"));
assertTrue("World", statelessMap.containsKey("Hello"));
assertFalse(statelessMap.isEmpty());
}
map1a.close();
}
}
@Test
public void testVersion() throws IOException, InterruptedException {
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
short channelID = (short) 1;
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
channelID);
builder.identifier(identifier);
builder.putReturnsNull(true);
try (ChronicleMap<String, String> statelessMap = builder.create()) {
String s = ((WiredStatelessChronicleMap) statelessMap).serverApplicationVersion();
assertNotNull(s);
}
map1a.close();
}
@Test
public void testLongSize() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
short channelID = (short) 1;
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
channelID);
builder.identifier(identifier);
builder.putReturnsNull(true);
try (ChronicleMap<String, String> statelessMap = builder.create()) {
assertTrue(statelessMap.isEmpty());
statelessMap.put("Hello", "World");
assertEquals(1, statelessMap.longSize());
assertFalse(statelessMap.isEmpty());
}
map1a.close();
}
}
@Test
public void testPutMarshallableWithChannels() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
map2a = of(byte[].class, byte[].class)
.entries(1000)
.instance().replicatedViaChannel(hubA.createChannel((short) 2)).create();
map3a = of(byte[].class, byte[].class)
.entries(1000)
.instance().replicatedViaChannel(hubA.createChannel((short) 3)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, Details> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
Details.class,
(short) 1);
builder.identifier(identifier);
builder.putReturnsNull(true);
final ChronicleMap<String, Details> statelessMap = builder.create();
final Details expected = new Details();
expected.keyClass = String.class;
expected.valueClass = String.class;
statelessMap.put("FirstMap", expected);
final Details actual = statelessMap.get("FirstMap");
assertEquals(expected, actual);
map1a.close();
map2a.close();
map3a.close();
}
}
@Test
public void testGetNullValue() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, String> statelessMap = builder.create();
final String actual = statelessMap.get("FirstMap");
assertEquals(null, actual);
map1a.close();
}
}
@Test
public void testKeySet_Values_EntrySet() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, String> statelessMap = builder.create();
statelessMap.put("Key1", "Value1");
statelessMap.put("Key2", "Value2");
Set<String> keys = statelessMap.keySet();
assertEquals(2, keys.size());
assertTrue(keys.contains("Key2"));
assertTrue(keys.contains("Key1"));
Collection<String> values = statelessMap.values();
assertEquals(2, values.size());
assertTrue(values.contains("Value1"));
assertTrue(values.contains("Value2"));
Set<Map.Entry<String, String>> entries = statelessMap.entrySet();
assertEquals(2, values.size());
assertTrue(entries.contains(Collections.singletonMap("Key1", "Value1").entrySet().iterator().next()));
assertTrue(entries.contains(Collections.singletonMap("Key2", "Value2").entrySet().iterator().next()));
map1a.close();
}
}
@Test
public void testPutAll() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, String> statelessMap = builder.create();
Map<String, String> map = new HashMap<>();
map.put("Key1", "Value1");
map.put("Key2", "Value2");
statelessMap.putAll(map);
Set<String> keys = statelessMap.keySet();
assertEquals(2, keys.size());
assertTrue(keys.contains("Key2"));
assertTrue(keys.contains("Key1"));
Collection<String> values = statelessMap.values();
assertEquals(2, values.size());
assertTrue(values.contains("Value1"));
assertTrue(values.contains("Value2"));
map1a.close();
}
}
public static class Details implements Marshallable {
Class keyClass;
Class valueClass;
short channelID;
public Details() {
}
public Details(Class<String> keyClass, Class<String> valueClass, short channelID) {
this.keyClass = keyClass;
this.valueClass = valueClass;
this.channelID = channelID;
}
@Override
public void writeMarshallable(WireOut wire) {
wire.write(() -> "keyClass").text(keyClass.getName());
wire.write(() -> "valueClass").text(valueClass.getName());
wire.write(() -> "channelID").int16(channelID);
}
@Override
public void readMarshallable(WireIn wire) throws IllegalStateException {
try {
keyClass = Class.forName(wire.read(() -> "keyClass").text());
valueClass = Class.forName(wire.read(() -> "valueClass").text());
channelID = wire.read(() -> "channelID").int16();
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Details details = (Details) o;
if (channelID != details.channelID) return false;
if (!keyClass.equals(details.keyClass)) return false;
if (!valueClass.equals(details.valueClass)) return false;
return true;
}
@Override
public int hashCode() {
int result = keyClass.hashCode();
result = 31 * result + valueClass.hashCode();
result = 31 * result + (int) channelID;
return result;
}
}
}