andreacavalli/rockserver
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: andreacavalli:aeadadd7b4d526bd7e8dc8eed9bbcf00a16a31f9
Branches Tags
andreacavalli:master
andreacavalli:sync-grpc
andreacavalli:migration
andreacavalli:v1.0.13
andreacavalli:v1.0.12
..
compare: andreacavalli:2abbc2984fbb652d88d6a52519182e552205dece
Branches Tags
andreacavalli:master
andreacavalli:sync-grpc
andreacavalli:migration
andreacavalli:v1.0.13
andreacavalli:v1.0.12

No commits in common. "aeadadd7b4d526bd7e8dc8eed9bbcf00a16a31f9" and "2abbc2984fbb652d88d6a52519182e552205dece" have entirely different histories.
aeadadd7b4 ... 2abbc2984f

15 changed files with 94 additions and 829 deletions
Show Stats Expand all files Collapse all files

48
pom.xml
View File

@ -17,9 +17,6 @@
<slf4j.version>2.0.12</slf4j.version> <slf4j.version>2.0.12</slf4j.version>
<imageName>rockserver-core</imageName> <imageName>rockserver-core</imageName>
<mainClass>it.cavallium.rockserver.core.Main</mainClass> <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> </properties>
<repositories> <repositories>
@ -83,21 +80,6 @@
<artifactId>slf4j-api</artifactId> <artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version> <version>${slf4j.version}</version>
</dependency> </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> <dependency>
<groupId>org.lz4</groupId> <groupId>org.lz4</groupId>
@ -124,13 +106,6 @@
<filtering>false</filtering> <filtering>false</filtering>
</resource> </resource>
</resources> </resources>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.6.1</version>
</extension>
</extensions>
<plugins> <plugins>
<plugin> <plugin>
<groupId>org.codehaus.mojo</groupId> <groupId>org.codehaus.mojo</groupId>
@ -196,29 +171,6 @@
</execution> </execution>
</executions> </executions>
</plugin> </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> </plugins>
</build> </build>

44
src/main/java/it/cavallium/rockserver/core/Main.java
View File

