HCOLL-239 - fixed :The JSON to Map import/export should work on the stateless client.

This commit is contained in:
Rob Austin
2014-12-26 12:11:53 +00:00
parent 06fcaa4c71
commit 6afd73bf30
7 changed files with 316 additions and 71 deletions
@@ -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 <K, V> void getAll(File toFile, Map<K, V> 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<K, V>(map));
xstream.registerConverter(new ByteBufferConverter());
xstream.registerConverter(new DataValueConverter());
static synchronized <K, V> void getAll(File toFile, Map<K, V> 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 <K, V> void putAll(File fromFile, Map<K, V> 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<K, V>(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 <K, V> XStream xStream(Map<K, V> 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(<file>) and map.putAll(<file>) 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" +
"<dependency>\n" +
" <groupId>xstream</groupId>\n" +
" <artifactId>xstream</artifactId>\n" +
" <version>1.2.2</version>\n" +
"</dependency>\n" +
"<dependency>\n" +
" <groupId>org.codehaus.jettison</groupId>\n" +
" <artifactId>jettison</artifactId>\n" +
" <version>1.3.6</version>\n" +
"</dependency>\n", e);
}
private static <K, V> void registerChronicleMapConverter(Map<K, V> map, XStream xstream) {
Converter converter = map instanceof StatelessChronicleMap ?
new StatelessChronicleMapConverter<K, V>(map) :
new VanillaChronicleMapConverter<K, V>(map);
xstream.registerConverter(converter);
}
}
@@ -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<K, V> {
void onEntry(K key, V value);
void onFinished();
}
@@ -190,13 +190,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
return result.entrySet();
}
/**
* @param callback each entry is passed to the callback
*/
void entrySet(@NotNull MapEntryCallback<K, V> 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<K> keyReader = keyReaderWithSize.readerForLoop(copies);
copies = valueReaderWithSize.getCopies(copies);
final BytesReader<V> 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<? extends K, ? extends V> map) {
@@ -1331,12 +1384,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long start = outBytes.position();
assert outBytes.position() == start;
outBytes.limit(outBytes.capacity());
return valueWriterWithSize.write(outBytes, value, copies);
}
@SuppressWarnings("UnusedReturnValue")
@@ -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<K, V> implements Converter {
public class AbstractChronicleMapConverter<K, V> implements Converter {
private final Class entrySetClass;
private final Class writeThroughEntryClass;
private final Map<K, V> map;
final Class mapClazz;
public ChronicleMapConverter(@NotNull Map<K, V> 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<K, V> 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.Entry>) ((Map) o).entrySet()) {
for (Map.Entry e : (Iterable<Map.Entry>) 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();
@@ -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<K, V> extends AbstractChronicleMapConverter<K, V> {
public StatelessChronicleMapConverter(@NotNull Map<K, V> 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);
}
}
}
@@ -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<K, V> extends AbstractChronicleMapConverter<K, V> {
public VanillaChronicleMapConverter(@NotNull Map<K, V> map) {
super(map);
}
@Override
public void marshal(Object o, HierarchicalStreamWriter writer, MarshallingContext
marshallingContext) {
for (Map.Entry e : (Iterable<Map.Entry>) ((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();
}
}
}