APISIX-38 Allow async plugins (#313)

* APISIX-38 Allow async plugins

* Fix E2eE tests

* APISIX-38 Document the threading model and async plugins

Author: chibenwa

.github/workflows/runner-e2e.yml

@@ -59,7 +59,7 @@ jobs:
 
     - name: startup apisix
       run: |
-        docker-compose -f ci/docker-compose.yml up -d
+        docker compose -f ci/docker-compose.yml up -d
         sleep 5
 
     - name: install ginkgo cli

docs/en/latest/the-internal-of-apisix-java-plugin-runner.md

@@ -78,3 +78,39 @@ The current type takes the following values
 * 2 means http_req_call
 
 The binary data generated by the flatbuffer serialization is placed in the data segment.
+
+## Threading model
+
+Apisix plugin runner will run your plugins directly onto the event loop.
+
+While this empower the best performance possible, as a plugin developer you will have the responsibility 
+never to block threads on the event loop. Doing so would result in catastrophic performance drop.
+
+Hopefully one can write asynchronous plugins easily: just call the `PluginFilterChain` as a callback once you 
+are done.
+
+For instance:
+
+```java
+@Component
+public class AsyncResponseFilter implements PluginFilter {
+    @Override
+    public String name() {
+        return "AyncResponseFilter";
+    }
+
+    @Override
+    public void postFilter(PostRequest request, PostResponse response, PluginFilterChain chain) {
+        callExternalService()
+            .thenAccept(body -> {
+                response.setBody(body);
+                chain.postFilter(request, response);
+            });
+    }
+
+    // This simulates calls to an external service
+    CompletableFuture<String> callExternalService() {
+        return CompletableFuture.completedFuture("response_body");
+    }
+}
+```

runner-core/src/main/java/org/apache/apisix/plugin/runner/handler/RpcCallHandler.java

@@ -207,11 +207,20 @@ private void doPostFilter(ChannelHandlerContext ctx) {
         postReq.initCtx(conf.getConfig());
         postReq.setVars(nginxVars);
 
-        PluginFilterChain chain = conf.getChain();
-        chain.postFilter(postReq, postResp);
+        PluginFilterChain chain = conf.getChain()
+            .addFilter(new PluginFilter() {
+                @Override
+                public String name() {
+                    return null;
+                }
 
-        ChannelFuture future = ctx.writeAndFlush(postResp);
-        future.addListeners(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+                @Override
+                public void postFilter(PostRequest request, PostResponse response, PluginFilterChain chain) {
+                    ChannelFuture future = ctx.writeAndFlush(postResp);
+                    future.addListeners(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+                }
+            });
+        chain.postFilter(postReq, postResp);
     }
 
     private void handleExtraInfo(ChannelHandlerContext ctx, ExtraInfoResponse request) {
@@ -256,12 +265,21 @@ private void doFilter(ChannelHandlerContext ctx) {
         currReq.initCtx(currResp, conf.getConfig());
         currReq.setVars(nginxVars);
 
-        PluginFilterChain chain = conf.getChain();
-        chain.filter(currReq, currResp);
+        PluginFilterChain chain = conf.getChain()
+                .addFilter(new PluginFilter() {
+                    @Override
+                    public String name() {
+                        return "writeFilter";
+                    }
 
-        ChannelFuture future = ctx.writeAndFlush(currResp);
-        future.addListeners(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+                    @Override
+                    public void filter(HttpRequest request, HttpResponse response, PluginFilterChain chain) {
+                        ChannelFuture future = ctx.writeAndFlush(currResp);
+                        future.addListeners(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+                    }
+                });
 
+        chain.filter(currReq, currResp);
     }
 
     private void handleHttpReqCall(ChannelHandlerContext ctx, HttpRequest request) {

runner-plugin-sdk/src/main/java/org/apache/apisix/plugin/runner/filter/PluginFilterChain.java

@@ -22,6 +22,7 @@
 import org.apache.apisix.plugin.runner.PostRequest;
 import org.apache.apisix.plugin.runner.PostResponse;
 
+import java.util.ArrayList;
 import java.util.List;
 
 public class PluginFilterChain {
@@ -47,6 +48,12 @@ public List<PluginFilter> getFilters() {
         return filters;
     }
 
+    public PluginFilterChain addFilter(PluginFilter filter) {
+        ArrayList<PluginFilter> pluginFilters = new ArrayList<>(filters);
+        pluginFilters.add(filter);
+        return new PluginFilterChain(pluginFilters);
+    }
+
     public void filter(HttpRequest request, HttpResponse response) {
         if (this.index < filters.size()) {
             PluginFilter filter = filters.get(this.index);

sample/src/main/java/org/apache/apisix/plugin/runner/filter/AsyncResponseFilter.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.apisix.plugin.runner.filter;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.apisix.plugin.runner.PostRequest;
+import org.apache.apisix.plugin.runner.PostResponse;
+import org.springframework.stereotype.Component;
+
+@Component
+public class AsyncResponseFilter implements PluginFilter {
+    @Override
+    public String name() {
+        return "AsyncResponseFilter";
+    }
+
+    @Override
+    public void postFilter(PostRequest request, PostResponse response, PluginFilterChain chain) {
+        callExternalService()
+            .thenAccept(body -> {
+                response.setBody(body);
+                chain.postFilter(request, response);
+            });
+    }
+
+    // This simulates calls to an external service
+    CompletableFuture<String> callExternalService() {
+        return CompletableFuture.completedFuture("response_body");
+    }
+}