Added support for dynamic discovery of files, https://github.com/OpenHFT/Chronicle-Core/issues/88

2021-09-13 01:26:12 +00:00 · 2019-08-28 17:59:17 +01:00
parent 1ed31d485c
commit 071bcf1d8e
18 changed files with 301 additions and 24 deletions
@@ -187,6 +187,13 @@
            <artifactId>commons-lang3</artifactId>
            <scope>test</scope>
        </dependency>
+
+        <dependency>
+            <groupId>io.hawt</groupId>
+            <artifactId>hawtio-embedded</artifactId>
+            <version>2.7.1</version>
+            <scope>test</scope>
+        </dependency>
    </dependencies>

    <build>
@@ -25,6 +25,7 @@ import net.openhft.chronicle.set.ChronicleSet;
 import org.jetbrains.annotations.NotNull;

 import java.io.File;
+import java.lang.reflect.Type;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Consumer;
@@ -96,6 +97,11 @@ public interface ChronicleHash<K, E extends HashEntry<K>, SC extends HashSegment
     */
    Class<K> keyClass();

+    /**
+     * @return Type as a Class or UnresolvedType if unknown.
+     */
+    Type keyType();
+
    /**
     * Returns a context to perform arbitrary operations with the given key in this store.
     * Conventionally, should be used in a try-with-resources block: <pre>{@code
@@ -38,6 +38,7 @@ import net.openhft.chronicle.map.ChronicleMapBuilder;
 import net.openhft.chronicle.values.Values;
 import net.openhft.chronicle.wire.Marshallable;
 import net.openhft.chronicle.wire.WireIn;
+import net.openhft.chronicle.wire.WireInternal;
 import net.openhft.chronicle.wire.WireOut;
 import org.jetbrains.annotations.NotNull;

@@ -46,6 +47,7 @@ import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.lang.ref.WeakReference;
+import java.lang.reflect.Type;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.ArrayList;
@@ -93,7 +95,7 @@ public abstract class VanillaChronicleHash<K,
    public transient boolean createdOrInMemory;
    /////////////////////////////////////////////////
    // Key Data model
-    public Class<K> keyClass;
+    public Type keyClass;
    public SizeMarshaller keySizeMarshaller;
    public SizedReader<K> keyReader;
    public DataAccess<K> keyDataAccess;
@@ -247,7 +249,7 @@ public abstract class VanillaChronicleHash<K,
        // that doesn't guarantee (?) to initialize fields with default values (false for boolean)
        createdOrInMemory = false;

-        keyClass = wireIn.read(() -> "keyClass").typeLiteral();
+        keyClass = wireIn.read(() -> "keyClass").lenientTypeLiteral();
        keySizeMarshaller = wireIn.read(() -> "keySizeMarshaller").object(SizeMarshaller.class);
        keyReader = wireIn.read(() -> "keyReader").object(SizedReader.class);
        keyDataAccess = wireIn.read(() -> "keyDataAccess").object(DataAccess.class);
@@ -682,8 +684,10 @@ public abstract class VanillaChronicleHash<K,
    }

    public final void checkKey(Object key) {
+        Class<K> keyClass = keyClass();
        if (!keyClass.isInstance(key)) {
-            // key.getClass will cause NPE exactly as needed
+            if (key == null)
+                throw new NullPointerException("null key not supported");
            throw new ClassCastException(toIdentityString() + ": Key must be a " +
                    keyClass.getName() + " but was a " + key.getClass());
        }
@@ -23,6 +23,8 @@ import net.openhft.chronicle.hash.serialization.DataAccess;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;

+import java.lang.reflect.Type;
+
 public class ByteableDataAccess<T extends Byteable> extends InstanceCreatingMarshaller<T>
        implements DataAccess<T>, Data<T> {

@@ -31,7 +33,7 @@ public class ByteableDataAccess<T extends Byteable> extends InstanceCreatingMars
     */
    private transient T instance;

-    public ByteableDataAccess(Class<T> tClass) {
+    public ByteableDataAccess(Type tClass) {
        super(tClass);
    }

@@ -91,6 +93,6 @@ public class ByteableDataAccess<T extends Byteable> extends InstanceCreatingMars

    @Override
    public DataAccess<T> copy() {
-        return new ByteableDataAccess<>(tClass());
+        return new ByteableDataAccess<>(tType());
    }
 }
@@ -23,6 +23,8 @@ import net.openhft.chronicle.wire.WireIn;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;

