diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 63409557b..50e2beaae 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -419,6 +419,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private void writeEntries(Map map) { + final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; @@ -605,20 +606,25 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { long sizeLocation = writeEvent(ENTRY_SET); // get the data back from the server - Bytes in = blockingFetch(sizeLocation); + final Bytes in = blockingFetch(sizeLocation); + final Set> result = new HashSet>(); - long size = in.readStopBit(); - LOG.info("size="+size); - Set> result = new HashSet>(); + boolean hasMoreEntries; + do { + hasMoreEntries = in.readBoolean(); - for (int i = 0; i < size; i++) { - K k = keyValueSerializer.readKey(in); - V v = keyValueSerializer.readValue(in); - LOG.info("k="+k); + // number of entries in the chunk + long size = in.readUnsignedShort(); - LOG.info("v="+v); - result.add(new Entry(k, v)); - } + LOG.info("size=" + size); + + for (int i = 0; i < size; i++) { + K k = keyValueSerializer.readKey(in); + V v = keyValueSerializer.readValue(in); + result.add(new Entry(k, v)); + } + + } while (hasMoreEntries); return result; } @@ -730,9 +736,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { // the number of bytes in the response int size = receive(SIZE_OF_SIZE, timeoutTime).readInt(); - if (bytes.capacity() < size) - bytes = new ByteBufferBytes(allocateDirect(size)); + int requiredSize = size + SIZE_OF_SIZE; + if (bytes.capacity() < requiredSize) { + bytes = new ByteBufferBytes(allocateDirect(requiredSize)); + } + // block until we have received all the byte in this chunk receive(size, timeoutTime); boolean isException = bytes.readBoolean(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 786d971e8..43efdeb39 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -548,9 +548,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { if (completed) attached.entryWriter.workCompleted(); - // attached.hasRemoteHeartbeatInterval = false; + // attached.hasRemoteHeartbeatInterval = false; - } else if (attached.remoteModificationIterator != null) + } else if (attached.remoteModificationIterator != null) attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); try { @@ -1646,15 +1646,18 @@ class StatelessServerConnector { private Work entrySet(Bytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); + final long transactionId = reader.readLong(); + + final Set> entries; try { entries = map.entrySet(); } catch (RuntimeException e) { + final long sizeLocation = reflectTransactionId(reader, writer); return sendException(writer, sizeLocation, e); } - writer.writeStopBit(entries.size()); + final Iterator> iterator = entries.iterator(); @@ -1664,23 +1667,71 @@ class StatelessServerConnector { @Override public boolean doWork(Bytes writer) { + final long sizeLocation = writer.position(); + + writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is + // exception + writer.writeLong(transactionId); + + // hasAnotherChunk + writer.skip(1); + + // count + writer.skip(2); + + int count = 0; while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (writer.remaining() <= maxEntrySizeBytes) + if (writer.remaining() <= maxEntrySizeBytes) { + writeHeader(writer, sizeLocation, count, true); + return false; + } + + count++; final Map.Entry next = iterator.next(); writeKey(next.getKey(), writer); writeValue(next.getValue(), writer); } - writeSizeAndFlags(sizeLocation, false, writer); + writeHeader(writer, sizeLocation, count, false); + return true; } + + private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) { + final long end = writer.position(); + final int size = (int) (end - sizeLocation); + writer.position(sizeLocation); + + // size in bytes + writer.writeInt(size); + + // is exception + writer.writeBoolean(false); + + //transaction id; + writer.skip(8); + + writer.writeBoolean(hasAnotherChunk); + + // count + writer.writeUnsignedShort(count); + writer.position(end); + } }; } + private void writeMoreData(long location, boolean hasMoreData, Bytes out) { + out.writeBoolean(location, hasMoreData); + } + + private void entryCount(long location, int count, Bytes out) { + + } + private Work putIfAbsent(Bytes reader, Bytes writer) { final K key = readKey(reader); final V v = readValue(reader); @@ -1726,7 +1777,7 @@ class StatelessServerConnector { private long reflectTransactionId(Bytes reader, Bytes writer) { final long sizeLocation = writer.position(); - writer.skip(1 + AbstractChannelReplicator.SIZE_OF_SIZE); // event size + SIZE_OF_SIZE + writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // isException + SIZE_OF_SIZE final long transactionId = reader.readLong(); writer.writeLong(transactionId); return sizeLocation; diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 8e90519a8..84a531b40 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -72,10 +72,7 @@ public class StatelessClientTest { Set> entries = statelessMap.entrySet(); - - Map.Entry next = entries.iterator().next(); - Assert.assertEquals("some value=" + next.getKey(), next.getValue()); diff --git a/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java b/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java new file mode 100644 index 000000000..2716fbfc1 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/example/ChannelExample.java @@ -0,0 +1,157 @@ +package net.openhft.chronicle.map.example; + +import net.openhft.chronicle.map.*; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.net.InetSocketAddress; + +import static org.junit.Assert.assertEquals; + +/** + * @author Rob Austin. + */ +public class ChannelExample { + + + + +/* + * 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. + */ + + + /** + * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket, but with 4 nodes + * + * @author Rob Austin. + */ + public class ChannelReplicationTest { + + private ChronicleMap favoriteComputerServer1; + private ChronicleMap favoriteComputerServer2; + + private ChronicleMap favoriteColourServer2; + private ChronicleMap favoriteColourServer1; + + + private ChannelProvider channelProviderServer1; + private ChannelProvider channelProviderServer2; + + + @Test + public void test() throws IOException, InterruptedException { + + // server 1 with identifier = 1 + { + byte identifier = (byte) 1; + + TcpReplicationConfig tcpConfig = TcpReplicationConfig + .of(8086, new InetSocketAddress("localhost", 8087)) + .heartBeatInterval(1, java.util.concurrent.TimeUnit.SECONDS); + + + channelProviderServer1 = new ChannelProviderBuilder() + .replicators(identifier, tcpConfig) + .create(); + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer1 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer1.createChannel(channel1)).create(); + + favoriteColourServer1.put("peter", "green"); + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer1 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer1.createChannel(channel2)).create(); + + favoriteComputerServer1.put("peter", "dell"); + + } + + // server 2 with identifier = 2 + { + + byte identifier = (byte) 2; + + TcpReplicationConfig tcpConfig = + TcpReplicationConfig.of(8087).heartBeatInterval(1, java.util.concurrent.TimeUnit.SECONDS); + + channelProviderServer2 = new ChannelProviderBuilder() + .replicators(identifier, tcpConfig) + .create(); + + + // this demotes favoriteColour + short channel1 = (short) 1; + + favoriteColourServer2 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer2.createChannel(channel1)).create(); + + + favoriteColourServer2.put("rob", "blue"); + + // this demotes favoriteComputer + short channel2 = (short) 2; + + favoriteComputerServer2 = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(1000) + .channel(channelProviderServer2.createChannel(channel2)).create(); + + favoriteComputerServer2.put("rob", "mac"); + favoriteComputerServer2.put("daniel", "mac"); + } + + + // allow time for the recompilation to resolve + for (int t = 0; t < 2500; t++) { + if (favoriteComputerServer2.equals(favoriteComputerServer1) && + favoriteColourServer2.equals(favoriteColourServer1)) + break; + Thread.sleep(1); + } + + + assertEquals(favoriteComputerServer1, favoriteComputerServer2); + Assert.assertEquals(3, favoriteComputerServer2.size()); + + + assertEquals(favoriteColourServer1, favoriteColourServer2); + Assert.assertEquals(2, favoriteColourServer1.size()); + + + favoriteColourServer1.close(); + favoriteComputerServer2.close(); + favoriteColourServer2.close(); + favoriteColourServer1.close(); +} + + + } + +} + + +