聊聊claudb的Database

public interface Database {
​
  int size();
​
  boolean isEmpty();
​
  boolean containsKey(DatabaseKey key);
​
  DatabaseValue get(DatabaseKey key);
​
  DatabaseValue put(DatabaseKey key, DatabaseValue value);
​
  DatabaseValue remove(DatabaseKey key);
​
  void clear();
​
  ImmutableSet<DatabaseKey> keySet();
​
  Sequence<DatabaseValue> values();
​
  ImmutableSet<Tuple2<DatabaseKey, DatabaseValue>> entrySet();
​
  default SafeString getString(SafeString key) {
    return getOrDefault(safeKey(key), DatabaseValue.EMPTY_STRING).getString();
  }
​
  default ImmutableList<SafeString> getList(SafeString key) {
    return getOrDefault(safeKey(key), DatabaseValue.EMPTY_LIST).getList();
  }
​
  default ImmutableSet<SafeString> getSet(SafeString key) {
    return getOrDefault(safeKey(key), DatabaseValue.EMPTY_SET).getSet();
  }
​
  default NavigableSet<Entry<Double, SafeString>> getSortedSet(SafeString key) {
    return getOrDefault(safeKey(key), DatabaseValue.EMPTY_ZSET).getSortedSet();
  }
​
  default ImmutableMap<SafeString, SafeString> getHash(SafeString key) {
    return getOrDefault(safeKey(key), DatabaseValue.EMPTY_HASH).getHash();
  }
​
  default void putAll(ImmutableMap<? extends DatabaseKey, ? extends DatabaseValue> map) {
    map.forEach(this::put);
  }
​
  default DatabaseValue putIfAbsent(DatabaseKey key, DatabaseValue value) {
    DatabaseValue oldValue = get(key);
    if (oldValue == null) {
        oldValue = put(key, value);
    }
    return oldValue;
  }
​
  default DatabaseValue merge(DatabaseKey key, DatabaseValue value,
      BiFunction<DatabaseValue, DatabaseValue, DatabaseValue> remappingFunction) {
    DatabaseValue oldValue = get(key);
    DatabaseValue newValue = oldValue == null ? value : remappingFunction.apply(oldValue, value);
    if(newValue == null) {
      remove(key);
    } else {
      put(key, newValue);
    }
    return newValue;
  }
​
  default DatabaseValue getOrDefault(DatabaseKey key, DatabaseValue defaultValue) {
    DatabaseValue value = get(key);
    return (value != null || containsKey(key)) ? value : defaultValue;
  }
​
  default boolean isType(DatabaseKey key, DataType type) {
    DatabaseValue value = get(key);
    return value != null ? value.getType() == type : true;
  }
​
  default boolean rename(DatabaseKey from, DatabaseKey to) {
    DatabaseValue value = remove(from);
    if (value != null) {
      put(to, value);
      return true;
    }
    return false;
  }
​
  default void overrideAll(ImmutableMap<DatabaseKey, DatabaseValue> value) {
    clear();
    putAll(value);
  }
​
  default ImmutableSet<DatabaseKey> evictableKeys(Instant now) {
    return entrySet()
        .filter(entry -> entry.get2().isExpired(now))
        .map(Tuple2::get1);
  }
}

public class OnHeapDatabase implements Database {
​
  private final Map<DatabaseKey, DatabaseValue> cache;
​
  public OnHeapDatabase(Map<DatabaseKey, DatabaseValue> cache) {
    this.cache = cache;
  }
​
  @Override
  public int size() {
    return cache.size();
  }
​
  @Override
  public boolean isEmpty() {
    return cache.isEmpty();
  }
​
  @Override
  public boolean containsKey(DatabaseKey key) {
    return cache.containsKey(key);
  }
​
  @Override
  public DatabaseValue get(DatabaseKey key) {
    DatabaseValue value = cache.get(key);
    if (value != null) {
      if (!value.isExpired(Instant.now())) {
        return value;
      }
      cache.remove(key);
    }
    return null;
  }
​
  @Override
  public DatabaseValue put(DatabaseKey key, DatabaseValue value) {
    DatabaseValue oldValue = cache.remove(key);
    cache.put(key, value);
    return oldValue;
  }
​
  @Override
  public DatabaseValue remove(DatabaseKey key) {
    return cache.remove(key);
  }
​
  @Override
  public void clear() {
    cache.clear();
  }
​
  @Override
  public ImmutableSet<DatabaseKey> keySet() {
    return ImmutableSet.from(cache.keySet());
  }
​
  @Override
  public Sequence<DatabaseValue> values() {
    return ImmutableSet.from(cache.values());
  }
​
  @Override
  public ImmutableSet<Tuple2<DatabaseKey, DatabaseValue>> entrySet() {
    return ImmutableSet.from(cache.entrySet()).map(Tuple::from);
  }
}

