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

Still using Java: Nginx infront of Quarkus Services

 

Today's goal:

- put a load balancer in front of our services.
- create a dummy domain name by modifying the hosts file.
- access the service using the dummy domain name.

First, let's check out all the services from github.

• https://github.com/devilkazuya99/auth-service.git 
• https://github.com/devilkazuya99/inventory-service.git
• https://github.com/devilkazuya99/cart-service.git
• https://github.com/devilkazuya99/payment-service.git
• https://github.com/devilkazuya99/transaction-service.git

Download nginx from

• https://nginx.org/en/download.html

Then open up all the services:

We try to start them up.

And we face our first problem.

We have 5 services trying to run on port 5005 for Quarkus debugging and port 8080 for the web application. To solve this, we will run our application on different ports. 

Setting port used in application.properties 

quarkus.http.port=9090

For Quarkus debugging port we defined it on the maven command.

mvn compile quarkus:dev -Ddebug=5006

Once the ports are set, we start all the services again.

This is the setup for now.

Next, we are going to edit the hosts file. 

127.0.0.1   mypetshop.local

We add this entry to let mypetshop.local points to the current host machine.

After the hosts file is edited, we can test it with Postman.

It will give us back a result as if we are hitting localhost.

Now we create a nginx config file, saved as mypetshop.conf

worker_processes  1;

events {
    worker_connections  1024;
}

http {
    upstream auth_service {
        server localhost:8080;
    }
    upstream cart_service {
        server localhost:8090;
    }
    upstream inventory_service {
        server localhost:8100;
    }
    upstream payment_service {
        server localhost:8110;
    }
    upstream transaction_service {
        server localhost:8120;
    }

    server {
        listen 80;
        location /auth {
            proxy_pass http://auth_service/auth;
        }
        location /cart {
            proxy_pass http://cart_service/cart;
        }
        location /inventory {
            proxy_pass http://inventory_service/inventory;
        }
        location /payment {
            proxy_pass http://payment_service/payment;
        }
        location /transaction {
            proxy_pass http://transaction_service/transaction;
        }
    }
}

then we start nginx with the config file.

nginx -c mypershop.conf

To test the outcome, in Postman we create a new Environment

  We create a new "host" variable with a value like below.  

Now open the Auth Service collection and change the List Users request.

We change the 'localhost:8080' into using the variable '{{host}}'.

Make sure to select the environment that we have created just now.

When we hit the request, we should get a 200 response.

Ok. now we try something more interesting. To demonstrate how nginx do load balancing:

Change the GET cart function to print the count in the console.

    static int count = 0;

    @GET
    public Cart getCart(@HeaderParam("X-User-Id") Long userId) {
        Cart cart = findCart(userId);
        logger.info(" Request count = {}", ++count);
        return cart;
    }
    

Run H2 Database in Server mode.

Download the h2 jar file from h2 website. In a terminal, start the server using the command above. 

For the application properties

    # change database url to connect in Server mode.
    quarkus.datasource.jdbc.url = jdbc:h2:tcp://localhost/~/cart-database
    
    # run an instance in port 8090
    quarkus.http.port=8090
    
    # run an instance in port 8091
    quarkus.http.port=8091
    
    

Update the nginx configuration.

Restart nginx and hit the GET cart request like crazy.

You should see the count printed alternately from the two instances.

Still using Java: Secured REST Resource with Quarkus Form Based Authentication

Quarkus comes with a lot of authentication mechanisms. Today we are going to implement a simple one with Form-Based Authentication. Quarkus has good documentation but the example for authentication is all over the place. We will create one here just for future reference.

You can visit the documentation here.

A quick description of what we want to do is:

- we have some REST endpoints
- we want to secure the endpoints
- with minimal effort
- username and password saved in the database

These are the basic requirement. 

To begin, check out my previous source code from https://github.com/devilkazuya99/inventory-service

It should have 5 existing endpoints in InventoryResource.java. 