+import java.lang.reflect.Type;
+
 import static net.openhft.chronicle.hash.serialization.impl.DefaultElasticBytes.DEFAULT_BYTES_CAPACITY;

 public class BytesMarshallableDataAccess<T extends BytesMarshallable>
@@ -42,7 +44,7 @@ public class BytesMarshallableDataAccess<T extends BytesMarshallable>
        this(tClass, DEFAULT_BYTES_CAPACITY);
    }

-    private BytesMarshallableDataAccess(Class<T> tClass, long bytesCapacity) {
+    private BytesMarshallableDataAccess(Type tClass, long bytesCapacity) {
        super(tClass);
        initTransients(bytesCapacity);
    }
@@ -140,6 +142,6 @@ public class BytesMarshallableDataAccess<T extends BytesMarshallable>

    @Override
    public DataAccess<T> copy() {
-        return new BytesMarshallableDataAccess<>(tClass(), bytes.realCapacity());
+        return new BytesMarshallableDataAccess<>(tType(), bytes.realCapacity());
    }
 }
@@ -27,6 +27,8 @@ import net.openhft.chronicle.wire.WireOut;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;

+import java.lang.reflect.Type;
+
 import static net.openhft.chronicle.hash.serialization.StatefulCopyable.copyIfNeeded;
 import static net.openhft.chronicle.hash.serialization.impl.DefaultElasticBytes.DEFAULT_BYTES_CAPACITY;

@@ -53,7 +55,7 @@ public class ExternalBytesMarshallableDataAccess<T> extends InstanceCreatingMars
    }

    private ExternalBytesMarshallableDataAccess(
-            Class<T> tClass, SizedReader<T> reader, BytesWriter<? super T> writer,
+            Type tClass, SizedReader<T> reader, BytesWriter<? super T> writer,
            long bytesCapacity) {
        super(tClass);
        this.writer = writer;
@@ -125,7 +127,7 @@ public class ExternalBytesMarshallableDataAccess<T> extends InstanceCreatingMars
    @Override
    public DataAccess<T> copy() {
        return new ExternalBytesMarshallableDataAccess<>(
-                tClass(), copyIfNeeded(reader), copyIfNeeded(writer), bytes.realCapacity());
+                tType(), copyIfNeeded(reader), copyIfNeeded(writer), bytes.realCapacity());
    }

    @Override
@@ -26,6 +26,8 @@ import net.openhft.chronicle.wire.WireIn;
 import net.openhft.chronicle.wire.WireOut;
 import org.jetbrains.annotations.NotNull;

