feat: add upstream request processing

Author: mattnichols

context/src/main/java/com/mx/path/core/context/RequestContext.java

@@ -3,6 +3,7 @@
 import java.util.function.Consumer;
 
 import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
 import lombok.Setter;
 import lombok.experimental.SuperBuilder;
@@ -43,9 +44,20 @@ public static RequestContext current() {
   private String userGuid;
   @Deprecated
   private String userId;
+
+  /**
+   * Headers for current request
+   */
   private SingleValueMap<String, Object> headers;
+
+  /**
+   * Function parameters for current request
+   */
   private SingleValueMap<String, Object> params;
 
+  @Builder.Default
+  private UpstreamRequestConfiguration upstreamRequestConfiguration = new UpstreamRequestConfiguration();
+
   public abstract static class RequestContextBuilder<C extends RequestContext, B extends RequestContext.RequestContextBuilder<C, B>> {
 
     private SingleValueMap<String, Object> headers = new SingleValueMap<>();

context/src/main/java/com/mx/path/core/context/UpstreamRequestConfiguration.java

@@ -0,0 +1,51 @@
+package com.mx.path.core.context;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+
+import com.mx.path.core.common.collection.SingleValueMap;
+
+/**
+ * Configuration to be applied to all upstream requests within {@link RequestContext#getUpstreamRequestConfiguration()}
+ */
+@Data
+@Builder
+@AllArgsConstructor
+public class UpstreamRequestConfiguration {
+
+  public UpstreamRequestConfiguration() {
+    forwardedRequestHeaders = new SingleValueMap<>();
+    upstreamRequestProcessors = new ArrayList<>();
+  }
+
+  /**
+   * Headers to be forwarded in all upstream {@link com.mx.path.core.common.connect.Request} executes
+   */
+  private SingleValueMap<String, Object> forwardedRequestHeaders;
+
+  /**
+   * Instances of upstream processors that will be invoked in all upstream {@link com.mx.path.core.common.connect.Request} executes
+   */
+  private List<UpstreamRequestProcessor> upstreamRequestProcessors;
+
+  public static class UpstreamRequestConfigurationBuilder {
+    private SingleValueMap<String, Object> forwardedRequestHeaders = new SingleValueMap<>();
+    private List<UpstreamRequestProcessor> upstreamRequestProcessors = new ArrayList<>();
+
+    public final UpstreamRequestConfigurationBuilder forwardedHeader(String key, String value) {
+      forwardedRequestHeaders.put(key, value);
+
+      return this;
+    }
+
+    public final UpstreamRequestConfigurationBuilder upstreamRequestProcessor(UpstreamRequestProcessor processor) {
+      upstreamRequestProcessors.add(processor);
+
+      return this;
+    }
+  }
+}

context/src/main/java/com/mx/path/core/context/UpstreamRequestProcessor.java

@@ -0,0 +1,28 @@
+package com.mx.path.core.context;
+
+import java.util.function.BiConsumer;
+
+import lombok.Builder;
+import lombok.Data;
+
+import com.mx.path.core.common.connect.Request;
+import com.mx.path.core.common.connect.Response;
+
+@Data
+@Builder
+public class UpstreamRequestProcessor {
+  private BiConsumer<Request<?, ?>, Response<?, ?>> after;
+  private BiConsumer<Request<?, ?>, Response<?, ?>> before;
+
+  public final void executeAfter(Request<?, ?> request, Response<?, ?> response) {
+    if (after != null) {
+      after.accept(request, response);
+    }
+  }
+
+  public final void executeBefore(Request<?, ?> request, Response<?, ?> response) {
+    if (before != null) {
+      before.accept(request, response);
+    }
+  }
+}

context/src/test/groovy/com/mx/path/core/context/RequestContextTest.groovy

@@ -64,6 +64,9 @@ class RequestContextTest extends Specification {
     subject.getHeaders().get("h1") == "v1"
     subject.getParams().get("p1") == "v1"
     subject.getParams().get("p2") == "v2"
+
+    and: "UpstreamRequestConfiguration is initialized"
+    subject.getUpstreamRequestConfiguration() != null
   }
 
   def "withSelfClearing"() {
@@ -95,6 +98,7 @@ class RequestContextTest extends Specification {
   def "setters"() {
     given:
     def subject = RequestContext.builder().build()
+    def upstreamRequestConfiguration = new UpstreamRequestConfiguration()
 
     when:
     subject.setClientGuid("clientAF")
@@ -103,6 +107,7 @@ class RequestContextTest extends Specification {
     subject.setOriginatingIP("127.0.0.1")
     subject.setPath("/accounts")
     subject.setSessionTraceId("abcde")
+    subject.setUpstreamRequestConfiguration(upstreamRequestConfiguration)
     subject.setUserGuid("userAF")
     subject.getHeaders().put("h2", "v2")
     subject.getHeaders().put("h1", "v1")
@@ -116,6 +121,7 @@ class RequestContextTest extends Specification {
     subject.getOriginatingIP() == "127.0.0.1"
     subject.getPath() == "/accounts"
     subject.getSessionTraceId() == "abcde"
+    subject.getUpstreamRequestConfiguration() == upstreamRequestConfiguration
     subject.getUserGuid() == "userAF"
     subject.getHeaders().get("h2") == "v2"
     subject.getHeaders().get("h1") == "v1"
@@ -126,6 +132,7 @@ class RequestContextTest extends Specification {
   def "toBuilder"() {
     given:
     def subject = RequestContext.builder().build()
+    def upstreamRequestConfiguration = new UpstreamRequestConfiguration()
 
     when:
     subject.setClientGuid("clientAF")
@@ -134,6 +141,7 @@ class RequestContextTest extends Specification {
     subject.setOriginatingIP("127.0.0.1")
     subject.setPath("/accounts")
     subject.setSessionTraceId("abcde")
+    subject.setUpstreamRequestConfiguration(upstreamRequestConfiguration)
     subject.setUserGuid("userAF")
     subject.getHeaders().put("h2", "v2")
     subject.getHeaders().put("h1", "v1")
@@ -148,6 +156,7 @@ class RequestContextTest extends Specification {
     result.getOriginatingIP() == "127.0.0.1"
     result.getPath() == "/accounts"
     result.getSessionTraceId() == "abcde"
+    subject.getUpstreamRequestConfiguration() == upstreamRequestConfiguration
     result.getUserGuid() == "userAF"
     result.getHeaders().get("h2") == "v2"
     result.getHeaders().get("h1") == "v1"

context/src/test/groovy/com/mx/path/core/context/UpstreamRequestConfigurationTest.groovy

@@ -0,0 +1,34 @@
+package com.mx.path.core.context
+
+import spock.lang.Specification
+
+class UpstreamRequestConfigurationTest extends Specification {
+  def "constructor"() {
+    when:
+    def subject = new UpstreamRequestConfiguration()
+
+    then:
+    subject.getForwardedRequestHeaders().isEmpty()
+    subject.getUpstreamRequestProcessors().isEmpty()
+  }
+
+  def "builder"() {
+    when: "take defaults"
+    def subject = UpstreamRequestConfiguration.builder().build()
+
+    then: "collections are initialized"
+    subject.getForwardedRequestHeaders().isEmpty()
+    subject.getUpstreamRequestProcessors().isEmpty()
+
+    when:
+    def upstreamRequestProcessors = UpstreamRequestProcessor.builder().build()
+    subject = UpstreamRequestConfiguration.builder()
+        .forwardedHeader("h1", "v1")
+        .upstreamRequestProcessor(upstreamRequestProcessors)
+        .build()
+
+    then:
+    subject.getForwardedRequestHeaders().get("h1") == "v1"
+    subject.getUpstreamRequestProcessors() == [upstreamRequestProcessors]
+  }
+}

context/src/test/groovy/com/mx/path/core/context/UpstreamRequestProcessorTest.groovy

@@ -0,0 +1,46 @@
+package com.mx.path.core.context
+
+import spock.lang.Specification
+
+class UpstreamRequestProcessorTest extends Specification {
+  def "builder"() {
+    when:
+    def subject = UpstreamRequestProcessor.builder().before({}).after({}).build()
+
+    then:
+    subject.getBefore() != null
+    subject.getAfter() != null
+  }
+
+  def "executeBefore"() {
+    given:
+    def beforeCalled = false
+
+    def subject = UpstreamRequestProcessor.builder().before({ request, response ->
+      beforeCalled = true
+    }).build()
+
+    when:
+    subject.executeBefore(null, null)
+    subject.executeAfter(null, null)
+
+    then:
+    beforeCalled
+  }
+
+  def "executeAfter"() {
+    given:
+    def afterCalled = false
+
+    def subject = UpstreamRequestProcessor.builder().after({ request, response ->
+      afterCalled = true
+    }).build()
+
+    when:
+    subject.executeBefore(null, null)
+    subject.executeAfter(null, null)
+
+    then:
+    afterCalled
+  }
+}

gateway/src/main/java/com/mx/path/gateway/configuration/AccessorStackConfigurator.java

@@ -35,6 +35,7 @@
 import com.mx.path.gateway.connect.filter.RequestFinishedFilter;
 import com.mx.path.gateway.connect.filter.TracingFilter;
 import com.mx.path.gateway.connect.filter.UpstreamRequestEventFilter;
+import com.mx.path.gateway.connect.filter.UpstreamRequestProcessorFilter;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -198,6 +199,7 @@ private void buildConnections(ObjectMap map, AccessorConfiguration.AccessorConfi
       connection.baseRequestFilter(new UpstreamRequestEventFilter());
       connection.baseRequestFilter(new ErrorHandlerFilter());
       connection.baseRequestFilter(new CallbacksFilter());
+      connection.baseRequestFilter(new UpstreamRequestProcessorFilter());
       connection.baseRequestFilter(new RequestFinishedFilter());
       connection.baseRequestFilter(new FaultTolerantRequestFilter());
 

gateway/src/main/java/com/mx/path/gateway/connect/filter/UpstreamRequestProcessorFilter.java

@@ -0,0 +1,34 @@
+package com.mx.path.gateway.connect.filter;
+
+import java.util.List;
+
+import com.mx.path.core.common.connect.Request;
+import com.mx.path.core.common.connect.RequestFilterBase;
+import com.mx.path.core.common.connect.Response;
+import com.mx.path.core.context.RequestContext;
+import com.mx.path.core.context.UpstreamRequestProcessor;
+
+public class UpstreamRequestProcessorFilter extends RequestFilterBase {
+  @Override
+  public final void execute(Request request, Response response) {
+    List<UpstreamRequestProcessor> processors = null;
+
+    if (RequestContext.current() != null && RequestContext.current().getUpstreamRequestConfiguration() != null) {
+      processors = RequestContext.current().getUpstreamRequestConfiguration().getUpstreamRequestProcessors();
+    }
+
+    if (processors != null && !processors.isEmpty()) {
+      processors.forEach(processor -> {
+        processor.executeBefore(request, response);
+      });
+
+      next(request, response);
+
+      processors.forEach(processor -> {
+        processor.executeAfter(request, response);
+      });
+    } else {
+      next(request, response);
+    }
+  }
+}

gateway/src/test/groovy/com/mx/path/gateway/connect/filter/UpstreamRequestProcessorFilterTest.groovy

@@ -0,0 +1,40 @@
+package com.mx.path.gateway.connect.filter
+
+import com.mx.path.core.context.RequestContext
+import com.mx.path.core.context.UpstreamRequestConfiguration
+import com.mx.path.core.context.UpstreamRequestProcessor
+
+import spock.lang.Specification
+
+class UpstreamRequestProcessorFilterTest extends Specification {
+  UpstreamRequestProcessorFilter subject
+
+  def setup() {
+    subject = new UpstreamRequestProcessorFilter()
+  }
+
+  def "executes before and after"() {
+    given:
+    def beforeCalled = false
+    def afterCalled = false
+
+    RequestContext.builder()
+        .upstreamRequestConfiguration(
+        UpstreamRequestConfiguration.builder()
+        .upstreamRequestProcessor(UpstreamRequestProcessor.builder()
+        .before({ request, response -> beforeCalled = true})
+        .after({ request, response -> afterCalled = true})
+        .build()
+        ).build()
+        )
+        .build()
+        .register()
+
+    when:
+    subject.execute(null, null)
+
+    then:
+    beforeCalled
+    afterCalled
+  }
+}