now supports the creating of channels

src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java

@@ -57,4 +57,6 @@ public interface ChronicleHashInstanceBuilder<H extends ChronicleHash> {
     ChronicleHashInstanceBuilder<H> name(String name);
 
     H create() throws IOException;
+
+
 }

src/main/java/net/openhft/chronicle/map/ChannelProvider.java

@@ -38,6 +38,7 @@ import java.util.concurrent.LinkedTransferQueue;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
 
 import static java.lang.Math.min;
 import static java.nio.ByteBuffer.wrap;
@@ -53,7 +54,8 @@ final class ChannelProvider implements Closeable {
 
     static final Map<ReplicationHub, ChannelProvider> implMapping = new IdentityHashMap<>();
 
-    static synchronized ChannelProvider getProvider(ReplicationHub hub) throws IOException {
+    static synchronized ChannelProvider getProvider(ReplicationHub hub,
+                                                    Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException {
         ChannelProvider channelProvider = implMapping.get(hub);
         if (channelProvider != null)
             return channelProvider;
@@ -70,7 +72,8 @@ final class ChannelProvider implements Closeable {
                     hub.remoteNodeValidator(),
                     null,
                     hub.name(),
-                    channelProvider.chronicleChannelList());
+                    channelProvider.chronicleChannelList(),
+                    statelessWiredConnectorSupplier);
 
             channelProvider.add(tcpReplicator);
         }
@@ -547,13 +550,6 @@ final class ChannelProvider implements Closeable {
             return localIdentifier;
         }
 
-        @Override
-        protected Closeable applyTo(ChronicleMapBuilder builder,
-                                    Replica map, EntryExternalizable entryExternalizable,
-                                    final ChronicleMap chronicleMap) {
-            add(chronicleChannel, map, entryExternalizable);
-            return this;
-        }
 
         @Override
         public void close() throws IOException {
@@ -570,5 +566,15 @@ final class ChannelProvider implements Closeable {
                 channelDataLock.writeLock().unlock();
             }
         }
+
+        @Override
+        protected Closeable applyTo(ChronicleMapBuilder builder,
+                                    Replica map,
+                                    EntryExternalizable entryExternalizable,
+                                    ChronicleMap chronicleMap,
+                                    Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException {
+            add(chronicleChannel, map, entryExternalizable);
+            return this;
+        }
     }
 }

src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java

@@ -22,12 +22,12 @@ import net.openhft.chronicle.hash.ChronicleHashBuilder;
 import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
 import net.openhft.chronicle.hash.impl.ChronicleHashBuilderImpl;
 import net.openhft.chronicle.hash.impl.hashlookup.HashLookup;
-import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
 import net.openhft.chronicle.hash.replication.*;
 import net.openhft.chronicle.hash.serialization.*;
 import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
 import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
 import net.openhft.chronicle.hash.serialization.internal.MetaProvider;
+import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
 import net.openhft.chronicle.set.ChronicleSetBuilder;
 import net.openhft.lang.Maths;
 import net.openhft.lang.io.Bytes;
@@ -54,6 +54,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 
 import static java.lang.Math.round;
 import static net.openhft.chronicle.hash.impl.util.Objects.builderEquals;
@@ -1458,14 +1459,21 @@ public final class ChronicleMapBuilder<K, V> implements
                 if (singleHashReplication.udpTransport() != null)
                     replicators.add(Replicators.udp(singleHashReplication.udpTransport()));
             } else {
-                ReplicationHub hub = channel.hub();
+                final ReplicationHub hub = channel.hub();
 
-                ChannelProvider provider = ChannelProvider.getProvider(hub);
+                Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier =
+                        () -> new StatelessWiredConnector(instance(), hub);
+
+                ChannelProvider provider = ChannelProvider.getProvider(hub, statelessWiredConnectorSupplier);
                 ChannelProvider.ChronicleChannel ch = provider.createChannel(channel.channelId());
                 replicators.add(ch);
             }
             for (Replicator replicator : replicators) {
-                Closeable token = replicator.applyTo(this, result, result, map);
+
+                Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier =
+                        () -> new StatelessWiredConnector(instance(), channel.hub());
+
+                Closeable token = replicator.applyTo(this, result, result, map, statelessWiredConnectorSupplier);
                 if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) {
                     LOG.warn(Replicators.ONLY_UDP_WARN_MESSAGE);
                 }

