Merge pull request #180 from tonivade/fix/single-threaded

fix #179

Author: tonivade

src/main/java/com/github/tonivade/resp/RespServerContext.java

@@ -43,8 +43,7 @@ public RespServerContext(String host, int port, CommandSuite commands) {
     this(host, port, commands, nullListener());
   }
 
-  public RespServerContext(String host, int port, CommandSuite commands,
-                           SessionListener sessionListener) {
+  public RespServerContext(String host, int port, CommandSuite commands, SessionListener sessionListener) {
     this.host = checkNonEmpty(host);
     this.port = checkRange(port, 1024, 65535);
     this.commands = checkNonNull(commands);
@@ -106,9 +105,9 @@ Session getSession(String sourceKey, Function<String, Session> factory) {
   void processCommand(Request request) {
     LOGGER.debug("received command: {}", request);
 
-    RespCommand command = getCommand(request.getCommand());
+    var command = getCommand(request.getCommand());
     try {
-      executeOn(execute(command, request))
+      enqueue(Observable.fromCallable(() -> executeCommand(command, request)))
         .subscribe(response -> processResponse(request, response),
                    ex -> LOGGER.error("error executing command: " + request, ex));
     } catch (RuntimeException ex) {
@@ -135,7 +134,7 @@ protected RedisToken executeCommand(RespCommand command, Request request) {
     return command.execute(request);
   }
 
-  protected <T> Observable<T> executeOn(Observable<T> observable) {
+  protected <T> Observable<T> enqueue(Observable<T> observable) {
     return observable.subscribeOn(scheduler);
   }
 
@@ -146,13 +145,6 @@ private void processResponse(Request request, RedisToken token) {
     }
   }
 
-  private Observable<RedisToken> execute(RespCommand command, Request request) {
-    return Observable.create(observer -> {
-      observer.onNext(executeCommand(command, request));
-      observer.onComplete();
-    });
-  }
-
   private void clear() {
     clients.clear();
     state.clear();

src/main/java/com/github/tonivade/resp/command/server/PingCommand.java

@@ -21,8 +21,7 @@ public class PingCommand implements RespCommand {
   public RedisToken execute(Request request) {
     if (request.getLength() > 0) {
       return string(request.getParam(0));
-    } else {
-      return status(PONG);
     }
+    return status(PONG);
   }
 }

src/main/java/com/github/tonivade/resp/protocol/RedisSerializer.java

@@ -18,7 +18,7 @@
 
 public class RedisSerializer {
 
-  private static final Recycler<ByteBufferBuilder> RECYCLER = new Recycler<ByteBufferBuilder>() {
+  private static final Recycler<ByteBufferBuilder> RECYCLER = new Recycler<>() {
     @Override
     protected ByteBufferBuilder newObject(Recycler.Handle<ByteBufferBuilder> handle) {
       return new ByteBufferBuilder(handle);

src/main/java/com/github/tonivade/resp/protocol/SafeString.java

@@ -38,7 +38,7 @@ public SafeString(ByteBuffer buffer) {
   }
 
   public byte[] getBytes() {
-    ByteBuffer copy = buffer.duplicate();
+    var copy = buffer.duplicate();
     byte[] bytes = new byte[copy.remaining()];
     copy.get(bytes);
     return bytes;
@@ -73,7 +73,7 @@ public String toString() {
   }
 
   public String toHexString() {
-    StringBuilder sb = new StringBuilder();
+    var sb = new StringBuilder();
     byte[] bytes = getBytes();
     for (int i = 0; i < bytes.length; i++) {
       int v = bytes[i] & 0xFF;
@@ -99,7 +99,7 @@ public static List<SafeString> safeAsList(String... strings) {
   }
 
   public static SafeString append(SafeString stringA, SafeString stringB) {
-    ByteBuffer byteBuffer = ByteBuffer.allocate(checkNonNull(stringA).length() + checkNonNull(stringB).length());
+    var byteBuffer = ByteBuffer.allocate(checkNonNull(stringA).length() + checkNonNull(stringB).length());
     byteBuffer.put(stringA.getBytes());
     byteBuffer.put(stringB.getBytes());
     ((Buffer) byteBuffer).rewind();

src/test/java/com/github/tonivade/resp/Main.java

@@ -0,0 +1,9 @@
+package com.github.tonivade.resp;
+
+public class Main {
+
+  public static void main(String[] args) {
+    RespServer.builder().build().start();
+  }
+
+}