Updated from main

Ahmad45123 · May 15, 2024 · 40a9f5a · 40a9f5a
2 parents 1ce486d + 43efcc4
commit 40a9f5a
Show file tree

Hide file tree

Showing 24 changed files with 696 additions and 191 deletions.
diff --git a/services/contracts/src/main/java/com/workup/contracts/ContractsApplication.java b/services/contracts/src/main/java/com/workup/contracts/ContractsApplication.java
@@ -1,11 +1,16 @@
 package com.workup.contracts;
 
-import static com.workup.contracts.tests.InitiateContractTests.initiateContractTest1;
 
+import static com.workup.contracts.tests.InitiateContractTests.initiateContractTest1;
 import com.workup.contracts.logger.ContractsLogger;
 import com.workup.contracts.logger.LoggingLevel;
+import com.workup.shared.enums.ControllerQueueNames;
 import com.workup.shared.enums.ServiceQueueNames;
 import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.AnonymousQueue;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.FanoutExchange;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
@@ -15,10 +20,13 @@
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 @SpringBootApplication
 @ComponentScan(basePackages = "com.workup")
 @EnableCaching
+@EnableAsync
 public class ContractsApplication {
 
   public static void main(String[] args) {
@@ -44,4 +52,30 @@ public Queue myQueue() {
   public MessageConverter messageConverter() {
     return new Jackson2JsonMessageConverter();
   }
+
+  @Bean
+  public Queue controllerQueue() {
+    return new AnonymousQueue();
+  }
+
+  @Bean
+  public FanoutExchange fanout() {
+    return new FanoutExchange(ControllerQueueNames.CONTRACTS);
+  }
+
+  @Bean
+  public Binding fanoutBinding(FanoutExchange fanout, Queue controllerQueue) {
+    return BindingBuilder.bind(controllerQueue).to(fanout);
+  }
+
+  @Bean
+  public ThreadPoolTaskExecutor taskExecutor() {
+    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+    executor.setCorePoolSize(50);
+    executor.setMaxPoolSize(50);
+    executor.setQueueCapacity(500);
+    executor.setThreadNamePrefix("contracts-");
+    executor.initialize();
+    return executor;
+  }
 }
diff --git a/services/contracts/src/main/java/com/workup/contracts/ControllerMQListener.java b/services/contracts/src/main/java/com/workup/contracts/ControllerMQListener.java
@@ -0,0 +1,35 @@
+package com.workup.contracts;
+
+import com.workup.shared.commands.controller.SetMaxThreadsRequest;
+import java.lang.reflect.Field;
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Service;
+
+@Service
+@RabbitListener(queues = "#{controllerQueue.name}")
+public class ControllerMQListener {
+
+  @Autowired public ThreadPoolTaskExecutor taskExecutor;
+
+  @Autowired private ApplicationContext context;
+
+  @RabbitHandler
+  public void receive(SetMaxThreadsRequest in) throws Exception {
+    try {
+      ThreadPoolTaskExecutor myBean = context.getBean(ThreadPoolTaskExecutor.class);
+      Field maxPoolSize = ThreadPoolTaskExecutor.class.getDeclaredField("maxPoolSize");
+      maxPoolSize.setAccessible(true);
+      maxPoolSize.set(myBean, in.getMaxThreads());
+      Field corePoolSize = ThreadPoolTaskExecutor.class.getDeclaredField("corePoolSize");
+      corePoolSize.setAccessible(true);
+      corePoolSize.set(myBean, in.getMaxThreads());
+    } catch (Exception e) {
+      System.out.println(e.getMessage());
+      e.printStackTrace();
+    }
+  }
+}
diff --git a/services/contracts/src/main/java/com/workup/contracts/RabbitMQListener.java b/services/contracts/src/main/java/com/workup/contracts/RabbitMQListener.java
@@ -22,9 +22,11 @@
 import com.workup.shared.commands.contracts.responses.ProgressMilestoneResponse;
 import com.workup.shared.commands.contracts.responses.ViewContractMilestonesResponse;
 import com.workup.shared.enums.ServiceQueueNames;
+import java.util.concurrent.CompletableFuture;
 import org.springframework.amqp.rabbit.annotation.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
 @Service
@@ -34,55 +36,77 @@ public class RabbitMQListener {
   @Autowired public ContractCommandMap commandMap;
 
   @RabbitHandler
-  public InitiateContractResponse receive(InitiateContractRequest in) throws Exception {
+  @Async
+  public CompletableFuture<InitiateContractResponse> receive(InitiateContractRequest in)
+      throws Exception {
     InitiateContractResponse response =
         ((InitiateContractCommand) commandMap.getCommand("InitiateContract")).Run(in);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public ContractTerminationResponse receive(ContractTerminationRequest in) throws Exception {
-    return ((RequestContractTerminationCommand) commandMap.getCommand("RequestContractTermination"))
-        .Run(in);
+  @Async
+  public CompletableFuture<ContractTerminationResponse> receive(ContractTerminationRequest in)
+      throws Exception {
+    return CompletableFuture.completedFuture(
+        ((RequestContractTerminationCommand) commandMap.getCommand("RequestContractTermination"))
+            .Run(in));
   }
 
   @RabbitHandler
-  public HandleTerminationResponse receive(HandleTerminationRequest in) throws Exception {
-    return ((HandleTerminationRequestCommand) commandMap.getCommand("HandleTerminationRequest"))
-        .Run(in);
+  @Async
+  public CompletableFuture<HandleTerminationResponse> receive(HandleTerminationRequest in)
+      throws Exception {
+    return CompletableFuture.completedFuture(
+        ((HandleTerminationRequestCommand) commandMap.getCommand("HandleTerminationRequest"))
+            .Run(in));
   }
 
   @RabbitHandler
-  public MarkPaymentCompletedResponse receive(MarkPaymentCompletedRequest in) throws Exception {
-    return ((MarkMilestoneAsPaidCommand) commandMap.getCommand("MarkMilestoneAsPaid")).Run(in);
+  @Async
+  public CompletableFuture<MarkPaymentCompletedResponse> receive(MarkPaymentCompletedRequest in)
+      throws Exception {
+    return CompletableFuture.completedFuture(
+        ((MarkMilestoneAsPaidCommand) commandMap.getCommand("MarkMilestoneAsPaid")).Run(in));
   }
 
   @RabbitHandler
-  public ViewContractMilestonesResponse receive(ViewContractMilestonesRequest in) throws Exception {
-    return ((ViewContractMilestonesCommand) commandMap.getCommand("ViewContractMilestones"))
-        .Run(in);
+  @Async
+  public CompletableFuture<ViewContractMilestonesResponse> receive(ViewContractMilestonesRequest in)
+      throws Exception {
+    return CompletableFuture.completedFuture(
+        ((ViewContractMilestonesCommand) commandMap.getCommand("ViewContractMilestones")).Run(in));
   }
 
   @RabbitHandler
-  public GetContractResponse receive(GetContractRequest in) throws Exception {
-    ContractsLogger.print("** ENTERED GET CONTRACT RABBITMQ", LoggingLevel.TRACE);
-    return ((GetContractCommand) commandMap.getCommand("GetContract")).Run(in);
+  @Async
+  public CompletableFuture<GetContractResponse> receive(GetContractRequest in) throws Exception {
+    return CompletableFuture.completedFuture(
+        ((GetContractCommand) commandMap.getCommand("GetContract")).Run(in));
   }
 
   @RabbitHandler
-  public EvaluateMilestoneResponse receive(EvaluateMilestoneRequest in) throws Exception {
-    return ((EvaluateMilestoneCommand) commandMap.getCommand("EvaluateMilestone")).Run(in);
+  @Async
+  public CompletableFuture<EvaluateMilestoneResponse> receive(EvaluateMilestoneRequest in)
+      throws Exception {
+    return CompletableFuture.completedFuture(
+        ((EvaluateMilestoneCommand) commandMap.getCommand("EvaluateMilestone")).Run(in));
   }
 
   @RabbitHandler
-  public ProgressMilestoneResponse receive(ProgressMilestoneRequest in) throws Exception {
-    return ((ProgressMilestoneCommand) commandMap.getCommand("ProgressMilestone")).Run(in);
+  @Async
+  public CompletableFuture<ProgressMilestoneResponse> receive(ProgressMilestoneRequest in)
+      throws Exception {
+    return CompletableFuture.completedFuture(
+        ((ProgressMilestoneCommand) commandMap.getCommand("ProgressMilestone")).Run(in));
   }
 
   @RabbitHandler
-  public GetPendingTerminationsResponse receive(GetPendingTerminationsRequest in) throws Exception {
-    return ((GetPendingTerminationsCommand) commandMap.getCommand("GetPendingTerminations"))
-        .Run(in);
+  @Async
+  public CompletableFuture<GetPendingTerminationsResponse> receive(GetPendingTerminationsRequest in)
+      throws Exception {
+    return CompletableFuture.completedFuture(
+        ((GetPendingTerminationsCommand) commandMap.getCommand("GetPendingTerminations")).Run(in));
   }
   // NEW_COMMAND_BOILERPLATE
 

diff --git a/services/jobs/pom.xml b/services/jobs/pom.xml
@@ -18,9 +18,30 @@
         <java.version>21</java.version>
     </properties>
     <dependencies>
+    <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>com.lmax</groupId>
+            <artifactId>disruptor</artifactId>
+            <version>3.4.2</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-amqp</artifactId>
+              <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>com.workup</groupId>
@@ -80,6 +101,10 @@
             <version>1.18.30</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-log4j2</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

diff --git a/services/jobs/src/main/java/com/workup/jobs/ControllerMQListener.java b/services/jobs/src/main/java/com/workup/jobs/ControllerMQListener.java
@@ -22,8 +22,6 @@ public class ControllerMQListener {
 
   @RabbitHandler
   public void receive(SetMaxThreadsRequest in) throws Exception {
-    System.out.println("Autowire: " + taskExecutor.getMaxPoolSize());
-    System.out.println("Input: " + in.getMaxThreads());
     try {
       ThreadPoolTaskExecutor myBean = context.getBean(ThreadPoolTaskExecutor.class);
       Field maxPoolSize = ThreadPoolTaskExecutor.class.getDeclaredField("maxPoolSize");

diff --git a/services/jobs/src/main/java/com/workup/jobs/JobsApplication.java b/services/jobs/src/main/java/com/workup/jobs/JobsApplication.java
@@ -5,6 +5,7 @@
 import org.springframework.amqp.core.*;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.boot.ApplicationRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.annotation.Bean;
@@ -19,6 +20,14 @@ public static void main(String[] args) {
     SpringApplication.run(JobsApplication.class, args);
   }
 
+  @Bean
+  public ApplicationRunner runner(AmqpTemplate template) {
+    return args -> {
+      System.out.println("ApplicationRunner is executing");
+      // Configurator.setLevel("com.workup.jobs", org.apache.logging.log4j.Level.ERROR);
+    };
+  }
+
   @Bean
   public Queue myQueue() {
     return new Queue(ServiceQueueNames.JOBS);

diff --git a/services/jobs/src/main/java/com/workup/jobs/RabbitMQListener.java b/services/jobs/src/main/java/com/workup/jobs/RabbitMQListener.java
@@ -26,9 +26,11 @@
 import com.workup.shared.commands.jobs.responses.GetMyJobsResponse;
 import com.workup.shared.commands.jobs.responses.SearchJobsResponse;
 import com.workup.shared.enums.ServiceQueueNames;
+import java.util.concurrent.CompletableFuture;
 import org.springframework.amqp.rabbit.annotation.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
 @Service
@@ -38,56 +40,69 @@ public class RabbitMQListener {
   @Autowired public JobCommandMap commandMap;
 
   @RabbitHandler
-  public CreateJobResponse receive(CreateJobRequest in) throws Exception {
-    return ((CreateJobCommand) commandMap.getCommand("CreateJob")).Run(in);
+  @Async
+  public CompletableFuture<CreateJobResponse> receive(CreateJobRequest in) throws Exception {
+    CreateJobResponse response = ((CreateJobCommand) commandMap.getCommand("CreateJob")).Run(in);
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public CreateProposalResponse receive(CreateProposalRequest in) throws Exception {
+  @Async
+  public CompletableFuture<CreateProposalResponse> receive(CreateProposalRequest in)
+      throws Exception {
     CreateProposalResponse response =
         ((CreateProposalCommand) commandMap.getCommand("CreateProposal")).Run(in);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public GetJobByIdResponse receive(GetJobByIdRequest request) throws Exception {
+  @Async
+  public CompletableFuture<GetJobByIdResponse> receive(GetJobByIdRequest request) throws Exception {
     GetJobByIdResponse response =
         ((GetJobByIdCommand) commandMap.getCommand("GetJobById")).Run(request);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public SearchJobsResponse receive(SearchJobsRequest request) throws Exception {
+  @Async
+  public CompletableFuture<SearchJobsResponse> receive(SearchJobsRequest request) throws Exception {
     SearchJobsResponse response =
         ((SearchJobsCommand) commandMap.getCommand("SearchJobs")).Run(request);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public GetMyJobsResponse receive(GetMyJobsRequest request) throws Exception {
+  @Async
+  public CompletableFuture<GetMyJobsResponse> receive(GetMyJobsRequest request) throws Exception {
     GetMyJobsResponse response =
         ((GetMyJobsCommand) commandMap.getCommand("GetMyJobs")).Run(request);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public AcceptProposalResponse receive(AcceptProposalRequest request) throws Exception {
+  @Async
+  public CompletableFuture<AcceptProposalResponse> receive(AcceptProposalRequest request)
+      throws Exception {
     AcceptProposalResponse response =
         ((AcceptProposalCommand) commandMap.getCommand("AcceptProposal")).Run(request);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public GetProposalsByJobIdResponse receive(GetProposalsByJobIdRequest request) throws Exception {
+  @Async
+  public CompletableFuture<GetProposalsByJobIdResponse> receive(GetProposalsByJobIdRequest request)
+      throws Exception {
     GetProposalsByJobIdResponse response =
         ((GetProposalsByJobIdCommand) commandMap.getCommand("GetProposalsByJobId")).Run(request);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 
   @RabbitHandler
-  public GetMyProposalsResponse receive(GetMyProposalsRequest request) throws Exception {
+  @Async
+  public CompletableFuture<GetMyProposalsResponse> receive(GetMyProposalsRequest request)
+      throws Exception {
     GetMyProposalsResponse response =
         ((GetMyProposalsCommand) commandMap.getCommand("GetMyProposals")).Run(request);
-    return response;
+    return CompletableFuture.completedFuture(response);
   }
 }