Merge remote-tracking branch 'origin/master'

2021-09-13 01:26:12 +00:00 · 2015-04-25 10:43:16 +01:00
parent 8f942239ec bdc6d0df90
commit 6d04f566ce
8 changed files with 453 additions and 283 deletions
@@ -40,6 +40,7 @@ import static java.util.Collections.emptyList;
 import static net.openhft.chronicle.engine.client.ClientWiredStatelessTcpConnectionHub.CoreFields.reply;
 import static net.openhft.chronicle.map.MapWireHandler.EventId;
 import static net.openhft.chronicle.map.MapWireHandler.EventId.*;
+import static net.openhft.chronicle.map.VanillaChronicleMap.newInstance;


 /**
@@ -51,24 +52,19 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
    private static final Logger LOG =
            LoggerFactory.getLogger(ClientWiredStatelessChronicleMap.class);

-
    public static final Consumer<ValueOut> VOID_PARAMETERS = out -> out.marshallable(AbstactStatelessClient.EMPTY);
    private final Class vClass;
-
    protected Class<K> kClass;
-
    private boolean putReturnsNull;
    private boolean removeReturnsNull;

-    // used with toString()
-    private static final int MAX_NUM_ENTRIES = 20;

    public ClientWiredStatelessChronicleMap(
            @NotNull final ClientWiredChronicleMapStatelessBuilder config,
            @NotNull final Class kClass,
            @NotNull final Class vClass,
            @NotNull final String channelName,
-            ClientWiredStatelessTcpConnectionHub hub) {
+            @NotNull final ClientWiredStatelessTcpConnectionHub hub) {
        super(channelName, hub, "MAP", 0, vClass);

        this.putReturnsNull = config.putReturnsNull();
@@ -96,12 +92,12 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev

    @Override
    public V newValueInstance() {
-        return (V) VanillaChronicleMap.newInstance(vClass, false);
+        return (V) newInstance(vClass, false);
    }

    @Override
    public K newKeyInstance() {
-        return VanillaChronicleMap.newInstance(kClass, true);
+        return newInstance(kClass, true);
    }

    @Override
@@ -210,8 +206,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev

    @NotNull
    public String toString() {
-
-        return "";
+        return "todo";
    }

    @NotNull
@@ -224,7 +219,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
    }

    public boolean containsKey(Object key) {
-        return proxyReturnBoolean(containsKey, out -> writeField(out, (K) key));
+        return proxyReturnBoolean(containsKey, out -> out.object(key));
    }

    @NotNull
@@ -233,7 +228,7 @@ class ClientWiredStatelessChronicleMap<K, V> extends MapStatelessClient<K, V, Ev
    }

    public boolean containsValue(Object value) {
-        return proxyReturnBoolean(containsValue, out -> writeField(out, (V) value));
+        return proxyReturnBoolean(containsValue, out -> out.object(value));
    }

    public void putAll(@NotNull Map<? extends K, ? extends V> map) {
@@ -80,7 +80,7 @@ public class EngineMap<K, V> implements ChronicleMap<K, V> {
        if (b == null)
            return null;
        final Wire wire = toWire();
-        AbstactStatelessClient.writeField(b, wire.getValueOut());
+        wire.getValueOut().object(b);
        wire.bytes().flip();

        return toWire().getValueIn().bytes();
@@ -160,13 +160,12 @@ public class EngineMap<K, V> implements ChronicleMap<K, V> {
            return null;
        final Wire wire = toWire(bytes);
        buffer.flip();
-        return AbstactStatelessClient.readObject(wire.getValueIn(), null, eClass);
+        return wire.getValueIn().object(eClass);
    }

    @Override
    public V put(final K key, final V value) {
        nullCheck(key);
-        //  nullCheck(value);
        return toObject(vClass, () -> map.put(bytes(key), bytes(value)));
    }

@@ -0,0 +1,62 @@
+package net.openhft.chronicle.map;
+
+/**
+ * Created by daniel on 23/04/15.
+ */
+public class FPMEvent {
+    public enum EventType {NEW, UPDATE, DELETE}
+
+    private EventType eventType;
+    private boolean programmatic;
+    private String key;
+    private String value;
+
+    public FPMEvent(EventType eventType, boolean programmatic, String key, String value) {
+        this.eventType = eventType;
+        this.programmatic = programmatic;
+        this.key = key;
+        this.value = value;
+    }
+
+    public EventType getEventType() {
+        return eventType;
+    }
+
+    public void setEventType(EventType eventType) {
+        this.eventType = eventType;
+    }
+
+    public boolean isProgrammatic() {
+        return programmatic;
+    }
+
+    public void setProgrammatic(boolean programmatic) {
+        this.programmatic = programmatic;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    @Override
+    public String toString() {
+        return "FPMEvent{" +
+                "eventType=" + eventType +
+                ", programmatic=" + programmatic +
+                ", key='" + key + '\'' +
+                ", value='" + value + '\'' +
+                '}';
+    }
+}
@@ -0,0 +1,199 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.hash.function.SerializableFunction;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+
+/**
+ * Created by daniel on 22/04/15.
+ */
+public class FilePerKeyChronicleMap implements ChronicleMap<String, String> {
+    private final ChronicleMap<String, String> chMap;
+    private final FilePerKeyMap fpMap;
+
+    public FilePerKeyChronicleMap(String dir) {
+        fpMap = new FilePerKeyMap(dir);
+        chMap = ChronicleMapBuilder.of(String.class, String.class)
+                .entries(500).averageValueSize(2_000_000).create();
+    }
+
+    @Override
+    public int size() {
+        return fpMap.size();
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return fpMap.isEmpty();
+    }
+
+    @Override
+    public boolean containsKey(Object key) {
+        return fpMap.containsKey(key);
+    }
+
+    @Override
+    public boolean containsValue(Object value) {
+        return fpMap.containsValue(value);
+    }
+
+    @Override
+    public String get(Object key) {
+        return fpMap.get(key);
+    }
+
+    @Override
+    public String put(String key, String value) {
+        fpMap.put(key, value);
+        return chMap.put(key,value);
+    }
+
+    @Override
+    public String remove(Object key) {
+        chMap.remove(key);
+        return fpMap.remove(key);
+    }
+
+    @Override
+    public void putAll(Map<? extends String, ? extends String> m) {
+        fpMap.putAll(m);
+        chMap.putAll(m);
+    }
+
+    @Override
+    public void clear() {
+        fpMap.clear();
+    }
+
+    @NotNull
+    @Override
+    public Set<String> keySet() {
+        return fpMap.keySet();
+    }
+
+    @NotNull
+    @Override
+    public Collection<String> values() {
+        return fpMap.values();
+    }
+
+    @NotNull
+    @Override
+    public Set<Entry<String, String>> entrySet() {
+        return fpMap.entrySet();
+    }
+
+    @Override
+    public String getUsing(String key, String usingValue) {
+        return chMap.getUsing(key, usingValue);
+    }
+
+    @Override
+    public String acquireUsing(@NotNull String key, String usingValue) {
+        //@todo ADD THE LOGIC FOR FPMAP
+        return chMap.acquireUsing(key, usingValue);
+    }
+
+    @NotNull
+    @Override
+    public MapKeyContext<String, String> acquireContext(@NotNull String key, @NotNull String usingValue) {
+        return chMap.acquireContext(key,usingValue);
+    }
+
+    @Override
+    public <R> R getMapped(String key, @NotNull SerializableFunction<? super String, R> function) {
+        return chMap.getMapped(key, function);
+    }
+
+    @Override
+    public String putMapped(String key, @NotNull UnaryOperator<String> unaryOperator) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void getAll(File toFile) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void putAll(File fromFile) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public String newValueInstance() {
+        return new String();
+    }
+
+    @Override
+    public String newKeyInstance() {
+        return new String();
+    }
+
+    @Override
+    public Class<String> valueClass() {
+        return String.class;
+    }
+
+    @Override
+    public File file() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public long longSize() {
+        return size();
+    }
+
+    @Override
+    public MapKeyContext<String, String> context(String key) {
+        return chMap.context(key);
+    }
+
+    @Override
+    public Class<String> keyClass() {
+        return String.class;
+    }
+
+    @Override
+    public boolean forEachEntryWhile(Predicate<? super MapKeyContext<String, String>> predicate) {
+        return chMap.forEachEntryWhile(predicate);
+    }
+
+    @Override
+    public void forEachEntry(Consumer<? super MapKeyContext<String, String>> action) {
+        chMap.forEachEntry(action);
+    }
+
+    @Override
+    public void close() {
+        chMap.close();
+    }
+
+    @Override
+    public String putIfAbsent(String key, String value) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public boolean remove(Object key, Object value) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public boolean replace(String key, String oldValue, String newValue) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public String replace(String key, String value) {
+        throw new UnsupportedOperationException();
+    }
+}
@@ -1,7 +1,6 @@
 package net.openhft.chronicle.map;

 import com.sun.nio.file.SensitivityWatchEventModifier;
-import net.openhft.chronicle.hash.function.SerializableFunction;
 import org.jetbrains.annotations.NotNull;

 import java.io.File;
@@ -10,112 +9,47 @@ import java.nio.file.*;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
-import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;

 /**
- * Created by daniel on 22/04/15.
+ *
 */
-public class FilePerKeyMap<K, V> implements ChronicleMap<K, V> {
-    private String dir;
-    private Path dirPath;
-    private Map<File, Long> lastModifiedByProgram = new ConcurrentHashMap<>();
+public class FilePerKeyMap implements Map<String, String> {
+    private final String dir;
+    private final Path dirPath;
+    private final Map<File, Long> lastModifiedByProgram = new ConcurrentHashMap<>();
+    private final List<Consumer<FPMEvent>> listeners = new ArrayList<>();
+    private final FPMWatcher fileFpmWatcher = new FPMWatcher();

-    public FilePerKeyMap(String dir) {
+    public FilePerKeyMap(String dir){
+        this.dir = dir;
+        this.dirPath = Paths.get(dir);
        try {
-            this.dir = dir;
-            dirPath = Paths.get(dir);
            Files.createDirectories(dirPath);
-
-            new Thread("WatcherThread") {
-                public void run() {
-                    try {
-                        WatchService watcher = FileSystems.getDefault().newWatchService();
-                        dirPath.register(watcher, new WatchEvent.Kind[]{
-                                    StandardWatchEventKinds.ENTRY_CREATE,
-                                    StandardWatchEventKinds.ENTRY_DELETE,
-                                    StandardWatchEventKinds.ENTRY_MODIFY},
-                                    SensitivityWatchEventModifier.HIGH
-                        );
-
-                        while(true){
-                            WatchKey key = null;
-                            try {
-                                key = watcher.take();
-                                System.out.println("Event Taken");
-                                for (WatchEvent<?> event : key.pollEvents()) {
-                                    // get event type
-                                    WatchEvent.Kind<?> kind = event.kind();
-
-                                    // get file name
-                                    @SuppressWarnings("unchecked")
-                                    WatchEvent<Path> ev = (WatchEvent<Path>) event;
-                                    Path fileName = ev.context();
-
-                                    String mapKey = fileName.toString();
-
-
-                                    if (kind == StandardWatchEventKinds.OVERFLOW) {
-                                        System.out.println("OVERFLOW");
-                                        continue;
-                                    } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) {
-                                        Path p = dirPath.resolve(fileName);
-                                        String mapVal = getFileContents(p);
-                                        if(isProgrammaticUpdate(p.toFile())){
-                                            System.out.println("Programmatic create to " + mapKey);// + " : " + mapVal);
-                                        }else {
-                                            System.out.println("created " + mapKey);// + " : " + mapVal);
-                                        }
-                                    } else if (kind == StandardWatchEventKinds.ENTRY_DELETE) {
-                                        Path p = dirPath.resolve(fileName);
-                                        if(isProgrammaticUpdate(p.toFile())){
-                                            System.out.println("Programmatic delete to " + mapKey);
-                                            lastModifiedByProgram.remove(p.toFile());
-                                        }else {
-                                            System.out.println("User delete " + mapKey);
-                                        }
-                                    } else if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
-                                        Path p = dirPath.resolve(fileName);
-                                        String mapVal = getFileContents(p);
-                                        if(isProgrammaticUpdate(p.toFile())){
-                                            System.out.println("Programmatic update to " + mapKey);// + " : " + mapVal);
-                                        }else {
-                                            System.out.println("User modified " + mapKey + " : ");// + mapVal);
-                                        }
-                                    }
-                                }
-
-                            } catch (InterruptedException e) {
-                                System.out.println("Watcher service exiting");
-                                return;
-                            }finally{
-                                key.reset();
-                            }
-                        }
-                    } catch (IOException e) {
-                        throw new AssertionError(e);
-                    }
-                }
-            }.start();
-
-
-        } catch (IOException e) {
-            throw new AssertionError(e);
+        }catch (IOException e){
+            throw new RuntimeException(e);
        }
+        fileFpmWatcher.start();
    }

-    private boolean isProgrammaticUpdate(File file){
-        if(lastModifiedByProgram.containsKey(file)){
-            return file.lastModified() == lastModifiedByProgram.get(file) ? true : false;
-        }else{
-            return false;
+    public void registerForEvents(Consumer<FPMEvent> listener){
+        listeners.add(listener);
+    }
+
+    public void unregisterForEvents(Consumer<FPMEvent> listener){
+        listeners.remove(listener);
+    }
+
+    private void fireEvent(FPMEvent event){
+        for (Consumer<FPMEvent> listener : listeners) {
+            listener.accept(event);
        }
    }

    @Override
    public int size() {
-        return (int)getFiles(dirPath).count();
+        return (int)getFiles().count();
    }

    @Override
@@ -125,184 +59,78 @@ public class FilePerKeyMap<K, V> implements ChronicleMap<K, V> {

    @Override
    public boolean containsKey(Object key) {
-        return getFiles(dirPath).anyMatch(p->p.getFileName().toString().equals(key));
+
+        return getFiles().anyMatch(p->p.getFileName().toString().equals(key));
    }

    @Override
    public boolean containsValue(Object value) {
-        return getFiles(dirPath).anyMatch(p->getFileContents(p).equals(value));
+        return getFiles().anyMatch(p->getFileContents(p).equals(value));
    }

    @Override
-    public V get(Object key) {
-        checkTypeIsString(key);
-
-        Path path = Paths.get(dir,(String)key);
-        return (V)getFileContents(path);
+    public String get(Object key) {
+        Path path = dirPath.resolve((String) key);
+        return getFileContents(path);
    }

    @Override
-    public V put(K key, V value) {
-        checkTypeIsString(key, value);
-
-        Path path = Paths.get(dir,(String)key);
+    public String put(String key, String value) {
+        Path path = dirPath.resolve(key);
        String existingValue = getFileContents(path);
-        writeToFile(path, (String) value);
+        writeToFile(path, value);

-        return (V) existingValue;
+        return existingValue;
    }

    @Override
-    public V remove(Object key) {
-        String existing = (String)get(key);
+    public String remove(Object key) {
+        String existing = get(key);
        if(existing != null){
            deleteFile(Paths.get(dir, (String)key));
        }
-        return (V)existing;
+        return existing;
+
    }

    @Override
-    public void putAll(Map<? extends K, ? extends V> m) {
+    public void putAll(Map<? extends String, ? extends String> m) {
        m.entrySet().stream().forEach(e->put(e.getKey(), e.getValue()));
    }

    @Override
    public void clear() {
-        getFiles(dirPath).forEach(this::deleteFile);
+        getFiles().forEach(this::deleteFile);
    }

    @NotNull
    @Override
-    public Set<K> keySet() {
-        return getFiles(dirPath).map(p -> (K) p.getFileName().toString())
-                                .collect(Collectors.toSet());
-    }
-
-    @NotNull
-    @Override
-    public Collection<V> values() {
-        return getFiles(dirPath).map(p -> (V) getFileContents(p))
+    public Set<String> keySet() {
+        return getFiles().map(p -> p.getFileName().toString())
                .collect(Collectors.toSet());
    }

    @NotNull
    @Override
-    public Set<Entry<K, V>> entrySet() {
-        return getFiles(dirPath).map(p ->
-                (Entry<K, V>) new FPMEntry(p.getFileName().toString(), getFileContents(p)))
+    public Collection<String> values() {
+        return getFiles().map(p -> getFileContents(p))
                .collect(Collectors.toSet());
-    }

-    @Override
-    public V getUsing(K key, V usingValue) {
-        return null;
-    }
-
-    @Override
-    public V acquireUsing(@NotNull K key, V usingValue) {
-        return null;
    }

    @NotNull
    @Override
-    public MapKeyContext<K, V> acquireContext(@NotNull K key, @NotNull V usingValue) {
-        return null;
-    }
-
-    @Override
-    public <R> R getMapped(K key, @NotNull SerializableFunction<? super V, R> function) {
-        return null;
-    }
-
-    @Override
-    public V putMapped(K key, @NotNull UnaryOperator<V> unaryOperator) {
-        return null;
-    }
-
-    @Override
-    public void getAll(File toFile) throws IOException {
-
-    }
-
-    @Override
-    public void putAll(File fromFile) throws IOException {
-
-    }
-
-    @Override
-    public V newValueInstance() {
-        return null;
-    }
-
-    @Override
-    public K newKeyInstance() {
-        return null;
-    }
-
-    @Override
-    public Class<V> valueClass() {
-        return null;
-    }
-
-    @Override
-    public File file() {
-        return null;
-    }
-
-    @Override
-    public long longSize() {
-        return 0;
-    }
-
-    @Override
-    public MapKeyContext<K, V> context(K key) {
-        return null;
-    }
-
-    @Override
-    public Class<K> keyClass() {
-        return null;
-    }
-
-    @Override
-    public boolean forEachEntryWhile(Predicate<? super MapKeyContext<K, V>> predicate) {
-        return false;
-    }
-
-    @Override
-    public void forEachEntry(Consumer<? super MapKeyContext<K, V>> action) {
-
-    }
-
-    @Override
-    public void close() {
-
-    }
-
-    @Override
-    public V putIfAbsent(K key, V value) {
-        return null;
-    }
-
-    @Override
-    public boolean remove(Object key, Object value) {
-        return false;
-    }
-
-    @Override
-    public boolean replace(K key, V oldValue, V newValue) {
-        return false;
-    }
-
-    @Override
-    public V replace(K key, V value) {
-        return null;
+    public Set<Entry<String, String>> entrySet() {
+        return getFiles().map(p ->
+                (Entry<String, String>) new FPMEntry(p.getFileName().toString(), getFileContents(p)))
+                .collect(Collectors.toSet());
    }


-    private Stream<Path> getFiles(Path path){
+
+    private Stream<Path> getFiles(){
        try {
-            return Files.walk(path).filter(p -> !Files.isDirectory(p));
+            return Files.walk(dirPath).filter(p -> !Files.isDirectory(p));
        } catch (IOException e) {
            throw new AssertionError(e);
        }
@@ -340,39 +168,35 @@ public class FilePerKeyMap<K, V> implements ChronicleMap<K, V> {
        }
    }

-
-
-    private void checkTypeIsString(Object... objs) {
-        for(Object o : objs){
-            if(!(o instanceof String))
-                throw new AssertionError("FilePerKeyMap only accepts Key and Values " +
-                    "of type String. Unsupported type: '" + o + "'");
-        }
+    public void close(){
+        fileFpmWatcher.interrupt();
    }

-    private static class FPMEntry<K,V> implements Entry
+    private static class FPMEntry<String> implements Entry<String, String>
    {
-        private K key;
-        private V value;
+        private String key;
+        private String value;

-        public FPMEntry(K key, V value){
+        public FPMEntry(String key, String value){
            this.key = key;
            this.value = value;
        }

        @Override
-        public K getKey() {
+        public String getKey() {
            return key;
        }

        @Override
-        public V getValue() {
+        public String getValue() {
            return value;
        }

        @Override
-        public Object setValue(Object value) {
-            throw new UnsupportedOperationException();
+        public String setValue(String value) {
+            String lastValue = this.value;
+            this.value = value;
+            return lastValue;
        }

        @Override
@@ -380,10 +204,10 @@ public class FilePerKeyMap<K, V> implements ChronicleMap<K, V> {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;

-            FPMEntry<?, ?> entry = (FPMEntry<?, ?>) o;
+            FPMEntry<?> fpmEntry = (FPMEntry<?>) o;

-            if (key != null ? !key.equals(entry.key) : entry.key != null) return false;
-            return !(value != null ? !value.equals(entry.value) : entry.value != null);
+            if (key != null ? !key.equals(fpmEntry.key) : fpmEntry.key != null) return false;
+            return !(value != null ? !value.equals(fpmEntry.value) : fpmEntry.value != null);

        }

@@ -394,4 +218,77 @@ public class FilePerKeyMap<K, V> implements ChronicleMap<K, V> {
            return result;
        }
    }
+
+    private class FPMWatcher extends Thread{
+        @Override
+        public void run(){
+            try {
+                WatchService watcher = FileSystems.getDefault().newWatchService();
+                dirPath.register(watcher, new WatchEvent.Kind[]{
+                                StandardWatchEventKinds.ENTRY_CREATE,
+                                StandardWatchEventKinds.ENTRY_DELETE,
+                                StandardWatchEventKinds.ENTRY_MODIFY},
+                        SensitivityWatchEventModifier.HIGH
+                );
+
+                while(true){
+                    WatchKey key = null;
+                    try {
+                        key = watcher.take();
+                        for (WatchEvent<?> event : key.pollEvents()) {
+                            // get event type
+                            WatchEvent.Kind<?> kind = event.kind();
+
+                            // get file name
+                            WatchEvent<Path> ev = (WatchEvent<Path>) event;
+                            Path fileName = ev.context();
+
+                            String mapKey = fileName.toString();
+
+
+                            if (kind == StandardWatchEventKinds.OVERFLOW) {
+                                continue;
+                            } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) {
+                                Path p = dirPath.resolve(fileName);
+                                String mapVal = getFileContents(p);
+                                if(isProgrammaticUpdate(p.toFile())){
+                                    fireEvent(new FPMEvent(FPMEvent.EventType.NEW, true, mapKey,mapVal));
+                                }else {
+                                    fireEvent(new FPMEvent(FPMEvent.EventType.NEW, false, mapKey, mapVal));
+                                }
+                            } else if (kind == StandardWatchEventKinds.ENTRY_DELETE) {
+                                Path p = dirPath.resolve(fileName);
+                                if(isProgrammaticUpdate(p.toFile())){
+                                    fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, true, mapKey, null));
+                                    lastModifiedByProgram.remove(p.toFile());
+                                }else {
+                                    fireEvent(new FPMEvent(FPMEvent.EventType.DELETE, false, mapKey, null));
+                                }
+                            } else if (kind == StandardWatchEventKinds.ENTRY_MODIFY) {
+                                Path p = dirPath.resolve(fileName);
+                                String mapVal = getFileContents(p);
+                                if(isProgrammaticUpdate(p.toFile())){
+                                    fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, true, mapKey,mapVal));
+                                }else {
+                                    fireEvent(new FPMEvent(FPMEvent.EventType.UPDATE, false, mapKey, mapVal));
+                                }
+                            }
+                        }
+                    } catch (InterruptedException e) {
+                        return;
+                    }finally{
+                        if(key!=null)key.reset();
+                    }
+                }
+            } catch (IOException e) {
+                throw new AssertionError(e);
+            }
+        }
+    }
+
+    private boolean isProgrammaticUpdate(File file) {
+        return lastModifiedByProgram.containsKey(file)
+                && (file.lastModified() == lastModifiedByProgram.get(file));
+    }
+
 }
@@ -54,10 +54,11 @@ public abstract class MapStatelessClient<K, V, E extends ParameterizeWireKey>

    @Nullable
    protected <R> R proxyReturnObject(@NotNull final Class<R> rClass,
-                                      @NotNull final E eventId, Object field) {
+                                      @NotNull final E eventId,
+                                      @NotNull final Object o) {

        final long startTime = System.currentTimeMillis();
-        final long tid = sendEvent(startTime, eventId, out -> writeField(out, field));
+        final long tid = sendEvent(startTime, eventId, out -> out.object(o));

        if (eventReturnsNull(eventId))
            return null;
@@ -897,16 +897,13 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
            serverSocket.bind(details.address());

            // these can be run on this thread
-            addPendingRegistration(new Runnable() {
-                @Override
-                public void run() {
-                    final Attached attached = new Attached();
-                    attached.connector = ServerConnector.this;
-                    try {
-                        serverChannel.register(TcpReplicator.this.selector, OP_ACCEPT, attached);
-                    } catch (ClosedChannelException e) {
-                        LOG.debug("", e);
-                    }
+            addPendingRegistration(() -> {
+                final Attached attached = new Attached();
+                attached.connector = ServerConnector.this;
+                try {
+                    serverChannel.register(TcpReplicator.this.selector, OP_ACCEPT, attached);
+                } catch (ClosedChannelException e) {
+                    LOG.debug("", e);
                }
            });

@@ -13,22 +13,13 @@ import static org.junit.Assert.*;
 * JUnit test class to support {@link FilePerKeyMapTest}
 */
 public class FilePerKeyMapTest {
-    /**
-     * FilePerMapKey only works with Strings
-     */
-    @Test(expected = AssertionError.class)
-    public void testWrongType(){
-        FilePerKeyMap<Integer, String> map = new FilePerKeyMap<>("/tmp/filepermaptests");
-        map.put(1, "test");
-    }
-
    /**
     * Testing all the methods of the map with simple tests.
     */
    @Test
    public void testMapMethods() {
        //There are no entries in the map so null should be returned
-        FilePerKeyMap<String, String> map = new FilePerKeyMap<>("/tmp/filepermaptests");
+        FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests");

        //just in case it hasn't been cleared up last time
        map.clear();
@@ -117,12 +108,16 @@ public class FilePerKeyMapTest {
        } catch (InterruptedException e) {
             e.printStackTrace();
        }
+
+        map.clear();
+        map.close();
+
    }

    @Test
    public void perfTest(){
        //There are no entries in the map so null should be returned
-        FilePerKeyMap<String, String> map = new FilePerKeyMap<>("/tmp/filepermaptests");
+        FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests");

        //just in case it hasn't been cleared up last time
        map.clear();
@@ -133,8 +128,8 @@ public class FilePerKeyMapTest {
        }
        String value = sb.toString();

-        //warmup
-        for (int i = 0; i < 1000; i++) {
+        //small warm-up
+        for (int i = 0; i < 10; i++) {
            map.put("big file", value);
        }

@@ -145,6 +140,31 @@ public class FilePerKeyMapTest {
            map.put("big file", value);
        }
        System.out.println("Time to update "+ iterations + " iterations " + (System.currentTimeMillis()-time));
+
+        map.clear();
+        map.close();
    }

+    @Test
+    public void eventTest() {
+        //There are no entries in the map so null should be returned
+        FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests");
+
+        //just in case it hasn't been cleared up last time
+        map.clear();
+
+        map.registerForEvents(e -> System.out.println(e));
+
+        map.put("one", "one");
+        map.put("one", "one");
+
+        try {
+            Thread.sleep(3000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        map.clear();
+        map.close();
+    }
 }