From c5c52b2e4ac2e278ee7ef42089f052a7aa216a3d Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 11 Mar 2015 16:29:33 +0000 Subject: [PATCH] added support for the WiredStatelesssChroncileMap to work using Chronicle-Network2 --- pom.xml | 14 +- .../chronicle/map/ChannelProvider.java | 4 +- .../map/StatelessWiredConnector.java | 198 ++++++++++++++++-- .../openhft/chronicle/map/TcpReplicator.java | 2 +- .../map/WiredStatelessChronicleMap.java | 8 +- 5 files changed, 196 insertions(+), 30 deletions(-) diff --git a/pom.xml b/pom.xml index f2a2de244..767259f31 100755 --- a/pom.xml +++ b/pom.xml @@ -75,6 +75,12 @@ 1.0.0-alpha-SNAPSHOT + + net.openhft + chronicle-network + 1.0.0-alpha-SNAPSHOT + + @@ -114,13 +120,11 @@ - com.thoughtworks.xstream xstream - org.codehaus.jettison jettison @@ -131,6 +135,12 @@ pax-url-aether + + net.openhft + chronicle-network + + + diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 8a8e3fe19..febf148dd 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -54,7 +54,7 @@ final class ChannelProvider implements Closeable { static final Map implMapping = new IdentityHashMap<>(); - static synchronized ChannelProvider getProvider(ReplicationHub hub, + static synchronized ChannelProvider getProvider(ReplicationHub hub, Supplier statelessWiredConnectorSupplier) throws IOException { ChannelProvider channelProvider = implMapping.get(hub); if (channelProvider != null) @@ -272,7 +272,7 @@ final class ChannelProvider implements Closeable { private final Replica[] chronicleChannels; - private List chronicleChannelList() { + List chronicleChannelList() { return Collections.unmodifiableList(chronicleChannelList); } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 9d9590b24..1da0beafd 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -27,6 +27,7 @@ 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; @@ -58,10 +59,14 @@ import static java.nio.channels.SelectionKey.OP_WRITE; /** * @author Rob Austin. */ -class StatelessWiredConnector { +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, @@ -87,8 +92,6 @@ class StatelessWiredConnector bytesChronicleMaps = new ArrayList<>(); - private final TextWire inWire = new TextWire(Bytes.elasticByteBuffer()); - private final TextWire outWire = new TextWire(Bytes.elasticByteBuffer()); @NotNull private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64); @@ -97,7 +100,6 @@ class StatelessWiredConnector channelList; @@ -128,8 +130,18 @@ class StatelessWiredConnector> 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) { @@ -137,7 +149,37 @@ class StatelessWiredConnector 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; @@ -185,8 +227,9 @@ class StatelessWiredConnector 4) { final long limit = inWire.bytes().limit(); @@ -205,7 +252,7 @@ class StatelessWiredConnector b.get(toReader(inWire, Fields.ARG_1))); + + writeValueUsingDelegate(map -> { + byte[] key1 = this.toByteArray(inWire, Fields.ARG_1); + return map.get(key1); + }); + return; + } if ("PUT".contentEquals(methodName)) { - writeValue(bytesMap -> { + writeValue(b -> { final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - bytesMap.put(reader, reader, timestamp, identifier()); + return b.put(reader, reader, timestamp, identifier()); }); + return; } @@ -426,31 +480,33 @@ class StatelessWiredConnector { final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - bytesMap.replace(reader, reader); + 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); - bytesMap.replace(reader, reader, reader); + 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); - bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); + return bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); }); return; } if ("REMOVE_WITH_VALUE".contentEquals(methodName)) { - writeValue(bytesMap -> { + write(bytesMap -> { final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); - bytesMap.remove(reader, reader, timestamp, identifier()); + outWire.write(Fields.RESULT).bool(bytesMap.remove(reader, reader, timestamp, identifier())); }); return; } @@ -496,6 +552,19 @@ class StatelessWiredConnector 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; @@ -574,7 +643,7 @@ class StatelessWiredConnector 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 * @@ -686,7 +797,52 @@ class StatelessWiredConnector process) { + 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); @@ -718,7 +874,6 @@ class StatelessWiredConnector c) { @@ -821,7 +976,7 @@ class StatelessWiredConnector extends AbstractChannelReplicator implements Clo @Nullable final StatelessClientParameters statelessClientParameters, String name, @Nullable final List channelList, - @NotNull Supplier statelessWiredConnectorSupplier) + @Nullable Supplier statelessWiredConnectorSupplier) throws IOException { super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig()); diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index 1203bda14..5262cdeaa 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -53,10 +53,10 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo private boolean removeReturnsNull; private short channelID; - WiredStatelessChronicleMap(@NotNull final WiredChronicleMapStatelessClientBuilder config, - @NotNull final Class kClass, - @NotNull final Class vClass, - 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();