Initial commit for FilePerMap implementation.

This commit is contained in:
danielshaya
2015-04-24 16:53:08 +01:00
parent 503ee11fc8
commit 1aa888f5f5
4 changed files with 51 additions and 52 deletions
@@ -1,10 +0,0 @@
package net.openhft.chronicle.map;
/**
* Created by daniel on 23/04/15.
*/
public class FPMListener {
public void onEvent(FPMEvent event){
}
}
@@ -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<String, String> {
private final String dir;
private final Map<File, Long> lastModifiedByProgram = new ConcurrentHashMap<>();
private final ChronicleMap<String, String> 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<String, String> {
@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<? extends String, ? extends String> m) {
fpMap.putAll(m);
chMap.putAll(m);
}
@Override
@@ -93,27 +92,28 @@ public class FilePerKeyChronicleMap implements ChronicleMap<String, String> {
@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<String, String> acquireContext(@NotNull String key, @NotNull String usingValue) {
throw new UnsupportedOperationException();
return chMap.acquireContext(key,usingValue);
}
@Override
public <R> R getMapped( String key, @NotNull SerializableFunction<? super String, R> function) {
throw new UnsupportedOperationException();
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) {
public String putMapped(String key, @NotNull UnaryOperator<String> unaryOperator) {
throw new UnsupportedOperationException();
}
@@ -129,17 +129,17 @@ public class FilePerKeyChronicleMap implements ChronicleMap<String, String> {
@Override
public String newValueInstance() {
return null;
return new String();
}
@Override
public String newKeyInstance() {
throw new UnsupportedOperationException();
return new String();
}
@Override
public Class<String> valueClass() {
throw new UnsupportedOperationException();
return String.class;
}
@Override
@@ -153,28 +153,28 @@ public class FilePerKeyChronicleMap implements ChronicleMap<String, String> {
}
@Override
public MapKeyContext<String, String> context( String key) {
throw new UnsupportedOperationException();
public MapKeyContext<String, String> context(String key) {
return chMap.context(key);
}
@Override
public Class<String> keyClass() {
throw new UnsupportedOperationException();
return String.class;
}
@Override
public boolean forEachEntryWhile(Predicate<? super MapKeyContext<String, String>> predicate) {
throw new UnsupportedOperationException();
return chMap.forEachEntryWhile(predicate);
}
@Override
public void forEachEntry(Consumer<? super MapKeyContext<String, String>> action) {
throw new UnsupportedOperationException();
chMap.forEachEntry(action);
}
@Override
public void close() {
throw new UnsupportedOperationException();
chMap.close();
}
@Override
@@ -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<String, String> {
private final String dir;
private final Path dirPath;
private final Map<File, Long> lastModifiedByProgram = new ConcurrentHashMap<>();
private final List<FPMListener> listeners = new ArrayList<>();
private final List<Consumer<FPMEvent>> 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<String, String> {
}catch (IOException e){
throw new RuntimeException(e);
}
new FPMWatcher().start();
fileFpmWatcher.start();
}
public void registerForEvents(FPMListener listener){
public void registerForEvents(Consumer<FPMEvent> listener){
listeners.add(listener);
}
public void unregisterForEvents(Consumer<FPMEvent> listener){
listeners.remove(listener);
}
private void fireEvent(FPMEvent event){
for (FPMListener listener : listeners) {
listener.onEvent(event);
for (Consumer<FPMEvent> listener : listeners) {
listener.accept(event);
}
}
@@ -162,6 +168,9 @@ public class FilePerKeyMap implements Map<String, String> {
}
}
public void close(){
fileFpmWatcher.interrupt();
}
private static class FPMEntry<String> implements Entry<String, String>
{
@@ -238,7 +247,6 @@ public class FilePerKeyMap implements Map<String, String> {
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<String, String> {
}
}
}
} catch (InterruptedException e) {
System.out.println("Watcher service exiting");
return;
}finally{
key.reset();
if(key!=null)key.reset();
}
}
} catch (IOException e) {