diff --git a/src/main/java/net/openhft/chronicle/map/ChroncileMapJsonSerializer.java b/src/main/java/net/openhft/chronicle/map/ChroncileMapJsonSerializer.java index 04cd4e486..369899cf2 100644 --- a/src/main/java/net/openhft/chronicle/map/ChroncileMapJsonSerializer.java +++ b/src/main/java/net/openhft/chronicle/map/ChroncileMapJsonSerializer.java @@ -18,10 +18,14 @@ package net.openhft.chronicle.map; import com.thoughtworks.xstream.XStream; +import com.thoughtworks.xstream.converters.Converter; import com.thoughtworks.xstream.io.json.JettisonMappedXmlDriver; import net.openhft.xstreem.convertors.ByteBufferConverter; -import net.openhft.xstreem.convertors.ChronicleMapConverter; import net.openhft.xstreem.convertors.DataValueConverter; +import net.openhft.xstreem.convertors.StatelessChronicleMapConverter; +import net.openhft.xstreem.convertors.VanillaChronicleMapConverter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.*; import java.util.Map; @@ -33,30 +37,21 @@ import java.util.zip.GZIPOutputStream; */ class JsonSerializer { - static synchronized void getAll(File toFile, Map map) throws IOException { - final XStream xstream = new XStream(new JettisonMappedXmlDriver()); - xstream.setMode(XStream.NO_REFERENCES); - xstream.alias("cmap", VanillaChronicleMap.EntrySet.class); + private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMap.class); - xstream.registerConverter(new ChronicleMapConverter(map)); - xstream.registerConverter(new ByteBufferConverter()); - xstream.registerConverter(new DataValueConverter()); + static synchronized void getAll(File toFile, Map map) throws IOException { + final XStream xstream = xStream(map); OutputStream outputStream = new FileOutputStream(toFile); if (toFile.getName().toLowerCase().endsWith(".gz")) outputStream = new GZIPOutputStream(outputStream); try (OutputStream out = outputStream) { - xstream.toXML(map.entrySet(), out); + xstream.toXML(map, out); } } static synchronized void putAll(File fromFile, Map map) throws IOException { - final XStream xstream = new XStream(new JettisonMappedXmlDriver()); - xstream.setMode(XStream.NO_REFERENCES); - xstream.alias("cmap", VanillaChronicleMap.EntrySet.class); - xstream.registerConverter(new ChronicleMapConverter(map)); - xstream.registerConverter(new ByteBufferConverter()); - xstream.registerConverter(new DataValueConverter()); + final XStream xstream = xStream(map); InputStream inputStream = new FileInputStream(fromFile); if (fromFile.getName().toLowerCase().endsWith(".gz")) @@ -65,4 +60,47 @@ class JsonSerializer { xstream.fromXML(out); } } + + private static XStream xStream(Map map) { + try { + final XStream xstream = new XStream(new JettisonMappedXmlDriver()); + xstream.setMode(XStream.NO_REFERENCES); + xstream.alias("cmap", map.getClass()); + + registerChronicleMapConverter(map, xstream); + xstream.registerConverter(new ByteBufferConverter()); + xstream.registerConverter(new DataValueConverter()); + return xstream; + } catch (NoClassDefFoundError e) { + logErrorSuggestXStreem(e); + throw e; + } + } + + static private void logErrorSuggestXStreem(Error e) { + LOG.error("map.getAll() and map.putAll() methods require the JSON XStream serializer, " + + "we don't include these artifacts by default as some users don't require this functionality. " + + "Please add the following artifacts to your project\n" + + "\n" + + " xstream\n" + + " xstream\n" + + " 1.2.2\n" + + "\n" + + "\n" + + " org.codehaus.jettison\n" + + " jettison\n" + + " 1.3.6\n" + + "\n", e); + } + + private static void registerChronicleMapConverter(Map map, XStream xstream) { + + Converter converter = map instanceof StatelessChronicleMap ? + new StatelessChronicleMapConverter(map) : + new VanillaChronicleMapConverter(map); + + xstream.registerConverter(converter); + } } + + diff --git a/src/main/java/net/openhft/chronicle/map/MapEntryCallback.java b/src/main/java/net/openhft/chronicle/map/MapEntryCallback.java new file mode 100644 index 000000000..3b1f5ed72 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/MapEntryCallback.java @@ -0,0 +1,28 @@ +/* + * 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 + */ +public interface MapEntryCallback { + + void onEntry(K key, V value); + + void onFinished(); +} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 28caf6d46..98da4a27e 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -190,13 +190,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } @Override - public void getAll(File toFile) { - throw new UnsupportedOperationException(); + public void getAll(File toFile) throws IOException { + JsonSerializer.getAll(toFile, this); } @Override - public void putAll(File fromFile) { - throw new UnsupportedOperationException(); + public void putAll(File fromFile) throws IOException { + JsonSerializer.putAll(fromFile, this); } @Override @@ -736,6 +736,59 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } return result.entrySet(); + + } + + /** + * @param callback each entry is passed to the callback + */ + void entrySet(@NotNull MapEntryCallback callback) { + final long transactionId; + final long timeoutTime; + + outBytesLock.lock(); + try { + final long sizeLocation = writeEventAnSkip(ENTRY_SET); + final long startTime = System.currentTimeMillis(); + timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } + + // get the data back from the server + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + final BytesReader keyReader = keyReaderWithSize.readerForLoop(copies); + copies = valueReaderWithSize.getCopies(copies); + final BytesReader valueReader = valueReaderWithSize.readerForLoop(copies); + + for (; ; ) { + + inBytesLock.lock(); + try { + + Bytes in = blockingFetchReadOnly(timeoutTime, transactionId); + + final boolean hasMoreEntries = in.readBoolean(); + + // number of entries in this chunk + final long size = in.readInt(); + + for (int i = 0; i < size; i++) { + final K k = keyReaderWithSize.readInLoop(in, keyReader); + final V v = valueReaderWithSize.readInLoop(in, valueReader); + callback.onEntry(k, v); + } + + if (!hasMoreEntries) + break; + + } finally { + inBytesLock.unlock(); + } + } + + } public void putAll(@NotNull Map map) { @@ -1331,12 +1384,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long start = outBytes.position(); - assert outBytes.position() == start; outBytes.limit(outBytes.capacity()); return valueWriterWithSize.write(outBytes, value, copies); - } @SuppressWarnings("UnusedReturnValue") diff --git a/src/main/java/net/openhft/xstreem/convertors/ChronicleMapConverter.java b/src/main/java/net/openhft/xstreem/convertors/AbstractChronicleMapConverter.java similarity index 72% rename from src/main/java/net/openhft/xstreem/convertors/ChronicleMapConverter.java rename to src/main/java/net/openhft/xstreem/convertors/AbstractChronicleMapConverter.java index 8c7bff28b..596945d28 100644 --- a/src/main/java/net/openhft/xstreem/convertors/ChronicleMapConverter.java +++ b/src/main/java/net/openhft/xstreem/convertors/AbstractChronicleMapConverter.java @@ -26,80 +26,58 @@ import com.thoughtworks.xstream.io.HierarchicalStreamWriter; import net.openhft.lang.model.DataValueClasses; import net.openhft.lang.model.constraints.NotNull; -import java.util.AbstractMap; import java.util.Collections; import java.util.Map; /** * @author Rob Austin. */ -public class ChronicleMapConverter implements Converter { +public class AbstractChronicleMapConverter implements Converter { + - private final Class entrySetClass; - private final Class writeThroughEntryClass; private final Map map; + final Class mapClazz; - public ChronicleMapConverter(@NotNull Map map) { - - final String vanillaChronicleMap = "net.openhft.chronicle.map.VanillaChronicleMap"; - - try { - this.entrySetClass = Class.forName(vanillaChronicleMap + "$EntrySet"); - this.writeThroughEntryClass = Class.forName(vanillaChronicleMap + "$WriteThroughEntry"); - } catch (ClassNotFoundException e) { - throw new ConversionException("", e); - } - + public AbstractChronicleMapConverter(@NotNull Map map) { this.map = map; + this.mapClazz = map.getClass(); } @Override public boolean canConvert(Class aClass) { - //noinspection unchecked - return entrySetClass.isAssignableFrom(aClass) || - writeThroughEntryClass.isAssignableFrom(aClass); - + return mapClazz.isAssignableFrom(aClass); } + @Override public void marshal(Object o, HierarchicalStreamWriter writer, MarshallingContext marshallingContext) { - if (entrySetClass.isAssignableFrom(o.getClass())) { + for (Map.Entry e : (Iterable) ((Map) o).entrySet()) { - for (Map.Entry e : (Iterable) o) { - writer.startNode("entry"); - marshallingContext.convertAnother(e); + writer.startNode("entry"); + { + final Object key = e.getKey(); + writer.startNode(key.getClass().getName()); + marshallingContext.convertAnother(key); + writer.endNode(); + + Object value = e.getValue(); + writer.startNode(value.getClass().getName()); + marshallingContext.convertAnother(value); writer.endNode(); } - return; + writer.endNode(); } - final AbstractMap.SimpleEntry e = (AbstractMap.SimpleEntry) o; - - final Object key = e.getKey(); - writer.startNode(key.getClass().getName()); - marshallingContext.convertAnother(key); - writer.endNode(); - - Object value = e.getValue(); - writer.startNode(value.getClass().getName()); - marshallingContext.convertAnother(value); - writer.endNode(); - } @Override public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext context) { - final String nodeName = reader.getNodeName(); - - if (!"cmap".equals(nodeName)) - return null; - while (reader.hasMoreChildren()) { reader.moveDown(); diff --git a/src/main/java/net/openhft/xstreem/convertors/StatelessChronicleMapConverter.java b/src/main/java/net/openhft/xstreem/convertors/StatelessChronicleMapConverter.java new file mode 100644 index 000000000..72aa1e7e3 --- /dev/null +++ b/src/main/java/net/openhft/xstreem/convertors/StatelessChronicleMapConverter.java @@ -0,0 +1,83 @@ +/* + * 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.xstreem.convertors; + +import com.thoughtworks.xstream.converters.ConversionException; +import com.thoughtworks.xstream.converters.MarshallingContext; +import com.thoughtworks.xstream.io.HierarchicalStreamWriter; +import net.openhft.chronicle.map.MapEntryCallback; +import net.openhft.lang.model.constraints.NotNull; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.Map; +import java.util.concurrent.CountDownLatch; + +/** + * @author Rob Austin. + */ +public class StatelessChronicleMapConverter extends AbstractChronicleMapConverter { + + public StatelessChronicleMapConverter(@NotNull Map map) { + super(map); + } + + @Override + public void marshal(Object o, final HierarchicalStreamWriter writer, final MarshallingContext + marshallingContext) { + + final CountDownLatch countDownLatch = new CountDownLatch(1); + + try { + + final Method entrySize = o.getClass().getDeclaredMethod("entrySet", MapEntryCallback.class); + entrySize.setAccessible(true); + entrySize.invoke(o, + + new MapEntryCallback() { + + @Override + public void onEntry(Object key, Object value) { + writer.startNode("entry"); + { + writer.startNode(key.getClass().getName()); + marshallingContext.convertAnother(key); + writer.endNode(); + + writer.startNode(value.getClass().getName()); + marshallingContext.convertAnother(value); + writer.endNode(); + } + writer.endNode(); + } + + @Override + public void onFinished() { + countDownLatch.countDown(); + } + }); + } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { + throw new ConversionException("", e); + } + + + } + + +} + diff --git a/src/main/java/net/openhft/xstreem/convertors/VanillaChronicleMapConverter.java b/src/main/java/net/openhft/xstreem/convertors/VanillaChronicleMapConverter.java new file mode 100644 index 000000000..d1cefaea2 --- /dev/null +++ b/src/main/java/net/openhft/xstreem/convertors/VanillaChronicleMapConverter.java @@ -0,0 +1,59 @@ +/* + * 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.xstreem.convertors; + +import com.thoughtworks.xstream.converters.MarshallingContext; +import com.thoughtworks.xstream.io.HierarchicalStreamWriter; +import net.openhft.lang.model.constraints.NotNull; + +import java.util.Map; + +/** + * @author Rob Austin. + */ +public class VanillaChronicleMapConverter extends AbstractChronicleMapConverter { + + public VanillaChronicleMapConverter(@NotNull Map map) { + super(map); + } + + @Override + public void marshal(Object o, HierarchicalStreamWriter writer, MarshallingContext + marshallingContext) { + + for (Map.Entry e : (Iterable) ((Map) o).entrySet()) { + + writer.startNode("entry"); + { + final Object key = e.getKey(); + writer.startNode(key.getClass().getName()); + marshallingContext.convertAnother(key); + writer.endNode(); + + Object value = e.getValue(); + writer.startNode(value.getClass().getName()); + marshallingContext.convertAnother(value); + writer.endNode(); + } + writer.endNode(); + } + + } + +} + diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java index 0f49892e9..3bbe93337 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java @@ -7,7 +7,6 @@ import net.openhft.chronicle.map.fromdocs.BondVOInterface; import net.openhft.lang.model.DataValueClasses; import net.openhft.lang.values.LongValue; import org.junit.Assert; -import org.junit.Ignore; import org.junit.Test; import java.io.File; @@ -178,7 +177,6 @@ public class ChronicleMapImportExportTest { xstream.toXML(map, new FileOutputStream(file)); - try (ChronicleMap expected = ChronicleMapBuilder.of(Integer.class, String.class).create()) { @@ -188,41 +186,51 @@ public class ChronicleMapImportExportTest { expected.getAll(file2); expected.putAll(file2); - Assert.assertEquals(2, expected.size()); Assert.assertEquals("one", expected.get(1)); Assert.assertEquals("two", expected.get(2)); } - //file.deleteOnExit(); + file.deleteOnExit(); } - @Ignore("HCOLL-239 - The JSON to Map import/export should work on the stateless client.") @Test - public void testToJsonWithStatlessClient() throws IOException, InterruptedException { + public void testToJsonWithStatelessClient() throws IOException, InterruptedException { File file = new File(TMP + "/chronicle-map-" + System.nanoTime() + ".json"); file.deleteOnExit(); + File file2 = new File(TMP + "/chronicle-map-" + System.nanoTime() + "-2.json"); + file.deleteOnExit(); try (ChronicleMap expected = ChronicleMapBuilder.of(CharSequence.class, CharSequence .class) .create()) { - try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + try (ChronicleMap serverMap = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { - try (ChronicleMap actual = ChronicleMapBuilder.of(Integer + try (ChronicleMap actual = ChronicleMapBuilder.of(CharSequence .class, CharSequence.class) .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { - expected.put("hello", "world"); + expected.put("hello", "world"); expected.getAll(file); actual.putAll(file); + + Assert.assertEquals(expected, actual); + + actual.getAll(file2); + actual.clear(); + actual.putAll(file2); + Assert.assertEquals(expected, actual); } } } + + file.delete(); + file2.delete(); }