Add GetAllInRange, RequestReduceRange, use reactive grpc server

pom.xml

@@ -119,6 +119,12 @@
       <groupId>io.grpc</groupId>
       <artifactId>grpc-netty</artifactId>
       <version>${grpc.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>jsr305</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>io.netty</groupId>
@@ -140,11 +146,23 @@
       <groupId>io.grpc</groupId>
       <artifactId>grpc-protobuf</artifactId>
       <version>${grpc.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>jsr305</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-stub</artifactId>
       <version>${grpc.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>jsr305</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.reactivestreams</groupId>
@@ -156,6 +174,18 @@
       <artifactId>reactor-core</artifactId>
       <version>3.6.4</version>
     </dependency>
+    <dependency>
+      <groupId>com.salesforce.servicelibs</groupId>
+      <artifactId>reactor-grpc-stub</artifactId>
+      <version>1.2.4</version>
+      <scope>provided</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>jsr305</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
 
     <dependency>
       <groupId>org.lz4</groupId>
@@ -244,6 +274,15 @@
           <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>
+          <protocPlugins>
+            <protocPlugin>
+              <id>reactor-grpc</id>
+              <groupId>com.salesforce.servicelibs</groupId>
+              <artifactId>reactor-grpc</artifactId>
+              <version>1.2.4</version>
+              <mainClass>com.salesforce.reactorgrpc.ReactorGrpcGenerator</mainClass>
+            </protocPlugin>
+          </protocPlugins>
         </configuration>
         <executions>
           <execution>

src/fatjar/java/module-info.java

@@ -13,7 +13,6 @@ module rockserver.core {
 	requires io.grpc.protobuf;
 	requires io.grpc.stub;
 	requires io.grpc;
-	requires jsr305;
 	requires com.google.common;
 	requires io.grpc.netty;
 	requires io.jstach.rainbowgum;
@@ -31,6 +30,8 @@ module rockserver.core {
 	requires org.reactivestreams;
 	requires io.netty.transport.unix.common;
     requires reactor.core;
+    requires reactor.grpc.stub;
+    requires java.annotation;
 
     exports it.cavallium.rockserver.core.client;
 	exports it.cavallium.rockserver.core.common;

src/library/java/module-info.java

@@ -13,7 +13,6 @@ module rockserver.core {
 	requires io.grpc.protobuf;
 	requires io.grpc.stub;
 	requires io.grpc;
-	requires jsr305;
 	requires com.google.common;
 	requires io.grpc.netty;
 	requires io.netty.common;
@@ -28,6 +27,8 @@ module rockserver.core {
 	requires io.netty.transport.classes.epoll;
 	requires org.reactivestreams;
 	requires io.netty.transport.unix.common;
+	requires reactor.grpc.stub;
+	requires java.annotation;
 
 	exports it.cavallium.rockserver.core.client;
 	exports it.cavallium.rockserver.core.common;

src/main/java/it/cavallium/rockserver/core/client/EmbeddedConnection.java

@@ -88,7 +88,7 @@ public class EmbeddedConnection extends BaseConnection implements RocksDBAPI {
 	}
 
 	@Override
-	public <R, RS> RS requestSync(RocksDBAPICommand<R, RS, ?> req) {
+	public <R, RS, RA> RS requestSync(RocksDBAPICommand<R, RS, RA> req) {
 		return req.handleSync(this);
 	}
 
@@ -176,7 +176,7 @@ public class EmbeddedConnection extends BaseConnection implements RocksDBAPI {
 	}
 
 	@Override
-	public <T> T reduceRange(Arena arena, long transactionId, long columnId, @Nullable Keys startKeysInclusive, @Nullable Keys endKeysExclusive, boolean reverse, RequestType.@NotNull RequestGetRange<? super KV, T> requestType, long timeoutMs) throws RocksDBException {
+	public <T> T reduceRange(Arena arena, long transactionId, long columnId, @Nullable Keys startKeysInclusive, @Nullable Keys endKeysExclusive, boolean reverse, RequestType.@NotNull RequestReduceRange<? super KV, T> requestType, long timeoutMs) throws RocksDBException {
 		return db.reduceRange(arena, transactionId, columnId, startKeysInclusive, endKeysExclusive, reverse, requestType, timeoutMs);
 	}
 }

src/main/java/it/cavallium/rockserver/core/client/GrpcConnection.java

@@ -73,6 +73,7 @@ public class GrpcConnection extends BaseConnection implements RocksDBAPI {
 	private final RocksDBServiceBlockingStub blockingStub;
 	private final RocksDBServiceStub asyncStub;
 	private final RocksDBServiceFutureStub futureStub;
+	private final ReactorRocksDBServiceGrpc.ReactorRocksDBServiceStub reactiveStub;
 	private final URI address;
 
 	private GrpcConnection(String name, SocketAddress socketAddress, URI address) {
@@ -102,6 +103,7 @@ public class GrpcConnection extends BaseConnection implements RocksDBAPI {
 		this.blockingStub = RocksDBServiceGrpc.newBlockingStub(channel);
 		this.asyncStub = RocksDBServiceGrpc.newStub(channel);
 		this.futureStub = RocksDBServiceGrpc.newFutureStub(channel);
+		this.reactiveStub = ReactorRocksDBServiceGrpc.newReactorStub(channel);
 		this.address = address;
 	}
 
@@ -516,7 +518,7 @@ public class GrpcConnection extends BaseConnection implements RocksDBAPI {
 
 	@SuppressWarnings("unchecked")
     @Override
-	public <T> CompletableFuture<T> reduceRangeAsync(Arena arena, long transactionId, long columnId, @NotNull Keys startKeysInclusive, @Nullable Keys endKeysExclusive, boolean reverse, RequestType.RequestGetRange<? super it.cavallium.rockserver.core.common.KV, T> requestType, long timeoutMs) throws RocksDBException {
+	public <T> CompletableFuture<T> reduceRangeAsync(Arena arena, long transactionId, long columnId, @Nullable Keys startKeysInclusive, @Nullable Keys endKeysExclusive, boolean reverse, RequestType.RequestReduceRange<? super it.cavallium.rockserver.core.common.KV, T> requestType, long timeoutMs) throws RocksDBException {
 		var request = GetRangeRequest.newBuilder()
 				.setTransactionId(transactionId)
 				.setColumnId(columnId)
@@ -527,16 +529,28 @@ public class GrpcConnection extends BaseConnection implements RocksDBAPI {
 				.build();
 		return (CompletableFuture<T>) switch (requestType) {
 			case RequestType.RequestGetFirstAndLast<?> _ ->
-					toResponse(this.futureStub.getRangeFirstAndLast(request), result -> new FirstAndLast<>(
+					toResponse(this.futureStub.reduceRangeFirstAndLast(request), result -> new FirstAndLast<>(
 							result.hasFirst() ? mapKV(arena, result.getFirst()) : null,
 							result.hasLast() ? mapKV(arena, result.getLast()) : null
 					));
 		};
 	}
 
+	@SuppressWarnings("unchecked")
     @Override
-	public <T> Publisher<T> getRangeStream(Arena arena, long transactionId, long columnId, @Nullable Keys startKeysInclusive, @Nullable Keys endKeysExclusive, boolean reverse, RequestType.RequestGetRange<? super it.cavallium.rockserver.core.common.KV, T> requestType, long timeoutMs) throws RocksDBException {
+	public <T> Publisher<T> getRangeAsync(Arena arena, long transactionId, long columnId, @Nullable Keys startKeysInclusive, @Nullable Keys endKeysExclusive, boolean reverse, RequestType.RequestGetRange<? super it.cavallium.rockserver.core.common.KV, T> requestType, long timeoutMs) throws RocksDBException {
-		// todo: implement
+		var request = GetRangeRequest.newBuilder()
+				.setTransactionId(transactionId)
+				.setColumnId(columnId)
+				.addAllStartKeysInclusive(mapKeys(startKeysInclusive))
+				.addAllEndKeysExclusive(mapKeys(endKeysExclusive))
+				.setReverse(reverse)
+				.setTimeoutMs(timeoutMs)
+				.build();
+		return (Publisher<T>) switch (requestType) {
+			case RequestType.RequestGetAllInRange<?> _ -> reactiveStub.getAllInRange(request)
+					.map(kv -> mapKV(arena, kv));
+		};
 	}
 
 	private static it.cavallium.rockserver.core.common.Delta<MemorySegment> mapDelta(Delta x) {

src/main/java/it/cavallium/rockserver/core/client/LoggingClient.java

@@ -6,11 +6,12 @@ import it.cavallium.rockserver.core.common.RocksDBSyncAPI;
 import java.io.IOException;
 import java.net.URI;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
+
-import java.util.function.Supplier;
+import org.reactivestreams.Publisher;
-import java.util.logging.Level;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+import reactor.core.publisher.Flux;
 
 public class LoggingClient implements RocksDBConnection {
 
@@ -55,9 +56,9 @@ public class LoggingClient implements RocksDBConnection {
 		}
 
 		@Override
-		public <R> R requestSync(RocksDBAPICommand<R> req) {
+		public <RESULT_ITEM_TYPE, SYNC_RESULT, ASYNC_RESULT> SYNC_RESULT requestSync(RocksDBAPICommand<RESULT_ITEM_TYPE, SYNC_RESULT, ASYNC_RESULT> req) {
 			logger.trace("Request input (sync): {}", req);
-			R result;
+			SYNC_RESULT result;
 			try {
 				result = syncApi.requestSync(req);
 			} catch (Throwable e) {
@@ -77,16 +78,37 @@ public class LoggingClient implements RocksDBConnection {
 			this.asyncApi = asyncApi;
 		}
 
+		@SuppressWarnings("unchecked")
         @Override
-		public <R> CompletableFuture<R> requestAsync(RocksDBAPICommand<R> req) {
+		public <RESULT_ITEM_TYPE, SYNC_RESULT, ASYNC_RESULT> ASYNC_RESULT requestAsync(RocksDBAPICommand<RESULT_ITEM_TYPE, SYNC_RESULT, ASYNC_RESULT> req) {
+			if (!logger.isEnabledForLevel(Level.TRACE)) {
+				return asyncApi.requestAsync(req);
+			} else {
 				logger.trace("Request input (async): {}", req);
-			return asyncApi.requestAsync(req).whenComplete((result, e) -> {
+				var r = asyncApi.requestAsync(req);
+				return switch (req) {
+					case RocksDBAPICommand.RocksDBAPICommandSingle<?> _ ->
+                            (ASYNC_RESULT) ((CompletableFuture<?>) r).whenComplete((result, e) -> {
                                 if (e != null) {
                                     logger.trace("Request failed: {}    Error: {}", req, e.getMessage());
                                 } else {
                                     logger.trace("Request executed: {}    Result: {}", req, result);
                                 }
                             });
+					case RocksDBAPICommand.RocksDBAPICommandStream<?> _ ->
+                            (ASYNC_RESULT) Flux.from((Publisher<?>) r).doOnEach(signal -> {
+                                if (signal.isOnNext()) {
+                                    logger.trace("Request: {}    Partial result: {}", req, signal);
+                                } else if (signal.isOnError()) {
+                                    var e = signal.getThrowable();
+                                    assert e != null;
+									logger.trace("Request failed: {}    Error: {}", req, e.getMessage());
+                                } else if (signal.isOnComplete()) {
+                                    logger.trace("Request executed: {}    Result: terminated successfully", req);
+                                }
+                            });
+				};
+			}
 		}
 	}
 }

src/main/java/it/cavallium/rockserver/core/common/RequestType.java

@@ -1,7 +1,6 @@
 package it.cavallium.rockserver.core.common;
 
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
 
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -99,6 +98,11 @@ public sealed interface RequestType<METHOD_DATA_TYPE, RESULT_TYPE> {
 		return (RequestGetFirstAndLast<T>) RequestGetFirstAndLast.INSTANCE;
 	}
 
+	@SuppressWarnings("unchecked")
+	static <T> RequestGetAllInRange<T> allInRange() {
+		return (RequestGetAllInRange<T>) RequestGetAllInRange.INSTANCE;
+	}
+
 	@SuppressWarnings("unchecked")
 	static <T> RequestNothing<T> none() {
 		return (RequestNothing<T>) RequestNothing.INSTANCE;
@@ -110,6 +114,8 @@ public sealed interface RequestType<METHOD_DATA_TYPE, RESULT_TYPE> {
 
 	sealed interface RequestGet<T, U> extends RequestType<T, U> {}
 
+	sealed interface RequestReduceRange<T, U> extends RequestType<T, U> {}
+
 	sealed interface RequestGetRange<T, U> extends RequestType<T, U> {}
 
 	sealed interface RequestIterate<T, U> extends RequestType<T, U> {}
@@ -205,7 +211,7 @@ public sealed interface RequestType<METHOD_DATA_TYPE, RESULT_TYPE> {
 		}
 	}
 
-	record RequestGetFirstAndLast<T>() implements RequestGetRange<T, FirstAndLast<T>> {
+	record RequestGetFirstAndLast<T>() implements RequestReduceRange<T, FirstAndLast<T>> {
 
 		private static final RequestGetFirstAndLast<Object> INSTANCE = new RequestGetFirstAndLast<>();
 
@@ -214,4 +220,14 @@ public sealed interface RequestType<METHOD_DATA_TYPE, RESULT_TYPE> {
 			return RequestTypeId.FIRST_AND_LAST;
 		}
 	}
+
+	record RequestGetAllInRange<T>() implements RequestGetRange<T, FirstAndLast<T>> {
+
+		private static final RequestGetAllInRange<Object> INSTANCE = new RequestGetAllInRange<>();
+
+		@Override
+		public RequestTypeId getRequestTypeId() {
+			return RequestTypeId.FIRST_AND_LAST;
+		}
+	}
 }

src/main/java/it/cavallium/rockserver/core/common/RocksDBAPICommand.java

@@ -410,7 +410,7 @@ public sealed interface RocksDBAPICommand<RESULT_ITEM_TYPE, SYNC_RESULT, ASYNC_R
 							  @Nullable Keys startKeysInclusive,
 							  @Nullable Keys endKeysExclusive,
 							  boolean reverse,
-							  RequestType.RequestGetRange<? super KV, T> requestType,
+							  RequestType.RequestReduceRange<? super KV, T> requestType,
 							  long timeoutMs) implements RocksDBAPICommandSingle<T> {
 
 			@Override

src/main/java/it/cavallium/rockserver/core/common/RocksDBAsyncAPI.java

@@ -17,6 +17,7 @@ import it.cavallium.rockserver.core.common.RocksDBAPICommand.RocksDBAPICommandSi
 import it.cavallium.rockserver.core.common.RocksDBAPICommand.RocksDBAPICommandSingle.PutBatch;
 import it.cavallium.rockserver.core.common.RocksDBAPICommand.RocksDBAPICommandSingle.SeekTo;
 import it.cavallium.rockserver.core.common.RocksDBAPICommand.RocksDBAPICommandSingle.Subsequent;
+import it.cavallium.rockserver.core.common.RocksDBAPICommand.RocksDBAPICommandStream.GetRange;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
 import java.util.List;
@@ -137,7 +138,7 @@ public interface RocksDBAsyncAPI extends RocksDBAsyncAPIRequestHandler {
 													  @Nullable Keys startKeysInclusive,
 													  @Nullable Keys endKeysExclusive,
 													  boolean reverse,
-													  RequestType.RequestGetRange<? super KV, T> requestType,
+													  RequestType.RequestReduceRange<? super KV, T> requestType,
 													  long timeoutMs) throws RocksDBException {
 		return requestAsync(new ReduceRange<>(arena,
 				transactionId,

src/main/java/it/cavallium/rockserver/core/common/RocksDBSyncAPI.java

@@ -131,7 +131,7 @@ public interface RocksDBSyncAPI extends RocksDBSyncAPIRequestHandler {
 							  @Nullable Keys startKeysInclusive,
 							  @Nullable Keys endKeysExclusive,
 							  boolean reverse,
-							  @NotNull RequestType.RequestGetRange<? super KV, T> requestType,
+							  @NotNull RequestType.RequestReduceRange<? super KV, T> requestType,
 							  long timeoutMs) throws RocksDBException {
 		return requestSync(new ReduceRange<>(arena, transactionId, columnId, startKeysInclusive, endKeysExclusive, reverse, requestType, timeoutMs));
 	}

src/main/java/it/cavallium/rockserver/core/impl/EmbeddedDB.java

@@ -955,7 +955,7 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
 							 @Nullable Keys startKeysInclusive,
 							 @Nullable Keys endKeysExclusive,
 							 boolean reverse,
-							 RequestType.@NotNull RequestGetRange<? super KV, T> requestType,
+							 RequestType.@NotNull RequestReduceRange<? super KV, T> requestType,
 							 long timeoutMs) throws it.cavallium.rockserver.core.common.RocksDBException {
 		ops.beginOp();
 		try {

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

@@ -48,8 +48,10 @@ import java.io.IOException;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
 import java.net.SocketAddress;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -62,6 +64,11 @@ import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 
 public class GrpcServer extends Server {
 
@@ -69,7 +76,7 @@ public class GrpcServer extends Server {
 
 	private final GrpcServerImpl grpc;
 	private final EventLoopGroup elg;
-	private final ExecutorService executor;
+	private final Scheduler executor;
 	private final io.grpc.Server server;
 
 	public GrpcServer(RocksDBConnection client, SocketAddress socketAddress) throws IOException {
@@ -85,7 +92,7 @@ public class GrpcServer extends Server {
 			channelType = NioServerSocketChannel.class;
 		}
 		this.elg = elg;
-		this.executor = Executors.newWorkStealingPool(Runtime.getRuntime().availableProcessors() * 2);
+		this.executor = Schedulers.newBoundedElastic(Runtime.getRuntime().availableProcessors() * 2, Schedulers.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE, "server-db-executor");
 		this.server = NettyServerBuilder
 				.forAddress(socketAddress)
 				.bossEventLoopGroup(elg)
@@ -104,7 +111,7 @@ public class GrpcServer extends Server {
 		server.start();
 	}
 
-	private final class GrpcServerImpl extends RocksDBServiceImplBase {
+	private final class GrpcServerImpl extends ReactorRocksDBServiceGrpc.RocksDBServiceImplBase {
 
 		private final RocksDBAsyncAPI asyncApi;
         private final RocksDBSyncAPI api;
@@ -116,93 +123,58 @@ public class GrpcServer extends Server {
 
 		// functions
 
+
 		@Override
-		public void openTransaction(OpenTransactionRequest request,
+		public Mono<OpenTransactionResponse> openTransaction(OpenTransactionRequest request) {
-				StreamObserver<OpenTransactionResponse> responseObserver) {
+			return executeSync(() -> {
-			executor.execute(() -> {
-				try {
 				var txId = api.openTransaction(request.getTimeoutMs());
-					responseObserver.onNext(OpenTransactionResponse.newBuilder().setTransactionId(txId).build());
+				return OpenTransactionResponse.newBuilder().setTransactionId(txId).build();
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void closeTransaction(CloseTransactionRequest request,
+		public Mono<CloseTransactionResponse> closeTransaction(CloseTransactionRequest request) {
-				StreamObserver<CloseTransactionResponse> responseObserver) {
+			return executeSync(() -> {
-			executor.execute(() -> {
-				try {
 				var committed = api.closeTransaction(request.getTransactionId(), request.getCommit());
-					var response = CloseTransactionResponse.newBuilder().setSuccessful(committed).build();
+                return CloseTransactionResponse.newBuilder().setSuccessful(committed).build();
-					responseObserver.onNext(response);
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void closeFailedUpdate(CloseFailedUpdateRequest request, StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> closeFailedUpdate(CloseFailedUpdateRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				api.closeFailedUpdate(request.getUpdateId());
-					responseObserver.onNext(Empty.getDefaultInstance());
+				return Empty.getDefaultInstance();
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void createColumn(CreateColumnRequest request, StreamObserver<CreateColumnResponse> responseObserver) {
+		public Mono<CreateColumnResponse> createColumn(CreateColumnRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				var colId = api.createColumn(request.getName(), mapColumnSchema(request.getSchema()));
-					var response = CreateColumnResponse.newBuilder().setColumnId(colId).build();
+				return CreateColumnResponse.newBuilder().setColumnId(colId).build();
-					responseObserver.onNext(response);
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void deleteColumn(DeleteColumnRequest request, StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> deleteColumn(DeleteColumnRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				api.deleteColumn(request.getColumnId());
-					responseObserver.onNext(Empty.getDefaultInstance());
+				return Empty.getDefaultInstance();
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void getColumnId(GetColumnIdRequest request, StreamObserver<GetColumnIdResponse> responseObserver) {
+		public Mono<GetColumnIdResponse> getColumnId(GetColumnIdRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				var colId = api.getColumnId(request.getName());
-					var response = GetColumnIdResponse.newBuilder().setColumnId(colId).build();
+				return GetColumnIdResponse.newBuilder().setColumnId(colId).build();
-					responseObserver.onNext(response);
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void put(PutRequest request, StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> put(PutRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					api.put(arena,
 							request.getTransactionOrUpdateId(),
@@ -212,209 +184,81 @@ public class GrpcServer extends Server {
 							new RequestNothing<>()
 					);
 				}
-					responseObserver.onNext(Empty.getDefaultInstance());
+				return Empty.getDefaultInstance();
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public StreamObserver<PutBatchRequest> putBatch(StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> putBatch(Flux<PutBatchRequest> request) {
-			final ServerCallStreamObserver<Empty> serverCallStreamObserver =
+			return request.switchOnFirst((firstSignal, nextRequests) -> {
-					(ServerCallStreamObserver<Empty>) responseObserver;
+				if (firstSignal.isOnNext()) {
-			serverCallStreamObserver.disableAutoRequest();
+					var firstValue = firstSignal.get();
-			serverCallStreamObserver.request(1);
+                    assert firstValue != null;
-			var requestObserver = new StreamObserver<PutBatchRequest>() {
+                    if (!firstValue.hasInitialRequest()) {
-				enum State {
+						return Mono.<Empty>error(RocksDBException.of(
-					BEFORE_INITIAL_REQUEST,
+								RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "Missing initial request"));
-					RECEIVING_DATA,
-					RECEIVED_ALL
 					}
-				private final ExecutorService sstExecutor = Executors.newSingleThreadExecutor();
+					var initialRequest = firstValue.getInitialRequest();
-				final AtomicInteger pendingRequests = new AtomicInteger();
+					var mode = switch (initialRequest.getMode()) {
-				State state = State.BEFORE_INITIAL_REQUEST;
-				private PutBatchInitialRequest initialRequest;
-				private Subscriber<? super KVBatch> putBatchInputsSubscriber;
-				@Override
-				public void onNext(PutBatchRequest putBatchRequest) {
-					if (state == State.BEFORE_INITIAL_REQUEST) {
-						if (!putBatchRequest.hasInitialRequest()) {
-							serverCallStreamObserver.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "Missing initial request"));
-						}
-
-						initialRequest = putBatchRequest.getInitialRequest();
-
-						try {
-							asyncApi.putBatchAsync(initialRequest.getColumnId(),
-									subscriber2 -> {
-										putBatchInputsSubscriber = subscriber2;
-										subscriber2.onSubscribe(new Subscription() {
-											@Override
-											public void request(long l) {
-												serverCallStreamObserver.request(Math.toIntExact(l));
-											}
-
-											@Override
-											public void cancel() {
-												serverCallStreamObserver.onError(new IOException("Cancelled"));
-
-											}
-										});
-									},
-									switch (initialRequest.getMode()) {
 						case WRITE_BATCH -> PutBatchMode.WRITE_BATCH;
 						case WRITE_BATCH_NO_WAL -> PutBatchMode.WRITE_BATCH_NO_WAL;
 						case SST_INGESTION -> PutBatchMode.SST_INGESTION;
 						case SST_INGEST_BEHIND -> PutBatchMode.SST_INGEST_BEHIND;
 						case UNRECOGNIZED -> throw new UnsupportedOperationException("Unrecognized request \"mode\"");
-									}
-							).whenComplete((_, ex) -> {
-								if (ex != null) {
-									handleError(serverCallStreamObserver, ex);
-								} else {
-									serverCallStreamObserver.onNext(Empty.getDefaultInstance());
-									serverCallStreamObserver.onCompleted();
-								}
-							});
-						} catch (Throwable ex) {
-							handleError(serverCallStreamObserver, ex);
-						}
-						state = State.RECEIVING_DATA;
-					} else if (state == State.RECEIVING_DATA) {
-						pendingRequests.incrementAndGet();
-						var kvBatch = putBatchRequest.getData();
-						sstExecutor.execute(() -> {
-							try {
-								try (var arena = Arena.ofConfined()) {
-									putBatchInputsSubscriber.onNext(mapKVBatch(arena, kvBatch.getEntriesCount(), kvBatch::getEntries));
-								}
-								checkCompleted(true);
-							} catch (Throwable ex) {
-								putBatchInputsSubscriber.onError(ex);
-							}
-						});
-					} else {
-						serverCallStreamObserver.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "Invalid request"));
-					}
-				}
-
-				@Override
-				public void onError(Throwable throwable) {
-					sstExecutor.execute(() -> {
-						state = State.RECEIVED_ALL;
-						doFinally();
-						if (putBatchInputsSubscriber != null) {
-							putBatchInputsSubscriber.onError(throwable);
-						} else {
-							serverCallStreamObserver.onError(throwable);
-						}
-					});
-				}
-
-				@Override
-				public void onCompleted() {
-					sstExecutor.execute(() -> {
-						if (state == State.BEFORE_INITIAL_REQUEST) {
-							serverCallStreamObserver.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "Missing initial request"));
-						} else if (state == State.RECEIVING_DATA) {
-							state = State.RECEIVED_ALL;
-							checkCompleted(false);
-						} else {
-							putBatchInputsSubscriber.onError(RocksDBException.of(RocksDBException.RocksDBErrorType.PUT_UNKNOWN_ERROR, "Unknown state during onComplete: " + state));
-						}
-					});
-				}
-
-				private void checkCompleted(boolean requestDone) {
-					if ((requestDone ? pendingRequests.decrementAndGet() : pendingRequests.get()) == 0
-							&& state == State.RECEIVED_ALL) {
-						doFinally();
-						putBatchInputsSubscriber.onComplete();
-					}
-				}
-
-				private void doFinally() {
-					sstExecutor.shutdown();
-				}
 					};
-			return requestObserver;
+
+					var batches = nextRequests.map(putBatchRequest -> {
+						var batch = putBatchRequest.getData();
+						return mapKVBatch(Arena.ofAuto(), batch.getEntriesCount(), batch::getEntries);
+					});
+
+					return Mono.fromFuture(asyncApi.putBatchAsync(initialRequest.getColumnId(), batches, mode));
+				} else if (firstSignal.isOnComplete()) {
+					return Mono.just(RocksDBException.of(
+							RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "No initial request"));
+				} else {
+					return nextRequests;
+				}
+			}).then(Mono.just(Empty.getDefaultInstance()));
 		}
 
 		@Override
-		public StreamObserver<PutMultiRequest> putMulti(StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> putMulti(Flux<PutMultiRequest> request) {
-			return new StreamObserver<>() {
+			return request.switchOnFirst((firstSignal, nextRequests) -> {
-				private boolean initialRequestDone = false;
+				if (firstSignal.isOnNext()) {
-				private long requestsCount = 0;
+					var firstValue = firstSignal.get();
-				private boolean requestsCountFinalized;
+					assert firstValue != null;
-				private final AtomicLong processedRequestsCount = new AtomicLong();
+					if (!firstValue.hasInitialRequest()) {
-				private PutMultiInitialRequest initialRequest;
+						return Mono.<Empty>error(RocksDBException.of(
+								RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "Missing initial request"));
+					}
+					var initialRequest = firstValue.getInitialRequest();
 
-				@Override
+                    return nextRequests
-				public void onNext(PutMultiRequest request) {
+                            .publishOn(executor)
-					switch (request.getPutMultiRequestTypeCase()) {
+                            .doOnNext(putRequest -> {
-						case INITIALREQUEST -> {
+                                var data = putRequest.getData();
-							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;
-							executor.execute(() -> {
-								try {
                                 try (var arena = Arena.ofConfined()) {
                                     api.put(arena,
                                             initialRequest.getTransactionOrUpdateId(),
                                             initialRequest.getColumnId(),
-                                                mapKeys(arena, request.getData().getKeysCount(), request.getData()::getKeys),
+                                            mapKeys(arena, data.getKeysCount(), data::getKeys),
-                                                toMemorySegment(arena, request.getData().getValue()),
+                                            toMemorySegment(arena, data.getValue()),
                                             new RequestNothing<>());
                                 }
-                                } catch (RocksDBException ex) {
-									handleError(responseObserver, ex);
-									return;
-								}
-
-								var newProcessedRequestCount = processedRequestsCount.incrementAndGet();
-								if (requestsCountFinalized) {
-									if (newProcessedRequestCount == requestsCount) {
-										responseObserver.onNext(Empty.getDefaultInstance());
-										responseObserver.onCompleted();
-									}
-								}
                             });
+				} else if (firstSignal.isOnComplete()) {
+					return Mono.just(RocksDBException.of(
+							RocksDBException.RocksDBErrorType.PUT_INVALID_REQUEST, "No initial request"));
+				} else {
+					return nextRequests;
 				}
-						case null, default ->
+			}).then(Mono.just(Empty.getDefaultInstance()));
-								throw new UnsupportedOperationException("Unsupported operation: "
-										+ request.getPutMultiRequestTypeCase());
-					}
 		}
 
 		@Override
-				public void onError(Throwable t) {
+		public Mono<Previous> putGetPrevious(PutRequest request) {
-					responseObserver.onError(t);
+			return executeSync(() -> {
-				}
-
-				@Override
-				public void onCompleted() {
-					requestsCountFinalized = true;
-					if (requestsCount == 0) {
-						responseObserver.onCompleted();
-					}
-				}
-			};
-		}
-
-		@Override
-		public void putGetPrevious(PutRequest request, StreamObserver<Previous> responseObserver) {
-			executor.execute(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var prev = api.put(arena,
 							request.getTransactionOrUpdateId(),
@@ -427,20 +271,14 @@ public class GrpcServer extends Server {
 					if (prev != null) {
 						prevBuilder.setPrevious(ByteString.copyFrom(prev.asByteBuffer()));
 					}
-                        var response = prevBuilder.build();
+                    return prevBuilder.build();
-						responseObserver.onNext(response);
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void putGetDelta(PutRequest request, StreamObserver<Delta> responseObserver) {
+		public Mono<Delta> putGetDelta(PutRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var delta = api.put(arena,
 							request.getTransactionOrUpdateId(),
@@ -456,20 +294,14 @@ public class GrpcServer extends Server {
 					if (delta.current() != null) {
 						deltaBuilder.setCurrent(ByteString.copyFrom(delta.current().asByteBuffer()));
 					}
-                        var response = deltaBuilder.build();
+					return deltaBuilder.build();
-                        responseObserver.onNext(response);
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void putGetChanged(PutRequest request, StreamObserver<Changed> responseObserver) {
+		public Mono<Changed> putGetChanged(PutRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
                     try (var arena = Arena.ofConfined()) {
                         var changed = api.put(arena,
                                 request.getTransactionOrUpdateId(),
@@ -478,20 +310,14 @@ public class GrpcServer extends Server {
                                 toMemorySegment(arena, request.getData().getValue()),
                                 new RequestChanged<>()
                         );
-                        var response = Changed.newBuilder().setChanged(changed).build();
+                        return Changed.newBuilder().setChanged(changed).build();
-                        responseObserver.onNext(response);
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
                     }
 			});
 		}
 
 		@Override
-		public void putGetPreviousPresence(PutRequest request, StreamObserver<PreviousPresence> responseObserver) {
+		public Mono<PreviousPresence> putGetPreviousPresence(PutRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var present = api.put(arena,
 							request.getTransactionOrUpdateId(),
@@ -500,20 +326,14 @@ public class GrpcServer extends Server {
 							toMemorySegment(arena, request.getData().getValue()),
 							new RequestPreviousPresence<>()
 					);
-                        var response = PreviousPresence.newBuilder().setPresent(present).build();
+					return PreviousPresence.newBuilder().setPresent(present).build();
-                        responseObserver.onNext(response);
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void get(GetRequest request, StreamObserver<GetResponse> responseObserver) {
+		public Mono<GetResponse> get(GetRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var current = api.get(arena,
 							request.getTransactionOrUpdateId(),
@@ -525,20 +345,14 @@ public class GrpcServer extends Server {
 					if (current != null) {
 						responseBuilder.setValue(ByteString.copyFrom(current.asByteBuffer()));
 					}
-                        var response = responseBuilder.build();
+					return responseBuilder.build();
-                        responseObserver.onNext(response);
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void getForUpdate(GetRequest request, StreamObserver<UpdateBegin> responseObserver) {
+		public Mono<UpdateBegin> getForUpdate(GetRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var forUpdate = api.get(arena,
 							request.getTransactionOrUpdateId(),
@@ -551,20 +365,14 @@ public class GrpcServer extends Server {
 					if (forUpdate.previous() != null) {
 						responseBuilder.setPrevious(ByteString.copyFrom(forUpdate.previous().asByteBuffer()));
 					}
-                        var response = responseBuilder.build();
+					return responseBuilder.build();
-                        responseObserver.onNext(response);
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void exists(GetRequest request, StreamObserver<PreviousPresence> responseObserver) {
+		public Mono<PreviousPresence> exists(GetRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var exists = api.get(arena,
 							request.getTransactionOrUpdateId(),
@@ -572,19 +380,14 @@ public class GrpcServer extends Server {
 							mapKeys(arena, request.getKeysCount(), request::getKeys),
 							new RequestExists<>()
 					);
-                        responseObserver.onNext(PreviousPresence.newBuilder().setPresent(exists).build());
+					return PreviousPresence.newBuilder().setPresent(exists).build();
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void openIterator(OpenIteratorRequest request, StreamObserver<OpenIteratorResponse> responseObserver) {
+		public Mono<OpenIteratorResponse> openIterator(OpenIteratorRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var iteratorId = api.openIterator(arena,
 							request.getTransactionId(),
@@ -594,84 +397,58 @@ public class GrpcServer extends Server {
 							request.getReverse(),
 							request.getTimeoutMs()
 					);
-                        responseObserver.onNext(OpenIteratorResponse.newBuilder().setIteratorId(iteratorId).build());
+					return OpenIteratorResponse.newBuilder().setIteratorId(iteratorId).build();
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void closeIterator(CloseIteratorRequest request, StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> closeIterator(CloseIteratorRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				api.closeIterator(request.getIteratorId());
-					responseObserver.onNext(Empty.getDefaultInstance());
+				return Empty.getDefaultInstance();
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void seekTo(SeekToRequest request, StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> seekTo(SeekToRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					api.seekTo(arena, request.getIterationId(), mapKeys(arena, request.getKeysCount(), request::getKeys));
 				}
-					responseObserver.onNext(Empty.getDefaultInstance());
+				return Empty.getDefaultInstance();
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void subsequent(SubsequentRequest request, StreamObserver<Empty> responseObserver) {
+		public Mono<Empty> subsequent(SubsequentRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					api.subsequent(arena, request.getIterationId(),
 							request.getSkipCount(),
 							request.getTakeCount(),
 							new RequestNothing<>());
 				}
-					responseObserver.onNext(Empty.getDefaultInstance());
+				return Empty.getDefaultInstance();
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
-				}
 			});
 		}
 
 		@Override
-		public void subsequentExists(SubsequentRequest request, StreamObserver<PreviousPresence> responseObserver) {
+		public Mono<PreviousPresence> subsequentExists(SubsequentRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					var exists = api.subsequent(arena, request.getIterationId(),
 							request.getSkipCount(),
 							request.getTakeCount(),
 							new RequestExists<>());
-                        var response = PreviousPresence.newBuilder().setPresent(exists).build();
+					return PreviousPresence.newBuilder().setPresent(exists).build();
-                        responseObserver.onNext(response);
-                    }
-					responseObserver.onCompleted();
-                } catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
 
 		@Override
-		public void subsequentMultiGet(SubsequentRequest request, StreamObserver<KV> responseObserver) {
+		public Flux<KV> subsequentMultiGet(SubsequentRequest request) {
-			executor.execute(() -> {
+			return Flux.create(emitter -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					int pageIndex = 0;
 					final long pageSize = 16L;
@@ -685,7 +462,7 @@ public class GrpcServer extends Server {
 						for (MemorySegment entry : response) {
 							Keys keys = null; // todo: implement
 							MemorySegment value = entry;
-                                responseObserver.onNext(KV.newBuilder()
+							emitter.next(KV.newBuilder()
 									.addAllKeys(null) // todo: implement
 									.setValue(ByteString.copyFrom(value.asByteBuffer()))
 									.build());
@@ -693,17 +470,13 @@ public class GrpcServer extends Server {
 						pageIndex++;
 					}
 				}
-					responseObserver.onCompleted();
+				emitter.complete();
-                } catch (Throwable ex) {
+			}, FluxSink.OverflowStrategy.BUFFER);
-					handleError(responseObserver, ex);
-				}
-			});
 		}
 
 		@Override
-		public void getRangeFirstAndLast(GetRangeRequest request, StreamObserver<FirstAndLast> responseObserver) {
+		public Mono<FirstAndLast> reduceRangeFirstAndLast(GetRangeRequest request) {
-			executor.execute(() -> {
+			return executeSync(() -> {
-				try {
 				try (var arena = Arena.ofConfined()) {
 					it.cavallium.rockserver.core.common.FirstAndLast<it.cavallium.rockserver.core.common.KV> firstAndLast
 							= api.reduceRange(arena,
@@ -715,14 +488,10 @@ public class GrpcServer extends Server {
 							RequestType.firstAndLast(),
 							request.getTimeoutMs()
 					);
-						responseObserver.onNext(FirstAndLast.newBuilder()
+					return FirstAndLast.newBuilder()
 							.setFirst(unmapKV(firstAndLast.first()))
 							.setLast(unmapKV(firstAndLast.last()))
-								.build());
+							.build();
-					}
-					responseObserver.onCompleted();
-				} catch (Throwable ex) {
-					handleError(responseObserver, ex);
 				}
 			});
 		}
@@ -737,6 +506,12 @@ public class GrpcServer extends Server {
 			}
 		}
 
+		// utils
+
+		private <T> Mono<T> executeSync(Callable<T> callable) {
+			return Mono.fromCallable(callable).subscribeOn(executor);
+		}
+
 		// mappers
 
 		private static KV unmapKV(it.cavallium.rockserver.core.common.KV kv) {
@@ -859,7 +634,11 @@ public class GrpcServer extends Server {
 			throw new RuntimeException(e);
 		}
 		elg.close();
-		executor.close();
+		executor.disposeGracefully().timeout(Duration.ofMinutes(2)).onErrorResume(ex -> {
+			LOG.error("Grpc server executor shutdown timed out, terminating...", ex);
+			executor.dispose();
+			return Mono.empty();
+		}).block();
 		super.close();
 	}
 }

src/main/proto/rocksdb.proto

@@ -124,6 +124,6 @@ service RocksDBService {
   rpc subsequent(SubsequentRequest) returns (google.protobuf.Empty);
   rpc subsequentExists(SubsequentRequest) returns (PreviousPresence);
   rpc subsequentMultiGet(SubsequentRequest) returns (stream KV);
-  rpc getRangeFirstAndLast(GetRangeRequest) returns (FirstAndLast);
+  rpc reduceRangeFirstAndLast(GetRangeRequest) returns (FirstAndLast);
-  rpc getRangeStream(GetRangeRequest) returns (stream KV);
+  rpc getAllInRange(GetRangeRequest) returns (stream KV);
 }

src/native/java/module-info.java

@@ -13,7 +13,6 @@ module rockserver.core {
 	requires io.grpc.protobuf;
 	requires io.grpc.stub;
 	requires io.grpc;
-	requires jsr305;
 	requires com.google.common;
 	requires io.grpc.netty;
 	requires io.jstach.rainbowgum;
@@ -30,6 +29,8 @@ module rockserver.core {
 	requires io.netty.transport.classes.epoll;
 	requires org.reactivestreams;
 	requires io.netty.transport.unix.common;
+	requires reactor.grpc.stub;
+	requires java.annotation;
 
 	exports it.cavallium.rockserver.core.client;
 	exports it.cavallium.rockserver.core.common;