src/main/java/net/openhft/chronicle/map/Replicator.java

@@ -18,6 +18,7 @@ package net.openhft.chronicle.map;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.function.Supplier;
 
 abstract class Replicator {
 
@@ -49,5 +50,6 @@ abstract class Replicator {
      */
     protected abstract Closeable applyTo(ChronicleMapBuilder builder,
                                          Replica map, Replica.EntryExternalizable entryExternalizable,
-                                         final ChronicleMap chronicleMap) throws IOException;
+                                         final ChronicleMap chronicleMap,
+                                         Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException;
 }

src/main/java/net/openhft/chronicle/map/Replicators.java

@@ -27,6 +27,7 @@ import java.io.Closeable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.util.function.Supplier;
 
 final class Replicators {
 
@@ -111,25 +112,28 @@ final class Replicators {
 
 
             @Override
-            protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
-                                        @NotNull final Replica replica,
-                                        @NotNull final Replica.EntryExternalizable entryExternalizable,
-                                        final ChronicleMap chronicleMap) throws IOException {
+            protected Closeable applyTo(ChronicleMapBuilder builder,
+                                        Replica map,
+                                        Replica.EntryExternalizable entryExternalizable,
+                                        ChronicleMap chronicleMap,
+                                        Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) 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(replica, entryExternalizable,
+                return new TcpReplicator(map, entryExternalizable,
                         tcpConfig,
                         replication.remoteNodeValidator(), statelessClientParameters,
                         replication.name(),
-                        null);
+                        null, statelessWiredConnectorSupplier);
             }
+
+
         };
     }
 
@@ -137,14 +141,15 @@ final class Replicators {
             final UdpTransportConfig replicationConfig) {
         return new Replicator() {
             @Override
-            protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
-                                        @NotNull final Replica map,
-                                        @NotNull final Replica.EntryExternalizable entryExternalizable,
-                                        final ChronicleMap chronicleMap)
-                    throws IOException {
+            protected Closeable applyTo(ChronicleMapBuilder builder,
+                                        Replica map,
+                                        Replica.EntryExternalizable entryExternalizable,
+                                        ChronicleMap chronicleMap,
+                                        Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier) throws IOException {
                 return new UdpReplicator(map, entryExternalizable, replicationConfig
                 );
             }
+
         };
     }
 }

src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java

@@ -24,7 +24,9 @@ package net.openhft.chronicle.map;
 
 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.wire.TextWire;
 import net.openhft.chronicle.wire.ValueIn;
 import net.openhft.chronicle.wire.Wire;
@@ -45,6 +47,7 @@ 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;
 
@@ -70,13 +73,16 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
     @NotNull
     private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64);
     private final StringBuilder methodName = new StringBuilder();
-    private final byte localIdentifier;
+    private final ChronicleHashInstanceBuilder<ChronicleMap<K, V>> chronicleHashInstanceBuilder;
+
+    private byte localIdentifier;
+
 
     private long timestamp;
     private short channelId;
-
     @NotNull