The first thing to do is to enable form-based authentication in the application.properties file.

This will unlock some forbidden dark arts into your system.

Now we need to write some entity class to hold our user's credentials.

        @Entity
        @UserDefinition
        public class AppUser extends PanacheEntity {
        
            @Username
            public String username;
            @Password
            public String password;
            @ManyToMany(cascade = CascadeType.ALL)
            @Roles
            public List<UserRole> roles = new ArrayList<>();;
            
            public static void add(String username, String password, String role) { 
                AppUser user = new AppUser();
                user.username = username;
                user.password = BcryptUtil.bcryptHash(password);
                UserRole userRole = new UserRole();
                userRole.role = role;
                user.roles.add(userRole);
                user.persist();
            }
                
        }

        @Entity
        public class UserRole extends PanacheEntity {
            
            @ManyToMany(mappedBy = "roles")
            public List<AppUser> users;
                
            @RolesValue
            public String role;
        }

Our AppUser has a username and password. It also can have a list of UserRoles.

In listCategory() endpoint in InventoryResource.java, add a @RolesAllowed("user") annotation to the function.

Lastly, add 2 static files, one for login and one for the error page.

The error page is just a dummy page with static content. We will explain more on the login page. 

    <div class="container">
        <section id="content">
            <form action="/j_security_check" method="POST">
                <h1>Login Form</h1>
                <div>
                    <input type="text" placeholder="Username" required="" id="username" name="j_username" />
                </div>
                <div>
                    <input type="password" placeholder="Password" required="" id="password" name="j_password" />
                </div>
                <div>
                    <input type="submit" value="Log in" />
                    <a href="#">Lost your password?</a>
                    <a href="#">Register</a>
                </div>
            </form><!-- form -->

        </section><!-- content -->
    </div><!-- container -->

In short, the login form will POST to '/j_security_check', which is provided by Quarkus dark art. The default username is 'j_username' and the password is 'j_password'. That's all it is.

These are default values that can be modified. For more details see here.

Now, starts the application with this command (that you need to remember for the rest of your life).

mvn compile quarkus:dev

Using a web browser we first browse the product page: http://localhost:8080/inventory/product

We will get an empty array as the response. 

[]

Now we browse the catalog page: http://localhost:8080/inventory/category

This time we got redirected to the login page.

The Quarkus dark art is working so far. We annotated the endpoint previously to only allows user with role "user" to access. It is looking good. 

We forgot one important thing. We don't have any user to login. Let's go back and create some.

We create a Startup class. (you can give it any name)

@Singleton
public class Startup {
	
    @Transactional
    public void loadUsers(@Observes StartupEvent evt) {
        // reset and load all test users
        AppUser.deleteAll();
        AppUser.add("admin", "admin", "admin");
        AppUser.add("user", "user", "user");
    }
	
}

Pay attention to the loadUsers function. The function has a @Observes annotation, which will listen to the Quarkus StartEvent. So every time Quarkus starts up, this function will be triggered. When it happened, we cleared the table, and add 2 dummy users to it.

Now, we go back to the category page, http://localhost:8080/inventory/category, key in the username and password, Voilà, we can see the empty JSON array.

[]

Still using Java: Working with the project code

This is a short post about how to check out the project from Github and work in an IDE. I'm using Eclipse as my IDE. Similar concepts and configurations should be applicable to other IDE. 

1st checkout the code from my Github repository.

On the [Package Explorer], right-clicks and click 'Import...'

Choose 'Existing Maven Project'.

Click 'Finish' to import.

When you import, the IDE will start to build the project. Make sure it is using Java 11 as JRE. The Jave version is defined in the pom.xml. 

After import, you should be able to run the project. Right-click on the project > Run As > Maven build...

 

Set the Goals to: compile quarkus:dev

Make sure the Maven Runtime version is 3.6.2+

Select the JRE tab (this is important). Make sure you use Java 11 from GraalVM. If you don't have it, go download a copy now. 

Once you hit the 'Run' button, wait for the message to be shown on the console. 

Now you can start coding. Yes. You hear me right. Start coding. No need to restart the application anymore. Add a new function? Add new class? Add a new database entity? Just hit the endpoint after the changes. Quarkus will reload your app and in a very fast manner. 

Still using Java: Cart, Payment, and Transaction Services

In order to get to the fun part, where we try to integrate all the services, I just quickly draft some requirement and roll them out. We will make changes to it accordingly ni the future.

 

We will create some endpoints for these services, persist the data into a database, that's it.

Create 3 projects using maven command

mvn io.quarkus:quarkus-maven-plugin:1.13.1.Final:create -DprojectGroupId=blog.technodeck -DprojectArtifactId=cart-service -Dverison=0.0.1-SNAPSHOT -Dextensions="resteasy-jackson,jdbc-h2,hibernate-orm-panache"
mvn io.quarkus:quarkus-maven-plugin:1.13.1.Final:create -DprojectGroupId=blog.technodeck -DprojectArtifactId=payment-service -Dverison=0.0.1-SNAPSHOT -Dextensions="resteasy-jackson,jdbc-h2,hibernate-orm-panache"
mvn io.quarkus:quarkus-maven-plugin:1.13.1.Final:create -DprojectGroupId=blog.technodeck -DprojectArtifactId=transaction-service -Dverison=0.0.1-SNAPSHOT -Dextensions="resteasy-jackson,jdbc-h2,hibernate-orm-panache"

then configure the application.properties accordingly.

Cart Service

# configure your datasource
quarkus.datasource.db-kind = h2
quarkus.datasource.username = sa
quarkus.datasource.password = sa
quarkus.datasource.jdbc.url = jdbc:h2:~/cart-database
# drop and create the database at startup (use `update` to only update the schema)
quarkus.hibernate-orm.database.generation = update

Payment Service

# configure your datasource
quarkus.datasource.db-kind = h2
quarkus.datasource.username = sa
quarkus.datasource.password = sa
quarkus.datasource.jdbc.url = jdbc:h2:~/payment-database
# drop and create the database at startup (use `update` to only update the schema)
quarkus.hibernate-orm.database.generation = update

Transaction Service

# configure your datasource
quarkus.datasource.db-kind = h2
quarkus.datasource.username = sa
quarkus.datasource.password = sa
quarkus.datasource.jdbc.url = jdbc:h2:~/transaction-database
# drop and create the database at startup (use `update` to only update the schema)
quarkus.hibernate-orm.database.generation = update

Cart Service Endpoint

We create simple entities.

@Entity
public class Cart extends PanacheEntity {

    public Long userId;
    @OneToMany(cascade = CascadeType.ALL, orphanRemoval = true)
    public List<CartItem> items;
    public ZonedDateTime createdDate;

}

@Entity
public class CartItem extends PanacheEntity {

    public Long productId;
    public String name;
    public Float price;
    public int quantity;

}

Then the resource class

@Path("/cart")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
public class CartResource {...}

The resource class can do basic CRUD operation. I won't show the full detail, but there are some interesting API that I found it useful and interesting.

Public functions that mapped to @GET, @PPOST, @PUT, and @DELETE

    Cart cart = (Cart) Cart.find("userId", userId)
                               .firstResultOptional()
                               .orElse(new Cart());

I like this firstResultOptional function from the PanacheQuery class. It returns an Optional which is very handy. Spring Data has this feature too.

Payment Service

Payment Service has a Wallet.

@Entity
public class Wallet extends PanacheEntity {
	
	@NotNull
	@Column(unique = true)
    public Long userId;
	@NotNull
    public Float balance;
	
	@NotBlank
	public String description;
	
}

It also has a few data classes.

public class CardInfo {

