Request/Reply asynchronous

[deepeshk12]

I am using gnatsd in my spring boot project when using the Request/Reply we are passing the message but its synchronous so other thread/User is waiting.

• How can I make it asynchronous

• [ nats-server version 1.0.2] Included gnatsd -DV output

• Code

@Component
public class NatsComponent {

    @Value("${nats.server.url}")
    private String natsServerUrl;

    private NatsConnector natsConnector;

    @PostConstruct
    public void init() {
        natsConnector=new NatsConnector()
                .addHost(natsServerUrl)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         ;

    }

    @Bean(name = "nc")
    public Nats nats() {
        return natsConnector.connect();
    }

    /**
     * @param subject
     * @param request
     * @param replyProtoClass
     * @param <T>
     * @throws IllegalArgumentException When T is not GeneratedMessageV3 or String
     * @return
     */
    public <T> CompletableFuture<T> request(String subject, GeneratedMessageV3 request, Class<T> replyProtoClass) {
        CompletableFuture completableFuture = new CompletableFuture<T>();
        if ( ( GeneratedMessageV3.class.isAssignableFrom(replyProtoClass) ) ||
                ( String.class.getName().equals(replyProtoClass.getName()) ) ) {}
        else {
            throw new IllegalArgumentException("only GeneratedMessageV3 or String class allowed as Class<T> " +
                    "replyProtoClass");
        }

        try( Nats nats = nats()) {

        nats.request(subject, protoSerialize(request), 1, TimeUnit.MINUTES, 1, (msg) -> {
            T type = null;

            try {
                if (String.class.getName().equals(replyProtoClass.getName())) {
                    type = (T) msg.getBody();
                } else {
                    type = (T) protoDeserialize(msg.getBody(), (Class<GeneratedMessageV3>) replyProtoClass);
                }
                completableFuture.complete(type);
            } catch (Exception e) {
                completableFuture.completeExceptionally(e);
            }
        });

            completableFuture.get(60,TimeUnit.SECONDS);
        }catch(CustomException e){
            throw e;
        }catch (Exception e){
            throw new CustomException("ERROR","INTERNAL_SERVER_ERROR");
        }
        return completableFuture;
    }

    /**
     * Method used to return Async Completable Proto responses
     *
     * @param subject
     * @param request
     * @param protoResponseClass
     * @param <T>
     * @param <R>
     * @return
     */
    public <T extends GeneratedMessageV3, R> CompletableFuture<T> protoToFutureResponse(String subject,
                                                            GeneratedMessageV3 request, Class<T> protoResponseClass) {
        return request(subject, request, protoResponseClass);
    }

    public <T extends GeneratedMessageV3, R> CompletableFuture<T> stringToFutureResponse(String subject,
                                                                                        String request,
                                                                                         Class<T> replyProtoClass) {
        CompletableFuture completableFuture = new CompletableFuture<T>();
        try( Nats nats = nats()) {
        nats.request(subject, request, 1, TimeUnit.MINUTES, 1, msg -> {
            T type = null;

            try {
                type = (T) protoDeserialize(msg.getBody(), (Class<GeneratedMessageV3>) replyProtoClass);
                completableFuture.complete(type);
            } catch (Exception e) {
                System.out.println(e.toString());
                completableFuture.completeExceptionally(e);
            }
        });
            completableFuture.get(60,TimeUnit.SECONDS);
        }catch(CustomException e){
            throw e;
        }catch (Exception e){
            throw new CustomException("ERROR","INTERNAL_SERVER_ERROR");
        }
        return completableFuture;
    }


}

[mheath]

I'm not sure what you're asking for. nats.request(...) is inherently asynchronous. The callback function gets invoked when a reply to the request is received and there aren't any threads blocking for that request.

When you say "but its synchronous so other thread/User is waiting". What is the "other thread/User" you are referring to?

[deepeshk12]

Hi @mheath,

Thanks for your reply.

In above code when Ist request calles completableFuture.get(60,TimeUnit.SECONDS); its calling a service, if we insert a debug inside that call and same time we call another request it's stuck in completableFuture.get(60,TimeUnit.SECONDS); line till not received response from Ist request.

This problem we are facing in our project, it seems the request is synchronously calling.

Hope you understand my problem.

If you request other code please let us know we will share.

[mheath]

The NATS client uses a single thread. If you block that thread (calling CompletableFuture.get() for example) then nothing will happen.

In your NatsConnector, you can specify an Executor using calllbackExecutor that will run all your callbacks in a different thread or pool of threads depending on how you configure the Executor.

[deepeshk12]

Thanks @mheath ,

We will try this and confirm.
Appreciate your help.

[deepeshk12]

Hi @mheath ,

We tried this, the calllbackExecutor executor is inside of NatsConnector.java when we are changing, its start giving the error in line 181 where we are trying to assign java.util.concurrent.Executor to java.util.concurrent.ThreadPoolExecutor .
Can you please eleborate in which class we need to change or is there any other way we can use instead of calling get method.

[deepeshk12]

Hi @mheath ,

We have done the below steps.

1. change the callbackExecutor in NatsConnector.java class .

 public NatsConnector() {
        this.reconnectWaitTime = Constants.DEFAULT_RECONNECT_TIME_WAIT;
        this.pedantic = false;
        this.maxFrameSize = 1048576;
        this.listeners = new ArrayList();
        this.idleTimeout = Constants.DEFAULT_IDLE_TIMEOUT;
        this.pingInterval = Constants.DEFAULT_PING_INTERVAL;
        **_this.callbackExecutor = new ThreadPoolExecutor(20, 400, 10L, TimeUnit.SECONDS, new ArrayBlockingQueue(2)) {_**
            private final Logger logger = LoggerFactory.getLogger(this.getClass());
            public void execute(Runnable command) {
                try {
                    command.run();
                } catch (Exception var3) {
                    this.logger.error("Error invoking callback", var3);
                }
           }
        };
    }

Still facing the same issue. please suggest.