From 1aa888f5f5aca0e7d5215a8e6daadd9d473dd17c Mon Sep 17 00:00:00 2001 From: danielshaya Date: Fri, 24 Apr 2015 16:53:08 +0100 Subject: [PATCH] Initial commit for FilePerMap implementation. --- .../openhft/chronicle/map/FPMListener.java | 10 ---- .../chronicle/map/FilePerKeyChronicleMap.java | 46 +++++++++---------- .../openhft/chronicle/map/FilePerKeyMap.java | 24 ++++++---- .../chronicle/map/FilePerKeyMapTest.java | 23 ++++++---- 4 files changed, 51 insertions(+), 52 deletions(-) delete mode 100644 src/main/java/net/openhft/chronicle/map/FPMListener.java diff --git a/src/main/java/net/openhft/chronicle/map/FPMListener.java b/src/main/java/net/openhft/chronicle/map/FPMListener.java deleted file mode 100644 index 7fba5460b..000000000 --- a/src/main/java/net/openhft/chronicle/map/FPMListener.java +++ /dev/null @@ -1,10 +0,0 @@ -package net.openhft.chronicle.map; - -/** - * Created by daniel on 23/04/15. - */ -public class FPMListener { - public void onEvent(FPMEvent event){ - - } -} diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java index 6648c4c90..84cf435e4 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java @@ -8,7 +8,6 @@ import java.io.IOException; import java.util.Collection; import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; import java.util.function.Predicate; @@ -16,16 +15,13 @@ import java.util.function.Predicate; * Created by daniel on 22/04/15. */ public class FilePerKeyChronicleMap implements ChronicleMap { - private final String dir; - private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); private final ChronicleMap chMap; private final FilePerKeyMap fpMap; public FilePerKeyChronicleMap(String dir) { - this.dir = dir; - fpMap = new FilePerKeyMap(dir); - chMap = ChronicleMapBuilder.of(String.class, String.class) - .entries(500).averageValueSize(2_000_000).create(); + fpMap = new FilePerKeyMap(dir); + chMap = ChronicleMapBuilder.of(String.class, String.class) + .entries(500).averageValueSize(2_000_000).create(); } @Override @@ -55,17 +51,20 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String put(String key, String value) { - return fpMap.put(key, 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 m) { fpMap.putAll(m); + chMap.putAll(m); } @Override @@ -93,27 +92,28 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String getUsing(String key, String usingValue) { - throw new UnsupportedOperationException(); + return chMap.getUsing(key, usingValue); } @Override public String acquireUsing(@NotNull String key, String usingValue) { - throw new UnsupportedOperationException(); + //@todo ADD THE LOGIC FOR FPMAP + return chMap.acquireUsing(key, usingValue); } @NotNull @Override public MapKeyContext acquireContext(@NotNull String key, @NotNull String usingValue) { - throw new UnsupportedOperationException(); + return chMap.acquireContext(key,usingValue); } @Override - public R getMapped( String key, @NotNull SerializableFunction function) { - throw new UnsupportedOperationException(); + public R getMapped(String key, @NotNull SerializableFunction function) { + return chMap.getMapped(key, function); } @Override - public String putMapped( String key, @NotNull UnaryOperator unaryOperator) { + public String putMapped(String key, @NotNull UnaryOperator unaryOperator) { throw new UnsupportedOperationException(); } @@ -129,17 +129,17 @@ public class FilePerKeyChronicleMap implements ChronicleMap { @Override public String newValueInstance() { - return null; + return new String(); } @Override public String newKeyInstance() { - throw new UnsupportedOperationException(); + return new String(); } @Override public Class valueClass() { - throw new UnsupportedOperationException(); + return String.class; } @Override @@ -153,28 +153,28 @@ public class FilePerKeyChronicleMap implements ChronicleMap { } @Override - public MapKeyContext context( String key) { - throw new UnsupportedOperationException(); + public MapKeyContext context(String key) { + return chMap.context(key); } @Override public Class keyClass() { - throw new UnsupportedOperationException(); + return String.class; } @Override public boolean forEachEntryWhile(Predicate> predicate) { - throw new UnsupportedOperationException(); + return chMap.forEachEntryWhile(predicate); } @Override public void forEachEntry(Consumer> action) { - throw new UnsupportedOperationException(); + chMap.forEachEntry(action); } @Override public void close() { - throw new UnsupportedOperationException(); + chMap.close(); } @Override diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index b94d7ab71..cd0925644 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -8,6 +8,7 @@ import java.io.IOException; import java.nio.file.*; import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -18,7 +19,8 @@ public class FilePerKeyMap implements Map { private final String dir; private final Path dirPath; private final Map lastModifiedByProgram = new ConcurrentHashMap<>(); - private final List listeners = new ArrayList<>(); + private final List> listeners = new ArrayList<>(); + private final FPMWatcher fileFpmWatcher = new FPMWatcher(); public FilePerKeyMap(String dir){ this.dir = dir; @@ -28,16 +30,20 @@ public class FilePerKeyMap implements Map { }catch (IOException e){ throw new RuntimeException(e); } - new FPMWatcher().start(); + fileFpmWatcher.start(); } - public void registerForEvents(FPMListener listener){ + public void registerForEvents(Consumer listener){ listeners.add(listener); } + public void unregisterForEvents(Consumer listener){ + listeners.remove(listener); + } + private void fireEvent(FPMEvent event){ - for (FPMListener listener : listeners) { - listener.onEvent(event); + for (Consumer listener : listeners) { + listener.accept(event); } } @@ -162,6 +168,9 @@ public class FilePerKeyMap implements Map { } } + public void close(){ + fileFpmWatcher.interrupt(); + } private static class FPMEntry implements Entry { @@ -238,7 +247,6 @@ public class FilePerKeyMap implements Map { if (kind == StandardWatchEventKinds.OVERFLOW) { - System.out.println("OVERFLOW"); continue; } else if (kind == StandardWatchEventKinds.ENTRY_CREATE) { Path p = dirPath.resolve(fileName); @@ -266,12 +274,10 @@ public class FilePerKeyMap implements Map { } } } - } catch (InterruptedException e) { - System.out.println("Watcher service exiting"); return; }finally{ - key.reset(); + if(key!=null)key.reset(); } } } catch (IOException e) { diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java index fa3590ea3..56c75ae50 100644 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java @@ -108,6 +108,10 @@ public class FilePerKeyMapTest { } catch (InterruptedException e) { e.printStackTrace(); } + + map.clear(); + map.close(); + } @Test @@ -136,32 +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(){ + 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(new FPMListener() { - @Override - public void onEvent(FPMEvent e) { - System.out.println(e); - } - - }); + map.registerForEvents(e -> System.out.println(e)); map.put("one", "one"); map.put("one", "one"); try { - Thread.sleep(300000); + Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } - } + map.clear(); + map.close(); + } }