From e9f995379d80ea5286ea1c7aa804c3d76b3592a9 Mon Sep 17 00:00:00 2001 From: danielshaya Date: Thu, 23 Apr 2015 18:37:53 +0100 Subject: [PATCH] Initial commit for FilePerMap implementation. --- .../net/openhft/chronicle/map/FPMEvent.java | 62 +++ .../openhft/chronicle/map/FPMListener.java | 10 + .../chronicle/map/FilePerKeyChronicleMap.java | 199 +++++++++ .../openhft/chronicle/map/FilePerKeyMap.java | 384 +++++++----------- .../chronicle/map/FilePerKeyMapTest.java | 43 +- 5 files changed, 440 insertions(+), 258 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/FPMEvent.java create mode 100644 src/main/java/net/openhft/chronicle/map/FPMListener.java create mode 100644 src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java diff --git a/src/main/java/net/openhft/chronicle/map/FPMEvent.java b/src/main/java/net/openhft/chronicle/map/FPMEvent.java new file mode 100644 index 000000000..bf7278e83 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/FPMEvent.java @@ -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 + '\'' + + '}'; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/FPMListener.java b/src/main/java/net/openhft/chronicle/map/FPMListener.java new file mode 100644 index 000000000..7fba5460b --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/FPMListener.java @@ -0,0 +1,10 @@ +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 new file mode 100644 index 000000000..6648c4c90 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyChronicleMap.java @@ -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.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; +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(); + } + + @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) { + return fpMap.put(key, value); + } + + @Override + public String remove(Object key) { + return fpMap.remove(key); + } + + @Override + public void putAll(Map m) { + fpMap.putAll(m); + } + + @Override + public void clear() { + fpMap.clear(); + } + + @NotNull + @Override + public Set keySet() { + return fpMap.keySet(); + } + + @NotNull + @Override + public Collection values() { + return fpMap.values(); + } + + @NotNull + @Override + public Set> entrySet() { + return fpMap.entrySet(); + } + + @Override + public String getUsing(String key, String usingValue) { + throw new UnsupportedOperationException(); + } + + @Override + public String acquireUsing(@NotNull String key, String usingValue) { + throw new UnsupportedOperationException(); + } + + @NotNull + @Override + public MapKeyContext acquireContext(@NotNull String key, @NotNull String usingValue) { + throw new UnsupportedOperationException(); + } + + @Override + public R getMapped( String key, @NotNull SerializableFunction function) { + throw new UnsupportedOperationException(); + } + + @Override + public String putMapped( String key, @NotNull UnaryOperator 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 null; + } + + @Override + public String newKeyInstance() { + throw new UnsupportedOperationException(); + } + + @Override + public Class valueClass() { + throw new UnsupportedOperationException(); + } + + @Override + public File file() { + throw new UnsupportedOperationException(); + } + + @Override + public long longSize() { + return size(); + } + + @Override + public MapKeyContext context( String key) { + throw new UnsupportedOperationException(); + } + + @Override + public Class keyClass() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean forEachEntryWhile(Predicate> predicate) { + throw new UnsupportedOperationException(); + } + + @Override + public void forEachEntry(Consumer> action) { + throw new UnsupportedOperationException(); + } + + @Override + public void close() { + throw new UnsupportedOperationException(); + } + + @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(); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java index 20ca46293..b94d7ab71 100644 --- a/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java +++ b/src/main/java/net/openhft/chronicle/map/FilePerKeyMap.java @@ -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; @@ -9,113 +8,42 @@ import java.io.IOException; 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 implements ChronicleMap { - private String dir; - private Path dirPath; - private Map lastModifiedByProgram = new ConcurrentHashMap<>(); +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<>(); - 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 ev = (WatchEvent) 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); } + new FPMWatcher().start(); } - private boolean isProgrammaticUpdate(File file){ - if(lastModifiedByProgram.containsKey(file)){ - return file.lastModified() == lastModifiedByProgram.get(file) ? true : false; - }else{ - return false; + public void registerForEvents(FPMListener listener){ + listeners.add(listener); + } + + private void fireEvent(FPMEvent event){ + for (FPMListener listener : listeners) { + listener.onEvent(event); } } @Override public int size() { - return (int)getFiles(dirPath).count(); + return (int)getFiles().count(); } @Override @@ -125,184 +53,78 @@ public class FilePerKeyMap implements ChronicleMap { @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 m) { + public void putAll(Map 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 keySet() { - return getFiles(dirPath).map(p -> (K) p.getFileName().toString()) - .collect(Collectors.toSet()); - } - - @NotNull - @Override - public Collection values() { - return getFiles(dirPath).map(p -> (V) getFileContents(p)) + public Set keySet() { + return getFiles().map(p -> p.getFileName().toString()) .collect(Collectors.toSet()); } @NotNull @Override - public Set> entrySet() { - return getFiles(dirPath).map(p -> - (Entry) new FPMEntry(p.getFileName().toString(), getFileContents(p))) + public Collection 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 acquireContext(@NotNull K key, @NotNull V usingValue) { - return null; - } - - @Override - public R getMapped(K key, @NotNull SerializableFunction function) { - return null; - } - - @Override - public V putMapped(K key, @NotNull UnaryOperator 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 valueClass() { - return null; - } - - @Override - public File file() { - return null; - } - - @Override - public long longSize() { - return 0; - } - - @Override - public MapKeyContext context(K key) { - return null; - } - - @Override - public Class keyClass() { - return null; - } - - @Override - public boolean forEachEntryWhile(Predicate> predicate) { - return false; - } - - @Override - public void forEachEntry(Consumer> 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> entrySet() { + return getFiles().map(p -> + (Entry) new FPMEntry(p.getFileName().toString(), getFileContents(p))) + .collect(Collectors.toSet()); } - private Stream getFiles(Path path){ + + private Stream 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); } @@ -341,38 +163,31 @@ public class FilePerKeyMap implements ChronicleMap { } - - 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 + "'"); - } - } - - private static class FPMEntry implements Entry + private static class FPMEntry implements Entry { - 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 +195,10 @@ public class FilePerKeyMap implements ChronicleMap { 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 +209,83 @@ public class FilePerKeyMap implements ChronicleMap { 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 ev = (WatchEvent) 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())){ + 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) { + System.out.println("Watcher service exiting"); + return; + }finally{ + key.reset(); + } + } + } catch (IOException e) { + throw new AssertionError(e); + } + } + } + + private boolean isProgrammaticUpdate(File file){ + if(lastModifiedByProgram.containsKey(file)){ + return file.lastModified() == lastModifiedByProgram.get(file) ? true : false; + }else{ + return false; + } + } + } diff --git a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java index cab4a770c..fa3590ea3 100644 --- a/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/FilePerKeyMapTest.java @@ -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 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 map = new FilePerKeyMap<>("/tmp/filepermaptests"); + FilePerKeyMap map = new FilePerKeyMap("/tmp/filepermaptests"); //just in case it hasn't been cleared up last time map.clear(); @@ -122,7 +113,7 @@ public class FilePerKeyMapTest { @Test public void perfTest(){ //There are no entries in the map so null should be returned - FilePerKeyMap 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 +124,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); } @@ -147,4 +138,30 @@ public class FilePerKeyMapTest { System.out.println("Time to update "+ iterations + " iterations " + (System.currentTimeMillis()-time)); } + @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(new FPMListener() { + @Override + public void onEvent(FPMEvent e) { + System.out.println(e); + } + + }); + + map.put("one", "one"); + map.put("one", "one"); + + try { + Thread.sleep(300000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + }