Refactor Flow End in SDKs (#3171)

- **Refactor** - Improved the efficiency of the `end` method by adding conditional checks to prevent unnecessary processing. - Cleaned up the codebase by removing redundant lines in the asynchronous flow management. --------- Co-authored-by: Hardik Shingala <[email protected]>
fluxninja · Jan 30, 2024 · f0ca2d7 · f0ca2d7
1 parent a135d73
commit f0ca2d7
Show file tree

Hide file tree

Showing 8 changed files with 47 additions and 5 deletions.
diff --git a/sdks/aperture-csharp/Core/FeatureFlow.cs b/sdks/aperture-csharp/Core/FeatureFlow.cs
@@ -259,6 +259,10 @@ public FeatureFlowEndResponse End()
 
             if (_checkResponse.LimiterDecisions[i].ConcurrencyLimiterInfo != null)
             {
+                if (_checkResponse.LimiterDecisions[i].ConcurrencyLimiterInfo.RequestId == "") {
+                    continue;
+                }
+
                 var inflightRequest = new InflightRequestRef
                 {
                     PolicyHash = _checkResponse.LimiterDecisions[i].PolicyHash,
@@ -277,6 +281,10 @@ public FeatureFlowEndResponse End()
 
             if (_checkResponse.LimiterDecisions[i].ConcurrencySchedulerInfo != null)
             {
+                if (_checkResponse.LimiterDecisions[i].ConcurrencySchedulerInfo.RequestId == "") {
+                    continue;
+                }
+
                 var inflightRequest = new InflightRequestRef
                 {
                     PolicyHash = _checkResponse.LimiterDecisions[i].PolicyHash,

diff --git a/sdks/aperture-go/sdk/flow.go b/sdks/aperture-go/sdk/flow.go
@@ -314,6 +314,9 @@ func (f *flow) End() EndResponse {
 
 	for _, decision := range f.checkResponse.GetLimiterDecisions() {
 		if decision.GetConcurrencyLimiterInfo() != nil {
+			if decision.GetConcurrencyLimiterInfo().GetRequestId() == "" {
+				continue
+			}
 			inflightRequest := &checkv1.InflightRequestRef{
 				PolicyName:  decision.PolicyName,
 				PolicyHash:  decision.PolicyHash,
@@ -324,11 +327,13 @@ func (f *flow) End() EndResponse {
 			if decision.GetConcurrencyLimiterInfo().GetTokensInfo() != nil {
 				inflightRequest.Tokens = decision.GetConcurrencyLimiterInfo().GetTokensInfo().GetConsumed()
 			}
-
 			inflightRequests = append(inflightRequests, inflightRequest)
 		}
 
 		if decision.GetConcurrencySchedulerInfo() != nil {
+			if decision.GetConcurrencySchedulerInfo().GetRequestId() == "" {
+				continue
+			}
 			ref := &checkv1.InflightRequestRef{
 				PolicyName:  decision.PolicyName,
 				PolicyHash:  decision.PolicyHash,
@@ -339,7 +344,6 @@ func (f *flow) End() EndResponse {
 			if decision.GetConcurrencySchedulerInfo().GetTokensInfo() != nil {
 				ref.Tokens = decision.GetConcurrencySchedulerInfo().GetTokensInfo().GetConsumed()
 			}
-
 			inflightRequests = append(inflightRequests, ref)
 		}
 	}

diff --git a/sdks/aperture-go/sdk/httpflow.go b/sdks/aperture-go/sdk/httpflow.go
@@ -125,6 +125,9 @@ func (f *httpflow) End() EndResponse {
 
 	for _, decision := range f.checkResponse.GetCheckResponse().GetLimiterDecisions() {
 		if decision.GetConcurrencyLimiterInfo() != nil {
+			if decision.GetConcurrencyLimiterInfo().GetRequestId() == "" {
+				continue
+			}
 			inflightRequest := &checkv1.InflightRequestRef{
 				PolicyName:  decision.PolicyName,
 				PolicyHash:  decision.PolicyHash,
@@ -140,6 +143,9 @@ func (f *httpflow) End() EndResponse {
 		}
 
 		if decision.GetConcurrencySchedulerInfo() != nil {
+			if decision.GetConcurrencySchedulerInfo().GetRequestId() == "" {
+				continue
+			}
 			ref := &checkv1.InflightRequestRef{
 				PolicyName:  decision.PolicyName,
 				PolicyHash:  decision.PolicyHash,

diff --git a/sdks/aperture-java/lib/core/src/main/java/com/fluxninja/aperture/sdk/Flow.java b/sdks/aperture-java/lib/core/src/main/java/com/fluxninja/aperture/sdk/Flow.java
@@ -419,6 +419,10 @@ public EndResponse end() {
 
         for (LimiterDecision decision : this.checkResponse.getLimiterDecisionsList()) {
             if (decision.getConcurrencyLimiterInfo() != null) {
+                if (decision.getConcurrencyLimiterInfo().getRequestId().isEmpty()) {
+                    continue;
+                }
+
                 InflightRequestRef.Builder refBuilder =
                         InflightRequestRef.newBuilder()
                                 .setPolicyName(decision.getPolicyName())
@@ -435,6 +439,10 @@ public EndResponse end() {
             }
 
             if (decision.getConcurrencySchedulerInfo() != null) {
+                if (decision.getConcurrencySchedulerInfo().getRequestId().isEmpty()) {
+                    continue;
+                }
+
                 InflightRequestRef.Builder refBuilder =
                         InflightRequestRef.newBuilder()
                                 .setPolicyName(decision.getPolicyName())

diff --git a/sdks/aperture-java/lib/core/src/main/java/com/fluxninja/aperture/sdk/TrafficFlow.java b/sdks/aperture-java/lib/core/src/main/java/com/fluxninja/aperture/sdk/TrafficFlow.java
@@ -182,6 +182,10 @@ public EndResponse end() {
         for (LimiterDecision decision :
                 this.checkResponse.getCheckResponse().getLimiterDecisionsList()) {
             if (decision.getConcurrencyLimiterInfo() != null) {
+                if (decision.getConcurrencyLimiterInfo().getRequestId().isEmpty()) {
+                    continue;
+                }
+
                 InflightRequestRef.Builder refBuilder =
                         InflightRequestRef.newBuilder()
                                 .setPolicyName(decision.getPolicyName())
@@ -198,6 +202,10 @@ public EndResponse end() {
             }
 
             if (decision.getConcurrencySchedulerInfo() != null) {
+                if (decision.getConcurrencySchedulerInfo().getRequestId().isEmpty()) {
+                    continue;
+                }
+
                 InflightRequestRef.Builder refBuilder =
                         InflightRequestRef.newBuilder()
                                 .setPolicyName(decision.getPolicyName())

diff --git a/sdks/aperture-js/sdk/flow.ts b/sdks/aperture-js/sdk/flow.ts
@@ -97,7 +97,7 @@ export class _Flow implements Flow {
     if (
       (!this.rampMode && this._checkResponse === null) ||
       this._checkResponse?.decisionType ===
-        _aperture_flowcontrol_check_v1_CheckResponse_DecisionType.DECISION_TYPE_ACCEPTED
+      _aperture_flowcontrol_check_v1_CheckResponse_DecisionType.DECISION_TYPE_ACCEPTED
     ) {
       return true;
     } else {
@@ -493,6 +493,9 @@ export class _Flow implements Flow {
 
           this._checkResponse.limiterDecisions.forEach((decision) => {
             if (decision.concurrencyLimiterInfo) {
+              if (decision.concurrencyLimiterInfo.requestId == "") {
+                return;
+              }
               inflightRequestRefs.push({
                 policyName: decision.policyName,
                 policyHash: decision.policyHash,
@@ -503,6 +506,9 @@ export class _Flow implements Flow {
               });
             }
             if (decision.concurrencySchedulerInfo) {
+              if (decision.concurrencySchedulerInfo.requestId == "") {
+                return;
+              }
               inflightRequestRefs.push({
                 policyName: decision.policyName,
                 policyHash: decision.policyHash,

diff --git a/sdks/aperture-py/aperture_sdk/flow.py b/sdks/aperture-py/aperture_sdk/flow.py
@@ -135,6 +135,8 @@ def end(self) -> EndResponse:
         if self.check_response:
             for decision in self.check_response.limiter_decisions:
                 if decision.WhichOneof("details") == "concurrency_limiter_info":
+                    if decision.concurrency_limiter_info.request_id == "":
+                        continue
                     ref: InflightRequestRef = InflightRequestRef(
                         policy_name=decision.policy_name,
                         policy_hash=decision.policy_hash,
@@ -148,6 +150,8 @@ def end(self) -> EndResponse:
                         )
                     inflight_request_ref.append(ref)
                 elif decision.WhichOneof("details") == "concurrency_scheduler_info":
+                    if decision.concurrency_scheduler_info.request_id == "":
+                        continue
                     ref: InflightRequestRef = InflightRequestRef(
                         policy_name=decision.policy_name,
                         policy_hash=decision.policy_hash,

diff --git a/sdks/aperture-py/aperture_sdk/flow_async.py b/sdks/aperture-py/aperture_sdk/flow_async.py
@@ -137,7 +137,6 @@ async def end(self) -> EndResponse:
                 if decision.WhichOneof("details") == "concurrency_limiter_info":
                     if decision.concurrency_limiter_info.request_id == "":
                         continue
-
                     ref: InflightRequestRef = InflightRequestRef(
                         policy_name=decision.policy_name,
                         policy_hash=decision.policy_hash,
@@ -153,7 +152,6 @@ async def end(self) -> EndResponse:
                 elif decision.WhichOneof("details") == "concurrency_scheduler_info":
                     if decision.concurrency_scheduler_info.request_id == "":
                         continue
-
                     ref: InflightRequestRef = InflightRequestRef(
                         policy_name=decision.policy_name,
                         policy_hash=decision.policy_hash,