Need help with a load balancer server translating REST requests to RPC servers

[ThusSpokeBieu]

I’m trying to create a server that acts as a load balancer. It should receive requests in the REST pattern and translate them to two other RPC servers, using the round-robin strategy.

However, I’m having difficulties getting it to work. It might be something I’m doing wrong.

I’ve made two different attempts to make the call with RPCClient:

• The first is to give a “start()”, and handle the call asynchronously after the start for each received Http request. The start() method seems to call a strategy exchange method, which results in a Nullpointer error because “newStrategy” is null.
• The second is simply to use client.sendRequest(). But the call hangs until it times out.

All instances start to work, and the client can connecto to both servers. But, I can't send any request.

The RPC servers are instances of the same code running on different ports.

Server code:

public class MyRpcServer extends Launcher {
  private static int PORT = 8080;
  public static final String PROPERTIES_FILE = "rpc-server.properties";

  @Inject RpcServer rpcServer;

  @SerializeRecord
  public static record FooRequest(int id) {}

  @SerializeRecord
  public static record FooResponse(int status, String body) {}

  @Provides
  NioReactor reactor() {
    return Eventloop.create();
  }

  @Provides
  Config config() {
    return Config.create()
        .overrideWith(ofClassPathProperties(PROPERTIES_FILE, true))
        .overrideWith(ofSystemProperties("config"));
  }

  @Provides
  RpcServer rpcServer(NioReactor reactor, Config config) throws Exception {
    PORT = config.get(ConfigConverters.ofInteger(), "http.port");
    return RpcServer.builder(reactor)
        .withMessageTypes(FooRequest.class, FooResponse.class)
        .withHandler(
            FooRequest.class, request -> Promise.of(new FooResponse(200, "it is working!!")))
        .withListenPort(PORT)
        .build();
  }

  @Override
  protected void run() throws Exception {
    rpcServer.listen();
    System.out.println("Listening to: " + PORT);
    awaitShutdown();
  }

  public static void main(String[] args) throws Exception {
    MyRpcServer server = new MyRpcServer();
    server.launch(args);
  }

}

Load balancer code:

public class MyLoadBalancer extends HttpServerLauncher {
  private static final Pattern PATH_REGEX =
      Pattern.compile("/([1-5])/(fooPath|notImplementedPath)");

  private static final Promise<HttpResponse> NOT_FOUND = HttpResponse.notFound404().toPromise();

  private static final InetSocketAddress ADDRESS_01 = new InetSocketAddress("localhost", 8081);
  private static final InetSocketAddress ADDRESS_02 = new InetSocketAddress("localhost", 8082);

  @SerializeRecord
  public static record FooRequest(int id) {}

  @SerializeRecord
  public static record FooResponse(int status, String body) {}

  @Provides
  RpcClient rpcClient(NioReactor reactor) {
    var server1 = RpcStrategies.server(ADDRESS_01);
    var server2 = RpcStrategies.server(ADDRESS_02);
    var roundRobin = RoundRobin.create(server1, server2); // also have tried
    // RpcStrategies#roundRobin

    return RpcClient.builder(reactor)
        .withMessageTypes(FooRequest.class, FooResponse.class)
        .withStrategy(roundRobin)
        .build();
  }

  @Provides
  AsyncServlet servlet(NioReactor reactor, RpcClient rpcClient) {
    return request -> {
      Matcher matcher = PATH_REGEX.matcher(request.getPath());

      boolean isValidPath = matcher.matches();
      if (!isValidPath) return NOT_FOUND;

      int id = Integer.parseInt(matcher.group(1));
      String path = matcher.group(2);

      if (path.equals("fooPath")) {

        return rpcClient
            .sendRequest(new FooRequest(id)) // the request will come up to here but it
            // will wait until timeout
            .then(
                response -> {
                  FooResponse foo = (FooResponse) response;
                  return HttpResponse.ofCode(foo.status()).withBody(foo.body()).toPromise();
                });

        /*
         * also tried: return rpcClient .start() .then(() -> rpcClient .sendRequest(new
         * ExtratoRequest(id))) .then( response -> { FooResponse foo = (FooResponse) response;
         * return HttpResponse.ofCode(foo.status()) .withBody(foo.body()) .toPromise(); });
         * .whenComplete(rpcClient::stop);
         *
         * but it will give me a null pointer since "rpcClient.start()" will call a "changeStrategy"
         * method.
         */

      } else {
        return NOT_FOUND;
      }
    };
  }

  @Override
  protected void run() throws Exception {
    awaitShutdown();
  }



  public static void main(String[] args) throws Exception {
    MyLoadBalancer launcher = new MyLoadBalancer();
    launcher.launch(args);
  }

}

Req Example that will not responds untill timeout:

Any help would be greatly appreciated. Thank you!

[eduard-vasinskyi]

Hi, @ThusSpokeBieu

First and foremost, it's essential to determine the threading model of the application. ActiveJ offers great flexibility in this regard, but it's crucial to understand how the application operates and what outcomes you aim to achieve. Most components in ActiveJ belong to one event loop or another, each running in its own separate thread. Additionally, there's the main application thread. We have several options:

• Running the event loop directly in the main thread. In this case, await() won't be necessary, and the application will operate within the main thread.
• Running the event loop in a separate thread while the main thread waits for its completion.
• Running multiple threads with parallel event loops, somehow synchronized with each other, while the main thread waits for them (this option may not be needed for your application). You can read more about this threading model here.

In the MyRpcServer class you actually never start your server’s Eventloop. You can do it manually, but we recommend using the ServiceGraphModule. It will automatically handle the service start and stop routines in the correct order. In this launcher, you have 2 services: Reactor(Eventloop) and RpcServer which need to be started and stopped in order.

In the MyLoadBalancer launcher you implicitly use the ServiceGrahModule as it is enabled in the HttpServerLauncher that your launcher extends. This is why the HTTP server is able to handle requests.

To enable the ServiceGraphModule in the MyRpcServer launcher you need to override the getModule method and return an instance of ServiceGraphModule:

@Override
protected Module getModule() {
   return ServiceGraphModule.create();
}

If you do so, you should also remove the rpcServer.listen(); line from the MyRpcServer#run method. The ServiceGraphModule will call listen() on the server for you when starting the service.

I also noticed that you use regex to route HTTP requests. You could achieve a similar result with the RoutingServlet. Here are some examples.

[ThusSpokeBieu]

Okay, now everything makes sense. I mistakenly thought that the parent class ‘Launcher’ already handled the event loop. I didn’t pay attention to that. Thank you very much for your help—it’s working perfectly now.

I’m using regex for paths just as a test to measure whether it’s more performant than using the RoutingServlet. I’m experimenting with various things focused on this, even though some of them don’t make much sense.

Good job on the excellent work with this project! I’m really enjoying using ActiveJ, and I hope it becomes a recognized tool in the market soon.

Thank you for taking the time to respond.