diff --git a/build.gradle b/build.gradle
index 21d42af..920db3a 100644
--- a/build.gradle
+++ b/build.gradle
@@ -75,6 +75,13 @@ tasks.named('test') {
useJUnitPlatform()
}
+tasks.register('loadGen', JavaExec) {
+ group = 'verification'
+ description = 'Run the plain gRPC load generator against an extproc server.'
+ mainClass = 'com.github.wrossmorrow.extproc.loadgen.LoadGen'
+ classpath = sourceSets.main.runtimeClasspath
+}
+
test {
useJUnit()
useJUnitPlatform()
diff --git a/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java
new file mode 100644
index 0000000..a1ecdff
--- /dev/null
+++ b/src/main/java/com/github/wrossmorrow/extproc/loadgen/LoadGen.java
@@ -0,0 +1,376 @@
+package com.github.wrossmorrow.extproc.loadgen;
+
+import build.buf.gen.envoy.config.core.v3.HeaderMap;
+import build.buf.gen.envoy.config.core.v3.HeaderValue;
+import build.buf.gen.envoy.config.core.v3.HeaderValueOption;
+import build.buf.gen.envoy.service.ext_proc.v3.CommonResponse;
+import build.buf.gen.envoy.service.ext_proc.v3.ExternalProcessorGrpc;
+import build.buf.gen.envoy.service.ext_proc.v3.HttpBody;
+import build.buf.gen.envoy.service.ext_proc.v3.HttpHeaders;
+import build.buf.gen.envoy.service.ext_proc.v3.ProcessingRequest;
+import build.buf.gen.envoy.service.ext_proc.v3.ProcessingResponse;
+import com.google.protobuf.ByteString;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.stub.StreamObserver;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+/**
+ * Plain gRPC load generator for the envoy ext_proc SDK (Java).
+ *
+ * Drives full ext_proc exchanges over a raw bidirectional gRPC stream to the SDK's ext_proc
+ * server (`envoy.service.ext_proc.v3.ExternalProcessor/Process`), without an envoy in the path, and
+ * reports per-exchange latency and errors.
+ *
+ *
Each exchange mirrors the phase sequence the envoy ext_proc filter drives for a buffered
+ * request/response: request headers -> request body -> response headers -> response body, then the
+ * stream is closed. Because envoy (and the HTTP machinery in between) are absent, the reported
+ * latency is the processor-side cost of the SDK's gRPC server path, complementing the containerized
+ * k6 load tests in `loadtest` that drive the full envoy -> extproc -> upstream chain.
+ */
+public class LoadGen {
+
+ private static final Logger logger = Logger.getLogger(LoadGen.class.getName());
+
+ /**
+ * The response header the server may set (via the ProcessingOptions.UpdateDurationHeader option)
+ * carrying the accumulated extproc processing time in nanoseconds.
+ */
+ static final String DURATION_HEADER = "x-extproc-duration-ns";
+
+ static final String DEFAULT_ADDR = "127.0.0.1:50051";
+ static final int DEFAULT_ITERATIONS = 100;
+ static final int DEFAULT_CONCURRENCY = 10;
+
+ public static void main(String[] args) throws Exception {
+ String addr = DEFAULT_ADDR;
+ int iterations = DEFAULT_ITERATIONS;
+ int concurrency = DEFAULT_CONCURRENCY;
+ for (int i = 0; i < args.length; i++) {
+ String arg = args[i];
+ if (arg.startsWith("--addr=")) {
+ addr = arg.substring("--addr=".length());
+ } else if (arg.startsWith("--iterations=")) {
+ iterations = Integer.parseInt(arg.substring("--iterations=".length()));
+ } else if (arg.startsWith("--concurrency=")) {
+ concurrency = Integer.parseInt(arg.substring("--concurrency=".length()));
+ }
+ }
+
+ if (iterations <= 0) {
+ System.err.println("LoadGen: iterations must be greater than zero");
+ System.exit(1);
+ }
+
+ System.out.println("LoadGen: target extproc gRPC server: http://" + addr);
+ System.out.println(
+ "LoadGen: running " + iterations + " exchanges with concurrency " + concurrency);
+
+ int port = 50051;
+ String host = addr;
+ int colon = addr.lastIndexOf(':');
+ if (colon >= 0) {
+ host = addr.substring(0, colon);
+ port = Integer.parseInt(addr.substring(colon + 1));
+ }
+ ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();
+ ExternalProcessorGrpc.ExternalProcessorStub stub = ExternalProcessorGrpc.newStub(channel);
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ try {
+ List workerResults = new ArrayList<>();
+ List> futures = new ArrayList<>();
+ for (int i = 0; i < concurrency; i++) {
+ final int base = i;
+ final ExternalProcessorGrpc.ExternalProcessorStub workerStub = stub;
+ final int workerIterations = iterations;
+ final int workerConcurrency = concurrency;
+ futures.add(
+ executor.submit(
+ () -> runWorker(workerStub, base, workerConcurrency, workerIterations)));
+ }
+ for (Future future : futures) {
+ workerResults.add(future.get(30, TimeUnit.SECONDS));
+ }
+
+ List latencies = new ArrayList<>();
+ List extprocDurationsNs = new ArrayList<>();
+ List errors = new ArrayList<>();
+ for (WorkerResult result : workerResults) {
+ latencies.addAll(result.latencies);
+ extprocDurationsNs.addAll(result.extprocDurationsNs);
+ errors.addAll(result.errors);
+ }
+ latencies.sort((left, right) -> left.compareTo(right));
+ double avgMs = 0.0;
+ if (!latencies.isEmpty()) {
+ Duration sum = Duration.ZERO;
+ for (Duration d : latencies) {
+ sum = sum.plus(d);
+ }
+ avgMs = sum.toMillis() / (double) latencies.size();
+ }
+ Double p50Ms = percentile(latencies, 0.50);
+ Double p95Ms = percentile(latencies, 0.95);
+ double maxMs = latencies.isEmpty() ? 0.0 : latencies.get(latencies.size() - 1).toMillis();
+ double avgDurationMs = 0.0;
+ if (!extprocDurationsNs.isEmpty()) {
+ long sumNs = 0;
+ for (long ns : extprocDurationsNs) {
+ sumNs += ns;
+ }
+ avgDurationMs = sumNs / ((double) extprocDurationsNs.size() * 1e6);
+ }
+
+ for (ExchangeError e : errors) {
+ System.out.println("LoadGen: exchange " + e.index + " failed: " + e.error);
+ }
+ System.out.println(
+ "LoadGen: finished: " + latencies.size() + " exchanges, " + errors.size() + " errors");
+ System.out.println(
+ "LoadGen: latency avg="
+ + avgMs
+ + "ms p50="
+ + p50Ms
+ + "ms p95="
+ + p95Ms
+ + "ms max="
+ + maxMs
+ + "ms");
+ System.out.println(
+ "LoadGen: extproc duration avg="
+ + avgDurationMs
+ + "ms ("
+ + extprocDurationsNs.size()
+ + " samples)");
+
+ if (!errors.isEmpty()) {
+ System.exit(1);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+
+ channel.shutdownNow();
+ }
+
+ static WorkerResult runWorker(
+ ExternalProcessorGrpc.ExternalProcessorStub stub, int base, int stride, int iterations) {
+ WorkerResult result = new WorkerResult(new ArrayList<>(), new ArrayList<>(), new ArrayList<>());
+ int index = base;
+ while (index < iterations) {
+ ExchangeOutcome outcome = runExchange(stub);
+ if (outcome.error != null) {
+ result.errors.add(new ExchangeError(index, outcome.error));
+ } else {
+ result.latencies.add(outcome.latency);
+ if (outcome.extprocDurationNs != null) {
+ result.extprocDurationsNs.add(outcome.extprocDurationNs);
+ }
+ }
+ index += stride;
+ }
+ return result;
+ }
+
+ static ExchangeOutcome runExchange(ExternalProcessorGrpc.ExternalProcessorStub stub) {
+ Instant start = Instant.now();
+ ResponseCollector responseObserver = new ResponseCollector();
+ StreamObserver requestObserver = stub.process(responseObserver);
+
+ // phase 1: request headers (not yet end of stream)
+ System.out.println("LoadGen: exchange starting, sending phase 1");
+ responseObserver.latch = new CountDownLatch(1);
+ requestObserver.onNext(
+ ProcessingRequest.newBuilder()
+ .setRequestHeaders(
+ HttpHeaders.newBuilder()
+ .setHeaders(requestHeaderMap("POST", "/test"))
+ .setEndOfStream(false)
+ .build())
+ .build());
+ awaitExchange(responseObserver);
+ ProcessingResponse response = responseObserver.lastResponse;
+ System.out.println("LoadGen: got response for phase 1: " + response.getResponseCase());
+
+ // phase 2: request body (end of stream)
+ responseObserver.latch = new CountDownLatch(1);
+ requestObserver.onNext(
+ ProcessingRequest.newBuilder()
+ .setRequestBody(
+ HttpBody.newBuilder()
+ .setBody(ByteString.copyFromUtf8("{}"))
+ .setEndOfStream(true)
+ .build())
+ .build());
+ awaitExchange(responseObserver);
+ response = responseObserver.lastResponse;
+ System.out.println("LoadGen: got response for phase 2: " + response.getResponseCase());
+
+ // phase 3: response headers (end of stream)
+ responseObserver.latch = new CountDownLatch(1);
+ requestObserver.onNext(
+ ProcessingRequest.newBuilder()
+ .setResponseHeaders(HttpHeaders.newBuilder().setEndOfStream(true).build())
+ .build());
+ awaitExchange(responseObserver);
+ response = responseObserver.lastResponse;
+ System.out.println("LoadGen: got response for phase 3: " + response.getResponseCase());
+
+ // phase 4: response body (end of stream)
+ responseObserver.latch = new CountDownLatch(1);
+ requestObserver.onNext(
+ ProcessingRequest.newBuilder()
+ .setResponseBody(
+ HttpBody.newBuilder()
+ .setBody(ByteString.copyFromUtf8("{}"))
+ .setEndOfStream(true)
+ .build())
+ .build());
+ requestObserver.onCompleted();
+ awaitExchange(responseObserver);
+ response = responseObserver.lastResponse;
+ System.out.println("LoadGen: got response for phase 4: " + response.getResponseCase());
+
+ Long extprocDurationNs = extractDurationNs(response);
+ Duration latency = Duration.between(start, Instant.now());
+ String error = responseObserver.error == null ? null : responseObserver.error.getMessage();
+ return new ExchangeOutcome(latency, extprocDurationNs, error);
+ }
+
+ static void awaitExchange(ResponseCollector responseObserver) {
+ System.out.println("LoadGen: awaiting processing response");
+ try {
+ if (!responseObserver.latch.await(30, TimeUnit.SECONDS)) {
+ throw new RuntimeException("Timed out waiting for processing response");
+ }
+ System.out.println("LoadGen: got processing response");
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for processing response", e);
+ }
+ }
+
+ static Long extractDurationNs(ProcessingResponse response) {
+ if (response == null) {
+ return null;
+ }
+ CommonResponse common;
+ if (response.getResponseCase() == ProcessingResponse.ResponseCase.RESPONSE_HEADERS) {
+ common = response.getResponseHeaders().getResponse();
+ } else if (response.getResponseCase() == ProcessingResponse.ResponseCase.RESPONSE_BODY) {
+ common = response.getResponseBody().getResponse();
+ } else {
+ return null;
+ }
+ if (common == null || !common.hasHeaderMutation()) {
+ return null;
+ }
+ for (HeaderValueOption option : common.getHeaderMutation().getSetHeadersList()) {
+ HeaderValue header = option.getHeader();
+ if (header == null || !DURATION_HEADER.equals(header.getKey())) {
+ continue;
+ }
+ ByteString raw = header.getRawValue();
+ if (raw == null || raw.size() == 0) {
+ return null;
+ }
+ try {
+ return Long.valueOf(
+ new String(raw.toByteArray(), java.nio.charset.StandardCharsets.UTF_8).trim());
+ } catch (Exception e) {
+ return null;
+ }
+ }
+ return null;
+ }
+
+ static HeaderMap requestHeaderMap(String method, String path) {
+ HeaderMap.Builder builder = HeaderMap.newBuilder();
+ builder.addHeaders(
+ HeaderValue.newBuilder()
+ .setKey(":method")
+ .setRawValue(ByteString.copyFromUtf8(method))
+ .build());
+ builder.addHeaders(
+ HeaderValue.newBuilder()
+ .setKey(":path")
+ .setRawValue(ByteString.copyFromUtf8(path))
+ .build());
+ return builder.build();
+ }
+
+ static double percentile(List sorted, double percentile) {
+ if (sorted.isEmpty()) {
+ return 0.0;
+ }
+ int rank = (int) Math.round((double) (sorted.size() - 1) * percentile);
+ return sorted.get(rank).toMillis();
+ }
+
+ static final class ResponseCollector implements StreamObserver {
+ ProcessingResponse lastResponse;
+ Throwable error;
+ boolean completed;
+ CountDownLatch latch = new CountDownLatch(0);
+
+ @Override
+ public void onNext(ProcessingResponse value) {
+ lastResponse = value;
+ latch.countDown();
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ error = t;
+ }
+
+ @Override
+ public void onCompleted() {
+ completed = true;
+ }
+ }
+
+ static final class WorkerResult {
+ final List latencies;
+ final List extprocDurationsNs;
+ final List errors;
+
+ WorkerResult(
+ List latencies, List extprocDurationsNs, List errors) {
+ this.latencies = latencies;
+ this.extprocDurationsNs = extprocDurationsNs;
+ this.errors = errors;
+ }
+ }
+
+ static final class ExchangeError {
+ final int index;
+ final String error;
+
+ ExchangeError(int index, String error) {
+ this.index = index;
+ this.error = error;
+ }
+ }
+
+ static final class ExchangeOutcome {
+ final Duration latency;
+ final Long extprocDurationNs;
+ final String error;
+
+ ExchangeOutcome(Duration latency, Long extprocDurationNs, String error) {
+ this.latency = latency;
+ this.extprocDurationNs = extprocDurationNs;
+ this.error = error;
+ }
+ }
+}
diff --git a/src/main/java/com/github/wrossmorrow/extproc/processors/TestingRequestProcessor.java b/src/main/java/com/github/wrossmorrow/extproc/processors/TestingRequestProcessor.java
new file mode 100644
index 0000000..eafaf62
--- /dev/null
+++ b/src/main/java/com/github/wrossmorrow/extproc/processors/TestingRequestProcessor.java
@@ -0,0 +1,270 @@
+package com.github.wrossmorrow.extproc.processors;
+
+import build.buf.gen.envoy.service.ext_proc.v3.ProcessingRequest.RequestCase;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.wrossmorrow.extproc.ProcessingOptions;
+import com.github.wrossmorrow.extproc.RequestContext;
+import com.github.wrossmorrow.extproc.RequestProcessor;
+import com.github.wrossmorrow.extproc.RequestProcessorHealthManager;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Request processor used by the testing harness. Mirrors the Go and Rust SDKs' tester processors,
+ * which drive full ext_proc exchanges (via the load generator) and mutate the stream through the
+ * processing actions encoded in the request body.
+ */
+public class TestingRequestProcessor implements RequestProcessor {
+
+ private static final Logger logger = Logger.getLogger(TestingRequestProcessor.class.getName());
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ /** Request processing actions applied by the tester across the phases of a single exchange. */
+ public static final class TesterRequestBody {
+
+ @JsonProperty("append_request_headers")
+ public Map appendRequestHeaders;
+
+ @JsonProperty("add_request_headers")
+ public Map addRequestHeaders;
+
+ @JsonProperty("overwrite_request_headers")
+ public Map overwriteRequestHeaders;
+
+ @JsonProperty("remove_request_headers")
+ public List removeRequestHeaders;
+
+ @JsonProperty("clear_request_body")
+ public boolean clearRequestBody;
+
+ @JsonProperty("replace_request_body")
+ public String replaceRequestBody;
+
+ @JsonProperty("append_response_headers")
+ public Map appendResponseHeaders;
+
+ @JsonProperty("add_response_headers")
+ public Map addResponseHeaders;
+
+ @JsonProperty("overwrite_response_headers")
+ public Map overwriteResponseHeaders;
+
+ @JsonProperty("remove_response_headers")
+ public List removeResponseHeaders;
+
+ @JsonProperty("clear_response_body")
+ public boolean clearResponseBody;
+
+ @JsonProperty("replace_response_body")
+ public String replaceResponseBody;
+
+ @JsonProperty("cancel_request")
+ public boolean cancelRequest;
+
+ @JsonProperty("cancel_request_status")
+ public int cancelRequestStatus;
+
+ @JsonProperty("cancel_request_headers")
+ public Map cancelRequestHeaders;
+
+ @JsonProperty("cancel_request_body")
+ public String cancelRequestBody;
+ }
+
+ private final ProcessingOptions options;
+ private volatile TesterRequestBody actions;
+
+ public TestingRequestProcessor() {
+ this(new ProcessingOptions());
+ }
+
+ public TestingRequestProcessor(ProcessingOptions options) {
+ this.options = options;
+ }
+
+ @Override
+ public String getName() {
+ return "testing";
+ }
+
+ @Override
+ public ProcessingOptions getOptions() {
+ return options;
+ }
+
+ @Override
+ public void setHealthManager(RequestProcessorHealthManager health) {}
+
+ @Override
+ public void shutdown() {
+ logger.info(TestingRequestProcessor.class.getSimpleName() + " shutting down");
+ }
+
+ @Override
+ public void close(int gracePeriodSeconds) {
+ logger.info("Closing tester");
+ }
+
+ @Override
+ public void errorHandler(RequestContext ctx, RequestCase phase, Throwable err) {
+ logger.log(Level.INFO, "Error in phase {0}: {1}", new Object[] {phase, err});
+ }
+
+ @Override
+ public void processRequestHeaders(RequestContext ctx, Map headers) {
+ reportHeaderParsing(ctx, headers);
+
+ // optional upstream override, exercising clear-route-cache driven upstream
+ // changes: rewrite the authority and clear the route cache so envoy
+ // recomputes the route
+ String upstream = getHeaderIgnoreCase(headers, "x-route-to-upstream");
+ if (upstream != null) {
+ ctx.overwriteHeader(":authority", upstream);
+ ctx.clearRouteCache();
+ }
+
+ ctx.continueAndReplace();
+ }
+
+ @Override
+ public void processRequestBody(RequestContext ctx, String body) {
+ // NOTE: assuming buffering, not streaming; if buffering we have to
+ // accumulate the request body chunks sent until the end of stream
+
+ TesterRequestBody parsed = parseActions(body);
+ ctx.setValue("actions", parsed);
+
+ if (parsed != null && parsed.cancelRequest) {
+ ctx.cancelRequest(
+ parsed.cancelRequestStatus, parsed.cancelRequestHeaders, parsed.cancelRequestBody);
+ return;
+ }
+
+ if (parsed != null) {
+ applyRequestHeaderActions(ctx, parsed);
+ applyRequestBodyActions(ctx, parsed);
+ }
+
+ ctx.continueAndReplace();
+ }
+
+ @Override
+ public void processRequestTrailers(RequestContext ctx, Map trailers) {
+ ctx.continueAndReplace();
+ }
+
+ @Override
+ public void processResponseHeaders(RequestContext ctx, Map headers) {
+ TesterRequestBody parsed = (TesterRequestBody) ctx.getValue("actions");
+ if (parsed != null) {
+ applyResponseHeaderActions(ctx, parsed);
+ }
+ ctx.continueAndReplace();
+ }
+
+ @Override
+ public void processResponseBody(RequestContext ctx, String body) {
+ TesterRequestBody parsed = (TesterRequestBody) ctx.getValue("actions");
+ if (parsed != null) {
+ applyResponseBodyActions(ctx, parsed);
+ }
+ ctx.continueAndReplace();
+ }
+
+ @Override
+ public void processResponseTrailers(RequestContext ctx, Map trailers) {
+ ctx.continueAndReplace();
+ }
+
+ /** Reflect on how the SDK parsed the inbound headers back via the request headers. */
+ protected void reportHeaderParsing(RequestContext ctx, Map headers) {
+ ctx.addHeader("x-parsed-str-count", String.valueOf(ctx.getAllRequestHeaders().size()));
+ String first = ctx.getAllRequestHeaders().get("x-probe-list");
+ ctx.addHeader("x-parsed-list-value", first == null ? "" : first);
+ List values = ctx.getAllRequestHeaders().values("x-probe-list");
+ ctx.addHeader("x-parsed-list-count", String.valueOf(values.size()));
+ List split = ctx.getAllRequestHeaders().splitList("x-probe-list");
+ ctx.addHeader("x-parsed-list-split", String.join("|", split));
+ }
+
+ protected TesterRequestBody parseActions(String body) {
+ if (body == null || body.isEmpty()) {
+ return null;
+ }
+ try {
+ JsonNode node = MAPPER.readTree(body);
+ if (node == null || node.isNull() || !node.isObject()) {
+ return null;
+ }
+ TesterRequestBody actions = MAPPER.convertValue(node, TesterRequestBody.class);
+ return actions;
+ } catch (Exception e) {
+ logger.log(Level.WARNING, "Error parsing request body: {0}", e.getMessage());
+ return null;
+ }
+ }
+
+ protected void applyRequestHeaderActions(RequestContext ctx, TesterRequestBody actions) {
+ if (actions.appendRequestHeaders != null) {
+ ctx.appendHeaders(actions.appendRequestHeaders);
+ }
+ if (actions.addRequestHeaders != null) {
+ ctx.addHeaders(actions.addRequestHeaders);
+ }
+ if (actions.overwriteRequestHeaders != null) {
+ ctx.overwriteHeaders(actions.overwriteRequestHeaders);
+ }
+ if (actions.removeRequestHeaders != null) {
+ ctx.removeHeaders(actions.removeRequestHeaders);
+ }
+ }
+
+ protected void applyResponseHeaderActions(RequestContext ctx, TesterRequestBody actions) {
+ if (actions.appendResponseHeaders != null) {
+ ctx.appendHeaders(actions.appendResponseHeaders);
+ }
+ if (actions.addResponseHeaders != null) {
+ ctx.addHeaders(actions.addResponseHeaders);
+ }
+ if (actions.overwriteResponseHeaders != null) {
+ ctx.overwriteHeaders(actions.overwriteResponseHeaders);
+ }
+ if (actions.removeResponseHeaders != null) {
+ ctx.removeHeaders(actions.removeResponseHeaders);
+ }
+ }
+
+ protected void applyRequestBodyActions(RequestContext ctx, TesterRequestBody actions) {
+ if (actions.clearRequestBody) {
+ ctx.clearBody();
+ } else if (actions.replaceRequestBody != null) {
+ ctx.replaceBody(actions.replaceRequestBody);
+ }
+ }
+
+ protected void applyResponseBodyActions(RequestContext ctx, TesterRequestBody actions) {
+ if (actions.clearResponseBody) {
+ ctx.clearBody();
+ } else if (actions.replaceResponseBody != null) {
+ ctx.replaceBody(actions.replaceResponseBody);
+ }
+ }
+
+ /** Case-insensitive header lookup over the phase header map. */
+ protected String getHeaderIgnoreCase(Map headers, String name) {
+ if (headers == null) {
+ return null;
+ }
+ for (Map.Entry entry : headers.entrySet()) {
+ if (entry.getKey() != null && entry.getKey().equalsIgnoreCase(name)) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+}