From a386fb0cad5fa9361e2613129f2240c751248d74 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 27 Nov 2014 11:40:07 +0000 Subject: [PATCH] HCOLL-219 support the importing and exporting of entries to a human readable file format like JSON --- pom.xml | 31 +- .../chronicle/map/StatelessChronicleMap.java | 12 +- .../chronicle/map/VanillaChronicleMap.java | 266 +++++++++++++++++- .../map/ChronicleMapImportExportTest.java | 47 ++++ .../chronicle/map/ReplicationCheckingMap.java | 10 + .../jrs166/map/StatelessChronicleMapTest.java | 10 + 6 files changed, 365 insertions(+), 11 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java diff --git a/pom.xml b/pom.xml index 72d5d7bdb..7931c08ca 100644 --- a/pom.xml +++ b/pom.xml @@ -17,13 +17,15 @@ ~ limitations under the License. --> - + net.openhft java-parent-pom 1.1.2 - + 4.0.0 @@ -61,12 +63,34 @@ 2.2.0 + + xstream + xstream + 1.2.2 + + + + org.codehaus.jettison + jettison + 1.3.6 + + + + xstream + xstream + + + + org.codehaus.jettison + jettison + + net.openhft compiler @@ -146,7 +170,8 @@ http://openhft.github.io/Chronicle-Map/apidocs/ - net.openhft.chronicle.hash.serialization.internal + net.openhft.chronicle.hash.serialization.internal + diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 0c0e6c297..3b2dfbb66 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -164,6 +164,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon }); } + @Override + public void getAll(File toFile) { + throw new UnsupportedOperationException(); + } + + @Override + public void putAll(File fromFile) { + throw new UnsupportedOperationException(); + } + private ExecutorService lazyExecutorService() { @@ -835,7 +845,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long startTime = System.currentTimeMillis(); final long transactionId = nextUniqueTransaction(startTime); - final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + final long timeoutTime = startTime + config.timeoutMs(); // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index e0dfd9b50..106f6f537 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -18,6 +18,14 @@ package net.openhft.chronicle.map; +import com.thoughtworks.xstream.XStream; +import com.thoughtworks.xstream.converters.ConversionException; +import com.thoughtworks.xstream.converters.Converter; +import com.thoughtworks.xstream.converters.MarshallingContext; +import com.thoughtworks.xstream.converters.UnmarshallingContext; +import com.thoughtworks.xstream.io.HierarchicalStreamReader; +import com.thoughtworks.xstream.io.HierarchicalStreamWriter; +import com.thoughtworks.xstream.io.json.JettisonMappedXmlDriver; import net.openhft.chronicle.hash.ChronicleHashErrorListener; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.Hasher; @@ -34,12 +42,14 @@ import net.openhft.lang.model.constraints.Nullable; import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -import java.io.File; -import java.io.IOException; -import java.io.ObjectInputStream; -import java.io.Serializable; +import java.io.*; import java.lang.reflect.Array; +import java.nio.BufferOverflowException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.util.*; import java.util.concurrent.ExecutionException; @@ -50,6 +60,7 @@ import java.util.concurrent.TimeoutException; import static java.lang.Long.numberOfTrailingZeros; import static java.lang.Math.max; import static java.lang.Thread.currentThread; +import static java.nio.ByteBuffer.allocateDirect; import static net.openhft.lang.MemoryUnit.*; @@ -57,6 +68,8 @@ class VanillaChronicleMap, V, VW, MVW extends MetaBytesWriter> extends AbstractMap implements ChronicleMap, Serializable { + private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class); + /** * Because DirectBitSet implementations couldn't find more than 64 continuous clear or set * bits. @@ -84,7 +97,7 @@ class VanillaChronicleMap, final Alignment alignment; final int actualSegments; final long entriesPerSegment; - final transient MapEventListener> eventListener; + transient MapEventListener> eventListener; // if set the ReturnsNull fields will cause some functions to return NULL // rather than as returning the Object can be expensive for something you probably don't use. final boolean putReturnsNull; @@ -105,9 +118,12 @@ class VanillaChronicleMap, transient long headerSize; transient Set> entrySet; + private transient SerializationBuilder keyBuilder; + private transient SerializationBuilder valueBuilder; + public VanillaChronicleMap(AbstractChronicleMapBuilder builder) throws IOException { - SerializationBuilder keyBuilder = builder.keyBuilder; + keyBuilder = builder.keyBuilder; kClass = keyBuilder.eClass; keySizeMarshaller = keyBuilder.sizeMarshaller(); originalKeyReader = keyBuilder.reader(); @@ -117,7 +133,7 @@ class VanillaChronicleMap, originalMetaKeyInterop = (MKI) keyBuilder.metaInterop(); metaKeyInteropProvider = (MetaProvider) keyBuilder.metaInteropProvider(); - SerializationBuilder valueBuilder = builder.valueBuilder; + valueBuilder = builder.valueBuilder; vClass = valueBuilder.eClass; isNativeValueClass = vClass.getName().endsWith("$$Native"); valueSizeMarshaller = valueBuilder.sizeMarshaller(); @@ -481,6 +497,242 @@ class VanillaChronicleMap, }; } + @Override + public void getAll(File toFile) throws IOException { + + final XStream xstream = new XStream(new JettisonMappedXmlDriver()); + xstream.setMode(XStream.NO_REFERENCES); + xstream.alias("key", kClass); + xstream.alias("value", vClass); + xstream.alias("entries", net.openhft.chronicle.map.VanillaChronicleMap.EntrySet.class); + + // ideally we will use java serialization if we can + + lazyXStreamConverter().registerConverter(xstream); + + + try (FileOutputStream out = new FileOutputStream(toFile)) { + xstream.toXML(entrySet, out); + } + + } + + @Override + public void putAll(File fromFile) throws IOException { + final XStream xstream = new XStream(new JettisonMappedXmlDriver()); + xstream.setMode(XStream.NO_REFERENCES); + xstream.alias("key", kClass); + xstream.alias("value", vClass); + xstream.alias("entries", net.openhft.chronicle.map.VanillaChronicleMap.EntrySet.class); + + // ideally we will use java serialization if we can + lazyXStreamConverter().registerConverter(xstream); + + try (FileInputStream out = new FileInputStream(fromFile)) { + xstream.fromXML(out); + } + + } + + + private XStreamConverter xStreamConverter; + + /** + * @return a lazily created XStreamConverter + */ + private XStreamConverter lazyXStreamConverter() { + + final KeyValueSerializer serializers = new KeyValueSerializer(keyBuilder, + valueBuilder); + + final ThreadLocalCopies threadLocalCopies = serializers.threadLocalCopies(); + + final Bytes buffer = new ByteBufferBytes(allocateDirect((int) entrySize) + .order(ByteOrder.nativeOrder())); + + return new XStreamConverter(threadLocalCopies, buffer, serializers); + } + + + private class XStreamConverter { + + + private final ThreadLocalCopies threadLocalCopies; + private Bytes buffer; + private final KeyValueSerializer keyValueSerializer; + + public XStreamConverter(ThreadLocalCopies threadLocalCopies, Bytes buffer, final KeyValueSerializer + keyValueSerializer) { + this.threadLocalCopies = threadLocalCopies; + this.buffer = buffer; + this.keyValueSerializer = keyValueSerializer; + } + + + private synchronized Bytes toKeyBytes(K key) { + for (; ; ) { + + try { + keyValueSerializer.writeKey(key, buffer, threadLocalCopies); + return buffer; + } catch (BufferOverflowException e) { + ByteBuffer b = allocateDirect((int) buffer.capacity() + (int) entrySize) + .order(ByteOrder.nativeOrder()); + buffer = new ByteBufferBytes(allocateDirect((int) buffer.capacity() + (int) entrySize) + .order(ByteOrder.nativeOrder())); + } + } + } + + private synchronized Bytes toValueBytes(V value) { + for (; ; ) { + + try { + keyValueSerializer.writeValue(value, buffer, threadLocalCopies); + return buffer; + } catch (BufferOverflowException e) { + ByteBuffer b = allocateDirect((int) buffer.capacity() + (int) entrySize) + .order(ByteOrder.nativeOrder()); + buffer = new ByteBufferBytes(b); + } + } + } + + + private synchronized Bytes append(CharSequence value) { + for (; ; ) { + + try { + buffer.clear(); + buffer.append(value); + buffer.flip(); + return buffer; + } catch (BufferOverflowException e) { + ByteBuffer b = allocateDirect((int) buffer.capacity() + (int) entrySize) + .order(ByteOrder.nativeOrder()); + buffer = new ByteBufferBytes(b); + } + } + } + + private void registerConverter(XStream xstream) { + final Converter converter = new Converter() { + + final Bytes buffer = XStreamConverter.this.buffer; + + @Override + public boolean canConvert(Class aClass) { + + if (EntrySet.class + .isAssignableFrom(aClass) || SimpleEntry.class + .isAssignableFrom(aClass)) + return true; + + if (Serializable.class + .isAssignableFrom(aClass)) + return false; + + return kClass.equals(aClass) || vClass.equals(aClass); + } + + @Override + public void marshal(Object o, HierarchicalStreamWriter writer, MarshallingContext + marshallingContext) { + + buffer.clear(); + + if (o.getClass().equals(kClass)) { + + //String text = toKeyBytes((K) o).asString().toString(); + // System.out.println("key=" + text); + // writer.setValue(text); + + + // writer.startNode("key"); + writer.setValue(o.toString()); + // writer.endNode(); + } else if (o.getClass().equals(vClass)) { + writer.setValue(o.toString()); + } else if (o instanceof Entry) { + final SimpleEntry e = (SimpleEntry) o; + + writer.startNode("key"); + marshallingContext.convertAnother(e.getKey()); + writer.endNode(); + writer.startNode("value"); + marshallingContext.convertAnother(e.getValue()); + writer.endNode(); + + } else if (EntrySet.class + .isAssignableFrom(o.getClass())) { + + for (Entry e : (EntrySet) o) { + writer.startNode("entry"); + marshallingContext.convertAnother(e); + writer.endNode(); + } + + } + } + + @Override + public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext unmarshallingContext) { + String value = reader.getValue(); + buffer.clear(); + append(value); + buffer.flip(); + + final String nodeName = reader.getNodeName(); + + + if (nodeName.equals("entries")) { + + while (reader.hasMoreChildren()) { + reader.moveDown(); + final String nodeName0 = reader + .getNodeName(); + + if (!nodeName0.equals("entry")) + throw new ConversionException("unable to convert node " + + "named=" + nodeName0); + K k = null; + V v = null; + + while (reader.hasMoreChildren()) { + reader.moveDown(); + final String nodeName1 = reader + .getNodeName(); + if ("key".equals(nodeName1)) + k = (K) unmarshallingContext.convertAnother(null, kClass); + else if ("value".equals(nodeName1)) + v = (V) unmarshallingContext.convertAnother(null, vClass); + else throw new ConversionException + ("expecting either a key or" + + " value"); + reader.moveUp(); + } + + if (k != null) + VanillaChronicleMap.this.put(k, v); + reader.moveUp(); + } + + + } else if (nodeName.equals("key")) + return keyValueSerializer.readKey(buffer, threadLocalCopies); + else if (nodeName.equals("value")) + return keyValueSerializer.readValue(buffer, threadLocalCopies); + + return null; + } + }; + + xstream.registerConverter(converter); + } + + } + + @NotNull @Override public WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java new file mode 100644 index 000000000..a45a2e8d6 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapImportExportTest.java @@ -0,0 +1,47 @@ +package net.openhft.chronicle.map; + +import junit.framework.Assert; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; + +/** + * @author Rob Austin. + */ +public class ChronicleMapImportExportTest { + + + public static final String TMP = System.getProperty("java.io.tmpdir"); + + @Test + public void test() throws IOException, InterruptedException { + + File file = new File(TMP + "/chronicle-map-" + System.nanoTime() + ".json"); + file.deleteOnExit(); + + try (ChronicleMap expected = ChronicleMapBuilder.of(String.class, String + .class) + .create()) { + + expected.put("hello", "world"); + expected.put("aKey", "aValue"); + + System.out.println(expected); + expected.getAll(file); + + try (ChronicleMap actual = ChronicleMapBuilder.of(String.class, String + .class) + .create()) { + + actual.putAll(file); + + Assert.assertEquals(expected, actual); + } + } finally { + file.delete(); + } + + + } +} diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index d4176cb45..ee30862a1 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -270,6 +270,16 @@ public class ReplicationCheckingMap implements ChronicleMap { throw new UnsupportedOperationException(); } + @Override + public void getAll(File toFile) { + throw new UnsupportedOperationException(); + } + + @Override + public void putAll(File fromFile) { + throw new UnsupportedOperationException(); + } + @Override public V acquireUsing(@NotNull K key, V usingValue) { return map1.acquireUsing(key, usingValue); diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index c67cf2d06..cf4f1b994 100755 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -192,6 +192,16 @@ public class StatelessChronicleMapTest extends JSR166TestCase { return d.removeLater(key); } + @Override + public void getAll(File toFile) throws IOException { + d.getAll(toFile); + } + + @Override + public void putAll(File fromFile) throws IOException { + d.putAll(fromFile); + } + @Override public V putIfAbsent(K key, V value) { return d.putIfAbsent(key, value);