Compare commits
2 Commits
2abbc2984f
...
aeadadd7b4
| Author | SHA1 | Date | |
|---|---|---|---|
|
aeadadd7b4 | ||
|
|
9ab08c54a3 |
48
pom.xml
48
pom.xml
@ -17,6 +17,9 @@
|
||||
<slf4j.version>2.0.12</slf4j.version>
|
||||
<imageName>rockserver-core</imageName>
|
||||
<mainClass>it.cavallium.rockserver.core.Main</mainClass>
|
||||
<protobuf-plugin.version>0.6.1</protobuf-plugin.version>
|
||||
<protobuf.version>3.25.3</protobuf.version>
|
||||
<grpc.version>1.65.0</grpc.version>
|
||||
</properties>
|
||||
|
||||
<repositories>
|
||||
@ -80,6 +83,21 @@
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
<version>${slf4j.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.grpc</groupId>
|
||||
<artifactId>grpc-netty</artifactId>
|
||||
<version>${grpc.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.grpc</groupId>
|
||||
<artifactId>grpc-protobuf</artifactId>
|
||||
<version>${grpc.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.grpc</groupId>
|
||||
<artifactId>grpc-stub</artifactId>
|
||||
<version>${grpc.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.lz4</groupId>
|
||||
@ -106,6 +124,13 @@
|
||||
<filtering>false</filtering>
|
||||
</resource>
|
||||
</resources>
|
||||
<extensions>
|
||||
<extension>
|
||||
<groupId>kr.motd.maven</groupId>
|
||||
<artifactId>os-maven-plugin</artifactId>
|
||||
<version>1.6.1</version>
|
||||
</extension>
|
||||
</extensions>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
@ -171,6 +196,29 @@
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.xolstice.maven.plugins</groupId>
|
||||
<artifactId>protobuf-maven-plugin</artifactId>
|
||||
<version>${protobuf-plugin.version}</version>
|
||||
<configuration>
|
||||
<protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
|
||||
<pluginId>grpc-java</pluginId>
|
||||
<pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
|
||||
</configuration>
|
||||
<executions>
|
||||
<execution>
|
||||
<configuration>
|
||||
<pluginParameter>
|
||||
@generated=omit
|
||||
</pluginParameter>
|
||||
</configuration>
|
||||
<goals>
|
||||
<goal>compile</goal>
|
||||
<goal>compile-custom</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
|
||||
@ -5,6 +5,7 @@ import static java.util.Objects.requireNonNull;
|
||||
|
||||
import it.cavallium.rockserver.core.common.Utils;
|
||||
import it.cavallium.rockserver.core.impl.rocksdb.RocksDBLoader;
|
||||
import it.cavallium.rockserver.core.server.ServerBuilder;
|
||||
import java.io.IOException;
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
@ -30,10 +31,14 @@ public class Main {
|
||||
.type(String.class)
|
||||
.setDefault(PRIVATE_MEMORY_URL.toString())
|
||||
.help("Specify database rocksdb://hostname:port, or unix://<path>, or file://<path>");
|
||||
parser.addArgument("-l", "--listen-url")
|
||||
parser.addArgument("-l", "--thrift-listen-url")
|
||||
.type(String.class)
|
||||
.setDefault("http://127.0.0.1:5332")
|
||||
.help("Specify database http://hostname:port, or unix://<path>, or file://<path>");
|
||||
parser.addArgument("-L", "--grpc-listen-url")
|
||||
.type(String.class)
|
||||
.setDefault("http://127.0.0.1:5333")
|
||||
.help("Specify database http://hostname:port, or unix://<path>, or file://<path>");
|
||||
parser.addArgument("-n", "--name")
|
||||
.type(String.class)
|
||||
.setDefault("main")
|
||||
@ -53,7 +58,6 @@ public class Main {
|
||||
System.exit(1);
|
||||
}
|
||||
var clientBuilder = new it.cavallium.rockserver.core.client.ClientBuilder();
|
||||
var serverBuilder = new it.cavallium.rockserver.core.server.ServerBuilder();
|
||||
|
||||
if (ns.getBoolean("print_default_config")) {
|
||||
requireNonNull(Main.class.getClassLoader()
|
||||
@ -67,12 +71,14 @@ public class Main {
|
||||
RocksDBLoader.loadLibrary();
|
||||
|
||||
var rawDatabaseUrl = ns.getString("database_url");
|
||||
var rawListenUrl = ns.getString("listen_url");
|
||||
var rawThriftListenUrl = ns.getString("thrift_listen_url");
|
||||
var rawGrpcListenUrl = ns.getString("grpc_listen_url");
|
||||
var name = ns.getString("name");
|
||||
var config = ns.getString("config");
|
||||
|
||||
var databaseUrl = new URI(rawDatabaseUrl);
|
||||
var listenUrl = new URI(rawListenUrl);
|
||||
var thriftListenUrl = new URI(rawThriftListenUrl);
|
||||
var grpcListenUrl = new URI(rawGrpcListenUrl);
|
||||
|
||||
if (config != null) {
|
||||
if (!databaseUrl.getScheme().equals("file")) {
|
||||
@ -93,24 +99,23 @@ public class Main {
|
||||
case null, default -> throw new IllegalArgumentException("Invalid scheme \"" + databaseUrlScheme + "\" for database url url: " + databaseUrl);
|
||||
}
|
||||
|
||||
var listenUrlScheme = listenUrl.getScheme();
|
||||
switch (listenUrlScheme) {
|
||||
case "unix" -> serverBuilder.setUnixSocket(UnixDomainSocketAddress.of(Path.of(listenUrl.getPath())));
|
||||
case "http" -> serverBuilder.setHttpAddress(listenUrl.getHost(), Utils.parsePort(listenUrl));
|
||||
case "rocksdb" -> serverBuilder.setAddress(Utils.parseHostAndPort(listenUrl));
|
||||
case null, default -> throw new IllegalArgumentException("Invalid scheme \"" + listenUrlScheme + "\" for listen url: " + listenUrl);
|
||||
}
|
||||
var thriftServerBuilder = new it.cavallium.rockserver.core.server.ServerBuilder();
|
||||
buildServerAddress(thriftServerBuilder, thriftListenUrl, true);
|
||||
var grpcServerBuilder = new it.cavallium.rockserver.core.server.ServerBuilder();
|
||||
buildServerAddress(grpcServerBuilder, grpcListenUrl, false);
|
||||
|
||||
clientBuilder.setName(name);
|
||||
try (var connection = clientBuilder.build()) {
|
||||
LOG.log(Level.INFO, "Connected to {0}", connection);
|
||||
|
||||
serverBuilder.setClient(connection);
|
||||
thriftServerBuilder.setClient(connection);
|
||||
grpcServerBuilder.setClient(connection);
|
||||
|
||||
CountDownLatch shutdownLatch = new CountDownLatch(1);
|
||||
Runtime.getRuntime().addShutdownHook(new Thread(shutdownLatch::countDown));
|
||||
|
||||
try (var server = serverBuilder.build()) {
|
||||
try (var _ = thriftServerBuilder.build();
|
||||
var _ = grpcServerBuilder.build()) {
|
||||
shutdownLatch.await();
|
||||
LOG.info("Shutting down...");
|
||||
}
|
||||
@ -119,4 +124,17 @@ public class Main {
|
||||
}
|
||||
LOG.info("Shut down successfully");
|
||||
}
|
||||
|
||||
private static void buildServerAddress(ServerBuilder serverBuilder, URI listenUrl, boolean useThrift) {
|
||||
var thriftListenUrlScheme = listenUrl.getScheme();
|
||||
switch (thriftListenUrlScheme) {
|
||||
case "unix" -> serverBuilder.setUnixSocket(UnixDomainSocketAddress.of(Path.of(listenUrl.getPath())));
|
||||
case "http" -> {
|
||||
serverBuilder.setHttpAddress(listenUrl.getHost(), Utils.parsePort(listenUrl));
|
||||
serverBuilder.setUseThrift(useThrift);
|
||||
}
|
||||
case "rocksdb" -> serverBuilder.setAddress(Utils.parseHostAndPort(listenUrl));
|
||||
case null, default -> throw new IllegalArgumentException("Invalid scheme \"" + thriftListenUrlScheme + "\" for listen url: " + listenUrl);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Autogenerated by Thrift Compiler (0.19.0)
|
||||
* Autogenerated by Thrift Compiler (0.20.0)
|
||||
*
|
||||
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
|
||||
* @generated
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Autogenerated by Thrift Compiler (0.19.0)
|
||||
* Autogenerated by Thrift Compiler (0.20.0)
|
||||
*
|
||||
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
|
||||
* @generated
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Autogenerated by Thrift Compiler (0.19.0)
|
||||
* Autogenerated by Thrift Compiler (0.20.0)
|
||||
*
|
||||
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
|
||||
* @generated
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Autogenerated by Thrift Compiler (0.19.0)
|
||||
* Autogenerated by Thrift Compiler (0.20.0)
|
||||
*
|
||||
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
|
||||
* @generated
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Autogenerated by Thrift Compiler (0.19.0)
|
||||
* Autogenerated by Thrift Compiler (0.20.0)
|
||||
*
|
||||
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
|
||||
* @generated
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Autogenerated by Thrift Compiler (0.19.0)
|
||||
* Autogenerated by Thrift Compiler (0.20.0)
|
||||
*
|
||||
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
|
||||
* @generated
|
||||
@ -13,7 +13,7 @@ public class RocksDB {
|
||||
|
||||
public long openTransaction(long timeoutMs) throws org.apache.thrift.TException;
|
||||
|
||||
public boolean closeTransaction(long timeoutMs, boolean commit) throws org.apache.thrift.TException;
|
||||
public boolean closeTransaction(long transactionId, boolean commit) throws org.apache.thrift.TException;
|
||||
|
||||
public void closeFailedUpdate(long updateId) throws org.apache.thrift.TException;
|
||||
|
||||
@ -61,7 +61,7 @@ public class RocksDB {
|
||||
|
||||
public void openTransaction(long timeoutMs, org.apache.thrift.async.AsyncMethodCallback<java.lang.Long> resultHandler) throws org.apache.thrift.TException;
|
||||
|
||||
public void closeTransaction(long timeoutMs, boolean commit, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler) throws org.apache.thrift.TException;
|
||||
public void closeTransaction(long transactionId, boolean commit, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler) throws org.apache.thrift.TException;
|
||||
|
||||
public void closeFailedUpdate(long updateId, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
|
||||
|
||||
@ -152,16 +152,16 @@ public class RocksDB {
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean closeTransaction(long timeoutMs, boolean commit) throws org.apache.thrift.TException
|
||||
public boolean closeTransaction(long transactionId, boolean commit) throws org.apache.thrift.TException
|
||||
{
|
||||
send_closeTransaction(timeoutMs, commit);
|
||||
send_closeTransaction(transactionId, commit);
|
||||
return recv_closeTransaction();
|
||||
}
|
||||
|
||||
public void send_closeTransaction(long timeoutMs, boolean commit) throws org.apache.thrift.TException
|
||||
public void send_closeTransaction(long transactionId, boolean commit) throws org.apache.thrift.TException
|
||||
{
|
||||
closeTransaction_args args = new closeTransaction_args();
|
||||
args.setTimeoutMs(timeoutMs);
|
||||
args.setTransactionId(transactionId);
|
||||
args.setCommit(commit);
|
||||
sendBase("closeTransaction", args);
|
||||
}
|
||||
@ -719,19 +719,19 @@ public class RocksDB {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void closeTransaction(long timeoutMs, boolean commit, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler) throws org.apache.thrift.TException {
|
||||
public void closeTransaction(long transactionId, boolean commit, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler) throws org.apache.thrift.TException {
|
||||
checkReady();
|
||||
closeTransaction_call method_call = new closeTransaction_call(timeoutMs, commit, resultHandler, this, ___protocolFactory, ___transport);
|
||||
closeTransaction_call method_call = new closeTransaction_call(transactionId, commit, resultHandler, this, ___protocolFactory, ___transport);
|
||||
this.___currentMethod = method_call;
|
||||
___manager.call(method_call);
|
||||
}
|
||||
|
||||
public static class closeTransaction_call extends org.apache.thrift.async.TAsyncMethodCall<java.lang.Boolean> {
|
||||
private long timeoutMs;
|
||||
private long transactionId;
|
||||
private boolean commit;
|
||||
public closeTransaction_call(long timeoutMs, boolean commit, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
|
||||
public closeTransaction_call(long transactionId, boolean commit, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
|
||||
super(client, protocolFactory, transport, resultHandler, false);
|
||||
this.timeoutMs = timeoutMs;
|
||||
this.transactionId = transactionId;
|
||||
this.commit = commit;
|
||||
}
|
||||
|
||||
@ -739,7 +739,7 @@ public class RocksDB {
|
||||
public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
|
||||
prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("closeTransaction", org.apache.thrift.protocol.TMessageType.CALL, 0));
|
||||
closeTransaction_args args = new closeTransaction_args();
|
||||
args.setTimeoutMs(timeoutMs);
|
||||
args.setTransactionId(transactionId);
|
||||
args.setCommit(commit);
|
||||
args.write(prot);
|
||||
prot.writeMessageEnd();
|
||||
@ -1673,7 +1673,7 @@ public class RocksDB {
|
||||
@Override
|
||||
public closeTransaction_result getResult(I iface, closeTransaction_args args) throws org.apache.thrift.TException {
|
||||
closeTransaction_result result = new closeTransaction_result();
|
||||
result.success = iface.closeTransaction(args.timeoutMs, args.commit);
|
||||
result.success = iface.closeTransaction(args.transactionId, args.commit);
|
||||
result.setSuccessIsSet(true);
|
||||
return result;
|
||||
}
|
||||
@ -2415,7 +2415,7 @@ public class RocksDB {
|
||||
|
||||
@Override
|
||||
public void start(I iface, closeTransaction_args args, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler) throws org.apache.thrift.TException {
|
||||
iface.closeTransaction(args.timeoutMs, args.commit,resultHandler);
|
||||
iface.closeTransaction(args.transactionId, args.commit,resultHandler);
|
||||
}
|
||||
}
|
||||
|
||||
@ -4481,18 +4481,18 @@ public class RocksDB {
|
||||
public static class closeTransaction_args implements org.apache.thrift.TBase<closeTransaction_args, closeTransaction_args._Fields>, java.io.Serializable, Cloneable, Comparable<closeTransaction_args> {
|
||||
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("closeTransaction_args");
|
||||
|
||||
private static final org.apache.thrift.protocol.TField TIMEOUT_MS_FIELD_DESC = new org.apache.thrift.protocol.TField("timeoutMs", org.apache.thrift.protocol.TType.I64, (short)1);
|
||||
private static final org.apache.thrift.protocol.TField TRANSACTION_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("transactionId", org.apache.thrift.protocol.TType.I64, (short)1);
|
||||
private static final org.apache.thrift.protocol.TField COMMIT_FIELD_DESC = new org.apache.thrift.protocol.TField("commit", org.apache.thrift.protocol.TType.BOOL, (short)2);
|
||||
|
||||
private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new closeTransaction_argsStandardSchemeFactory();
|
||||
private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new closeTransaction_argsTupleSchemeFactory();
|
||||
|
||||
public long timeoutMs; // required
|
||||
public long transactionId; // required
|
||||
public boolean commit; // required
|
||||
|
||||
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
|
||||
public enum _Fields implements org.apache.thrift.TFieldIdEnum {
|
||||
TIMEOUT_MS((short)1, "timeoutMs"),
|
||||
TRANSACTION_ID((short)1, "transactionId"),
|
||||
COMMIT((short)2, "commit");
|
||||
|
||||
private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
|
||||
@ -4509,8 +4509,8 @@ public class RocksDB {
|
||||
@org.apache.thrift.annotation.Nullable
|
||||
public static _Fields findByThriftId(int fieldId) {
|
||||
switch(fieldId) {
|
||||
case 1: // TIMEOUT_MS
|
||||
return TIMEOUT_MS;
|
||||
case 1: // TRANSACTION_ID
|
||||
return TRANSACTION_ID;
|
||||
case 2: // COMMIT
|
||||
return COMMIT;
|
||||
default:
|
||||
@ -4556,13 +4556,13 @@ public class RocksDB {
|
||||
}
|
||||
|
||||
// isset id assignments
|
||||
private static final int __TIMEOUTMS_ISSET_ID = 0;
|
||||
private static final int __TRANSACTIONID_ISSET_ID = 0;
|
||||
private static final int __COMMIT_ISSET_ID = 1;
|
||||
private byte __isset_bitfield = 0;
|
||||
public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
|
||||
static {
|
||||
java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
|
||||
tmpMap.put(_Fields.TIMEOUT_MS, new org.apache.thrift.meta_data.FieldMetaData("timeoutMs", org.apache.thrift.TFieldRequirementType.REQUIRED,
|
||||
tmpMap.put(_Fields.TRANSACTION_ID, new org.apache.thrift.meta_data.FieldMetaData("transactionId", org.apache.thrift.TFieldRequirementType.REQUIRED,
|
||||
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I64)));
|
||||
tmpMap.put(_Fields.COMMIT, new org.apache.thrift.meta_data.FieldMetaData("commit", org.apache.thrift.TFieldRequirementType.REQUIRED,
|
||||
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL)));
|
||||
@ -4574,12 +4574,12 @@ public class RocksDB {
|
||||
}
|
||||
|
||||
public closeTransaction_args(
|
||||
long timeoutMs,
|
||||
long transactionId,
|
||||
boolean commit)
|
||||
{
|
||||
this();
|
||||
this.timeoutMs = timeoutMs;
|
||||
setTimeoutMsIsSet(true);
|
||||
this.transactionId = transactionId;
|
||||
setTransactionIdIsSet(true);
|
||||
this.commit = commit;
|
||||
setCommitIsSet(true);
|
||||
}
|
||||
@ -4589,7 +4589,7 @@ public class RocksDB {
|
||||
*/
|
||||
public closeTransaction_args(closeTransaction_args other) {
|
||||
__isset_bitfield = other.__isset_bitfield;
|
||||
this.timeoutMs = other.timeoutMs;
|
||||
this.transactionId = other.transactionId;
|
||||
this.commit = other.commit;
|
||||
}
|
||||
|
||||
@ -4600,33 +4600,33 @@ public class RocksDB {
|
||||
|
||||
@Override
|
||||
public void clear() {
|
||||
setTimeoutMsIsSet(false);
|
||||
this.timeoutMs = 0;
|
||||
setTransactionIdIsSet(false);
|
||||
this.transactionId = 0;
|
||||
setCommitIsSet(false);
|
||||
this.commit = false;
|
||||
}
|
||||
|
||||
public long getTimeoutMs() {
|
||||
return this.timeoutMs;
|
||||
public long getTransactionId() {
|
||||
return this.transactionId;
|
||||
}
|
||||
|
||||
public closeTransaction_args setTimeoutMs(long timeoutMs) {
|
||||
this.timeoutMs = timeoutMs;
|
||||
setTimeoutMsIsSet(true);
|
||||
public closeTransaction_args setTransactionId(long transactionId) {
|
||||
this.transactionId = transactionId;
|
||||
setTransactionIdIsSet(true);
|
||||
return this;
|
||||
}
|
||||
|
||||
public void unsetTimeoutMs() {
|
||||
__isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID);
|
||||
public void unsetTransactionId() {
|
||||
__isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __TRANSACTIONID_ISSET_ID);
|
||||
}
|
||||
|
||||
/** Returns true if field timeoutMs is set (has been assigned a value) and false otherwise */
|
||||
public boolean isSetTimeoutMs() {
|
||||
return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID);
|
||||
/** Returns true if field transactionId is set (has been assigned a value) and false otherwise */
|
||||
public boolean isSetTransactionId() {
|
||||
return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __TRANSACTIONID_ISSET_ID);
|
||||
}
|
||||
|
||||
public void setTimeoutMsIsSet(boolean value) {
|
||||
__isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID, value);
|
||||
public void setTransactionIdIsSet(boolean value) {
|
||||
__isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __TRANSACTIONID_ISSET_ID, value);
|
||||
}
|
||||
|
||||
public boolean isCommit() {
|
||||
@ -4655,11 +4655,11 @@ public class RocksDB {
|
||||
@Override
|
||||
public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
|
||||
switch (field) {
|
||||
case TIMEOUT_MS:
|
||||
case TRANSACTION_ID:
|
||||
if (value == null) {
|
||||
unsetTimeoutMs();
|
||||
unsetTransactionId();
|
||||
} else {
|
||||
setTimeoutMs((java.lang.Long)value);
|
||||
setTransactionId((java.lang.Long)value);
|
||||
}
|
||||
break;
|
||||
|
||||
@ -4678,8 +4678,8 @@ public class RocksDB {
|
||||
@Override
|
||||
public java.lang.Object getFieldValue(_Fields field) {
|
||||
switch (field) {
|
||||
case TIMEOUT_MS:
|
||||
return getTimeoutMs();
|
||||
case TRANSACTION_ID:
|
||||
return getTransactionId();
|
||||
|
||||
case COMMIT:
|
||||
return isCommit();
|
||||
@ -4696,8 +4696,8 @@ public class RocksDB {
|
||||
}
|
||||
|
||||
switch (field) {
|
||||
case TIMEOUT_MS:
|
||||
return isSetTimeoutMs();
|
||||
case TRANSACTION_ID:
|
||||
return isSetTransactionId();
|
||||
case COMMIT:
|
||||
return isSetCommit();
|
||||
}
|
||||
@ -4717,12 +4717,12 @@ public class RocksDB {
|
||||
if (this == that)
|
||||
return true;
|
||||
|
||||
boolean this_present_timeoutMs = true;
|
||||
boolean that_present_timeoutMs = true;
|
||||
if (this_present_timeoutMs || that_present_timeoutMs) {
|
||||
if (!(this_present_timeoutMs && that_present_timeoutMs))
|
||||
boolean this_present_transactionId = true;
|
||||
boolean that_present_transactionId = true;
|
||||
if (this_present_transactionId || that_present_transactionId) {
|
||||
if (!(this_present_transactionId && that_present_transactionId))
|
||||
return false;
|
||||
if (this.timeoutMs != that.timeoutMs)
|
||||
if (this.transactionId != that.transactionId)
|
||||
return false;
|
||||
}
|
||||
|
||||
@ -4742,7 +4742,7 @@ public class RocksDB {
|
||||
public int hashCode() {
|
||||
int hashCode = 1;
|
||||
|
||||
hashCode = hashCode * 8191 + org.apache.thrift.TBaseHelper.hashCode(timeoutMs);
|
||||
hashCode = hashCode * 8191 + org.apache.thrift.TBaseHelper.hashCode(transactionId);
|
||||
|
||||
hashCode = hashCode * 8191 + ((commit) ? 131071 : 524287);
|
||||
|
||||
@ -4757,12 +4757,12 @@ public class RocksDB {
|
||||
|
||||
int lastComparison = 0;
|
||||
|
||||
lastComparison = java.lang.Boolean.compare(isSetTimeoutMs(), other.isSetTimeoutMs());
|
||||
lastComparison = java.lang.Boolean.compare(isSetTransactionId(), other.isSetTransactionId());
|
||||
if (lastComparison != 0) {
|
||||
return lastComparison;
|
||||
}
|
||||
if (isSetTimeoutMs()) {
|
||||
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.timeoutMs, other.timeoutMs);
|
||||
if (isSetTransactionId()) {
|
||||
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.transactionId, other.transactionId);
|
||||
if (lastComparison != 0) {
|
||||
return lastComparison;
|
||||
}
|
||||
@ -4801,8 +4801,8 @@ public class RocksDB {
|
||||
java.lang.StringBuilder sb = new java.lang.StringBuilder("closeTransaction_args(");
|
||||
boolean first = true;
|
||||
|
||||
sb.append("timeoutMs:");
|
||||
sb.append(this.timeoutMs);
|
||||
sb.append("transactionId:");
|
||||
sb.append(this.transactionId);
|
||||
first = false;
|
||||
if (!first) sb.append(", ");
|
||||
sb.append("commit:");
|
||||
@ -4814,7 +4814,7 @@ public class RocksDB {
|
||||
|
||||
public void validate() throws org.apache.thrift.TException {
|
||||
// check for required fields
|
||||
// alas, we cannot check 'timeoutMs' because it's a primitive and you chose the non-beans generator.
|
||||
// alas, we cannot check 'transactionId' because it's a primitive and you chose the non-beans generator.
|
||||
// alas, we cannot check 'commit' because it's a primitive and you chose the non-beans generator.
|
||||
// check for sub-struct validity
|
||||
}
|
||||
@ -4857,10 +4857,10 @@ public class RocksDB {
|
||||
break;
|
||||
}
|
||||
switch (schemeField.id) {
|
||||
case 1: // TIMEOUT_MS
|
||||
case 1: // TRANSACTION_ID
|
||||
if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
|
||||
struct.timeoutMs = iprot.readI64();
|
||||
struct.setTimeoutMsIsSet(true);
|
||||
struct.transactionId = iprot.readI64();
|
||||
struct.setTransactionIdIsSet(true);
|
||||
} else {
|
||||
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
|
||||
}
|
||||
@ -4881,8 +4881,8 @@ public class RocksDB {
|
||||
iprot.readStructEnd();
|
||||
|
||||
// check for required fields of primitive type, which can't be checked in the validate method
|
||||
if (!struct.isSetTimeoutMs()) {
|
||||
throw new org.apache.thrift.protocol.TProtocolException("Required field 'timeoutMs' was not found in serialized data! Struct: " + toString());
|
||||
if (!struct.isSetTransactionId()) {
|
||||
throw new org.apache.thrift.protocol.TProtocolException("Required field 'transactionId' was not found in serialized data! Struct: " + toString());
|
||||
}
|
||||
if (!struct.isSetCommit()) {
|
||||
throw new org.apache.thrift.protocol.TProtocolException("Required field 'commit' was not found in serialized data! Struct: " + toString());
|
||||
@ -4895,8 +4895,8 @@ public class RocksDB {
|
||||
struct.validate();
|
||||
|
||||
oprot.writeStructBegin(STRUCT_DESC);
|
||||
oprot.writeFieldBegin(TIMEOUT_MS_FIELD_DESC);
|
||||
oprot.writeI64(struct.timeoutMs);
|
||||
oprot.writeFieldBegin(TRANSACTION_ID_FIELD_DESC);
|
||||
oprot.writeI64(struct.transactionId);
|
||||
oprot.writeFieldEnd();
|
||||
oprot.writeFieldBegin(COMMIT_FIELD_DESC);
|
||||
oprot.writeBool(struct.commit);
|
||||
@ -4919,15 +4919,15 @@ public class RocksDB {
|
||||
@Override
|
||||
public void write(org.apache.thrift.protocol.TProtocol prot, closeTransaction_args struct) throws org.apache.thrift.TException {
|
||||
org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
|
||||
oprot.writeI64(struct.timeoutMs);
|
||||
oprot.writeI64(struct.transactionId);
|
||||
oprot.writeBool(struct.commit);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void read(org.apache.thrift.protocol.TProtocol prot, closeTransaction_args struct) throws org.apache.thrift.TException {
|
||||
org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
|
||||
struct.timeoutMs = iprot.readI64();
|
||||
struct.setTimeoutMsIsSet(true);
|
||||
struct.transactionId = iprot.readI64();
|
||||
struct.setTransactionIdIsSet(true);
|
||||
struct.commit = iprot.readBool();
|
||||
struct.setCommitIsSet(true);
|
||||
}
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Autogenerated by Thrift Compiler (0.19.0)
|
||||
* Autogenerated by Thrift Compiler (0.20.0)
|
||||
*
|
||||
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
|
||||
* @generated
|
||||
|
||||
@ -0,0 +1,532 @@
|
||||
package it.cavallium.rockserver.core.server;
|
||||
|
||||
import com.google.protobuf.ByteString;
|
||||
import com.google.protobuf.Empty;
|
||||
import io.grpc.netty.NettyServerBuilder;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import it.cavallium.rockserver.core.client.RocksDBConnection;
|
||||
import it.cavallium.rockserver.core.common.ColumnHashType;
|
||||
import it.cavallium.rockserver.core.common.ColumnSchema;
|
||||
import it.cavallium.rockserver.core.common.Keys;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestChanged;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestCurrent;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestDelta;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestExists;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestForUpdate;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestMulti;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestNothing;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestPrevious;
|
||||
import it.cavallium.rockserver.core.common.RequestType.RequestPreviousPresence;
|
||||
import it.cavallium.rockserver.core.common.api.proto.Changed;
|
||||
import it.cavallium.rockserver.core.common.api.proto.CloseFailedUpdateRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.CloseIteratorRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.CloseTransactionRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.CloseTransactionResponse;
|
||||
import it.cavallium.rockserver.core.common.api.proto.CreateColumnRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.CreateColumnResponse;
|
||||
import it.cavallium.rockserver.core.common.api.proto.DeleteColumnRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.Delta;
|
||||
import it.cavallium.rockserver.core.common.api.proto.GetColumnIdRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.GetColumnIdResponse;
|
||||
import it.cavallium.rockserver.core.common.api.proto.GetRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.GetResponse;
|
||||
import it.cavallium.rockserver.core.common.api.proto.KV;
|
||||
import it.cavallium.rockserver.core.common.api.proto.OpenIteratorRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.OpenIteratorResponse;
|
||||
import it.cavallium.rockserver.core.common.api.proto.OpenTransactionRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.OpenTransactionResponse;
|
||||
import it.cavallium.rockserver.core.common.api.proto.Previous;
|
||||
import it.cavallium.rockserver.core.common.api.proto.PreviousPresence;
|
||||
import it.cavallium.rockserver.core.common.api.proto.PutMultiInitialRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.PutMultiRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.PutRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.RocksDBServiceGrpc.RocksDBServiceImplBase;
|
||||
import it.cavallium.rockserver.core.common.api.proto.SeekToRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.SubsequentRequest;
|
||||
import it.cavallium.rockserver.core.common.api.proto.UpdateBegin;
|
||||
import it.unimi.dsi.fastutil.ints.Int2IntFunction;
|
||||
import it.unimi.dsi.fastutil.ints.Int2ObjectFunction;
|
||||
import it.unimi.dsi.fastutil.ints.IntArrayList;
|
||||
import it.unimi.dsi.fastutil.ints.IntList;
|
||||
import it.unimi.dsi.fastutil.objects.ObjectArrayList;
|
||||
import it.unimi.dsi.fastutil.objects.ObjectList;
|
||||
import java.io.IOException;
|
||||
import java.lang.foreign.Arena;
|
||||
import java.lang.foreign.MemorySegment;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class GrpcServer extends Server {
|
||||
|
||||
private static final Logger LOG = Logger.getLogger(GrpcServer.class.getName());
|
||||
|
||||
private final GrpcServerImpl grpc;
|
||||
private final io.grpc.Server server;
|
||||
|
||||
public GrpcServer(RocksDBConnection client, String http2Host, int http2Port) throws IOException {
|
||||
super(client);
|
||||
this.grpc = new GrpcServerImpl(this.getClient());
|
||||
this.server = NettyServerBuilder.forAddress(new InetSocketAddress(http2Host, http2Port))
|
||||
.addService(grpc)
|
||||
.build();
|
||||
server.start();
|
||||
LOG.info("GRPC RocksDB server is listening at " + http2Host + ":" + http2Port);
|
||||
}
|
||||
|
||||
private static final class GrpcServerImpl extends RocksDBServiceImplBase {
|
||||
|
||||
private static final Function<? super Void, Empty> MAP_EMPTY = _ -> Empty.getDefaultInstance();
|
||||
private final RocksDBConnection client;
|
||||
private final Arena autoArena;
|
||||
|
||||
public GrpcServerImpl(RocksDBConnection client) {
|
||||
this.client = client;
|
||||
this.autoArena = Arena.ofAuto();
|
||||
}
|
||||
|
||||
// functions
|
||||
|
||||
@Override
|
||||
public void openTransaction(OpenTransactionRequest request,
|
||||
StreamObserver<OpenTransactionResponse> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.openTransactionAsync(request.getTimeoutMs())
|
||||
.whenComplete(handleResponseObserver(
|
||||
txId -> OpenTransactionResponse.newBuilder().setTransactionId(txId).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void closeTransaction(CloseTransactionRequest request,
|
||||
StreamObserver<CloseTransactionResponse> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.closeTransactionAsync(request.getTransactionId(), request.getCommit())
|
||||
.whenComplete(handleResponseObserver(
|
||||
committed -> CloseTransactionResponse.newBuilder().setSuccessful(committed).build(),
|
||||
responseObserver
|
||||
));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void closeFailedUpdate(CloseFailedUpdateRequest request, StreamObserver<Empty> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.closeFailedUpdateAsync(request.getUpdateId())
|
||||
.whenComplete(handleResponseObserver(MAP_EMPTY, responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void createColumn(CreateColumnRequest request, StreamObserver<CreateColumnResponse> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.createColumnAsync(request.getName(), mapColumnSchema(request.getSchema()))
|
||||
.whenComplete(handleResponseObserver(
|
||||
colId -> CreateColumnResponse.newBuilder().setColumnId(colId).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deleteColumn(DeleteColumnRequest request, StreamObserver<Empty> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.deleteColumnAsync(request.getColumnId())
|
||||
.whenComplete(handleResponseObserver(MAP_EMPTY, responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void getColumnId(GetColumnIdRequest request, StreamObserver<GetColumnIdResponse> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.getColumnIdAsync(request.getName())
|
||||
.whenComplete(handleResponseObserver(
|
||||
colId -> GetColumnIdResponse.newBuilder().setColumnId(colId).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void put(PutRequest request, StreamObserver<Empty> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.putAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getData().getKeysCount(), request.getData()::getKeys),
|
||||
MemorySegment.ofBuffer(request.getData().getValue().asReadOnlyByteBuffer()),
|
||||
new RequestNothing<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(MAP_EMPTY, responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public StreamObserver<PutMultiRequest> putMulti(StreamObserver<Empty> responseObserver) {
|
||||
return new StreamObserver<>() {
|
||||
private boolean initialRequestDone = false;
|
||||
private long requestsCount = 0;
|
||||
private boolean requestsCountFinalized;
|
||||
private final AtomicLong processedRequestsCount = new AtomicLong();
|
||||
private PutMultiInitialRequest initialRequest;
|
||||
|
||||
@Override
|
||||
public void onNext(PutMultiRequest request) {
|
||||
switch (request.getPutMultiRequestTypeCase()) {
|
||||
case INITIALREQUEST -> {
|
||||
if (initialRequestDone) {
|
||||
throw new UnsupportedOperationException("Initial request already done!");
|
||||
}
|
||||
this.initialRequest = request.getInitialRequest();
|
||||
this.initialRequestDone = true;
|
||||
}
|
||||
case DATA -> {
|
||||
if (!initialRequestDone) {
|
||||
throw new UnsupportedOperationException("Initial request already done!");
|
||||
}
|
||||
++requestsCount;
|
||||
client.getAsyncApi()
|
||||
.putAsync(autoArena,
|
||||
initialRequest.getTransactionOrUpdateId(),
|
||||
initialRequest.getColumnId(),
|
||||
mapKeys(request.getData().getKeysCount(), request.getData()::getKeys),
|
||||
MemorySegment.ofBuffer(request.getData().getValue().asReadOnlyByteBuffer()),
|
||||
new RequestNothing<>()
|
||||
)
|
||||
.whenComplete((_, error) -> {
|
||||
if (error != null) {
|
||||
responseObserver.onError(error);
|
||||
} else {
|
||||
var newProcessedRequestCount = processedRequestsCount.incrementAndGet();
|
||||
if (requestsCountFinalized) {
|
||||
if (newProcessedRequestCount == requestsCount) {
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
case null, default ->
|
||||
throw new UnsupportedOperationException("Unsupported operation: "
|
||||
+ request.getPutMultiRequestTypeCase());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
responseObserver.onError(t);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
requestsCountFinalized = true;
|
||||
if (requestsCount == 0) {
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public void putGetPrevious(PutRequest request, StreamObserver<Previous> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.putAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getData().getKeysCount(), request.getData()::getKeys),
|
||||
MemorySegment.ofBuffer(request.getData().getValue().asReadOnlyByteBuffer()),
|
||||
new RequestPrevious<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
prev -> {
|
||||
var prevBuilder = Previous.newBuilder();
|
||||
if (prev != null) {
|
||||
prevBuilder.setPrevious(ByteString.copyFrom(prev.asByteBuffer()));
|
||||
}
|
||||
return prevBuilder.build();
|
||||
},
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void putGetDelta(PutRequest request, StreamObserver<Delta> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.putAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getData().getKeysCount(), request.getData()::getKeys),
|
||||
MemorySegment.ofBuffer(request.getData().getValue().asReadOnlyByteBuffer()),
|
||||
new RequestDelta<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
delta -> {
|
||||
var deltaBuilder = Delta.newBuilder();
|
||||
if (delta.previous() != null) {
|
||||
deltaBuilder.setPrevious(ByteString.copyFrom(delta.previous().asByteBuffer()));
|
||||
}
|
||||
if (delta.current() != null) {
|
||||
deltaBuilder.setCurrent(ByteString.copyFrom(delta.current().asByteBuffer()));
|
||||
}
|
||||
return deltaBuilder.build();
|
||||
},
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void putGetChanged(PutRequest request, StreamObserver<Changed> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.putAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getData().getKeysCount(), request.getData()::getKeys),
|
||||
MemorySegment.ofBuffer(request.getData().getValue().asReadOnlyByteBuffer()),
|
||||
new RequestChanged<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
changed -> Changed.newBuilder().setChanged(changed).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void putGetPreviousPresence(PutRequest request, StreamObserver<PreviousPresence> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.putAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getData().getKeysCount(), request.getData()::getKeys),
|
||||
MemorySegment.ofBuffer(request.getData().getValue().asReadOnlyByteBuffer()),
|
||||
new RequestPreviousPresence<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
present -> PreviousPresence.newBuilder().setPresent(present).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void get(GetRequest request, StreamObserver<GetResponse> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.getAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getKeysCount(), request::getKeys),
|
||||
new RequestCurrent<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
current -> {
|
||||
var response = GetResponse.newBuilder();
|
||||
if (current != null) {
|
||||
response.setValue(ByteString.copyFrom(current.asByteBuffer()));
|
||||
}
|
||||
return response.build();
|
||||
},
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void getForUpdate(GetRequest request, StreamObserver<UpdateBegin> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.getAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getKeysCount(), request::getKeys),
|
||||
new RequestForUpdate<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
forUpdate -> {
|
||||
var response = UpdateBegin.newBuilder();
|
||||
response.setUpdateId(forUpdate.updateId());
|
||||
if (forUpdate.previous() != null) {
|
||||
response.setPrevious(ByteString.copyFrom(forUpdate.previous().asByteBuffer()));
|
||||
}
|
||||
return response.build();
|
||||
},
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void exists(GetRequest request, StreamObserver<PreviousPresence> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.getAsync(autoArena,
|
||||
request.getTransactionOrUpdateId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getKeysCount(), request::getKeys),
|
||||
new RequestExists<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
exists -> PreviousPresence.newBuilder().setPresent(exists).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void openIterator(OpenIteratorRequest request, StreamObserver<OpenIteratorResponse> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.openIteratorAsync(autoArena,
|
||||
request.getTransactionId(),
|
||||
request.getColumnId(),
|
||||
mapKeys(request.getStartKeysInclusiveCount(), request::getStartKeysInclusive),
|
||||
mapKeys(request.getEndKeysExclusiveCount(), request::getEndKeysExclusive),
|
||||
request.getReverse(),
|
||||
request.getTimeoutMs()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
iteratorId -> OpenIteratorResponse.newBuilder().setIteratorId(iteratorId).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void closeIterator(CloseIteratorRequest request, StreamObserver<Empty> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.closeIteratorAsync(request.getIteratorId())
|
||||
.whenComplete(handleResponseObserver(MAP_EMPTY, responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void seekTo(SeekToRequest request, StreamObserver<Empty> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.seekToAsync(autoArena, request.getIterationId(), mapKeys(request.getKeysCount(), request::getKeys))
|
||||
.whenComplete(handleResponseObserver(MAP_EMPTY, responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subsequent(SubsequentRequest request, StreamObserver<Empty> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.subsequentAsync(autoArena,
|
||||
request.getIterationId(),
|
||||
request.getSkipCount(),
|
||||
request.getTakeCount(),
|
||||
new RequestNothing<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(MAP_EMPTY, responseObserver));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subsequentExists(SubsequentRequest request, StreamObserver<PreviousPresence> responseObserver) {
|
||||
client.getAsyncApi()
|
||||
.subsequentAsync(autoArena,
|
||||
request.getIterationId(),
|
||||
request.getSkipCount(),
|
||||
request.getTakeCount(),
|
||||
new RequestExists<>()
|
||||
)
|
||||
.whenComplete(handleResponseObserver(
|
||||
exists -> PreviousPresence.newBuilder().setPresent(exists).build(),
|
||||
responseObserver));
|
||||
}
|
||||
|
||||
public void subsequentMultiPage(SubsequentRequest request, StreamObserver<KV> responseObserver, int pageIndex) {
|
||||
final long pageSize = 16L;
|
||||
if (request.getTakeCount() > pageIndex * pageSize) {
|
||||
client.getAsyncApi()
|
||||
.subsequentAsync(autoArena,
|
||||
request.getIterationId(),
|
||||
pageIndex == 0 ? request.getSkipCount() : 0,
|
||||
Math.min(request.getTakeCount() - pageIndex * pageSize, pageSize),
|
||||
new RequestMulti<>()
|
||||
)
|
||||
.whenComplete((response, ex) -> {
|
||||
if (ex != null) {
|
||||
responseObserver.onError(ex);
|
||||
} else {
|
||||
for (MemorySegment entry : response) {
|
||||
Keys keys = null; // todo: implement
|
||||
MemorySegment value = entry;
|
||||
responseObserver.onNext(KV.newBuilder()
|
||||
.addAllKeys(null) // todo: implement
|
||||
.setValue(ByteString.copyFrom(value.asByteBuffer()))
|
||||
.build());
|
||||
}
|
||||
subsequentMultiPage(request, responseObserver, pageIndex + 1);
|
||||
}
|
||||
});
|
||||
} else {
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void subsequentMultiGet(SubsequentRequest request, StreamObserver<KV> responseObserver) {
|
||||
subsequentMultiPage(request, responseObserver, 0);
|
||||
}
|
||||
|
||||
// mappers
|
||||
|
||||
private static ColumnSchema mapColumnSchema(it.cavallium.rockserver.core.common.api.proto.ColumnSchema schema) {
|
||||
return ColumnSchema.of(mapKeysLength(schema.getFixedKeysCount(), schema::getFixedKeys),
|
||||
mapVariableTailKeys(schema.getVariableTailKeysCount(), schema::getVariableTailKeys),
|
||||
schema.getHasValue()
|
||||
);
|
||||
}
|
||||
|
||||
private static IntList mapKeysLength(int count, Int2IntFunction keyGetterAt) {
|
||||
var l = new IntArrayList(count);
|
||||
for (int i = 0; i < count; i++) {
|
||||
l.add(keyGetterAt.apply(i));
|
||||
}
|
||||
return l;
|
||||
}
|
||||
|
||||
private static ObjectList<ColumnHashType> mapVariableTailKeys(int count,
|
||||
Int2ObjectFunction<it.cavallium.rockserver.core.common.api.proto.ColumnHashType> variableTailKeyGetterAt) {
|
||||
var l = new ObjectArrayList<ColumnHashType>(count);
|
||||
for (int i = 0; i < count; i++) {
|
||||
l.add(switch (variableTailKeyGetterAt.apply(i)) {
|
||||
case XXHASH32 -> ColumnHashType.XXHASH32;
|
||||
case XXHASH8 -> ColumnHashType.XXHASH8;
|
||||
case ALLSAME8 -> ColumnHashType.ALLSAME8;
|
||||
case UNRECOGNIZED -> throw new UnsupportedOperationException();
|
||||
});
|
||||
}
|
||||
return l;
|
||||
}
|
||||
|
||||
private static Keys mapKeys(int count, Int2ObjectFunction<ByteString> keyGetterAt) {
|
||||
var segments = new MemorySegment[count];
|
||||
for (int i = 0; i < count; i++) {
|
||||
segments[i] = MemorySegment.ofBuffer(keyGetterAt.apply(i).asReadOnlyByteBuffer());
|
||||
}
|
||||
return new Keys(segments);
|
||||
}
|
||||
|
||||
// utils
|
||||
|
||||
private static <T> BiConsumer<? super T, Throwable> handleResponseObserver(StreamObserver<T> responseObserver) {
|
||||
return (value, ex) -> {
|
||||
if (ex != null) {
|
||||
responseObserver.onError(ex);
|
||||
} else {
|
||||
if (value != null) {
|
||||
responseObserver.onNext(value);
|
||||
}
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private static <PREV, T> BiConsumer<? super PREV, Throwable> handleResponseObserver(Function<PREV, T> resultMapper,
|
||||
StreamObserver<T> responseObserver) {
|
||||
return (value, ex) -> {
|
||||
if (ex != null) {
|
||||
responseObserver.onError(ex);
|
||||
} else {
|
||||
T mapped;
|
||||
try {
|
||||
mapped = resultMapper.apply(value);
|
||||
} catch (Throwable ex2) {
|
||||
responseObserver.onError(ex2);
|
||||
return;
|
||||
}
|
||||
if (mapped != null) {
|
||||
responseObserver.onNext(mapped);
|
||||
}
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
LOG.info("GRPC server is shutting down...");
|
||||
server.shutdown();
|
||||
try {
|
||||
server.awaitTermination();
|
||||
} catch (InterruptedException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
super.close();
|
||||
}
|
||||
}
|
||||
@ -14,6 +14,7 @@ public class ServerBuilder {
|
||||
private UnixDomainSocketAddress unixAddress;
|
||||
private String http2Host;
|
||||
private int http2Port;
|
||||
private boolean useThrift;
|
||||
private RocksDBConnection client;
|
||||
|
||||
public void setUnixSocket(UnixDomainSocketAddress address) {
|
||||
@ -29,13 +30,21 @@ public class ServerBuilder {
|
||||
this.http2Port = port;
|
||||
}
|
||||
|
||||
public void setUseThrift(boolean useThrift) {
|
||||
this.useThrift = useThrift;
|
||||
}
|
||||
|
||||
public void setClient(RocksDBConnection client) {
|
||||
this.client = client;
|
||||
}
|
||||
|
||||
public Server build() throws IOException {
|
||||
if (http2Host != null) {
|
||||
return new ThriftServer(client, http2Host, http2Port);
|
||||
if (useThrift) {
|
||||
return new ThriftServer(client, http2Host, http2Port);
|
||||
} else {
|
||||
return new GrpcServer(client, http2Host, http2Port);
|
||||
}
|
||||
} else if (unixAddress != null) {
|
||||
throw new UnsupportedOperationException("Not implemented: unix socket");
|
||||
} else if (iNetAddress != null) {
|
||||
|
||||
@ -22,6 +22,8 @@ import java.net.InetSocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
import java.util.List;
|
||||
import java.util.logging.Logger;
|
||||
import org.apache.thrift.TException;
|
||||
import org.apache.thrift.server.TThreadedSelectorServer;
|
||||
import org.apache.thrift.transport.TNonblockingServerSocket;
|
||||
import org.apache.thrift.transport.TTransportException;
|
||||
@ -29,19 +31,26 @@ import org.jetbrains.annotations.NotNull;
|
||||
|
||||
public class ThriftServer extends Server {
|
||||
|
||||
private static final Logger LOG = Logger.getLogger(ThriftServer.class.getName());
|
||||
private static final OfByte BYTE_BE = ValueLayout.JAVA_BYTE.withOrder(ByteOrder.BIG_ENDIAN);
|
||||
|
||||
private final Thread thriftThread;
|
||||
private final TThreadedSelectorServer server;
|
||||
|
||||
public ThriftServer(RocksDBConnection client, String http2Host, int http2Port) throws IOException {
|
||||
super(client);
|
||||
var handler = new ThriftHandler(this.getClient());
|
||||
|
||||
try {
|
||||
var serverTransport = new TNonblockingServerSocket(new InetSocketAddress(http2Host, http2Port));
|
||||
var server = new TThreadedSelectorServer(new TThreadedSelectorServer.Args(serverTransport)
|
||||
this.server = new TThreadedSelectorServer(new TThreadedSelectorServer.Args(serverTransport)
|
||||
.processor(new Processor<>(handler))
|
||||
);
|
||||
|
||||
server.serve();
|
||||
var thriftThread = new Thread(server::serve);
|
||||
thriftThread.setName("Thrift server thread");
|
||||
this.thriftThread = thriftThread;
|
||||
LOG.info("Thrift RocksDB server is listening at " + http2Host + ":" + http2Port);
|
||||
} catch (TTransportException e) {
|
||||
throw new IOException("Can't open server socket", e);
|
||||
}
|
||||
@ -131,8 +140,8 @@ public class ThriftServer extends Server {
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean closeTransaction(long timeoutMs, boolean commit) {
|
||||
return client.getSyncApi().closeTransaction(timeoutMs, commit);
|
||||
public boolean closeTransaction(long transactionId, boolean commit) throws TException {
|
||||
return client.getSyncApi().closeTransaction(transactionId, commit);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -299,4 +308,11 @@ public class ThriftServer extends Server {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
LOG.info("Thrift server is shutting down...");
|
||||
this.server.stop();
|
||||
super.close();
|
||||
}
|
||||
}
|
||||
|
||||
@ -9,6 +9,13 @@ module rockserver.core {
|
||||
requires it.unimi.dsi.fastutil;
|
||||
requires org.apache.thrift;
|
||||
requires org.slf4j;
|
||||
requires protobuf.java;
|
||||
requires io.grpc.protobuf;
|
||||
requires io.grpc.stub;
|
||||
requires io.grpc;
|
||||
requires jsr305;
|
||||
requires com.google.common;
|
||||
requires io.grpc.netty;
|
||||
|
||||
exports it.cavallium.rockserver.core.client;
|
||||
exports it.cavallium.rockserver.core.common;
|
||||
|
||||
105
src/main/proto/rocksdb.proto
Normal file
105
src/main/proto/rocksdb.proto
Normal file
@ -0,0 +1,105 @@
|
||||
syntax = "proto3";
|
||||
|
||||
import "google/protobuf/empty.proto";
|
||||
|
||||
option java_multiple_files = true;
|
||||
package it.cavallium.rockserver.core.common.api.proto;
|
||||
|
||||
message ColumnSchema {
|
||||
repeated int32 fixedKeys = 1;
|
||||
repeated ColumnHashType variableTailKeys = 2;
|
||||
bool hasValue = 3;
|
||||
}
|
||||
|
||||
enum ColumnHashType {
|
||||
XXHASH32 = 0;
|
||||
XXHASH8 = 1;
|
||||
ALLSAME8 = 2;
|
||||
}
|
||||
|
||||
enum Operation {
|
||||
NOTHING = 0;
|
||||
PREVIOUS = 1;
|
||||
CURRENT = 2;
|
||||
FOR_UPDATE = 3;
|
||||
EXISTS = 4;
|
||||
DELTA = 5;
|
||||
MULTI = 6;
|
||||
CHANGED = 7;
|
||||
PREVIOUS_PRESENCE = 8;
|
||||
}
|
||||
|
||||
message Delta {
|
||||
optional bytes previous = 1;
|
||||
optional bytes current = 2;
|
||||
}
|
||||
message Previous {optional bytes previous = 1;}
|
||||
message Changed {bool changed = 1;}
|
||||
message PreviousPresence {bool present = 1;}
|
||||
|
||||
message UpdateBegin {
|
||||
optional bytes previous = 1;
|
||||
optional int64 updateId = 2;
|
||||
}
|
||||
|
||||
message KV {
|
||||
repeated bytes keys = 1;
|
||||
bytes value = 2;
|
||||
}
|
||||
|
||||
message OpenTransactionRequest {int64 timeoutMs = 1;}
|
||||
message OpenTransactionResponse {int64 transactionId = 1;}
|
||||
|
||||
message CloseTransactionRequest {int64 transactionId = 1; int64 timeoutMs = 2; bool commit = 3;}
|
||||
message CloseTransactionResponse {bool successful = 1;}
|
||||
|
||||
message CloseFailedUpdateRequest {int64 updateId = 1;}
|
||||
|
||||
message CreateColumnRequest {string name = 1; ColumnSchema schema = 2;}
|
||||
message CreateColumnResponse {int64 columnId = 1;}
|
||||
|
||||
message DeleteColumnRequest {int64 columnId = 1;}
|
||||
|
||||
message GetColumnIdRequest {string name = 1;}
|
||||
message GetColumnIdResponse {int64 columnId = 1;}
|
||||
|
||||
message PutRequest {int64 transactionOrUpdateId = 1; int64 columnId = 2; KV data = 3;}
|
||||
|
||||
message PutMultiInitialRequest {int64 transactionOrUpdateId = 1; int64 columnId = 2;}
|
||||
message PutMultiRequest {oneof putMultiRequestType {PutMultiInitialRequest initialRequest = 1;KV data = 2;}}
|
||||
|
||||
message GetRequest {int64 transactionOrUpdateId = 1; int64 columnId = 2; repeated bytes keys = 3;}
|
||||
message GetResponse {optional bytes value = 1;}
|
||||
|
||||
message OpenIteratorRequest {int64 transactionId = 1; int64 columnId = 2; repeated bytes startKeysInclusive = 3; repeated bytes endKeysExclusive = 4; bool reverse = 5; int64 timeoutMs = 6;}
|
||||
message OpenIteratorResponse {int64 iteratorId = 1;}
|
||||
|
||||
message CloseIteratorRequest {int64 iteratorId = 1;}
|
||||
|
||||
message SeekToRequest {int64 iterationId = 1; repeated bytes keys = 2;}
|
||||
|
||||
message SubsequentRequest {int64 iterationId = 1; int64 skipCount = 2; int64 takeCount = 3;}
|
||||
|
||||
service RocksDBService {
|
||||
rpc openTransaction(OpenTransactionRequest) returns (OpenTransactionResponse);
|
||||
rpc closeTransaction(CloseTransactionRequest) returns (CloseTransactionResponse);
|
||||
rpc closeFailedUpdate(CloseFailedUpdateRequest) returns (google.protobuf.Empty);
|
||||
rpc createColumn(CreateColumnRequest) returns (CreateColumnResponse);
|
||||
rpc deleteColumn(DeleteColumnRequest) returns (google.protobuf.Empty);
|
||||
rpc getColumnId(GetColumnIdRequest) returns (GetColumnIdResponse);
|
||||
rpc put(PutRequest) returns (google.protobuf.Empty);
|
||||
rpc putMulti(stream PutMultiRequest) returns (google.protobuf.Empty);
|
||||
rpc putGetPrevious(PutRequest) returns (Previous);
|
||||
rpc putGetDelta(PutRequest) returns (Delta);
|
||||
rpc putGetChanged(PutRequest) returns (Changed);
|
||||
rpc putGetPreviousPresence(PutRequest) returns (PreviousPresence);
|
||||
rpc get(GetRequest) returns (GetResponse);
|
||||
rpc getForUpdate(GetRequest) returns (UpdateBegin);
|
||||
rpc exists(GetRequest) returns (PreviousPresence);
|
||||
rpc openIterator(OpenIteratorRequest) returns (OpenIteratorResponse);
|
||||
rpc closeIterator(CloseIteratorRequest) returns (google.protobuf.Empty);
|
||||
rpc seekTo(SeekToRequest) returns (google.protobuf.Empty);
|
||||
rpc subsequent(SubsequentRequest) returns (google.protobuf.Empty);
|
||||
rpc subsequentExists(SubsequentRequest) returns (PreviousPresence);
|
||||
rpc subsequentMultiGet(SubsequentRequest) returns (stream KV);
|
||||
}
|
||||
@ -42,7 +42,7 @@ service RocksDB {
|
||||
|
||||
i64 openTransaction(1: required i64 timeoutMs),
|
||||
|
||||
bool closeTransaction(1: required i64 timeoutMs, 2: required bool commit),
|
||||
bool closeTransaction(1: required i64 transactionId, 2: required bool commit),
|
||||
|
||||
void closeFailedUpdate(1: required i64 updateId),
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user