Alibaba RSocket Broker in Action

 

I have been playing with Alibaba RSocket Broker recently. If you are interested, read: 

• https://dzone.com/articles/rsocket-broker-use-case-in-alibaba-cloud
  
• https://dzone.com/articles/rsocket-vs-grpc-benchmark
  

RSocket broker has proven to be reliable and high performance. Check out their site at Github. https://github.com/alibaba/alibaba-rsocket-broker/blob/master/README-en.md

Running on your local machine

Prerequisite:

• JDK 1.8.0+
• Maven 3.5.x
• Node 10+

For a quick start, check out the source code from git, then run:

   mvn -DskipTests clean package

 

Go to alibaba-broker-server > target folder. 

3

Execute the jar file to get Alibaba RSocket Broker running.

java -jar target/alibaba-rsocket-broker.jar

And you will see this in the console.

Browse to http://localhost:9998 and you will see the web console.

Register a Service

Next, we are going to register a service to the broker. To do this we look at the example projects included in the source code.

When I first look at the rsocket-responder project, you will see that it is complaining about the com.alibaba.user.Account class not found. 

The required class is located in user-service-api project and it is a ProtoBuf file. It needs to be manually generated. 

For the purpose of the experiment, we just run 'mvn -DskipTest package' on the example folder. This will solve the dependency problem and generates the jar file that can be used. 

Then we go to rsocket-responder folder and run:

   java -jar target/rsocket-responder-1.0.0-SNAPSHOT.jar

We can see a Spring Boot application started and all services published to the Broker.

Now let's go back to the Broker web console. 

We can see our app is listed here now.

We can also see it in App Instances List. Click on the app will show us more detail.

The Service Testing is also very cool. I can use it to test the methods in the service.

This is the method implementation.

And the response return from the service.

Integrate 2 Services

Go to example/rsocket-requester. Run the jar file to start the service.

    java -jar target/rsocket-requester-1.0.0-SNAPSHOT.jar

This will starts a Spring Boot application with a few endpoints available. For a quick test run:

  curl http://localhost:8181/user/2

And here is the formatted response:

Here is what happened.

In rsocket-requester > UserController there is a GET endpoint that will call userService.findById(id).

The UserService is autowired.

In the Bean configuration, we can see that the UserService instance is constructed using RSocketRemoteServiceBuilder. 

This setting tells the rsocket-requester to call the remote service from the broker. 

At the rsocket-responder, there is an implementation class of the UserService interface. The class needs to be annotated with @RScoketService. This will register the service to the broker when the application is deployed and running.

-------------------------------------------------------------------------------------------------------------------------