	@NotBlank
	public String cardNumber;
	@NotBlank
	public String nameOnCard;
	@NotNull
	public Integer expMonth;
	@NotNull
	public Integer expYear;
	@NotNull
	public Integer ccv;
	
}

//

public class PaymentDetail {

	@NotNull
	public Long walletId;
	@NotNull
	public Float amount;
	public String transactionId;
	public ZonedDateTime createdDate;
	public ZonedDateTime updatedDate;
	public String status;
	
	public PaymentDetail() {}
	
	public PaymentDetail(@NotNull Long walletId, @NotNull Float amount, String transactionId, ZonedDateTime createdDate,
	        ZonedDateTime updatedDate, String status) {
		super();
		this.walletId = walletId;
		this.amount = amount;
		this.transactionId = transactionId;
		this.createdDate = createdDate;
		this.updatedDate = updatedDate;
		this.status = status;
	}
	
}

//

public class TopupInfo {

	@NotNull
	public Long walletId;
	@NotNull
	public Float amount;
	@NotNull
	public CardInfo card;
	
}

In this service, we are experimenting with the Quarkus Validation extension.

        <dependency>
            <groupId>io.quarkus</groupId>
            <artifactId>quarkus-hibernate-validator</artifactId>
        </dependency>

By including this dependency, it allows us to validate the request body by simply adding a @Valid annotation in the parameter. See the createWallet() function below. 

Here is the PaymentResource

Let's test the validation by creating a wallet without a required field.

You will get a build-in report. How cool is that?

Lastly Transaction Service

The Entity classes.

@Entity
public class ShippingInfo extends PanacheEntity{

    @NotBlank
    public String recieverName;
    public String company;
    @NotBlank
    public String addressLine1;
    public String addressLine2;
    public String addressLine3;
    public String addressLine4;
    public Integer postcode;
    @NotBlank
    @Column(length = 50)
    public String state;
    @NotBlank
    @Column(length = 50)
    public String country;
    
}
// ---------------------------------
@Entity
public class PurchaseTransaction extends PanacheEntity {

    @OneToOne(fetch = FetchType.LAZY)
    @MapsId
    @NotNull
    @Valid
    public ShippingInfo shippingInfo;
    @NotNull
    @Column(nullable = false)
    public Long cartId;
    @NotNull
    @Column(nullable = false)
    public Long paymentId;
    @NotNull
    @Column(nullable = false)
    public Double amount;

    public ZonedDateTime createdDate;
    
    @PrePersist
    public void beforePersist() {
        createdDate = ZonedDateTime.now();
    }
}
// ---------------------------------
@Entity
public class WalletTransaction extends PanacheEntity {

    @NotNull
    @Column(nullable = false)
    public Long walletId;
    @NotNull
    @Column(nullable = false)
    public Long userId;
    @NotNull
    @Column(nullable = false)
    public Long txId;
    @NotNull
    @Column(nullable = false)
    public String txType;
    public Double credit;
    public Double debit;

    public ZonedDateTime createdDate;
    
    @PrePersist
    public void beforePersist() {
        createdDate = ZonedDateTime.now();
    }
    
}

The resource endpoint.

@Path("/tx")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
public class TransactionResource {

    @Transactional
    @POST
    @Path("/purchase")
    public void purchaseTx(@Valid PurchaseTransaction purchaseTransaction) {
        purchaseTransaction.persist();
    }

    @GET
    @Path("/purchase")
    public List<PurchaseTransaction> getPurchaseTransaction() {
        return PurchaseTransaction.findAll(Sort.by("id").ascending()).list();
    }
    
    @Transactional
    @POST
    @Path("/wallet")
    public void walletTx(@Valid WalletTransaction walletTransaction) {
        walletTransaction.persist();   
    }
    
    @GET
    @Path("/wallet")
    public List<WalletTransaction> getWalletTransaction() {
        return WalletTransaction.findAll(Sort.by("id").ascending()).list();
    }
    
}

Making some POST requests