-    private final List<Replica> channelList;
+    private List<Replica> channelList;
+    private ReplicationHub hub;
 
     final Consumer writeElement = new Consumer<Iterator<byte[]>>() {
 
@@ -108,19 +114,27 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
     private byte localIdentifer;
     private Runnable out = null;
 
-    public StatelessWiredConnector(@Nullable List<Replica> channelList, byte localIdentifier) {
-
-        if (channelList == null) {
-            throw new IllegalStateException("The StatelessWiredConnector currently only support maps " +
-                    "that are set up via the replication channel. localIdentifier=" + localIdentifier);
-        }
-        this.channelList = channelList;
-        this.localIdentifier = localIdentifier;
 
+    public StatelessWiredConnector(ChronicleHashInstanceBuilder<ChronicleMap<K, V>> chronicleHashInstanceBuilder, ReplicationHub hub) {
+        this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder;
+        this.hub = hub;
     }
 
     private SelectionKey key;
 
+    public void setChannelList(@NotNull List<Replica> channelList) {
+        if (channelList == null)
+            throw new IllegalStateException("The StatelessWiredConnector currently only support maps " +
+                    "that are set up via the replication channel. localIdentifier=" + localIdentifier);
+
+        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) {
@@ -312,7 +326,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
 
         try {
 
-
             if ("KEY_SET".contentEquals(methodName)) {
                 writeChunked(transactionId, map -> map.keySet().iterator(), writeElement);
                 return;
@@ -337,6 +350,17 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
             // write the transaction id
             outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
 
+            if ("CREATE_CHANNEL".contentEquals(methodName)) {
+
+                writeVoid(() -> {
+                    short channelId1 = inWire.read(() -> "ARG_1").int16();
+                    chronicleHashInstanceBuilder.replicatedViaChannel(hub.createChannel(channelId1)).create();
+                    return null;
+                });
+                return;
+            }
+
+
             if ("REMOTE_IDENTIFIER".contentEquals(methodName)) {
                 this.remoteIdentifier = inWire.read(() -> "RESULT").int8();
                 return;
@@ -719,6 +743,37 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
 
     }
 
+    @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(() -> "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(() -> "IS_EXCEPTION").bool(true);
+            outWire.write(() -> "EXCEPTION").text(toString(e));
+            LOG.error("", e);
+            return;
+        }
+
+        nofityDataWritten();
+
+    }
+
+
     /**
      * only used for debugging
      */
@@ -759,6 +814,10 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
         return sw.toString();
     }
 
+    public void localIdentifier(byte localIdentifier) {
+        this.localIdentifer = localIdentifier;
+    }
+
 
     /**
      * used to send data to the wired stateless client, this class converts/adapts a chronicle map

src/main/java/net/openhft/chronicle/map/TcpReplicator.java

@@ -45,6 +45,7 @@ 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;
@@ -86,8 +87,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
     private final BitSet activeKeys = new BitSet(selectionKeysStore.length);
     private final long heartBeatIntervalMillis;
 
-    @Nullable
-    final List<Replica> channelList;
 
     private long largestEntrySoFar = 128;
 
@@ -110,6 +109,9 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
 
     StatelessClientParameters<K, V> statelessClientParameters;
 
+    private Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier;
+    private List<Replica> channelList;
+
     static class StatelessClientParameters<K, V> {
         public StatelessClientParameters(VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
                                          SerializationBuilder<K> keySerializationBuilder,
@@ -133,7 +135,8 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
                          @Nullable final RemoteNodeValidator remoteNodeValidator,
                          @Nullable final StatelessClientParameters statelessClientParameters,
                          String name,
-                         @Nullable final List<Replica> channelList)
+                         @Nullable final List<Replica> channelList,
+                         @NotNull Supplier<? extends StatelessWiredConnector> statelessWiredConnectorSupplier)
             throws IOException {
 
         super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig());
@@ -141,6 +144,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
         this.statelessClientParameters = statelessClientParameters;
         this.channelList = channelList;
 
+        this.statelessWiredConnectorSupplier = statelessWiredConnectorSupplier;
         final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig();
         long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS);
 
@@ -156,6 +160,8 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
 
         this.remoteNodeValidator = remoteNodeValidator;
         this.name = name;
+
+
         start();
     }
 
@@ -810,7 +816,16 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
             if (b == WIRED_CONNECTION)
                 attached.useWire = true;
             forceRead = true;
-            attached.statelessWiredConnector = new StatelessWiredConnector(channelList, localIdentifier);
+
+            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;
         }
 
@@ -1411,26 +1426,6 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
          * @throws InterruptedException
          */
         void entriesFromBuffer(@NotNull Attached attached, @NotNull SelectionKey key) {
-/*
-            if (attached.useWire) {
-
-                if (out.remaining() >= 4) {
-                    int size = out.readInt(out.position());
-                    if (out.remaining() >= size) {
-
-                        long end = out.position() + size;
-
-                        try {
-                            out.skip(4); // for the size
-                            wiredStatelessServer.processStatelessEvent(out, attached.entryWriter);
-                        } finally {
-                            out.position(end);
-                        }
-
-                    }
-                }
-                return;
-            }*/
 
             int entriesRead = 0;
             try {

src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java

@@ -33,6 +33,7 @@ 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.*;
 
 
@@ -269,6 +270,15 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
         return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key);
     }
 
+    @NotNull
+    public void createChannel(short channelID) {
+        proxyReturnVoid(() -> {
+            writeField(METHOD_NAME, "CREATE_CHANNEL");
+            hub.outWire().write(() -> "ARG_1").int16(channelID);
+        });
+    }
+
+
     public void putAll(@NotNull Map<? extends K, ? extends V> map) {
 
         final long startTime = System.currentTimeMillis();
@@ -341,7 +351,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
     }
 
     public void clear() {
-        proxyReturnVoid(CLEAR.toString());
+        proxyReturnVoid(() -> writeField(METHOD_NAME, CLEAR.toString()));
     }
 
 
@@ -674,16 +684,16 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
     }
 
     @SuppressWarnings("SameParameterValue")
-    private void proxyReturnVoid(@NotNull final String methodName) {
+    private void proxyReturnVoid(Runnable r) {
+
 
         final long startTime = System.currentTimeMillis();
-
         long transactionId;
 
         hub.outBytesLock().lock();
         try {
             transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
-            writeField(METHOD_NAME, methodName);
+            r.run();
             hub.writeSocket(hub.outWire());
         } finally {
             hub.outBytesLock().unlock();

src/test/java/net/openhft/chronicle/wire/TestWire.java

@@ -47,6 +47,51 @@ public class TestWire {
     private ReplicationHub hubA;
 
 
+    @Test
+    public void testCreateChannel() throws IOException, InterruptedException {
+
+        {
+            final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
+                    .of(++SERVER_PORT)
+                    .heartBeatInterval(1, SECONDS);
+
+            hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
+                    .createWithId((byte) 1);
+
+            // this is how you add maps after the custer is created
+            map1a = of(byte[].class, byte[].class)
+                    .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
+
+
+            byte identifier = (byte) 2;
+
+            final WiredChronicleMapStatelessClientBuilder<String, String> builder =
+                    new WiredChronicleMapStatelessClientBuilder<>(
+                            new InetSocketAddress("localhost", SERVER_PORT),
+                            String.class,
+                            String.class,
+                            (short) 1);
+
+            builder.identifier(identifier);
+            final ChronicleMap<String, String> statelessMap = builder.create();
+            ((WiredStatelessChronicleMap) statelessMap).createChannel((short) 2);
+
+
+            ChronicleMap<String, String> map2 = new WiredChronicleMapStatelessClientBuilder<String, String>(
+                    new InetSocketAddress("localhost", SERVER_PORT),
+                    String.class,
+                    String.class,
+                    (short) 2).create();
+
+            map2.put("hello", "world");
+
+            assertEquals(1, map2.size());
+
+            map1a.close();
+        }
+
+    }
+
     @Test
     public void testSimplePutStringWithChannels() throws IOException, InterruptedException {