rate limiter implemented inside the outbox processor

RamziHaddad · Dec 28, 2024 · 7c9fc21 · 7c9fc21
1 parent e42b719
commit 7c9fc21
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 10 deletions.
diff --git a/payment/pom.xml b/payment/pom.xml
@@ -53,6 +53,12 @@
             <artifactId>rest-assured</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.quarkiverse.bucket4j</groupId>
+            <artifactId>quarkus-bucket4j</artifactId>
+            <version>1.0.4</version>
+        </dependency>
+
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-jdbc-postgresql</artifactId>
@@ -71,9 +77,9 @@
             <version>3.15.2</version>
         </dependency>
         <dependency>
-    <groupId>io.quarkus</groupId>
-    <artifactId>quarkus-smallrye-fault-tolerance</artifactId>
-</dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-smallrye-fault-tolerance</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>io.quarkus</groupId>

diff --git a/payment/src/main/java/payment/api/PaymentResource.java b/payment/src/main/java/payment/api/PaymentResource.java
@@ -5,6 +5,7 @@
 import java.util.Optional;
 import java.util.UUID;
 
+import io.quarkiverse.bucket4j.runtime.RateLimited;
 import jakarta.inject.Inject;
 import jakarta.ws.rs.Consumes;
 import jakarta.ws.rs.DELETE;
@@ -30,6 +31,7 @@ public class PaymentResource {
     private PaymentService paymentService;
     int i=0 ; 
     @POST
+
     public Response processPayment(PaymentRequestDTO paymentRequest) {
         i++ ; 
         System.out.println(i);

diff --git a/payment/src/main/java/payment/services/outBoxProcessor/PaymentOutBoxProcessorImpl.java b/payment/src/main/java/payment/services/outBoxProcessor/PaymentOutBoxProcessorImpl.java
@@ -40,19 +40,28 @@ public class PaymentOutBoxProcessorImpl implements PaymentOutBoxProcessor {
     BankClient bankClient;
 
     @Override
-    @Retry(maxRetries = 3, delay = 500, jitter = 200) // Retry mechanism for failures
-    @Timeout(1000) // Timeout after 1 second
+    @Retry(maxRetries = 2, delay = 500, jitter = 200) // Retry mechanism for failures
+    @Timeout(500) // Timeout after 1 second
     public Response processPaymentWithRetry(BankPaymentRequest request) {
         return bankClient.makeNewPayment(request);
     }
-
+    int i = 0 ; 
     @Override
     @Scheduled(every = "10s")
-    @Transactional
+    @Transactional(Transactional.TxType.REQUIRES_NEW)
     public void processOutboxEvents() {
         List<PaymentOutBox> unprocessedEvents = boxRepository.findUnprocessedEvents();
-
+        int maxEventsToProcess = 200;
+
+    int processedCount = 0;
         for (PaymentOutBox event : unprocessedEvents) {
+            if (processedCount >= maxEventsToProcess) {
+                i++ ; 
+                System.out.println("oppa 33la sel3t oropa"+i);
+
+
+                break; // Exit loop if we have processed the maximum number of events
+            }
             JsonObject payloadJson = Json.createReader(new StringReader(event.getPayload())).readObject();
 
             BigDecimal amount = payloadJson.getJsonNumber("amount").bigDecimalValue();
@@ -67,6 +76,7 @@ public void processOutboxEvents() {
                 paymentService.cancelPayment(event.getPaymentId());
                 event.setProcessed(true);
                 boxRepository.update(event);
+                processedCount++; 
                 continue;
             }
 
@@ -83,21 +93,25 @@ public void processOutboxEvents() {
 
                 if (response.getStatus() == Response.Status.OK.getStatusCode()) {
                     paymentService.completePayment(payment.getPaymentId());
+                }
+                else if (response.getStatus() == Response.Status.CONFLICT.getStatusCode()) {
+                paymentService.cancelPayment(payment.getPaymentId()) ;  
+                System.out.println("oooooh ya 3alm");
                 } else {
                     paymentService.cancelPayment(payment.getPaymentId());
                 }
 
                 // Mark the event as processed
                 event.setProcessed(true);
                 boxRepository.update(event);
+                processedCount++; 
 
             } catch (Exception e) {
                 System.err.println("Error processing payment: " + e.getMessage());
-
-                // Handle failure due to timeout or other exceptions
                 paymentService.cancelPayment(event.getPaymentId());
                 event.setProcessed(true);
                 boxRepository.update(event);
+                processedCount++; 
             }
         }
     }

diff --git a/payment/src/main/resources/application.properties b/payment/src/main/resources/application.properties
@@ -11,6 +11,14 @@ quarkus.http.cors.origins=*
 mp.fault.tolerance.retry.max-retries=3
 mp.fault.tolerance.retry.delay=500
 mp.fault.tolerance.retry.jitter=200
+# burst protection
+quarkus.rate-limiter.buckets.group1.limits[0].permitted-uses: 10
+quarkus.rate-limiter.buckets.group1.limits[0].period: 1S
+# fair use
+quarkus.rate-limiter.buckets.group1.limits[1].permitted-uses: 100
+quarkus.rate-limiter.buckets.group1.limits[1].period: 5M 
+quarkus.transaction-manager.default-transaction-timeout=120
+