public class OffHeapDatabase implements Database {
​
  private OHCache<DatabaseKey, DatabaseValue> cache;
​
  public OffHeapDatabase(OHCache<DatabaseKey, DatabaseValue> cache) {
    this.cache = cache;
  }
​
  @Override
  public int size() {
    return (int) cache.size();
  }
​
  @Override
  public boolean isEmpty() {
    return cache.size() == 0;
  }
​
  @Override
  public boolean containsKey(DatabaseKey key) {
    return cache.containsKey(key);
  }
​
  @Override
  public DatabaseValue get(DatabaseKey key) {
    DatabaseValue value = cache.get(key);
    if (value != null) {
      if (!value.isExpired(Instant.now())) {
        return value;
      }
      cache.remove(key);
    }
    return null;
  }
​
  @Override
  public DatabaseValue put(DatabaseKey key, DatabaseValue value) {
    cache.put(key, value);
    return value;
  }
​
  @Override
  public DatabaseValue remove(DatabaseKey key) {
    DatabaseValue value = get(key);
    cache.remove(key);
    return value;
  }
​
  @Override
  public void clear() {
    cache.clear();
  }
​
  @Override
  public ImmutableSet<DatabaseKey> keySet() {
    Set<DatabaseKey> keys = new HashSet<>();
    try (CloseableIterator<DatabaseKey> iterator = cache.keyIterator()) {
      while (iterator.hasNext()) {
        keys.add(iterator.next());
      }
    } catch(IOException e) {
      throw new UncheckedIOException(e);
    }
    return ImmutableSet.from(keys);
  }
​
  @Override
  public Sequence<DatabaseValue> values() {
    List<DatabaseValue> values = new LinkedList<>();
    for (DatabaseKey key : keySet()) {
      values.add(cache.get(key));
    }
    return ImmutableList.from(values);
  }
​
  @Override
  public ImmutableSet<Tuple2<DatabaseKey, DatabaseValue>> entrySet() {
    return keySet().map(key -> Tuple.of(key, get(key)));
  }
}

public interface DatabaseFactory {
  Database create(String name);
  void clear();
}

public class OnHeapDatabaseFactory implements DatabaseFactory {
​
  @Override
  public Database create(String name) {
    return new OnHeapDatabase(new HashMap<>());
  }
​
  @Override
  public void clear() {
    // nothing to clear
  }
}

public class OffHeapDatabaseFactory implements DatabaseFactory {
​
  @Override
  public Database create(String name) {
    return new OffHeapDatabase(createCache());
  }
​
  private OHCache<DatabaseKey, DatabaseValue> createCache() {
    return builder()
        .eviction(Eviction.NONE)
        .throwOOME(true)
        .keySerializer(new FSTSerializer<>())
        .valueSerializer(new FSTSerializer<>())
        .build();
  }
​
  private OHCacheBuilder<DatabaseKey, DatabaseValue> builder() {
    return OHCacheBuilder.newBuilder();
  }
​
  @Override
  public void clear() {
    // nothing to do
  }
​
  private static class FSTSerializer<E> implements CacheSerializer<E> {
​
    private static final FSTConfiguration FST = FSTConfiguration.createDefaultConfiguration();
​
    static {
      FST.registerClass(DatabaseValue.class);
      FST.registerClass(DatabaseKey.class);
      FST.registerClass(SafeString.class);
      FST.registerClass(SortedSet.class);
    }
​
    @Override
    public void serialize(E value, ByteBuffer buf) {
      byte[] array = FST.asByteArray(value);
      buf.putInt(array.length);
      buf.put(array);
    }
​
    @SuppressWarnings("unchecked")
    @Override
    public E deserialize(ByteBuffer buf) {
      int length = buf.getInt();
      byte[] array = new byte[length];
      buf.get(array);
      return (E) FST.asObject(array);
    }
​
    @Override
    public int serializedSize(E value) {
      return FST.asByteArray(value).length + Integer.BYTES;
    }
  }
}