CavalliumDBEngine/src/main/java/it/cavallium/dbengine/database/disk/LLLocalDatabaseConnection.java

154 lines
4.5 KiB
Java
Raw Normal View History

2020-12-07 22:15:18 +01:00
package it.cavallium.dbengine.database.disk;
import io.micrometer.core.instrument.MeterRegistry;
2022-03-16 13:47:56 +01:00
import io.netty5.buffer.api.BufferAllocator;
import it.cavallium.dbengine.database.LLDatabaseConnection;
import it.cavallium.dbengine.database.LLLuceneIndex;
import it.cavallium.dbengine.lucene.LuceneHacks;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.lucene.LuceneRocksDBManager;
2021-05-03 18:07:18 +02:00
import it.cavallium.dbengine.netty.JMXNettyMonitoringManager;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.rpc.current.data.Column;
import it.cavallium.dbengine.rpc.current.data.DatabaseOptions;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.rpc.current.data.IndicizerAnalyzers;
import it.cavallium.dbengine.rpc.current.data.IndicizerSimilarities;
import it.cavallium.dbengine.rpc.current.data.LuceneIndexStructure;
import it.cavallium.dbengine.rpc.current.data.LuceneOptions;
2020-12-07 22:15:18 +01:00
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.jetbrains.annotations.Nullable;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
2020-12-07 22:15:18 +01:00
public class LLLocalDatabaseConnection implements LLDatabaseConnection {
2021-05-03 18:07:18 +02:00
static {
2021-05-03 21:41:51 +02:00
JMXNettyMonitoringManager.initialize();
2021-05-03 18:07:18 +02:00
}
private final AtomicBoolean connected = new AtomicBoolean();
2021-08-29 23:18:03 +02:00
private final BufferAllocator allocator;
private final MeterRegistry meterRegistry;
2020-12-07 22:15:18 +01:00
private final Path basePath;
2022-03-02 12:34:30 +01:00
private final boolean inMemory;
2022-03-05 15:46:40 +01:00
private final LuceneRocksDBManager rocksDBManager;
private final AtomicReference<LLTempLMDBEnv> env = new AtomicReference<>();
2020-12-07 22:15:18 +01:00
2022-03-02 12:34:30 +01:00
public LLLocalDatabaseConnection(BufferAllocator allocator,
MeterRegistry meterRegistry,
Path basePath,
2022-03-05 15:46:40 +01:00
boolean inMemory,
LuceneRocksDBManager rocksDBManager) {
2021-05-03 21:41:51 +02:00
this.allocator = allocator;
this.meterRegistry = meterRegistry;
2020-12-07 22:15:18 +01:00
this.basePath = basePath;
2022-03-02 12:34:30 +01:00
this.inMemory = inMemory;
2022-03-05 15:46:40 +01:00
this.rocksDBManager = rocksDBManager;
2020-12-07 22:15:18 +01:00
}
2021-05-03 21:41:51 +02:00
@Override
2021-08-29 23:18:03 +02:00
public BufferAllocator getAllocator() {
2021-05-03 21:41:51 +02:00
return allocator;
}
public MeterRegistry getMeterRegistry() {
return meterRegistry;
}
2020-12-07 22:15:18 +01:00
@Override
public Mono<LLDatabaseConnection> connect() {
return Mono
.<LLDatabaseConnection>fromCallable(() -> {
if (!connected.compareAndSet(false, true)) {
throw new IllegalStateException("Already connected");
}
if (Files.notExists(basePath)) {
Files.createDirectories(basePath);
}
var prev = env.getAndSet(new LLTempLMDBEnv());
if (prev != null) {
throw new IllegalStateException("Env was already set");
}
return this;
})
.subscribeOn(Schedulers.boundedElastic());
2020-12-07 22:15:18 +01:00
}
@Override
public Mono<LLLocalKeyValueDatabase> getDatabase(String name,
List<Column> columns,
2021-06-27 15:40:56 +02:00
DatabaseOptions databaseOptions) {
return Mono
2021-05-03 21:41:51 +02:00
.fromCallable(() -> new LLLocalKeyValueDatabase(
allocator,
meterRegistry,
2021-05-03 21:41:51 +02:00
name,
2022-03-02 12:34:30 +01:00
inMemory,
basePath.resolve("database_" + name),
columns,
new LinkedList<>(),
2021-06-27 15:40:56 +02:00
databaseOptions
))
.subscribeOn(Schedulers.boundedElastic());
2020-12-07 22:15:18 +01:00
}
@Override
2022-03-05 15:46:40 +01:00
public Mono<? extends LLLuceneIndex> getLuceneIndex(String clusterName,
LuceneIndexStructure indexStructure,
2021-05-28 16:04:59 +02:00
IndicizerAnalyzers indicizerAnalyzers,
IndicizerSimilarities indicizerSimilarities,
LuceneOptions luceneOptions,
@Nullable LuceneHacks luceneHacks) {
return Mono
.fromCallable(() -> {
2021-11-21 12:31:23 +01:00
var env = this.env.get();
2022-03-05 15:46:40 +01:00
if (clusterName == null) {
throw new IllegalArgumentException("Cluster name must be set");
2021-12-30 17:28:06 +01:00
}
2022-03-05 15:46:40 +01:00
if (indexStructure.activeShards().size() != 1) {
Objects.requireNonNull(env, "Environment not set");
return new LLLocalMultiLuceneIndex(env,
meterRegistry,
2021-12-30 17:28:06 +01:00
clusterName,
2022-03-05 15:46:40 +01:00
indexStructure.activeShards(),
indexStructure.totalShards(),
2021-05-28 16:04:59 +02:00
indicizerAnalyzers,
indicizerSimilarities,
luceneOptions,
2022-03-05 15:46:40 +01:00
luceneHacks,
rocksDBManager
);
} else {
2022-02-26 22:51:22 +01:00
return new LLLocalLuceneIndex(env,
meterRegistry,
2021-12-30 17:28:06 +01:00
clusterName,
2022-03-05 15:46:40 +01:00
indexStructure.activeShards().getInt(0),
2021-05-28 16:04:59 +02:00
indicizerAnalyzers,
indicizerSimilarities,
luceneOptions,
2022-03-05 15:46:40 +01:00
luceneHacks,
rocksDBManager
);
}
})
.subscribeOn(Schedulers.boundedElastic());
2020-12-07 22:15:18 +01:00
}
@Override
public Mono<Void> disconnect() {
return Mono.<Void>fromCallable(() -> {
if (connected.compareAndSet(true, false)) {
var env = this.env.get();
if (env != null) {
2021-10-16 14:59:38 +02:00
env.close();
}
}
return null;
}).subscribeOn(Schedulers.boundedElastic());
2020-12-07 22:15:18 +01:00
}
}