added rabbitMQ configuration

pom.xml

@@ -113,6 +113,10 @@
 			<version>1.18.1</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-amqp</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>pulsar</artifactId>

src/main/java/edu/stanford/protege/webprotege/ipc/CommandExecutor.java

@@ -13,4 +13,6 @@
 public interface CommandExecutor<Q extends Request<R>, R extends Response> {
 
     CompletableFuture<R> execute(Q request, ExecutionContext executionContext);
+
+    CompletableFuture<R> executeRabbit(Q request, ExecutionContext executionContext);
 }

src/main/java/edu/stanford/protege/webprotege/ipc/CommandHandler.java

@@ -11,6 +11,7 @@
  * Stanford Center for Biomedical Informatics Research
  * 2021-08-06
  */
+@WebProtegeHandler
 public interface CommandHandler<Q extends Request<R>, R extends Response> {
 
     /**

src/main/java/edu/stanford/protege/webprotege/ipc/pulsar/PulsarCommandExecutor.java

@@ -13,6 +13,7 @@
 import org.apache.pulsar.client.api.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 
@@ -52,6 +53,9 @@ public class PulsarCommandExecutor<Q extends Request<R>, R extends Response> imp
     @Autowired
     private ObjectMapper objectMapper;
 
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
     private Producer<byte[]> producer;
 
     private Consumer<byte[]> consumer;
@@ -107,6 +111,37 @@ public CompletableFuture<R> execute(Q request, ExecutionContext executionContext
         }
     }
 
+    @Override
+    public CompletableFuture<R> executeRabbit(Q request, ExecutionContext executionContext) {
+        try {
+            var json = objectMapper.writeValueAsBytes(request);
+            org.springframework.amqp.core.Message rabbitRequest = new org.springframework.amqp.core.Message(json);
+            rabbitRequest.getMessageProperties().getHeaders().put(Headers.ACCESS_TOKEN, executionContext.jwt());
+            rabbitRequest.getMessageProperties().getHeaders().put(Headers.USER_ID, executionContext.userId());
+            rabbitRequest.getMessageProperties().setConsumerQueue(request.getChannel());
+
+            org.springframework.amqp.core.Message rabbitResponse = rabbitTemplate.sendAndReceive(rabbitRequest);
+
+            CompletableFuture<R> replyHandler = new CompletableFuture<>();
+
+            assert rabbitResponse != null;
+            var error = (String) rabbitResponse.getMessageProperties().getHeaders().get(Headers.ERROR);
+            if (error != null) {
+                var executionException = objectMapper.readValue(error, CommandExecutionException.class);
+                replyHandler.completeExceptionally(executionException);
+            }
+            else {
+                var response = objectMapper.readValue(rabbitResponse.getBody(), responseClass);
+                logger.info("ALEX raspund la reply handler {} cu response {}", replyHandler, response);
+                replyHandler.complete(response);
+            }
+            return replyHandler;
+        } catch (Exception e) {
+            logger.error("Error ", e);
+            throw new RuntimeException(e);
+        }
+    }
+
     private synchronized Producer<byte[]> getProducer(Q request) {
         try {
             if (producer != null) {

src/main/java/edu/stanford/protege/webprotege/ipc/pulsar/PulsarCommandHandlerWrapper.java

@@ -195,7 +195,7 @@ private void authorizeAndReplyToRequest(String replyChannel,
                                                             subject,
                                                             requiredActionId.stream().findFirst().orElse(null));
         var executionContext = new ExecutionContext(new UserId(userId), "");
-        var authResponseFuture = authorizationStatusExecutor.execute(authRequest, executionContext);
+        var authResponseFuture = authorizationStatusExecutor.executeRabbit(authRequest, executionContext);
         authResponseFuture.whenComplete((authResponse, authError) -> {
             if (authError != null) {
                 // The call to the authorization service failed

src/main/java/edu/stanford/protege/webprotege/ipc/pulsar/PulsarCommandHandlersConfiguration.java

@@ -1,14 +1,29 @@
 package edu.stanford.protege.webprotege.ipc.pulsar;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.rabbitmq.client.Channel;
+import edu.stanford.protege.webprotege.authorization.GetAuthorizationStatusRequest;
+import edu.stanford.protege.webprotege.authorization.GetAuthorizationStatusResponse;
+import edu.stanford.protege.webprotege.common.Request;
+import edu.stanford.protege.webprotege.common.Response;
 import edu.stanford.protege.webprotege.ipc.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.connection.Connection;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import javax.annotation.PostConstruct;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
 
 /**
  * Matthew Horridge
@@ -21,11 +36,76 @@ public class PulsarCommandHandlersConfiguration {
     private static final Logger logger = LoggerFactory.getLogger(PulsarCommandHandlersConfiguration.class);
 
     @Autowired(required = false)
-    private List<CommandHandler<?, ?>> commandHandlers = new ArrayList<>();
+    private List<CommandHandler<? extends Request, ? extends Response>> commandHandlers = new ArrayList<>();
 
     @Autowired
     private PulsarCommandHandlerWrapperFactory wrapperFactory;
 
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+    @Autowired
+    private ObjectMapper objectMapper;
+
+    @Autowired
+    CommandExecutor<GetAuthorizationStatusRequest, GetAuthorizationStatusResponse> authorizationStatusExecutor;
+
+    @Autowired
+    private ConnectionFactory connectionFactory;
+
+    public static final String RPC_QUEUE1 = "webprotege-rpc-queue";
+    public static final String RPC_RESPONSE_QUEUE = "webprotege-backend-response-queue";
+
+    public static final String RPC_EXCHANGE = "webprotege-exchange";
+
+    @Bean
+    Queue msgQueue() {
+        return new Queue(RPC_QUEUE1, true);
+    }
+
+    @Bean
+    Queue replyQueue() {
+        return new Queue(RPC_RESPONSE_QUEUE, true);
+    }
+
+    @Bean
+    DirectExchange exchange() {
+        return new DirectExchange(RPC_EXCHANGE, true, false);
+    }
+
+    @Bean
+    public List<Binding> bindings(DirectExchange directExchange, Queue msgQueue){
+        try (Connection connection = connectionFactory.createConnection();
+             Channel channel = connection.createChannel(true)) {
+            channel.exchangeDeclare(RPC_EXCHANGE, "direct", true);
+            channel.queueDeclare(RPC_QUEUE1,true,false, false,null);
+
+            var response = new ArrayList<Binding>();
+
+            for(CommandHandler handler: commandHandlers) {
+                logger.info("Declaring binding {} {} " + handler.getChannelName());
+                channel.queueBind(RPC_QUEUE1, RPC_EXCHANGE, handler.getChannelName());
+                response.add(BindingBuilder.bind(msgQueue).to(directExchange).with(handler.getChannelName()));
+            }
+            return response;
+
+        } catch (Exception e) {
+            logger.error("Error ", e);
+            throw new RuntimeException(e);
+        }
+
+
+    }
+
+    @Bean
+    public SimpleMessageListenerContainer messageListenerContainers(ConnectionFactory connectionFactory){
+
+        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+        container.setQueueNames(RPC_QUEUE1);
+        container.setConnectionFactory(connectionFactory);
+        container.setMessageListener(new RabbitMqHandlerWrapper<>(commandHandlers, objectMapper, authorizationStatusExecutor, rabbitTemplate));
+
+        return container;
+    }
 
     @PostConstruct
     private void postConstruct() {