CavalliumDBEngine/src/main/java/it/cavallium/dbengine/database/remote/LLQuicConnection.java

500 lines
16 KiB
Java
Raw Normal View History

2022-03-02 12:34:30 +01:00
package it.cavallium.dbengine.database.remote;
2022-03-05 15:46:40 +01:00
import com.google.common.collect.Multimap;
2022-03-02 12:34:30 +01:00
import io.micrometer.core.instrument.MeterRegistry;
2022-03-20 14:33:27 +01:00
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.netty.incubator.codec.quic.QuicSslContextBuilder;
2022-03-16 13:47:56 +01:00
import io.netty5.buffer.api.Buffer;
import io.netty5.buffer.api.BufferAllocator;
import io.netty5.buffer.api.Send;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.client.MemoryStats;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.client.query.current.data.Query;
import it.cavallium.dbengine.client.query.current.data.QueryParams;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.database.LLDatabaseConnection;
2022-03-20 14:33:27 +01:00
import it.cavallium.dbengine.database.LLDelta;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.database.LLDictionary;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.database.LLIndexRequest;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.database.LLKeyValueDatabase;
import it.cavallium.dbengine.database.LLLuceneIndex;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.database.LLSearchResultShard;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.database.LLSingleton;
import it.cavallium.dbengine.database.LLSnapshot;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.database.LLTerm;
import it.cavallium.dbengine.database.LLUpdateDocument;
2022-04-09 02:45:42 +02:00
import it.cavallium.dbengine.database.TableWithProperties;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.database.UpdateMode;
import it.cavallium.dbengine.database.UpdateReturnMode;
import it.cavallium.dbengine.database.disk.BinarySerializationFunction;
2022-03-04 01:26:18 +01:00
import it.cavallium.dbengine.database.remote.RPCCodecs.RPCEventCodec;
import it.cavallium.dbengine.database.serialization.SerializationException;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.database.serialization.SerializationFunction;
import it.cavallium.dbengine.lucene.LuceneHacks;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.lucene.collector.Buckets;
import it.cavallium.dbengine.lucene.searcher.BucketParams;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.rpc.current.data.BinaryOptional;
import it.cavallium.dbengine.rpc.current.data.ClientBoundRequest;
import it.cavallium.dbengine.rpc.current.data.ClientBoundResponse;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.rpc.current.data.CloseDatabase;
import it.cavallium.dbengine.rpc.current.data.CloseLuceneIndex;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.rpc.current.data.Column;
import it.cavallium.dbengine.rpc.current.data.DatabaseOptions;
import it.cavallium.dbengine.rpc.current.data.GeneratedEntityId;
import it.cavallium.dbengine.rpc.current.data.GetDatabase;
2022-03-05 15:46:40 +01:00
import it.cavallium.dbengine.rpc.current.data.GetLuceneIndex;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.rpc.current.data.GetSingleton;
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;
2022-03-04 01:26:18 +01:00
import it.cavallium.dbengine.rpc.current.data.RPCEvent;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.rpc.current.data.ServerBoundRequest;
import it.cavallium.dbengine.rpc.current.data.ServerBoundResponse;
import it.cavallium.dbengine.rpc.current.data.SingletonGet;
import it.cavallium.dbengine.rpc.current.data.SingletonSet;
2022-03-04 01:26:18 +01:00
import it.cavallium.dbengine.rpc.current.data.SingletonUpdateEnd;
import it.cavallium.dbengine.rpc.current.data.SingletonUpdateInit;
import it.cavallium.dbengine.rpc.current.data.SingletonUpdateOldData;
2022-03-20 14:33:27 +01:00
import it.cavallium.dbengine.rpc.current.data.nullables.NullableBytes;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.rpc.current.data.nullables.NullableLLSnapshot;
import it.unimi.dsi.fastutil.bytes.ByteList;
import java.io.File;
import java.net.SocketAddress;
import java.time.Duration;
import java.util.List;
2022-03-05 15:46:40 +01:00
import java.util.Map.Entry;
2022-03-02 12:34:30 +01:00
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
2022-03-05 15:46:40 +01:00
import org.jetbrains.annotations.NotNull;
2022-03-02 12:34:30 +01:00
import org.jetbrains.annotations.Nullable;
2022-04-09 02:45:42 +02:00
import org.rocksdb.DBOptions;
import org.rocksdb.TableProperties;
2022-03-05 15:46:40 +01:00
import reactor.core.publisher.Flux;
2022-03-02 12:34:30 +01:00
import reactor.core.publisher.Mono;
import reactor.netty.incubator.quic.QuicClient;
import reactor.netty.incubator.quic.QuicConnection;
public class LLQuicConnection implements LLDatabaseConnection {
private final BufferAllocator allocator;
private final MeterRegistry meterRegistry;
private final SocketAddress bindAddress;
private final SocketAddress remoteAddress;
private volatile QuicConnection quicConnection;
private final ConcurrentHashMap<String, Mono<Long>> databases = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, Mono<Long>> indexes = new ConcurrentHashMap<>();
private Mono<Void> connectionMono = Mono.error(new IllegalStateException("Not connected"));
public LLQuicConnection(BufferAllocator allocator,
MeterRegistry meterRegistry,
SocketAddress bindAddress,
SocketAddress remoteAddress) {
this.allocator = allocator;
this.meterRegistry = meterRegistry;
this.bindAddress = bindAddress;
this.remoteAddress = remoteAddress;
}
@Override
public BufferAllocator getAllocator() {
return allocator;
}
@Override
public MeterRegistry getMeterRegistry() {
return meterRegistry;
}
@Override
public Mono<? extends LLDatabaseConnection> connect() {
String keyFileLocation = System.getProperty("it.cavalliumdb.keyFile", null);
String certFileLocation = System.getProperty("it.cavalliumdb.certFile", null);
String keyStorePassword = System.getProperty("it.cavalliumdb.keyPassword", null);
String certChainLocation = System.getProperty("it.cavalliumdb.caFile", null);
File keyFile;
File certFile;
File certChain;
if (keyFileLocation != null) {
keyFile = new File(keyFileLocation);
} else {
keyFile = null;
}
if (certFileLocation != null) {
certFile = new File(certFileLocation);
} else {
certFile = null;
}
if (certChainLocation != null) {
certChain = new File(certChainLocation);
} else {
certChain = null;
}
var sslContextBuilder = QuicSslContextBuilder.forClient();
if (keyFileLocation != null || certFileLocation != null) {
sslContextBuilder.keyManager(keyFile, keyStorePassword, certFile);
}
if (certChainLocation != null) {
sslContextBuilder.trustManager(certChain);
} else {
sslContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);
}
var sslContext = sslContextBuilder
.applicationProtocols("db/0.9")
.build();
return QuicClient.create()
.bindAddress(() -> bindAddress)
.remoteAddress(() -> remoteAddress)
.secure(sslContext)
.idleTimeout(Duration.ofSeconds(30))
.initialSettings(spec -> spec
.maxData(10000000)
.maxStreamDataBidirectionalLocal(1000000)
)
.connect()
.doOnNext(conn -> quicConnection = conn)
.thenReturn(this);
}
2022-03-04 01:26:18 +01:00
@SuppressWarnings("unchecked")
private <T extends ClientBoundResponse> Mono<T> sendRequest(ServerBoundRequest serverBoundRequest) {
return QuicUtils.<RPCEvent, RPCEvent>sendSimpleRequest(quicConnection,
RPCEventCodec::new,
RPCEventCodec::new,
serverBoundRequest
).map(event -> (T) event);
2022-03-02 12:34:30 +01:00
}
2022-03-05 15:46:40 +01:00
private Mono<Void> sendEvent(ServerBoundRequest serverBoundRequest) {
return QuicUtils.<RPCEvent>sendSimpleEvent(quicConnection,
RPCEventCodec::new,
serverBoundRequest
);
}
2022-03-04 01:26:18 +01:00
private <T extends ClientBoundResponse, U extends ClientBoundRequest> Mono<T> sendUpdateRequest(ServerBoundRequest serverBoundReq,
2022-03-02 12:34:30 +01:00
Function<U, ServerBoundResponse> updaterFunction) {
2022-03-04 01:26:18 +01:00
return Mono.empty();
/*
return Mono.defer(() -> {
Empty<Void> streamTerminator = Sinks.empty();
return QuicUtils.createStream(quicConnection, stream -> {
Mono<Void> serverReq = Mono.defer(() -> stream.out()
.withConnection(conn -> conn.addHandler(new RPCCodecs.RPCServerBoundRequestDecoder()))
.sendObject(serverBoundReq)
.then()).doOnSubscribe(s -> System.out.println("out1"));
//noinspection unchecked
Mono<U> clientBoundReqMono = Mono.defer(() -> stream.in()
.withConnection(conn -> conn.addHandler(new RPCClientBoundRequestDecoder()))
.receiveObject()
.log("TO_CLIENT_REQ", Level.INFO)
.take(1, true)
.singleOrEmpty()
.map(req -> (U) req)
.doOnSubscribe(s -> System.out.println("in1"))
.switchIfEmpty((Mono<U>) QuicUtils.NO_RESPONSE_ERROR)
);
Mono<Void> serverBoundRespFlux = clientBoundReqMono
.map(updaterFunction)
.transform(respMono -> Mono.defer(() -> stream.out()
.withConnection(conn -> conn.addHandler(new RPCServerBoundResponseDecoder()))
.sendObject(respMono)
.then()
.doOnSubscribe(s -> System.out.println("out2"))
));
//noinspection unchecked
Mono<T> clientBoundResponseMono = Mono.defer(() -> stream.in()
.withConnection(conn -> conn.addHandler(new RPCClientBoundResponseDecoder()))
.receiveObject()
.map(resp -> (T) resp)
.log("TO_SERVER_RESP", Level.INFO)
.take(1, true)
.doOnSubscribe(s -> System.out.println("out2"))
.singleOrEmpty()
.switchIfEmpty((Mono<T>) QuicUtils.NO_RESPONSE_ERROR));
return serverReq
.then(serverBoundRespFlux)
.then(clientBoundResponseMono)
.doFinally(s -> streamTerminator.tryEmitEmpty());
}, streamTerminator.asMono()).single();
});
*/
2022-03-02 12:34:30 +01:00
}
@Override
public Mono<? extends LLKeyValueDatabase> getDatabase(String databaseName,
List<Column> columns, DatabaseOptions databaseOptions) {
return sendRequest(new GetDatabase(databaseName, columns, databaseOptions))
.cast(GeneratedEntityId.class)
.map(GeneratedEntityId::id)
.map(id -> new LLKeyValueDatabase() {
@Override
public Mono<? extends LLSingleton> getSingleton(byte[] singletonListColumnName,
byte[] name,
2022-03-20 14:33:27 +01:00
byte @Nullable[] defaultValue) {
2022-03-02 12:34:30 +01:00
return sendRequest(new GetSingleton(id,
ByteList.of(singletonListColumnName),
ByteList.of(name),
2022-03-20 14:33:27 +01:00
defaultValue == null ? NullableBytes.empty() : NullableBytes.of(ByteList.of(defaultValue))
2022-03-02 12:34:30 +01:00
)).cast(GeneratedEntityId.class).map(GeneratedEntityId::id).map(singletonId -> new LLSingleton() {
@Override
public BufferAllocator getAllocator() {
return allocator;
}
@Override
2022-03-20 14:33:27 +01:00
public Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot) {
2022-03-02 12:34:30 +01:00
return sendRequest(new SingletonGet(singletonId, NullableLLSnapshot.ofNullable(snapshot)))
.cast(BinaryOptional.class)
2022-03-20 14:33:27 +01:00
.mapNotNull(result -> {
if (result.val().isPresent()) {
return allocator.copyOf(QuicUtils.toArrayNoCopy(result.val().get().val())).send();
} else {
return null;
}
});
2022-03-02 12:34:30 +01:00
}
@Override
2022-03-20 14:33:27 +01:00
public Mono<Void> set(Mono<Send<Buffer>> valueMono) {
return QuicUtils.toBytes(valueMono)
.flatMap(valueSendOpt -> sendRequest(new SingletonSet(singletonId, valueSendOpt)).then());
2022-03-02 12:34:30 +01:00
}
@Override
public Mono<Send<Buffer>> update(BinarySerializationFunction updater, UpdateReturnMode updateReturnMode) {
2022-03-04 01:26:18 +01:00
return LLQuicConnection.this.<BinaryOptional, SingletonUpdateOldData>sendUpdateRequest(new SingletonUpdateInit(singletonId, updateReturnMode), prev -> {
byte[] oldData = toArrayNoCopy(prev);
Buffer oldDataBuf;
2022-03-04 01:26:18 +01:00
if (oldData != null) {
oldDataBuf = allocator.copyOf(oldData);
2022-03-04 01:26:18 +01:00
} else {
oldDataBuf = null;
}
try (oldDataBuf) {
try (var result = updater.apply(oldDataBuf)) {
if (result == null) {
return new SingletonUpdateEnd(false, ByteList.of());
} else {
byte[] resultArray = new byte[result.readableBytes()];
result.readBytes(resultArray, 0, resultArray.length);
return new SingletonUpdateEnd(true, ByteList.of(resultArray));
}
}
} catch (SerializationException e) {
throw new IllegalStateException(e);
}
}).mapNotNull(result -> {
if (result.val().isPresent()) {
return allocator.copyOf(QuicUtils.toArrayNoCopy(result.val().get().val())).send();
} else {
return null;
}
});
2022-03-02 12:34:30 +01:00
}
2022-03-20 14:33:27 +01:00
@Override
public Mono<Send<LLDelta>> updateAndGetDelta(BinarySerializationFunction updater) {
2022-03-20 14:33:27 +01:00
return Mono.error(new UnsupportedOperationException());
}
2022-03-02 12:34:30 +01:00
@Override
public String getDatabaseName() {
return databaseName;
}
2022-03-20 14:33:27 +01:00
@Override
public String getColumnName() {
return new String(singletonListColumnName);
}
@Override
public String getName() {
return new String(name);
}
2022-03-02 12:34:30 +01:00
});
}
@Override
public Mono<? extends LLDictionary> getDictionary(byte[] columnName, UpdateMode updateMode) {
return null;
}
@Override
public Mono<Long> getProperty(String propertyName) {
return null;
}
@Override
public Mono<MemoryStats> getMemoryStats() {
return null;
}
2022-04-09 02:45:42 +02:00
@Override
public Mono<String> getRocksDBStats() {
return null;
}
@Override
public Flux<TableWithProperties> getTableProperties() {
return null;
}
2022-03-02 12:34:30 +01:00
@Override
public Mono<Void> verifyChecksum() {
return null;
}
@Override
public BufferAllocator getAllocator() {
2022-03-05 15:46:40 +01:00
return allocator;
2022-03-02 12:34:30 +01:00
}
@Override
public MeterRegistry getMeterRegistry() {
2022-03-05 15:46:40 +01:00
return meterRegistry;
2022-03-02 12:34:30 +01:00
}
@Override
public Mono<Void> close() {
2022-03-05 15:46:40 +01:00
return sendRequest(new CloseDatabase(id)).then();
2022-03-02 12:34:30 +01:00
}
@Override
public String getDatabaseName() {
2022-03-05 15:46:40 +01:00
return databaseName;
2022-03-02 12:34:30 +01:00
}
@Override
public Mono<LLSnapshot> takeSnapshot() {
return null;
}
@Override
public Mono<Void> releaseSnapshot(LLSnapshot snapshot) {
return null;
}
});
}
2022-03-04 01:26:18 +01:00
@Nullable
private static byte[] toArrayNoCopy(SingletonUpdateOldData oldData) {
if (oldData.exist()) {
return QuicUtils.toArrayNoCopy(oldData.oldValue());
2022-03-02 12:34:30 +01:00
} else {
2022-03-04 01:26:18 +01:00
return null;
2022-03-02 12:34:30 +01:00
}
}
@Override
2022-03-05 15:46:40 +01:00
public Mono<? extends LLLuceneIndex> getLuceneIndex(String clusterName,
LuceneIndexStructure indexStructure,
2022-03-02 12:34:30 +01:00
IndicizerAnalyzers indicizerAnalyzers,
IndicizerSimilarities indicizerSimilarities,
LuceneOptions luceneOptions,
@Nullable LuceneHacks luceneHacks) {
2022-03-05 15:46:40 +01:00
return sendRequest(new GetLuceneIndex(clusterName, indexStructure, indicizerAnalyzers, indicizerSimilarities, luceneOptions))
.cast(GeneratedEntityId.class)
.map(GeneratedEntityId::id)
.map(id -> new LLLuceneIndex() {
@Override
public String getLuceneIndexName() {
return clusterName;
}
@Override
public Mono<Void> addDocument(LLTerm id, LLUpdateDocument doc) {
return null;
}
@Override
2022-03-19 00:08:23 +01:00
public Mono<Long> addDocuments(boolean atomic, Flux<Entry<LLTerm, LLUpdateDocument>> documents) {
2022-03-05 15:46:40 +01:00
return null;
}
@Override
public Mono<Void> deleteDocument(LLTerm id) {
return null;
}
@Override
public Mono<Void> update(LLTerm id, LLIndexRequest request) {
return null;
}
@Override
2022-03-19 00:08:23 +01:00
public Mono<Long> updateDocuments(Flux<Entry<LLTerm, LLUpdateDocument>> documents) {
2022-03-05 15:46:40 +01:00
return null;
}
@Override
public Mono<Void> deleteAll() {
return null;
}
@Override
public Flux<LLSearchResultShard> moreLikeThis(@Nullable LLSnapshot snapshot,
QueryParams queryParams,
@Nullable String keyFieldName,
Multimap<String, String> mltDocumentFields) {
return null;
}
@Override
public Flux<LLSearchResultShard> search(@Nullable LLSnapshot snapshot,
QueryParams queryParams,
@Nullable String keyFieldName) {
return null;
}
@Override
public Mono<Buckets> computeBuckets(@Nullable LLSnapshot snapshot,
@NotNull List<Query> queries,
@Nullable Query normalizationQuery,
BucketParams bucketParams) {
return null;
}
@Override
public boolean isLowMemoryMode() {
return false;
}
@Override
public Mono<Void> close() {
return sendRequest(new CloseLuceneIndex(id)).then();
}
@Override
public Mono<Void> flush() {
return null;
}
@Override
public Mono<Void> refresh(boolean force) {
return null;
}
@Override
public Mono<LLSnapshot> takeSnapshot() {
return null;
}
@Override
public Mono<Void> releaseSnapshot(LLSnapshot snapshot) {
return null;
}
});
2022-03-02 12:34:30 +01:00
}
@Override
public Mono<Void> disconnect() {
2022-03-05 15:46:40 +01:00
return sendDisconnect().then(Mono.fromRunnable(() -> quicConnection.dispose())).then(quicConnection.onDispose());
2022-03-02 12:34:30 +01:00
}
private Mono<Void> sendDisconnect() {
return Mono.empty();
}
}