+import java.lang.reflect.Type;
+
 /**
 * Holds new instance creation logic, common for many {@link DataAccess} and {@link SizedReader}
 * implementations
@@ -34,7 +36,7 @@ import org.jetbrains.annotations.NotNull;
 */
 public abstract class InstanceCreatingMarshaller<T> implements Marshallable {

-    private Class<T> tClass;
+    private Type tClass;

    /**
     * Constructor for use in subclasses.
@@ -45,11 +47,19 @@ public abstract class InstanceCreatingMarshaller<T> implements Marshallable {
        this.tClass = tClass;
    }

+    protected InstanceCreatingMarshaller(Type tClass) {
+        this.tClass = tClass;
+    }
+
    /**
     * Returns the class of objects deserialized.
     */
    protected Class<T> tClass() {
-        return tClass;
+        return (Class<T>) tClass;
+    }
+
+    protected Type tType() {
+        return  tClass;
    }

    /**
@@ -63,7 +73,7 @@ public abstract class InstanceCreatingMarshaller<T> implements Marshallable {
     */
    protected T createInstance() {
        try {
-            return ObjectUtils.newInstance(tClass);
+            return ObjectUtils.newInstance(tClass());
        } catch (Exception e) {
            throw new IllegalStateException("Some of default marshallers, chosen for the type\n" +
                    tClass + " by default, delegate to \n" +
@@ -80,7 +90,7 @@ public abstract class InstanceCreatingMarshaller<T> implements Marshallable {
    @Override
    public void readMarshallable(@NotNull WireIn wireIn) {
        //noinspection unchecked
-        tClass = wireIn.read(() -> "tClass").typeLiteral();
+        tClass = wireIn.read(() -> "tClass").lenientTypeLiteral();
    }

    @Override
@@ -20,13 +20,16 @@ import net.openhft.chronicle.hash.serialization.DataAccess;
 import net.openhft.chronicle.hash.serialization.SizedReader;
 import net.openhft.chronicle.hash.serialization.SizedWriter;

+import java.lang.reflect.Type;
+
 import static net.openhft.chronicle.hash.serialization.StatefulCopyable.copyIfNeeded;
+import static net.openhft.chronicle.hash.serialization.impl.DefaultElasticBytes.DEFAULT_BYTES_CAPACITY;

 public class NotReusingSizedMarshallableDataAccess<T> extends SizedMarshallableDataAccess<T> {

    public NotReusingSizedMarshallableDataAccess(
-            Class<T> tClass, SizedReader<T> sizedReader, SizedWriter<? super T> sizedWriter) {
-        super(tClass, sizedReader, sizedWriter);
+            Type tClass, SizedReader<T> sizedReader, SizedWriter<? super T> sizedWriter) {
+        super(tClass, sizedReader, sizedWriter, DEFAULT_BYTES_CAPACITY);
    }

    @Override
@@ -37,6 +40,6 @@ public class NotReusingSizedMarshallableDataAccess<T> extends SizedMarshallableD
    @Override
    public DataAccess<T> copy() {
        return new NotReusingSizedMarshallableDataAccess<>(
-                tClass(), copyIfNeeded(sizedReader()), copyIfNeeded(sizedWriter()));
+                tType(), copyIfNeeded(sizedReader()), copyIfNeeded(sizedWriter()));
    }
 }
@@ -26,6 +26,8 @@ import net.openhft.chronicle.wire.WireOut;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;

+import java.lang.reflect.Type;
+
 import static net.openhft.chronicle.hash.serialization.StatefulCopyable.copyIfNeeded;
 import static net.openhft.chronicle.hash.serialization.impl.DefaultElasticBytes.DEFAULT_BYTES_CAPACITY;

@@ -52,8 +54,8 @@ public class SizedMarshallableDataAccess<T> extends InstanceCreatingMarshaller<T
        this(tClass, sizedReader, sizedWriter, DEFAULT_BYTES_CAPACITY);
    }

-    private SizedMarshallableDataAccess(
-            Class<T> tClass, SizedReader<T> sizedReader, SizedWriter<? super T> sizedWriter,
+    protected SizedMarshallableDataAccess(
+            Type tClass, SizedReader<T> sizedReader, SizedWriter<? super T> sizedWriter,
            long bytesCapacity) {
        super(tClass);
        this.sizedWriter = sizedWriter;
@@ -151,15 +153,15 @@ public class SizedMarshallableDataAccess<T> extends InstanceCreatingMarshaller<T
    @Override
    public DataAccess<T> copy() {
        return new SizedMarshallableDataAccess<>(
-                tClass(), copyIfNeeded(sizedReader), copyIfNeeded(sizedWriter),
+                tType(), copyIfNeeded(sizedReader), copyIfNeeded(sizedWriter),
                bytes.realCapacity());
    }

    @Override
    public void readMarshallable(@NotNull WireIn wireIn) {
        super.readMarshallable(wireIn);
-        sizedReader = wireIn.read(() -> "sizedReader").typedMarshallable();
-        sizedWriter = wireIn.read(() -> "sizedWriter").typedMarshallable();
+        sizedReader = wireIn.read(() -> "sizedReader").object(SizedReader.class);
+        sizedWriter = wireIn.read(() -> "sizedWriter").object(SizedWriter.class);
        initTransients(DEFAULT_BYTES_CAPACITY);
    }

@@ -27,6 +27,7 @@ import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
+import java.lang.reflect.Type;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;

@@ -207,5 +208,10 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>,
     * @return the class of {@code <V>}
     */
    Class<V> valueClass();
+
+    /**
+     * @return the value Class or UnresolvedType if unknown.
+     */
+    Type valueType();
 }

@@ -44,6 +44,7 @@ import org.jetbrains.annotations.NotNull;
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.lang.reflect.Type;
 import java.util.Set;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -73,7 +74,7 @@ public class VanillaChronicleMap<K, V, R>
    public transient DefaultValueProvider<K, V> defaultValueProvider;
    /////////////////////////////////////////////////
    // Value Data model
-    Class<V> valueClass;
+    Type valueClass;
    /////////////////////////////////////////////////
    // Behavior
    transient boolean putReturnsNull;
@@ -116,7 +117,7 @@ public class VanillaChronicleMap<K, V, R>
    protected void readMarshallableFields(@NotNull WireIn wireIn) {
        super.readMarshallableFields(wireIn);

-        valueClass = wireIn.read(() -> "valueClass").typeLiteral();
+        valueClass = wireIn.read(() -> "valueClass").lenientTypeLiteral();
        valueSizeMarshaller = wireIn.read(() -> "valueSizeMarshaller").object(SizeMarshaller.class);
        valueReader = wireIn.read(() -> "valueReader").object(SizedReader.class);
        valueDataAccess = wireIn.read(() -> "valueDataAccess").object(DataAccess.class);
@@ -185,6 +186,7 @@ public class VanillaChronicleMap<K, V, R>
    }

    public final V checkValue(Object value) {
+        Class<V> valueClass = valueClass();
        if (!valueClass.isInstance(value)) {
            throw new ClassCastException(toIdentityString() + ": Value must be a " +
                    valueClass.getName() + " but was a " + value.getClass());
@@ -206,11 +208,21 @@ public class VanillaChronicleMap<K, V, R>

    @Override
    public Class<K> keyClass() {
+        return (Class<K>) keyClass;
+    }
+
+    @Override
+    public Type keyType() {
        return keyClass;
    }

    @Override
    public Class<V> valueClass() {
+        return (Class<V>) valueClass;
+    }
+
+    @Override
+    public Type valueType() {
        return valueClass;
    }

@@ -44,7 +44,7 @@ public class EntryValueBytesData<V> extends AbstractData<V> {

    @Stage("CachedEntryValue")
    private V cachedEntryValue =
-            mh.m().valueClass() == CharSequence.class ? (V) new StringBuilder() : null;
+            mh.m().valueType() == CharSequence.class ? (V) new StringBuilder() : null;
    @Stage("CachedEntryValue")
    private boolean cachedEntryValueRead = false;

@@ -24,6 +24,7 @@ import net.openhft.chronicle.map.VanillaChronicleMap;
 import org.jetbrains.annotations.NotNull;

 import java.io.File;
+import java.lang.reflect.Type;
 import java.util.AbstractSet;
 import java.util.Collection;
 import java.util.Iterator;
@@ -134,6 +135,11 @@ class SetFromMap<E> extends AbstractSet<E> implements ChronicleSet<E> {
        return m.keyClass();
    }

+    @Override
+    public Type keyType() {
+        return m.keyType();
+    }
+
    // TODO test queryContext methods

    @NotNull
@@ -24,6 +24,7 @@ import org.jetbrains.annotations.NotNull;
 import org.junit.Assert;

 import java.io.File;
+import java.lang.reflect.Type;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Set;
@@ -243,6 +244,11 @@ public class ReplicationCheckingMap<K, V> implements ChronicleMap<K, V> {
        return map1.keyClass();
    }

+    @Override
+    public Type keyType() {
+        return map1.keyType();
+    }
+
    @Override
    public boolean forEachEntryWhile(Predicate<? super MapEntry<K, V>> predicate) {
        return map1.forEachEntryWhile(predicate);
@@ -258,6 +264,11 @@ public class ReplicationCheckingMap<K, V> implements ChronicleMap<K, V> {
        return map1.valueClass();
    }

+    @Override
+    public Type valueType() {
+        return map1.valueType();
+    }
+
    @Override
    public V acquireUsing(@NotNull K key, V usingValue) {
        return map1.acquireUsing(key, usingValue);
@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2016-2019 Chronicle Software Ltd
+ */
+
+package net.openhft.chronicle.map.watcher;
+
+import net.openhft.chronicle.core.watcher.FileClassifier;
+import net.openhft.chronicle.core.watcher.FileManager;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+public class MapFileClassifier implements FileClassifier {
+    @Override
+    public FileManager classify(String base, String relative) {
+        Path path = Paths.get(base, relative);
+        if (Files.isRegularFile(path) && relative.endsWith(".cmap")) {
+            return new MapFileManager(base, relative);
+        }
+        return null;
+    }
+}
@@ -0,0 +1,123 @@
+/*
+ * Copyright (c) 2016-2019 Chronicle Software Ltd
+ */
+
+package net.openhft.chronicle.map.watcher;
+
+import net.openhft.chronicle.bytes.MappedBytes;
+import net.openhft.chronicle.bytes.MappedFile;
+import net.openhft.chronicle.core.Jvm;
+import net.openhft.chronicle.core.watcher.JMXFileManager;
+import net.openhft.chronicle.hash.impl.PersistedChronicleHashResources;
+import net.openhft.chronicle.hash.impl.SizePrefixedBlob;
+import net.openhft.chronicle.map.VanillaChronicleMap;
+import net.openhft.chronicle.wire.ReadAnyWire;
+import net.openhft.chronicle.wire.Wires;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.lang.reflect.Type;
+import java.nio.file.Paths;
+
+public class MapFileManager extends JMXFileManager implements MapFileManagerMBean {
+    private static final long TIME_OUT = 5_000;
+    private String header;
+    private Type keyClass;
+    private Type valueClass;
+    private long size;
+    private String name;
+    private long dataStoreSize;
+    private int segments;
+
+    public MapFileManager(String basePath, String relativePath) {
+        super(basePath, relativePath);
+    }
+
+    @Override
+    protected String type() {
+        return "maps";
+    }
+
+    private long lastUpdate = 0;
+
+    public String getHeader() {
+        update();
+        return header;
+    }
+
+    public String getKeyClass() {
+        return ""+keyClass;
+    }
+
+    public String getValueClass() {
+        return ""+valueClass;
+    }
+
+    public long getSize() {
+        return size;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public long getDataStoreSize() {
+        return dataStoreSize;
+    }
+
+    public int getSegments() {
+        return segments;
+    }
+
+    public long getLastUpdate() {
+        return lastUpdate;
+    }
+
+    private void update() {
+        long now = System.currentTimeMillis();
+        if (lastUpdate + TIME_OUT > now)
+            return;
+
+        File file = Paths.get(basePath, relativePath).toFile();
+
+        MappedFile mf;
+        try {
+            mf = MappedFile.mappedFile(file, file.length());
+        } catch (FileNotFoundException e) {
+            Jvm.warn().on(getClass(), e);
+            return;
+        }
+        try (MappedBytes bytes = MappedBytes.mappedBytes(mf)) {
+            int length = bytes.readInt(SizePrefixedBlob.SIZE_WORD_OFFSET);
+            if (!Wires.isReady(length)) {
+                header = "not ready";
+            } else {
+                bytes.readPositionRemaining(SizePrefixedBlob.SIZE_WORD_OFFSET, Wires.lengthOf(length) + 4);
+                header = Wires.fromSizePrefixedBlobs(bytes);
+
+                bytes.readPositionRemaining(SizePrefixedBlob.SELF_BOOTSTRAPPING_HEADER_OFFSET, Wires.lengthOf(length));
+                ReadAnyWire wire = new ReadAnyWire(bytes);
+                VanillaChronicleMap map = wire.getValueIn()
+                        .object(VanillaChronicleMap.class);
+                int headerEnd = SizePrefixedBlob.SELF_BOOTSTRAPPING_HEADER_OFFSET + Wires.lengthOf(length);
+                map.initBeforeMapping(file, mf.raf(), headerEnd, false);
+                map.createMappedStoreAndSegments(
+                        new PersistedChronicleHashResources(file));
+
+
+                keyClass = map.keyType();
+                valueClass = map.valueType();
+                size = map.longSize();
+                name = map.name();
+                dataStoreSize = map.dataStoreSize();
+                segments = map.segments();
+            }
+        } catch (Exception e) {
+            Jvm.warn().on(getClass(), "Unable to update", e);
+            header = e.toString();
+        } finally {
+            mf.release();
+            assert mf.refCount() == 0;
+        }
+    }
+}
@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2016-2019 Chronicle Software Ltd
+ */
+
+package net.openhft.chronicle.map.watcher;
+
+import java.lang.reflect.Type;
+import java.util.Set;
+
+@SuppressWarnings("unused")
+public interface MapFileManagerMBean {
+
+    public String getHeader();
+
+    public String getKeyClass();
+
+    public String getValueClass();
+
+//    public long getSize();
+
+    public String getName();
+
+    public long getDataStoreSize();
+
+    public int getSegments();
+
+    public long getLastUpdate();
+}
@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2016-2019 Chronicle Software Ltd
+ */
+
+package net.openhft.chronicle.map.watcher;
+
+import io.hawt.embedded.Main;
+import net.openhft.chronicle.core.Jvm;
+import net.openhft.chronicle.core.watcher.ClassifyingWatcherListener;
+import net.openhft.chronicle.core.watcher.FileSystemWatcher;
+
+import java.io.File;
+
+public class MapWatcherMain {
+    public static void main(String[] args) throws Exception {
+        FileSystemWatcher fsw = new FileSystemWatcher();
+        String absolutePath = new File(".").getAbsolutePath();
+        System.out.println("Watching " + absolutePath);
+        fsw.addPath(absolutePath);
+        ClassifyingWatcherListener listener = new ClassifyingWatcherListener();
+        listener.addClassifier(new MapFileClassifier());
+        fsw.addListener(listener);
+        fsw.start();
+
+        System.setProperty("hawtio.authenticationEnabled", "false");
+        Main main = new Main();
+        main.setWar(Jvm.userHome() +"/OpenHFT/hawtio-default-2.7.1.war");
+        main.run();
+    }
+}