From 894e4ccde679e330286e24c89f9058d96b1a7c71 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 12 Mar 2015 09:54:32 +0000 Subject: [PATCH] CE-14 moved the stateswire connector for map into engine --- .../chronicle/map/ChannelProvider.java | 11 +- .../chronicle/map/ChronicleMapBuilder.java | 8 +- .../net/openhft/chronicle/map/Replicator.java | 4 +- .../openhft/chronicle/map/Replicators.java | 34 +- .../map/StatelessWiredConnector.java | 1093 ----------------- .../openhft/chronicle/map/TcpReplicator.java | 128 +- ...redChronicleMapStatelessClientBuilder.java | 166 --- .../map/WiredStatelessChronicleMap.java | 911 -------------- .../WiredStatelessClientTcpConnectionHub.java | 667 ---------- .../net/openhft/chronicle/map/TestWire.java | 528 -------- 10 files changed, 41 insertions(+), 3509 deletions(-) delete mode 100644 src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java delete mode 100644 src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java delete mode 100644 src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java delete mode 100644 src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java delete mode 100644 src/test/java/net/openhft/chronicle/map/TestWire.java diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index febf148dd..a6efc7525 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -54,8 +54,7 @@ final class ChannelProvider implements Closeable { static final Map implMapping = new IdentityHashMap<>(); - static synchronized ChannelProvider getProvider(ReplicationHub hub, - Supplier 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 statelessWiredConnectorSupplier) throws IOException { + ChronicleMap chronicleMap + ) throws IOException { add(chronicleChannel, map, entryExternalizable); return this; } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 03f1e0cea..15c3a7f6f 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -1461,19 +1461,15 @@ public final class ChronicleMapBuilder implements } else { final ReplicationHub hub = channel.hub(); - Supplier 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 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); } diff --git a/src/main/java/net/openhft/chronicle/map/Replicator.java b/src/main/java/net/openhft/chronicle/map/Replicator.java index 48112095c..a6307c3f2 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicator.java +++ b/src/main/java/net/openhft/chronicle/map/Replicator.java @@ -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 statelessWiredConnectorSupplier) throws IOException; + final ChronicleMap chronicleMap) throws IOException; } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index 2b1587ac5..d72209de9 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -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 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 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 ); } - }; } } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java deleted file mode 100644 index 04bdee598..000000000 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ /dev/null @@ -1,1093 +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; - -/** - * Created by Rob Austin - */ - -import net.openhft.chronicle.bytes.Bytes; -import net.openhft.chronicle.bytes.BytesMarshallable; -import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; -import net.openhft.chronicle.hash.impl.util.BuildVersion; -import net.openhft.chronicle.hash.replication.ReplicationHub; -import net.openhft.chronicle.network2.TcpHandler; -import net.openhft.chronicle.wire.TextWire; -import net.openhft.chronicle.wire.ValueIn; -import net.openhft.chronicle.wire.Wire; -import net.openhft.chronicle.wire.WireKey; -import net.openhft.lang.io.ByteBufferBytes; -import net.openhft.lang.io.DirectStore; -import net.openhft.lang.io.IByteBufferBytes; -import net.openhft.lang.io.NativeBytes; -import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.io.PrintWriter; -import java.io.StringWriter; -import java.nio.BufferOverflowException; -import java.nio.ByteBuffer; -import java.nio.channels.SelectionKey; -import java.nio.channels.SocketChannel; -import java.util.*; -import java.util.concurrent.Callable; -import java.util.function.Consumer; -import java.util.function.Function; - -import static java.nio.channels.SelectionKey.OP_READ; -import static java.nio.channels.SelectionKey.OP_WRITE; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; - -/** - * @author Rob Austin. - */ -public class StatelessWiredConnector implements TcpHandler { - - private static final Logger LOG = LoggerFactory.getLogger(StatelessWiredConnector.class); - - private final Wire inWire = new TextWire(Bytes.elasticByteBuffer()); - boolean usingTcpReplicator = false; - private Wire outWire = new TextWire(Bytes.elasticByteBuffer()); - - static enum Fields implements WireKey { - HAS_NEXT, - TIME_STAMP, - CHANNEL_ID, - METHOD_NAME, - TRANSACTION_ID, - RESULT, - RESULT_KEY, - RESULT_VALUE, - ARG_1, - ARG_2, - ARG_3, - IS_EXCEPTION, - EXCEPTION, - RESULT_IS_NULL - } - - private final NativeBytes inLangBytes = new NativeBytes(0, 0); - - @NotNull - private final OutMessageAdapter outMessageAdapter = new OutMessageAdapter(); - - @NotNull - private final ArrayList bytesChronicleMaps = new ArrayList<>(); - - - @NotNull - private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64); - private final StringBuilder methodName = new StringBuilder(); - private final ChronicleHashInstanceBuilder> chronicleHashInstanceBuilder; - - private byte localIdentifier; - - private long timestamp; - private short channelId; - private List channelList; - private ReplicationHub hub; - - final Consumer writeElement = new Consumer>() { - - @Override - public void accept(Iterator iterator) { - outWire.write(Fields.RESULT); - outWire.bytes().write(iterator.next()); - } - }; - - final Consumer writeEntry = new Consumer>>() { - - @Override - public void accept(Iterator> iterator) { - - final Map.Entry entry = iterator.next(); - - outWire.write(Fields.RESULT_KEY); - outWire.bytes().write(entry.getKey()); - - outWire.write(Fields.RESULT_VALUE); - outWire.bytes().write(entry.getValue()); - } - }; - - private byte remoteIdentifier; - private boolean hasEatenFirstByte = false; - private Runnable out = null; - private SelectionKey key = null; - - public StatelessWiredConnector(@NotNull final ChronicleHashInstanceBuilder> chronicleHashInstanceBuilder, - @NotNull final ReplicationHub hub, - byte localIdentifier, - @NotNull final List channelList) { - this(chronicleHashInstanceBuilder, hub); - this.channelList = channelList; - this.localIdentifier = localIdentifier; - } - - - public StatelessWiredConnector(ChronicleHashInstanceBuilder> chronicleHashInstanceBuilder, ReplicationHub hub) { - this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder; - this.hub = hub; - } - - @Override - public void process(Bytes in, Bytes out) { - this.inWire.bytes().clear().write(in); - this.inWire.bytes().flip(); - - this.outWire = new TextWire(out); - - - if (inWire.bytes().remaining() > 0) { - - // we will skip the first byte as this is used by the TCP Replicator - if (!hasEatenFirstByte) { - inWire.bytes().readByte(); - hasEatenFirstByte = true; - if (inWire.bytes().remaining() > 0) { - System.out.println("server read:\n" + Bytes.toDebugString(inWire.bytes())); - processMessage(); - return; - } - } - - System.out.println("server read:\n" + Bytes.toDebugString(inWire.bytes())); - processMessage(); - - - } - - // if (inWire.bytes().position() > 0) - // processMessage(); - - } - - public void setChannelList(@NotNull List channelList) { - this.channelList = channelList; - } - - public void setLocalIdentifier(byte localIdentifier) { - this.localIdentifier = localIdentifier; - } - - - public void onWrite(@NotNull SocketChannel socketChannel, @NotNull SelectionKey key) throws IOException { - - if (out != null) { - - long markStart = outWire.bytes().position(); - outWire.bytes().skip(4); - - try { - out.run(); - } finally { - int len = (int) (outWire.bytes().position() - markStart); - outWire.bytes().writeInt(markStart, len); - } - - } - - long position = outWire.bytes().position(); - - if (position > Integer.MAX_VALUE) - throw new BufferOverflowException(); - - ByteBuffer buffer = (ByteBuffer) outWire.bytes().underlyingObject(); - buffer.limit((int) position); - - socketChannel.write(buffer); - - - if (buffer.remaining() == 0) { - buffer.clear(); - outWire.bytes().clear(); - if (out == null) - key.interestOps(OP_READ); - outMessageAdapter.clear(); - } - - } - - - public int onRead(@NotNull SocketChannel socketChannel, SelectionKey key) throws IOException { - usingTcpReplicator = true; - - this.key = key; - - final int len = readSocket(socketChannel); - - if (len == -1 || len == 0) - return len; - - processMessage(); - return len; - } - - private void processMessage() { - while (inWire.bytes().remaining() > SIZE_OF_SIZE) { - - final long limit = inWire.bytes().limit(); - - if (nextWireMessage() == null) { - if (shouldCompactInBuffer()) - compactInBuffer(); - else if (shouldClearInBuffer()) - clearInBuffer(); - return; - } - - long nextPosition = inWire.bytes().limit(); - try { - onEvent(); - } finally { - inWire.bytes().position(nextPosition); - inWire.bytes().limit(limit); - } - } - - } - - private void clearInBuffer() { - inWire.bytes().clear(); - inWireBuffer().clear(); - } - - private boolean shouldClearInBuffer() { - return inWire.bytes().position() == inWireBuffer().position(); - } - - /** - * @return true if remaining space is less than 50% - */ - private boolean shouldCompactInBuffer() { - return inWire.bytes().position() > 0 && inWire.bytes().remaining() < (inWireBuffer().capacity() / 2); - } - - - private void compactInBuffer() { - inWireBuffer().position((int) inWire.bytes().position()); - inWireBuffer().limit(inWireBuffer().position()); - inWireBuffer().compact(); - - inWire.bytes().position(0); - inWire.bytes().limit(0); - } - - @NotNull - private ByteBuffer inWireBuffer() { - return (ByteBuffer) inWire.bytes().underlyingObject(); - } - - private int readSocket(@NotNull SocketChannel socketChannel) throws IOException { - ByteBuffer dst = inWireBuffer(); - int len = socketChannel.read(dst); - int readUpTo = dst.position(); - inWire.bytes().limit(readUpTo); - return len; - } - - @Nullable - private Wire nextWireMessage() { - if (inWire.bytes().remaining() < SIZE_OF_SIZE) - return null; - - final Bytes bytes = inWire.bytes(); - - // the size of the next wire message - int size = bytes.readUnsignedShort(bytes.position()); - - inWire.bytes().ensureCapacity(bytes.position() + size); - - if (bytes.remaining() < size) { - assert size < 100000; - return null; - } - - inWire.bytes().limit(bytes.position() + size); - - // skip the size - inWire.bytes().skip(SIZE_OF_SIZE); - - return inWire; - } - - - Map incompleteWork = new HashMap<>(); - - - private void writeChunked(long transactionId, - @NotNull final Function> function, - @NotNull final Consumer> c) { - - final BytesChronicleMap m = bytesMap(channelId); - final Iterator iterator = function.apply(m); - - // this allows us to write more data than the buffer will allow - out = () -> { - - // each chunk has its own transaction-id - outWire.write(Fields.TRANSACTION_ID).int64(transactionId); - - write(map -> { - - boolean hasNext = iterator.hasNext(); - outWire.write(Fields.HAS_NEXT).bool(hasNext); - - if (hasNext) - c.accept(iterator); - else - // setting out to NULL denotes that there are no more chunks - out = null; - }); - - }; - - out.run(); - - } - - @SuppressWarnings("UnusedReturnValue") - void onEvent() { - - // it is assumed by this point that the buffer has all the bytes in it for this message - long transactionId = inWire.read(Fields.TRANSACTION_ID).int64(); - timestamp = inWire.read(Fields.TIME_STAMP).int64(); - channelId = inWire.read(Fields.CHANNEL_ID).int16(); - inWire.read(Fields.METHOD_NAME).text(methodName); - - if ("PUT_WITHOUT_ACC".contentEquals(methodName)) { - writeVoid(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - bytesMap.put(reader, reader, timestamp, identifier()); - }); - return; - } - - // for the length - long markStart = outWire.bytes().position(); - outWire.bytes().skip(4); - - try { - - if ("KEY_SET".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.keySet().iterator(), writeElement); - return; - } - - if ("VALUES".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.values().iterator(), writeElement); - return; - } - - if ("ENTRY_SET".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.entrySet().iterator(), writeEntry); - return; - } - - if ("PUT_ALL".contentEquals(methodName)) { - putAll(transactionId); - return; - } - - // write the transaction id - outWire.write(() -> "TRANSACTION_ID").int64(transactionId); - - if ("CREATE_CHANNEL".contentEquals(methodName)) { - - writeVoid(() -> { - short channelId1 = inWire.read(Fields.ARG_1).int16(); - chronicleHashInstanceBuilder.replicatedViaChannel(hub.createChannel(channelId1)).create(); - return null; - }); - return; - } - - - if ("REMOTE_IDENTIFIER".contentEquals(methodName)) { - this.remoteIdentifier = inWire.read(Fields.RESULT).int8(); - return; - } - - if ("LONG_SIZE".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT).int64(b.longSize())); - return; - } - - if ("IS_EMPTY".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT).bool(b.isEmpty())); - return; - } - - if ("CONTAINS_KEY".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT). - bool(b.containsKey(toReader(inWire, Fields.ARG_1)))); - return; - } - - if ("CONTAINS_VALUE".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT). - bool(b.containsKey(toReader(inWire, Fields.ARG_1)))); - return; - } - - if ("GET".contentEquals(methodName)) { - - writeValueUsingDelegate(map -> { - byte[] key1 = this.toByteArray(inWire, Fields.ARG_1); - return map.get(key1); - }); - - return; - - } - - if ("PUT".contentEquals(methodName)) { - writeValue(b -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - return b.put(reader, reader, timestamp, identifier()); - }); - - return; - } - - if ("REMOVE".contentEquals(methodName)) { - writeValue(b -> b.remove(toReader(inWire, Fields.ARG_1))); - return; - } - - if ("CLEAR".contentEquals(methodName)) { - writeVoid(BytesChronicleMap::clear); - return; - } - - if ("REPLACE".contentEquals(methodName)) { - writeValue(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - return bytesMap.replace(reader, reader); - }); - return; - } - - if ("REPLACE_WITH_OLD_AND_NEW_VALUE".contentEquals(methodName)) { - - writeValue(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2, Fields.ARG_3); - return bytesMap.replace(reader, reader); - }); - - return; - } - - if ("PUT_IF_ABSENT".contentEquals(methodName)) { - writeValue(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - return bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); - }); - return; - } - - if ("REMOVE_WITH_VALUE".contentEquals(methodName)) { - write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - outWire.write(Fields.RESULT).bool(bytesMap.remove(reader, reader, timestamp, identifier())); - }); - return; - } - - if ("TO_STRING".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT).text(b.toString())); - return; - } - - if ("APPLICATION_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT).text(applicationVersion())); - return; - } - - if ("PERSISTED_DATA_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT).text(persistedDataVersion())); - - return; - } - - if ("HASH_CODE".contentEquals(methodName)) { - write(b -> outWire.write(Fields.RESULT).int32(b.hashCode())); - return; - } - - throw new IllegalStateException("unsupported event=" + methodName); - - } finally { - - - int len = (int) (outWire.bytes().position() - markStart); - - if (len == 4) - // remove the 4 byte len as nothing was written - outWire.bytes().position(markStart); - else - outWire.bytes().writeInt(markStart, len); - - if (len > 4) - System.out.println("\n----------------------------\n\nserver wrote:\n" + Bytes.toDebugString(outWire.bytes(), markStart + 4, len - 4)); - } - - - } - - - private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) { - if (bytes == null || bytes.remaining() == 0) - return new byte[]{}; - - if (bytes.remaining() > Integer.MAX_VALUE) - throw new BufferOverflowException(); - - byte[] result = new byte[(int) bytes.remaining()]; - bytes.write(result); - return result; - } - - private byte[] toBytes(WireKey fieldName) { - - Wire wire = inWire; - System.out.println(Bytes.toDebugString(inWire.bytes())); - - ValueIn read = wire.read(fieldName); - - long l = read.readLength(); - if (l > Integer.MAX_VALUE) - throw new BufferOverflowException(); - - int fieldLength = (int) l; - - long endPos = wire.bytes().position() + fieldLength; - long limit = wire.bytes().limit(); - - try { - byte[] bytes = new byte[fieldLength]; - - wire.bytes().read(bytes); - return bytes; - - } finally - - { - wire.bytes().position(endPos); - wire.bytes().limit(limit); - } - } - - private void putAll(long transactionId) { - - final BytesChronicleMap bytesMap = bytesMap(StatelessWiredConnector.this.channelId); - - if (bytesMap == null) - return; - - Runnable runnable = incompleteWork.get(transactionId); - - if (runnable != null) { - runnable.run(); - return; - } - - runnable = new Runnable() { - - // we should try and collect the data and then apply it atomically as quickly possible - final Map collectData = new HashMap<>(); - - @Override - public void run() { - if (inWire.read(Fields.HAS_NEXT).bool()) { - collectData.put(toBytes(Fields.ARG_1), toBytes(Fields.ARG_2)); - } else { - // the old code assumed that all the data would fit into a single buffer - // this assumption is invalid - if (!collectData.isEmpty()) { - bytesMap.delegate.putAll((Map) collectData); - incompleteWork.remove(transactionId); - outWire.write(Fields.TRANSACTION_ID).int64(transactionId); - - // todo handle the case where there is an exception - outWire.write(Fields.IS_EXCEPTION).bool(false); - - nofityDataWritten(); - } - } - } - }; - - incompleteWork.put(transactionId, runnable); - runnable.run(); - - } - - - private byte identifier() { - // if the client provides a remote identifier then we will use that otherwise we will use the local identifier - return remoteIdentifier == 0 ? localIdentifier : remoteIdentifier; - } - - @NotNull - private CharSequence persistedDataVersion() { - BytesChronicleMap bytesChronicleMap = bytesMap(channelId); - if (bytesChronicleMap == null) - return ""; - return bytesChronicleMap.persistedDataVersion(); - } - - @NotNull - private CharSequence applicationVersion() { - return BuildVersion.version(); - } - - /** - * creates a lang buffer that holds just the payload of the args - * - * @param wire the inbound wire - * @param args the key names of the {@code wire} args - * @return a new lang buffer containing the bytes of the args - */ - private net.openhft.lang.io.Bytes toReader(@NotNull Wire wire, @NotNull WireKey... args) { - - long inSize = wire.bytes().limit(); - final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize).bytes(); - - // copy the bytes to the reader - for (final WireKey field : args) { - - ValueIn read = wire.read(field); - long fieldLength = read.readLength(); - - long endPos = wire.bytes().position() + fieldLength; - long limit = wire.bytes().limit(); - - try { - - final Bytes source = wire.bytes(); - source.limit(endPos); - - // write the size - bytes.writeStopBit(source.remaining()); - - while (source.remaining() > 0) { - if (source.remaining() >= 8) - bytes.writeLong(source.readLong()); - else - bytes.writeByte(source.readByte()); - } - - } finally { - wire.bytes().position(endPos); - wire.bytes().limit(limit); - } - - } - - return bytes.flip(); - } - - - /** - * creates a lang buffer that holds just the payload of the args - * - * @param wire the inbound wire - * @return a new lang buffer containing the bytes of the args - */ - - // todo remove this method - just added to get it to work for now - private byte[] toByteArray(@NotNull Wire wire, @NotNull WireKey field) { - - long inSize = wire.bytes().limit(); - // final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize).bytes(); - - - ValueIn read = wire.read(field); - long fieldLength = read.readLength(); - - long endPos = wire.bytes().position() + fieldLength; - long limit = wire.bytes().limit(); - byte[] result = new byte[]{}; - try { - - final Bytes source = wire.bytes(); - source.limit(endPos); - - if (source.remaining() > Integer.MAX_VALUE) - throw new BufferOverflowException(); - - // write the size - result = new byte[(int) source.remaining()]; - source.read(result); - - - } finally { - wire.bytes().position(endPos); - wire.bytes().limit(limit); - } - - - return result; - } - - /** - * gets the map for this channel id - * - * @param channelId the ID of the map - * @return the chronicle map with this {@code channelId} - */ - @NotNull - private ReplicatedChronicleMap map(short channelId) { - - // todo this cast is a bit of a hack, improve later - final ReplicatedChronicleMap replicas = - (ReplicatedChronicleMap) channelList.get(channelId); - - if (replicas != null) - return replicas; - - throw new IllegalStateException(); - } - - /** - * this is used to push the data straight into the entry in memory - * - * @param channelId the ID of the map - * @return a BytesChronicleMap used to update the memory which holds the chronicle map - */ - @Nullable - private BytesChronicleMap bytesMap(short channelId) { - - final BytesChronicleMap bytesChronicleMap = (channelId < bytesChronicleMaps.size()) - ? bytesChronicleMaps.get(channelId) - : null; - - if (bytesChronicleMap != null) - return bytesChronicleMap; - - // grow the array - for (int i = bytesChronicleMaps.size(); i <= channelId; i++) { - bytesChronicleMaps.add(null); - } - - final ReplicatedChronicleMap delegate = map(channelId); - final BytesChronicleMap element = new BytesChronicleMap(delegate); - bytesChronicleMaps.set(channelId, element); - return element; - - } - - - @SuppressWarnings("SameReturnValue") - private void writeValue(final Function f) { - - write(b -> { - - byte[] fromBytes = toByteArray(f.apply(b)); - boolean isNull = fromBytes.length == 0; - outWire.write(Fields.RESULT_IS_NULL).bool(isNull); - if (isNull) - return; - - outWire.write(Fields.RESULT); - outWire.bytes().write(fromBytes); - - }); - - - nofityDataWritten(); - - } - - - @SuppressWarnings("SameReturnValue") - private void writeValueUsingDelegate(final Function, byte[]> f) { - - write(b -> { - - byte[] result = f.apply((ChronicleMap) b.delegate); - boolean isNull = result == null; - - outWire.write(Fields.RESULT_IS_NULL).bool(isNull); - if (isNull) - return; - - isNull = result.length == 0; - - outWire.write(Fields.RESULT); - outWire.bytes().write(result); - - }); - - - nofityDataWritten(); - - } - - private void writeValueTcpRep(@NotNull Consumer process) { - - final BytesChronicleMap bytesMap = bytesMap(channelId); - - if (bytesMap == null) { - LOG.error("no map for channelId=" + channelId + " can be found."); - return; - } - - inLanByteBuffer.clear(); - inLangBytes.clear(); - outMessageAdapter.markStartOfMessage(); - - bytesMap.output = outMessageAdapter.outBuffer; - - try { - process.accept(bytesMap); - } catch (Exception e) { - - // the idea of wire is that is platform independent, - // so we wil have to send the exception as a String - outWire.write(Fields.IS_EXCEPTION).bool(true); - outWire.write(Fields.EXCEPTION).text(toString(e)); - LOG.error("", e); - return; - } - - outMessageAdapter.accept(outWire); - nofityDataWritten(); - - } - - @SuppressWarnings("SameReturnValue") - private void write(@NotNull Consumer c) { - - final BytesChronicleMap bytesMap = bytesMap(channelId); - - if (bytesMap == null) { - LOG.error("no map for channelId=" + channelId + " can be found."); - return; - } - - bytesMap.output = null; - outWire.bytes().mark(); - outWire.write(Fields.IS_EXCEPTION).bool(false); - - try { - c.accept(bytesMap); - } catch (Exception e) { - outWire.bytes().reset(); - // the idea of wire is that is platform independent, - // so we wil have to send the exception as a String - outWire.write(Fields.IS_EXCEPTION).bool(true); - outWire.write(Fields.EXCEPTION).text(toString(e)); - LOG.error("", e); - return; - } - - nofityDataWritten(); - - } - - @SuppressWarnings("SameReturnValue") - private void writeVoid(@NotNull Callable r) { - - final BytesChronicleMap bytesMap = bytesMap(channelId); - - if (bytesMap == null) { - LOG.error("no map for channelId=" + channelId + " can be found."); - return; - } - - bytesMap.output = null; - outWire.bytes().mark(); - outWire.write(Fields.IS_EXCEPTION).bool(false); - - try { - r.call(); - } catch (Exception e) { - outWire.bytes().reset(); - // the idea of wire is that is platform independent, - // so we wil have to send the exception as a String - outWire.write(Fields.IS_EXCEPTION).bool(true); - outWire.write(Fields.EXCEPTION).text(toString(e)); - LOG.error("", e); - return; - } - - nofityDataWritten(); - - } - - - /** - * only used for debugging - */ - @SuppressWarnings("UnusedDeclaration") - private void showOutWire() { - System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); - } - - @SuppressWarnings("SameReturnValue") - private void writeVoid(@NotNull Consumer process) { - - // skip 4 bytes where we will write the size - final BytesChronicleMap bytesMap = bytesMap(channelId); - - if (bytesMap == null) { - LOG.error("no map for channelId=" + channelId + " can be found."); - return; - } - - bytesMap.output = null; - try { - process.accept(bytesMap); - } catch (Exception e) { - LOG.error("", e); - return; - } - - - } - - /** - * converts the exception into a String, so that it can be sent to c# clients - */ - private String toString(@NotNull Throwable t) { - final StringWriter sw = new StringWriter(); - final PrintWriter pw = new PrintWriter(sw); - t.printStackTrace(pw); - return sw.toString(); - } - - public void localIdentifier(byte localIdentifier) { - this.localIdentifier = localIdentifier; - } - - - /** - * used to send data to the wired stateless client, this class converts/adapts a chronicle map - * lang-bytes output into chronicle-bytes - */ - private static class OutMessageAdapter { - - - @NotNull - private final IByteBufferBytes outLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate(1024)); - private Bytes outChronBytes = Bytes.wrap(outLangBytes.buffer()); - - // chronicle map will write its result in here - final MapIOBuffer outBuffer = new MapIOBuffer() { - - @Override - public void ensureBufferSize(long l) { - - if (outLangBytes.remaining() >= l) - return; - - long size = outLangBytes.capacity() + l; - if (size > Integer.MAX_VALUE) - throw new BufferOverflowException(); - - // record the current position and limit - long position = outLangBytes.position(); - - // create a new buffer and copy the data into it - outLangBytes.clear(); - final IByteBufferBytes newOutLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate((int) size)); - newOutLangBytes.write(outLangBytes); - - outChronBytes = Bytes.wrap(outLangBytes.buffer()); - - newOutLangBytes.limit(newOutLangBytes.capacity()); - newOutLangBytes.position(position); - - outChronBytes.limit(outChronBytes.capacity()); - outChronBytes.position(position); - } - - @NotNull - @Override - public net.openhft.lang.io.Bytes in() { - return outLangBytes; - } - }; - private long startChunk; - - - void clear() { - outChronBytes.clear(); - outLangBytes.clear(); - } - - - /** - * adapts the chronicle out lang bytes to chroncile bytes - * - * @param wire the wire that we wish to append data to - */ - void accept(@NotNull Wire wire) { - - wire.write(Fields.IS_EXCEPTION).bool(false); - - // flips calls flip on this message so that we can read it - flipMessage(); - - // set the chron-bytes and the lang-bytes to be the same - outChronBytes.position(outLangBytes.position()); - outChronBytes.limit(outLangBytes.limit()); - - // is Null - boolean isNull = outChronBytes.readBoolean(); - - // read the size - not used - outChronBytes.readStopBit(); - - wire.write(Fields.RESULT_IS_NULL).bool(isNull); - if (!isNull) { - // write the result - wire.write(Fields.RESULT); - wire.bytes().write(outChronBytes); - } - } - - private void flipMessage() { - long position = outLangBytes.position(); - outLangBytes.position(startChunk); - outLangBytes.limit(position); - } - - /** - * marks the start of the message - */ - public void markStartOfMessage() { - startChunk = outLangBytes.position(); - } - - } - - private void nofityDataWritten() { - if (key != null) - key.interestOps(OP_WRITE | OP_READ); - } - - -} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 5ec7c9871..4333c2470 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -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 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 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 extends AbstractChannelReplicator implements Clo StatelessClientParameters statelessClientParameters; - private Supplier statelessWiredConnectorSupplier; - private List channelList; - static class StatelessClientParameters { public StatelessClientParameters(VanillaChronicleMap map, SerializationBuilder keySerializationBuilder, @@ -134,17 +127,13 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo @NotNull final TcpTransportAndNetworkConfig replicationConfig, @Nullable final RemoteNodeValidator remoteNodeValidator, @Nullable final StatelessClientParameters statelessClientParameters, - String name, - @Nullable final List channelList, - @Nullable Supplier 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 extends AbstractChannelReplicator implements Clo this.remoteNodeValidator = remoteNodeValidator; this.name = name; - - start(); } @@ -562,7 +549,6 @@ final class TcpReplicator 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 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 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 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 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 extends AbstractChannelReplicator implements Clo return; } - if (len == 0 && !forceRead) + if (len == 0) return; if (attached.entryWriter.isWorkIncomplete()) @@ -1081,22 +996,20 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo */ class Attached 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 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 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++) { diff --git a/src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java b/src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java deleted file mode 100644 index 68f17ec8e..000000000 --- a/src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java +++ /dev/null @@ -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 implements - ChronicleHashStatelessClientBuilder, - ChronicleMap>, - MapBuilder> { - - 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..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 putReturnsNull(boolean putReturnsNull) { - this.putReturnsNull = putReturnsNull; - return this; - } - - boolean putReturnsNull() { - return putReturnsNull; - } - - @Override - public WiredChronicleMapStatelessClientBuilder removeReturnsNull(boolean removeReturnsNull) { - this.removeReturnsNull = removeReturnsNull; - return this; - } - - boolean removeReturnsNull() { - return removeReturnsNull; - } - - @Override - public WiredChronicleMapStatelessClientBuilder timeout(long timeout, TimeUnit units) { - this.timeoutMs = units.toMillis(timeout); - return this; - } - - long timeoutMs() { - return timeoutMs; - } - - @Override - public WiredChronicleMapStatelessClientBuilder name(String name) { - this.name = name; - return this; - } - - String name() { - return name; - } - - @Override - public WiredChronicleMapStatelessClientBuilder tcpBufferSize(int tcpBufferSize) { - this.tcpBufferSize = tcpBufferSize; - return this; - } - - int tcpBufferSize() { - return tcpBufferSize; - } - - @Override - public ChronicleMap create() throws IOException { - - // todo clean this up - if (hub == null) - hub = new WiredStatelessClientTcpConnectionHub(this, localIdentifier); - - if (!used.getAndSet(true)) { - return new WiredStatelessChronicleMap(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 hub(WiredStatelessClientTcpConnectionHub hub) { - this.hub = hub; - return this; - } - -} diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java deleted file mode 100644 index 5262cdeaa..000000000 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ /dev/null @@ -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 implements ChronicleMap, Cloneable { - - private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class); - - private final WiredStatelessClientTcpConnectionHub hub; - - protected Class kClass; - protected Class 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 keyClass() { - return kClass; - } - - @Override - public boolean forEachEntryWhile(Predicate> predicate) { - // TODO implement! - throw new UnsupportedOperationException(); - } - - @Override - public void forEachEntry(Consumer> action) { - // TODO implement! - throw new UnsupportedOperationException(); - } - - - public String serverApplicationVersion() { - return hub.serverApplicationVersion(channelID); - } - - @Override - public void close() { - // todo add ref count - } - - @Override - public Class 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 that = (Map) object; - - final int size = size(); - - if (that.size() != size) - return false; - - final Set> 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 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 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 map) { - - final long startTime = System.currentTimeMillis(); - long transactionId = hub.nextUniqueTransaction(startTime); - - - Set> entries = map.entrySet(); - Iterator> 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 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 getMapped(@Nullable K key, @NotNull SerializableFunction function) { - throw new UnsupportedOperationException(); - } - - @Nullable - @Override - public V putMapped(@Nullable K key, @NotNull UnaryOperator unaryOperator) { - throw new UnsupportedOperationException(); - } - - public void clear() { - proxyReturnVoid(() -> writeField(StatelessWiredConnector.Fields.METHOD_NAME, CLEAR.toString())); - } - - - public Set keySet() { - final long startTime = System.currentTimeMillis(); - final Set result = new HashSet(); - // 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 values() { - final long startTime = System.currentTimeMillis(); - final Set result = new HashSet(); - // 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> entrySet() { - final long startTime = System.currentTimeMillis(); - final Map result = new HashMap(); - - // 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 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 proxyReturnObject(@NotNull final String methodName, K key, V value, Class 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 proxyReturnObject(Class 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 { - - 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(); - } - } -} - diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java deleted file mode 100644 index c1eed05e5..000000000 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java +++ /dev/null @@ -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; - } -} diff --git a/src/test/java/net/openhft/chronicle/map/TestWire.java b/src/test/java/net/openhft/chronicle/map/TestWire.java deleted file mode 100644 index 4e86856be..000000000 --- a/src/test/java/net/openhft/chronicle/map/TestWire.java +++ /dev/null @@ -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 map1a; - private ChronicleMap map2a; - private ChronicleMap 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - Details.class, - (short) 1); - - builder.identifier(identifier); - final ChronicleMap 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 colours = new WiredChronicleMapStatelessClientBuilder( - 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - (short) 1); - - builder.identifier(identifier); - final ChronicleMap statelessMap = builder.create(); - ((WiredStatelessChronicleMap) statelessMap).createChannel((short) 2); - - WiredChronicleMapStatelessClientBuilder localhost = new WiredChronicleMapStatelessClientBuilder( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - (short) 2).hub(builder.hub()); - - ChronicleMap 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - channelID); - - builder.identifier(identifier); - builder.putReturnsNull(true); - - try (ChronicleMap 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - channelID); - - builder.identifier(identifier); - builder.putReturnsNull(true); - - try (ChronicleMap 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - channelID); - - builder.identifier(identifier); - builder.putReturnsNull(true); - - try (ChronicleMap 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - Details.class, - (short) 1); - - builder.identifier(identifier); - builder.putReturnsNull(true); - final ChronicleMap 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - (short) 1); - - builder.identifier(identifier); - final ChronicleMap 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - (short) 1); - - builder.identifier(identifier); - final ChronicleMap statelessMap = builder.create(); - - statelessMap.put("Key1", "Value1"); - statelessMap.put("Key2", "Value2"); - - - Set keys = statelessMap.keySet(); - - assertEquals(2, keys.size()); - assertTrue(keys.contains("Key2")); - assertTrue(keys.contains("Key1")); - - Collection values = statelessMap.values(); - - assertEquals(2, values.size()); - assertTrue(values.contains("Value1")); - assertTrue(values.contains("Value2")); - - - Set> 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 builder = - new WiredChronicleMapStatelessClientBuilder<>( - new InetSocketAddress("localhost", SERVER_PORT), - String.class, - String.class, - (short) 1); - - builder.identifier(identifier); - final ChronicleMap statelessMap = builder.create(); - - - Map map = new HashMap<>(); - map.put("Key1", "Value1"); - map.put("Key2", "Value2"); - - statelessMap.putAll(map); - - Set keys = statelessMap.keySet(); - - assertEquals(2, keys.size()); - assertTrue(keys.contains("Key2")); - assertTrue(keys.contains("Key1")); - - Collection 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 keyClass, Class 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; - } - } -}