聊聊claudb的DatabaseCleaner

public class DatabaseCleaner {
​
  private static final Logger LOGGER = LoggerFactory.getLogger(DatabaseCleaner.class);
​
  private final DBServerContext server;
  private final DBConfig config;
​
  private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
​
  public DatabaseCleaner(DBServerContext server, DBConfig config) {
    this.server = server;
    this.config = config;
  }
​
  public void start() {
    executor.scheduleWithFixedDelay(this::clean,
        config.getCleanPeriod(), config.getCleanPeriod(), TimeUnit.SECONDS);
  }
​
  public void stop() {
    executor.shutdown();
  }
​
  private void clean() {
    LOGGER.debug("cleaning database: running");
    server.clean(Instant.now());
    LOGGER.debug("cleaning database: done");
  }
}

public class ClauDB extends RespServerContext implements DBServerContext {
​
    //......
​
  public void clean(Instant now) {
    executeOn(Observable.create(observable -> {
      getState().evictExpired(now);
      observable.onComplete();
    })).blockingSubscribe();
  }
​
    //......
​
}

public class DBServerState {
​
    //......
​
  public void evictExpired(Instant now) {
    for (Database database : databases) {
      database.evictableKeys(now).forEach(database::remove);
    }
  }
​
    //......
​
}

public interface Database {
​
  //......
​
  default ImmutableSet<DatabaseKey> evictableKeys(Instant now) {
    return entrySet()
        .filter(entry -> entry.get2().isExpired(now))
        .map(Tuple2::get1);
  }
​
  //......
​
}