@ -5,7 +5,6 @@ import static java.util.Objects.requireNonNull;
import it.cavallium.rockserver.core.common.Utils; import it.cavallium.rockserver.core.common.Utils;
import it.cavallium.rockserver.core.impl.rocksdb.RocksDBLoader; import it.cavallium.rockserver.core.impl.rocksdb.RocksDBLoader;
import it.cavallium.rockserver.core.server.ServerBuilder;
import java.io.IOException; import java.io.IOException;
import java.net.URI; import java.net.URI;
import java.net.URISyntaxException; import java.net.URISyntaxException;
@ -31,14 +30,10 @@ public class Main {
.type(String.class) .type(String.class)
.setDefault(PRIVATE_MEMORY_URL.toString()) .setDefault(PRIVATE_MEMORY_URL.toString())
.help("Specify database rocksdb://hostname:port, or unix://<path>, or file://<path>"); .help("Specify database rocksdb://hostname:port, or unix://<path>, or file://<path>");
parser.addArgument("-l", "--thrift-listen-url") parser.addArgument("-l", "--listen-url")
.type(String.class) .type(String.class)
.setDefault("http://127.0.0.1:5332") .setDefault("http://127.0.0.1:5332")
.help("Specify database http://hostname:port, or unix://<path>, or file://<path>"); .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") parser.addArgument("-n", "--name")
.type(String.class) .type(String.class)
.setDefault("main") .setDefault("main")
@ -58,6 +53,7 @@ public class Main {
System.exit(1); System.exit(1);
} }
var clientBuilder = new it.cavallium.rockserver.core.client.ClientBuilder(); var clientBuilder = new it.cavallium.rockserver.core.client.ClientBuilder();
var serverBuilder = new it.cavallium.rockserver.core.server.ServerBuilder();
if (ns.getBoolean("print_default_config")) { if (ns.getBoolean("print_default_config")) {
requireNonNull(Main.class.getClassLoader() requireNonNull(Main.class.getClassLoader()
@ -71,14 +67,12 @@ public class Main {
RocksDBLoader.loadLibrary(); RocksDBLoader.loadLibrary();
var rawDatabaseUrl = ns.getString("database_url"); var rawDatabaseUrl = ns.getString("database_url");
var rawThriftListenUrl = ns.getString("thrift_listen_url"); var rawListenUrl = ns.getString("listen_url");
var rawGrpcListenUrl = ns.getString("grpc_listen_url");
var name = ns.getString("name"); var name = ns.getString("name");
var config = ns.getString("config"); var config = ns.getString("config");
var databaseUrl = new URI(rawDatabaseUrl); var databaseUrl = new URI(rawDatabaseUrl);
var thriftListenUrl = new URI(rawThriftListenUrl); var listenUrl = new URI(rawListenUrl);
var grpcListenUrl = new URI(rawGrpcListenUrl);
if (config != null) { if (config != null) {
if (!databaseUrl.getScheme().equals("file")) { if (!databaseUrl.getScheme().equals("file")) {
@ -99,23 +93,24 @@ public class Main {
case null, default -> throw new IllegalArgumentException("Invalid scheme \"" + databaseUrlScheme + "\" for database url url: " + databaseUrl); case null, default -> throw new IllegalArgumentException("Invalid scheme \"" + databaseUrlScheme + "\" for database url url: " + databaseUrl);
} }
var thriftServerBuilder = new it.cavallium.rockserver.core.server.ServerBuilder(); var listenUrlScheme = listenUrl.getScheme();
buildServerAddress(thriftServerBuilder, thriftListenUrl, true); switch (listenUrlScheme) {
var grpcServerBuilder = new it.cavallium.rockserver.core.server.ServerBuilder(); case "unix" -> serverBuilder.setUnixSocket(UnixDomainSocketAddress.of(Path.of(listenUrl.getPath())));
buildServerAddress(grpcServerBuilder, grpcListenUrl, false); 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);
}
clientBuilder.setName(name); clientBuilder.setName(name);
try (var connection = clientBuilder.build()) { try (var connection = clientBuilder.build()) {
LOG.log(Level.INFO, "Connected to {0}", connection); LOG.log(Level.INFO, "Connected to {0}", connection);
thriftServerBuilder.setClient(connection); serverBuilder.setClient(connection);
grpcServerBuilder.setClient(connection);
CountDownLatch shutdownLatch = new CountDownLatch(1); CountDownLatch shutdownLatch = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(new Thread(shutdownLatch::countDown)); Runtime.getRuntime().addShutdownHook(new Thread(shutdownLatch::countDown));
try (var _ = thriftServerBuilder.build(); try (var server = serverBuilder.build()) {
var _ = grpcServerBuilder.build()) {
shutdownLatch.await(); shutdownLatch.await();
LOG.info("Shutting down..."); LOG.info("Shutting down...");
} }
@ -124,17 +119,4 @@ public class Main {
} }
LOG.info("Shut down successfully"); 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);
}
}
} }

2
src/main/java/it/cavallium/rockserver/core/common/api/ColumnHashType.java
View File

@ -1,5 +1,5 @@
/** /**
* Autogenerated by Thrift Compiler (0.20.0) * Autogenerated by Thrift Compiler (0.19.0)
* *
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated * @generated

2
src/main/java/it/cavallium/rockserver/core/common/api/ColumnSchema.java
View File

@ -1,5 +1,5 @@
/** /**
* Autogenerated by Thrift Compiler (0.20.0) * Autogenerated by Thrift Compiler (0.19.0)
* *
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated * @generated

2
src/main/java/it/cavallium/rockserver/core/common/api/Delta.java
View File

@ -1,5 +1,5 @@
/** /**
* Autogenerated by Thrift Compiler (0.20.0) * Autogenerated by Thrift Compiler (0.19.0)
* *
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated * @generated

2
src/main/java/it/cavallium/rockserver/core/common/api/Operation.java
View File

@ -1,5 +1,5 @@
/** /**
* Autogenerated by Thrift Compiler (0.20.0) * Autogenerated by Thrift Compiler (0.19.0)
* *
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated * @generated

2
src/main/java/it/cavallium/rockserver/core/common/api/OptionalBinary.java
View File

@ -1,5 +1,5 @@
/** /**
* Autogenerated by Thrift Compiler (0.20.0) * Autogenerated by Thrift Compiler (0.19.0)
* *
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated * @generated

138
src/main/java/it/cavallium/rockserver/core/common/api/RocksDB.java
View File

@ -1,5 +1,5 @@
/** /**
* Autogenerated by Thrift Compiler (0.20.0) * Autogenerated by Thrift Compiler (0.19.0)
* *
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated * @generated
@ -13,7 +13,7 @@ public class RocksDB {
public long openTransaction(long timeoutMs) throws org.apache.thrift.TException; public long openTransaction(long timeoutMs) throws org.apache.thrift.TException;
public boolean closeTransaction(long transactionId, boolean commit) throws org.apache.thrift.TException; public boolean closeTransaction(long timeoutMs, boolean commit) throws org.apache.thrift.TException;
public void closeFailedUpdate(long updateId) 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 openTransaction(long timeoutMs, org.apache.thrift.async.AsyncMethodCallback<java.lang.Long> 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 closeTransaction(long timeoutMs, 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; 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 @Override
public boolean closeTransaction(long transactionId, boolean commit) throws org.apache.thrift.TException public boolean closeTransaction(long timeoutMs, boolean commit) throws org.apache.thrift.TException
{ {
send_closeTransaction(transactionId, commit); send_closeTransaction(timeoutMs, commit);
return recv_closeTransaction(); return recv_closeTransaction();
} }
public void send_closeTransaction(long transactionId, boolean commit) throws org.apache.thrift.TException public void send_closeTransaction(long timeoutMs, boolean commit) throws org.apache.thrift.TException
{ {
closeTransaction_args args = new closeTransaction_args(); closeTransaction_args args = new closeTransaction_args();
args.setTransactionId(transactionId); args.setTimeoutMs(timeoutMs);
args.setCommit(commit); args.setCommit(commit);
sendBase("closeTransaction", args); sendBase("closeTransaction", args);
} }
@ -719,19 +719,19 @@ public class RocksDB {
} }
@Override @Override
public void closeTransaction(long transactionId, boolean commit, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> 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 {
checkReady(); checkReady();
closeTransaction_call method_call = new closeTransaction_call(transactionId, commit, resultHandler, this, ___protocolFactory, ___transport); closeTransaction_call method_call = new closeTransaction_call(timeoutMs, commit, resultHandler, this, ___protocolFactory, ___transport);
this.___currentMethod = method_call; this.___currentMethod = method_call;
___manager.call(method_call); ___manager.call(method_call);
} }
public static class closeTransaction_call extends org.apache.thrift.async.TAsyncMethodCall<java.lang.Boolean> { public static class closeTransaction_call extends org.apache.thrift.async.TAsyncMethodCall<java.lang.Boolean> {
private long transactionId; private long timeoutMs;
private boolean commit; private boolean commit;
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 { 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 {
super(client, protocolFactory, transport, resultHandler, false); super(client, protocolFactory, transport, resultHandler, false);
this.transactionId = transactionId; this.timeoutMs = timeoutMs;
this.commit = commit; 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 { 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)); prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("closeTransaction", org.apache.thrift.protocol.TMessageType.CALL, 0));
closeTransaction_args args = new closeTransaction_args(); closeTransaction_args args = new closeTransaction_args();
args.setTransactionId(transactionId); args.setTimeoutMs(timeoutMs);
args.setCommit(commit); args.setCommit(commit);
args.write(prot); args.write(prot);
prot.writeMessageEnd(); prot.writeMessageEnd();
@ -1673,7 +1673,7 @@ public class RocksDB {
@Override @Override
public closeTransaction_result getResult(I iface, closeTransaction_args args) throws org.apache.thrift.TException { public closeTransaction_result getResult(I iface, closeTransaction_args args) throws org.apache.thrift.TException {
closeTransaction_result result = new closeTransaction_result(); closeTransaction_result result = new closeTransaction_result();
result.success = iface.closeTransaction(args.transactionId, args.commit); result.success = iface.closeTransaction(args.timeoutMs, args.commit);
result.setSuccessIsSet(true); result.setSuccessIsSet(true);
return result; return result;
} }
@ -2415,7 +2415,7 @@ public class RocksDB {
@Override @Override
public void start(I iface, closeTransaction_args args, org.apache.thrift.async.AsyncMethodCallback<java.lang.Boolean> resultHandler) throws org.apache.thrift.TException { 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.transactionId, args.commit,resultHandler); iface.closeTransaction(args.timeoutMs, 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> { 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.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("closeTransaction_args");
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 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 COMMIT_FIELD_DESC = new org.apache.thrift.protocol.TField("commit", org.apache.thrift.protocol.TType.BOOL, (short)2); 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 STANDARD_SCHEME_FACTORY = new closeTransaction_argsStandardSchemeFactory();
private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new closeTransaction_argsTupleSchemeFactory(); private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new closeTransaction_argsTupleSchemeFactory();
public long transactionId; // required public long timeoutMs; // required
public boolean commit; // required public boolean commit; // required
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
public enum _Fields implements org.apache.thrift.TFieldIdEnum { public enum _Fields implements org.apache.thrift.TFieldIdEnum {
TRANSACTION_ID((short)1, "transactionId"), TIMEOUT_MS((short)1, "timeoutMs"),
COMMIT((short)2, "commit"); COMMIT((short)2, "commit");
private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>(); 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 @org.apache.thrift.annotation.Nullable
public static _Fields findByThriftId(int fieldId) { public static _Fields findByThriftId(int fieldId) {
switch(fieldId) { switch(fieldId) {
case 1: // TRANSACTION_ID case 1: // TIMEOUT_MS
return TRANSACTION_ID; return TIMEOUT_MS;
case 2: // COMMIT case 2: // COMMIT
return COMMIT; return COMMIT;
default: default:
@ -4556,13 +4556,13 @@ public class RocksDB {
} }
// isset id assignments // isset id assignments
private static final int __TRANSACTIONID_ISSET_ID = 0; private static final int __TIMEOUTMS_ISSET_ID = 0;
private static final int __COMMIT_ISSET_ID = 1; private static final int __COMMIT_ISSET_ID = 1;
private byte __isset_bitfield = 0; private byte __isset_bitfield = 0;
public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap; public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
static { 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); 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.TRANSACTION_ID, new org.apache.thrift.meta_data.FieldMetaData("transactionId", org.apache.thrift.TFieldRequirementType.REQUIRED, tmpMap.put(_Fields.TIMEOUT_MS, new org.apache.thrift.meta_data.FieldMetaData("timeoutMs", org.apache.thrift.TFieldRequirementType.REQUIRED,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I64))); 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, 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))); new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL)));
@ -4574,12 +4574,12 @@ public class RocksDB {
} }
public closeTransaction_args( public closeTransaction_args(
long transactionId, long timeoutMs,
boolean commit) boolean commit)
{ {
this(); this();
this.transactionId = transactionId; this.timeoutMs = timeoutMs;
setTransactionIdIsSet(true); setTimeoutMsIsSet(true);
this.commit = commit; this.commit = commit;
setCommitIsSet(true); setCommitIsSet(true);
} }
@ -4589,7 +4589,7 @@ public class RocksDB {
*/ */
public closeTransaction_args(closeTransaction_args other) { public closeTransaction_args(closeTransaction_args other) {
__isset_bitfield = other.__isset_bitfield; __isset_bitfield = other.__isset_bitfield;
this.transactionId = other.transactionId; this.timeoutMs = other.timeoutMs;
this.commit = other.commit; this.commit = other.commit;
} }
@ -4600,33 +4600,33 @@ public class RocksDB {
@Override @Override
public void clear() { public void clear() {
setTransactionIdIsSet(false); setTimeoutMsIsSet(false);
this.transactionId = 0; this.timeoutMs = 0;
setCommitIsSet(false); setCommitIsSet(false);
this.commit = false; this.commit = false;
} }
public long getTransactionId() { public long getTimeoutMs() {
return this.transactionId; return this.timeoutMs;
} }
public closeTransaction_args setTransactionId(long transactionId) { public closeTransaction_args setTimeoutMs(long timeoutMs) {
this.transactionId = transactionId; this.timeoutMs = timeoutMs;
setTransactionIdIsSet(true); setTimeoutMsIsSet(true);
return this; return this;
} }
public void unsetTransactionId() { public void unsetTimeoutMs() {
__isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __TRANSACTIONID_ISSET_ID); __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID);
} }
/** Returns true if field transactionId is set (has been assigned a value) and false otherwise */ /** Returns true if field timeoutMs is set (has been assigned a value) and false otherwise */
public boolean isSetTransactionId() { public boolean isSetTimeoutMs() {
return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __TRANSACTIONID_ISSET_ID); return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID);
} }
public void setTransactionIdIsSet(boolean value) { public void setTimeoutMsIsSet(boolean value) {
__isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __TRANSACTIONID_ISSET_ID, value); __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID, value);
} }
public boolean isCommit() { public boolean isCommit() {
@ -4655,11 +4655,11 @@ public class RocksDB {
@Override @Override
public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) { public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
switch (field) { switch (field) {
case TRANSACTION_ID: case TIMEOUT_MS:
if (value == null) { if (value == null) {
unsetTransactionId(); unsetTimeoutMs();
} else { } else {
setTransactionId((java.lang.Long)value); setTimeoutMs((java.lang.Long)value);
} }
break; break;
@ -4678,8 +4678,8 @@ public class RocksDB {
@Override @Override
public java.lang.Object getFieldValue(_Fields field) { public java.lang.Object getFieldValue(_Fields field) {
switch (field) { switch (field) {
case TRANSACTION_ID: case TIMEOUT_MS:
return getTransactionId(); return getTimeoutMs();
case COMMIT: case COMMIT:
return isCommit(); return isCommit();
@ -4696,8 +4696,8 @@ public class RocksDB {
} }
switch (field) { switch (field) {
case TRANSACTION_ID: case TIMEOUT_MS:
return isSetTransactionId(); return isSetTimeoutMs();
case COMMIT: case COMMIT:
return isSetCommit(); return isSetCommit();
} }
@ -4717,12 +4717,12 @@ public class RocksDB {
if (this == that) if (this == that)
return true; return true;
boolean this_present_transactionId = true; boolean this_present_timeoutMs = true;
boolean that_present_transactionId = true; boolean that_present_timeoutMs = true;
if (this_present_transactionId || that_present_transactionId) { if (this_present_timeoutMs || that_present_timeoutMs) {
if (!(this_present_transactionId && that_present_transactionId)) if (!(this_present_timeoutMs && that_present_timeoutMs))
return false; return false;
if (this.transactionId != that.transactionId) if (this.timeoutMs != that.timeoutMs)
return false; return false;
} }
@ -4742,7 +4742,7 @@ public class RocksDB {
public int hashCode() { public int hashCode() {
int hashCode = 1; int hashCode = 1;
hashCode = hashCode * 8191 + org.apache.thrift.TBaseHelper.hashCode(transactionId); hashCode = hashCode * 8191 + org.apache.thrift.TBaseHelper.hashCode(timeoutMs);
hashCode = hashCode * 8191 + ((commit) ? 131071 : 524287); hashCode = hashCode * 8191 + ((commit) ? 131071 : 524287);
@ -4757,12 +4757,12 @@ public class RocksDB {
int lastComparison = 0; int lastComparison = 0;
lastComparison = java.lang.Boolean.compare(isSetTransactionId(), other.isSetTransactionId()); lastComparison = java.lang.Boolean.compare(isSetTimeoutMs(), other.isSetTimeoutMs());
if (lastComparison != 0) { if (lastComparison != 0) {
return lastComparison; return lastComparison;
} }
if (isSetTransactionId()) { if (isSetTimeoutMs()) {
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.transactionId, other.transactionId); lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.timeoutMs, other.timeoutMs);
if (lastComparison != 0) { if (lastComparison != 0) {
return lastComparison; return lastComparison;
} }
@ -4801,8 +4801,8 @@ public class RocksDB {
java.lang.StringBuilder sb = new java.lang.StringBuilder("closeTransaction_args("); java.lang.StringBuilder sb = new java.lang.StringBuilder("closeTransaction_args(");
boolean first = true; boolean first = true;
sb.append("transactionId:"); sb.append("timeoutMs:");
sb.append(this.transactionId); sb.append(this.timeoutMs);
first = false; first = false;
if (!first) sb.append(", "); if (!first) sb.append(", ");
sb.append("commit:"); sb.append("commit:");
@ -4814,7 +4814,7 @@ public class RocksDB {
public void validate() throws org.apache.thrift.TException { public void validate() throws org.apache.thrift.TException {
// check for required fields // check for required fields
// alas, we cannot check 'transactionId' because it's a primitive and you chose the non-beans generator. // alas, we cannot check 'timeoutMs' 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. // alas, we cannot check 'commit' because it's a primitive and you chose the non-beans generator.
// check for sub-struct validity // check for sub-struct validity
} }
@ -4857,10 +4857,10 @@ public class RocksDB {
break; break;
} }
switch (schemeField.id) { switch (schemeField.id) {
case 1: // TRANSACTION_ID case 1: // TIMEOUT_MS
if (schemeField.type == org.apache.thrift.protocol.TType.I64) { if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
struct.transactionId = iprot.readI64(); struct.timeoutMs = iprot.readI64();
struct.setTransactionIdIsSet(true); struct.setTimeoutMsIsSet(true);
} else { } else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type); org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
} }
@ -4881,8 +4881,8 @@ public class RocksDB {
iprot.readStructEnd(); iprot.readStructEnd();
// check for required fields of primitive type, which can't be checked in the validate method // check for required fields of primitive type, which can't be checked in the validate method
if (!struct.isSetTransactionId()) { if (!struct.isSetTimeoutMs()) {
throw new org.apache.thrift.protocol.TProtocolException("Required field 'transactionId' was not found in serialized data! Struct: " + toString()); throw new org.apache.thrift.protocol.TProtocolException("Required field 'timeoutMs' was not found in serialized data! Struct: " + toString());
} }
if (!struct.isSetCommit()) { if (!struct.isSetCommit()) {
throw new org.apache.thrift.protocol.TProtocolException("Required field 'commit' was not found in serialized data! Struct: " + toString()); 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(); struct.validate();
oprot.writeStructBegin(STRUCT_DESC); oprot.writeStructBegin(STRUCT_DESC);
oprot.writeFieldBegin(TRANSACTION_ID_FIELD_DESC); oprot.writeFieldBegin(TIMEOUT_MS_FIELD_DESC);
oprot.writeI64(struct.transactionId); oprot.writeI64(struct.timeoutMs);
oprot.writeFieldEnd(); oprot.writeFieldEnd();
oprot.writeFieldBegin(COMMIT_FIELD_DESC); oprot.writeFieldBegin(COMMIT_FIELD_DESC);
oprot.writeBool(struct.commit); oprot.writeBool(struct.commit);
@ -4919,15 +4919,15 @@ public class RocksDB {
@Override @Override
public void write(org.apache.thrift.protocol.TProtocol prot, closeTransaction_args struct) throws org.apache.thrift.TException { 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; org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
oprot.writeI64(struct.transactionId); oprot.writeI64(struct.timeoutMs);
oprot.writeBool(struct.commit); oprot.writeBool(struct.commit);
} }
@Override @Override
public void read(org.apache.thrift.protocol.TProtocol prot, closeTransaction_args struct) throws org.apache.thrift.TException { 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; org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
struct.transactionId = iprot.readI64(); struct.timeoutMs = iprot.readI64();
struct.setTransactionIdIsSet(true); struct.setTimeoutMsIsSet(true);
struct.commit = iprot.readBool(); struct.commit = iprot.readBool();
struct.setCommitIsSet(true); struct.setCommitIsSet(true);
} }

2
src/main/java/it/cavallium/rockserver/core/common/api/UpdateBegin.java
View File

@ -1,5 +1,5 @@
/** /**
* Autogenerated by Thrift Compiler (0.20.0) * Autogenerated by Thrift Compiler (0.19.0)
* *
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated * @generated

532
src/main/java/it/cavallium/rockserver/core/server/GrpcServer.java
View File

@ -1,532 +0,0 @@
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();
}
}

11
src/main/java/it/cavallium/rockserver/core/server/ServerBuilder.java
View File

@ -14,7 +14,6 @@ public class ServerBuilder {
private UnixDomainSocketAddress unixAddress; private UnixDomainSocketAddress unixAddress;
private String http2Host; private String http2Host;
private int http2Port; private int http2Port;
private boolean useThrift;
private RocksDBConnection client; private RocksDBConnection client;
public void setUnixSocket(UnixDomainSocketAddress address) { public void setUnixSocket(UnixDomainSocketAddress address) {
@ -30,21 +29,13 @@ public class ServerBuilder {
this.http2Port = port; this.http2Port = port;
} }
public void setUseThrift(boolean useThrift) {
this.useThrift = useThrift;
}
public void setClient(RocksDBConnection client) { public void setClient(RocksDBConnection client) {
this.client = client; this.client = client;
} }
public Server build() throws IOException { public Server build() throws IOException {
if (http2Host != null) { if (http2Host != null) {
if (useThrift) { return new ThriftServer(client, http2Host, http2Port);
return new ThriftServer(client, http2Host, http2Port);
} else {
return new GrpcServer(client, http2Host, http2Port);
}
} else if (unixAddress != null) { } else if (unixAddress != null) {
throw new UnsupportedOperationException("Not implemented: unix socket"); throw new UnsupportedOperationException("Not implemented: unix socket");
} else if (iNetAddress != null) { } else if (iNetAddress != null) {

24
src/main/java/it/cavallium/rockserver/core/server/ThriftServer.java
View File

@ -22,8 +22,6 @@ import java.net.InetSocketAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.ByteOrder; import java.nio.ByteOrder;
import java.util.List; import java.util.List;
import java.util.logging.Logger;
import org.apache.thrift.TException;
import org.apache.thrift.server.TThreadedSelectorServer; import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TNonblockingServerSocket; import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException; import org.apache.thrift.transport.TTransportException;
@ -31,26 +29,19 @@ import org.jetbrains.annotations.NotNull;
public class ThriftServer extends Server { 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 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 { public ThriftServer(RocksDBConnection client, String http2Host, int http2Port) throws IOException {
super(client); super(client);
var handler = new ThriftHandler(this.getClient()); var handler = new ThriftHandler(this.getClient());
try { try {
var serverTransport = new TNonblockingServerSocket(new InetSocketAddress(http2Host, http2Port)); var serverTransport = new TNonblockingServerSocket(new InetSocketAddress(http2Host, http2Port));
this.server = new TThreadedSelectorServer(new TThreadedSelectorServer.Args(serverTransport) var server = new TThreadedSelectorServer(new TThreadedSelectorServer.Args(serverTransport)
.processor(new Processor<>(handler)) .processor(new Processor<>(handler))
); );
var thriftThread = new Thread(server::serve); server.serve();
thriftThread.setName("Thrift server thread");
this.thriftThread = thriftThread;
LOG.info("Thrift RocksDB server is listening at " + http2Host + ":" + http2Port);
} catch (TTransportException e) { } catch (TTransportException e) {
throw new IOException("Can't open server socket", e); throw new IOException("Can't open server socket", e);
} }
@ -140,8 +131,8 @@ public class ThriftServer extends Server {
} }
@Override @Override
public boolean closeTransaction(long transactionId, boolean commit) throws TException { public boolean closeTransaction(long timeoutMs, boolean commit) {
return client.getSyncApi().closeTransaction(transactionId, commit); return client.getSyncApi().closeTransaction(timeoutMs, commit);
} }
@Override @Override
@ -308,11 +299,4 @@ public class ThriftServer extends Server {
} }
} }
} }
@Override
public void close() throws IOException {
LOG.info("Thrift server is shutting down...");
this.server.stop();
super.close();
}
} }

7
src/main/java/module-info.java
View File

@ -9,13 +9,6 @@ module rockserver.core {
requires it.unimi.dsi.fastutil; requires it.unimi.dsi.fastutil;
requires org.apache.thrift; requires org.apache.thrift;
requires org.slf4j; 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.client;
exports it.cavallium.rockserver.core.common; exports it.cavallium.rockserver.core.common;

105
src/main/proto/rocksdb.proto
View File

@ -1,105 +0,0 @@
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);
}

2
src/main/resources/it/cavallium/rockserver/core/resources/rocksdb.thrift
View File

@ -42,7 +42,7 @@ service RocksDB {
i64 openTransaction(1: required i64 timeoutMs), i64 openTransaction(1: required i64 timeoutMs),
bool closeTransaction(1: required i64 transactionId, 2: required bool commit), bool closeTransaction(1: required i64 timeoutMs, 2: required bool commit),
void closeFailedUpdate(1: required i64 updateId), void closeFailedUpdate(1: required i64 updateId),