Implement GRPC server

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>
 

src/main/java/it/cavallium/rockserver/core/Main.java

@@ -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();
+		var thriftServerBuilder = new it.cavallium.rockserver.core.server.ServerBuilder();
-		switch (listenUrlScheme) {
+		buildServerAddress(thriftServerBuilder, thriftListenUrl, true);
-			case "unix" -> serverBuilder.setUnixSocket(UnixDomainSocketAddress.of(Path.of(listenUrl.getPath())));
+		var grpcServerBuilder = new it.cavallium.rockserver.core.server.ServerBuilder();
-			case "http" -> serverBuilder.setHttpAddress(listenUrl.getHost(), Utils.parsePort(listenUrl));
+		buildServerAddress(grpcServerBuilder, grpcListenUrl, false);
-			case "rocksdb" -> serverBuilder.setAddress(Utils.parseHostAndPort(listenUrl));
-			case null, default -> throw new IllegalArgumentException("Invalid scheme \"" + listenUrlScheme + "\" for listen url: " + listenUrl);
-		}
 
 		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);
+		}
+	}
 }

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

@@ -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

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

@@ -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

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

@@ -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

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

@@ -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

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

@@ -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

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

@@ -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
+          case 1: // TRANSACTION_ID
-            return TIMEOUT_MS;
+            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;
+      this.transactionId = transactionId;
-      setTimeoutMsIsSet(true);
+      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);
+      setTransactionIdIsSet(false);
-      this.timeoutMs = 0;
+      this.transactionId = 0;
       setCommitIsSet(false);
       this.commit = false;
     }
 
-    public long getTimeoutMs() {
+    public long getTransactionId() {
-      return this.timeoutMs;
+      return this.transactionId;
     }
 
-    public closeTransaction_args setTimeoutMs(long timeoutMs) {
+    public closeTransaction_args setTransactionId(long transactionId) {
-      this.timeoutMs = timeoutMs;
+      this.transactionId = transactionId;
-      setTimeoutMsIsSet(true);
+      setTransactionIdIsSet(true);
       return this;
     }
 
-    public void unsetTimeoutMs() {
+    public void unsetTransactionId() {
-      __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID);
+      __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 */
+    /** Returns true if field transactionId is set (has been assigned a value) and false otherwise */
-    public boolean isSetTimeoutMs() {
+    public boolean isSetTransactionId() {
-      return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID);
+      return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __TRANSACTIONID_ISSET_ID);
     }
 
-    public void setTimeoutMsIsSet(boolean value) {
+    public void setTransactionIdIsSet(boolean value) {
-      __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __TIMEOUTMS_ISSET_ID, 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:
+      case TRANSACTION_ID:
-        return getTimeoutMs();
+        return getTransactionId();
 
       case COMMIT:
         return isCommit();
@@ -4696,8 +4696,8 @@ public class RocksDB {
       }
 
       switch (field) {
-      case TIMEOUT_MS:
+      case TRANSACTION_ID:
-        return isSetTimeoutMs();
+        return isSetTransactionId();
       case COMMIT:
         return isSetCommit();
       }
@@ -4717,12 +4717,12 @@ public class RocksDB {
       if (this == that)
         return true;
 
-      boolean this_present_timeoutMs = true;
+      boolean this_present_transactionId = true;
-      boolean that_present_timeoutMs = true;
+      boolean that_present_transactionId = true;
-      if (this_present_timeoutMs || that_present_timeoutMs) {
+      if (this_present_transactionId || that_present_transactionId) {
-        if (!(this_present_timeoutMs && that_present_timeoutMs))
+        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()) {
+      if (isSetTransactionId()) {
-        lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.timeoutMs, other.timeoutMs);
+        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("transactionId:");
-      sb.append(this.timeoutMs);
+      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.transactionId = iprot.readI64();
-                struct.setTimeoutMsIsSet(true);
+                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()) {
+        if (!struct.isSetTransactionId()) {
-          throw new org.apache.thrift.protocol.TProtocolException("Required field 'timeoutMs' was not found in serialized data! Struct: " + toString());
+          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.writeFieldBegin(TRANSACTION_ID_FIELD_DESC);
-        oprot.writeI64(struct.timeoutMs);
+        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.transactionId = iprot.readI64();
-        struct.setTimeoutMsIsSet(true);
+        struct.setTransactionIdIsSet(true);
         struct.commit = iprot.readBool();
         struct.setCommitIsSet(true);
       }

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

@@ -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

src/main/java/it/cavallium/rockserver/core/server/GrpcServer.java

@@ -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();
+	}
+}

src/main/java/it/cavallium/rockserver/core/server/ServerBuilder.java

@@ -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) {

src/main/java/it/cavallium/rockserver/core/server/ThriftServer.java

@@ -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) {
+		public boolean closeTransaction(long transactionId, boolean commit) throws TException {
-			return client.getSyncApi().closeTransaction(timeoutMs, commit);
+			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();
+	}
 }

src/main/java/module-info.java

@@ -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;

src/main/proto/rocksdb.proto

@@ -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);
+}

src/main/resources/it/cavallium/rockserver/core/resources/rocksdb.thrift

@@ -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),