Alibaba RSocket Broker seems to be well designed and it makes it easy for a developer to use. Compared to using the auto-generated stubs from gRPC, I think it is more cleaner and less cognitive burden. (yes, I don't like those stubs.  :p)

Hope to see better documentation in the future, (I might try to write some). That will make more people adopting the system. 

RSocket Java Example

Come across RSocket here. 

RSocket is a binary protocol for use on byte stream transports such as TCP, WebSockets, and Aeron.

It could be a good alternative to REST for microservice communications. 

Let's write some code and try it out. All you need is these libraries in your pom file. Then you are good to go.

        <dependency>
            <groupId>io.rsocket</groupId>
            <artifactId>rsocket-core</artifactId>
            <version>1.1.0</version>
        </dependency>
        <dependency>
            <groupId>io.rsocket</groupId>
            <artifactId>rsocket-transport-netty</artifactId>
            <version>1.1.0</version>
        </dependency>

For the demo, we will write a Server that returns the message that is sent by the client.

Here is the Server code.

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.frame.decoder.PayloadDecoder;
import io.rsocket.transport.netty.server.CloseableChannel;
import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.core.publisher.Mono;

public class ServerApplication {

    public static final int TCP_PORT = 8910;
    
    public static void main(String[] args) {
        CloseableChannel channel = 
                RSocketServer.create(SocketAcceptor.with(new MyRSocket()))
                             // Enable Zero Copy
                             .payloadDecoder(PayloadDecoder.ZERO_COPY)
                             .bind(TcpServerTransport.create(TCP_PORT))
                             .block();
        if(channel != null) {
            channel.onClose().block();
        }
    }
    
    private static class MyRSocket implements RSocket {
        @Override
        public Mono<Payload%gt; requestResponse(Payload payload) {
            return Mono.just(payload);
        }
    }
    
}

Just start the ServerApplication and let it listen to port 8910.

Now we write the Client code.

import java.nio.ByteBuffer;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;

public class ReqResClient {

    private final RSocket socket;

    public ReqResClient() {
        this.socket = RSocketFactory.connect()
                .transport(TcpClientTransport.create(HOST, TCP_PORT))
                .start()
                .block();
    }

    public ByteBuffer callBlocking(byte[] byteArray) {
        return socket
                .requestResponse(DefaultPayload.create(byteArray))
                .map(Payload::getData)
                .block();
    }
    
    public void dispose() {
        this.socket.dispose();
    }

}

And the class for the object we want to send to the Server. 

import java.io.Serializable;

import lombok.AllArgsConstructor;
import lombok.Data;

/* ISO-99999999 standard user */
@Data
@AllArgsConstructor
public class User implements Serializable {

    private static final long serialVersionUID = -8731669541043173364L;
    private String username;
    private String password;
    
}

Now we run the client.

import java.nio.ByteBuffer;
import org.apache.commons.lang3.SerializationException;
import org.apache.commons.lang3.SerializationUtils;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class Application {

    public static final String HOST = "localhost";
    public static final int TCP_PORT = 8910;

    public static void main(String[] args) {

        ReqResClient client = new ReqResClient();
        
        User user = new User("devil", "kazuya");
        try {
            // serializing object to byte array.
            byte[] bytes = SerializationUtils.serialize(user);
            
            log.info("Sending 'user' to server:   ----  {}", user);
            ByteBuffer response = client.callBlocking(bytes);
            
            // deserializing byte array back to object.
            User clone = (User) SerializationUtils.deserialize(response.array());
            log.info("Getting 'user' from server: ----  {}", clone);
            
        } catch (NullPointerException | SerializationException e) {
            log.error("something wrong", e);
        }
        
        client.dispose();
    }

}

For the result, we see some warnings messages. We ignore them for now. We can see your object went to the Server and back in one piece. How nice.

[main] WARN reactor.netty.tcp.TcpResources - [tcp] resources will use the default LoopResources: DefaultLoopResources {prefix=reactor-tcp, daemon=true, selectCount=4, workerCount=4}
[main] WARN reactor.netty.tcp.TcpResources - [tcp] resources will use the default ConnectionProvider: PooledConnectionProvider {name=tcp, poolFactory=reactor.netty.resources.ConnectionProvider$$Lambda$7/13138721@39f31e}
[main] INFO ong.ternchow.rsocketclient.Application - Sending 'user' to server:   ----  User(username=devil, password=kazuya)
[main] INFO ong.ternchow.rsocketclient.Application - Getting 'user' from server: ----  User(username=devil, password=kazuya)

Few thoughts come across my mind.

• I don't have to parse my object to JSON in order to talk to other microservice. 
• There will be a lot of Serializing/Deserializing (the good/bad old RMI days). 
• Sending data in binaries will be faster than text(JSON). 

The Interaction Models of RSocket are very interesting as well. Read more here. 

Fire-and-Forget

Future<Void> completionSignalOfSend = socketClient.fireAndForget(message);

Request/Response

Future<Payload> response = socketClient.requestResponse(requestPayload);

Request/Stream

Publisher<Payload> response = socketClient.requestStream(requestPayload);

Channel

Publisher<Payload> output = socketClient.requestChannel(Publisher<Payload> input);

We will play with each model in the future. Stay tuned.

---

Source code:

https://github.com/devilkazuya99/rsocketserver 

https://github.com/devilkazuya99/rsocketclient

Math Test: Visual Multiply Game

{{num1}}	❌	{{num2}}
{{answer}} {{outcome}}