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 extends StatelessWiredConnector> 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 extends StatelessWiredConnector> statelessWiredConnectorSupplier)
+ @Nullable Supplier extends StatelessWiredConnector> 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();