feat(apiman-gateway) Simple HTTP backpressure first pass impl; only implemented in Vert.x platform.
MODIFY
gateway/engine/core/src/main/java/io/apiman/gateway/engine/async/IAsyncHandler.java
gateway/engine/core/src/main/java/io/apiman/gateway/engine/async/IAsyncHandler.java
@@ -22,6 +22,7 @@ package io.apiman.gateway.engine.async;
*
* @param <T> The event to handle
*/
+@FunctionalInterface
public interface IAsyncHandler<T> {
/**

MODIFY
gateway/engine/core/src/main/java/io/apiman/gateway/engine/impl/ApiRequestExecutorImpl.java
gateway/engine/core/src/main/java/io/apiman/gateway/engine/impl/ApiRequestExecutorImpl.java
@@ -482,6 +482,16 @@ public class ApiRequestExecutorImpl implements IApiRequestExecutor {
}
}
}
+
+ @Override
+ public void drainHandler(IAsyncHandler<Void> drainHandler) {
+ apiConnection.drainHandler(drainHandler);
+ }
+
+ @Override
+ public boolean isFull() {
+ return apiConnection.isFull();
+ }
});
}
@@ -732,6 +742,16 @@ public class ApiRequestExecutorImpl implements IApiRequestExecutor {
public boolean isFinished() {
return streamFinished;
}
+
+ @Override
+ public void drainHandler(IAsyncHandler<Void> drainHandler) {
+ apiConnection.drainHandler(drainHandler);
+ }
+
+ @Override
+ public boolean isFull() {
+ return apiConnection.isFull();
+ }
});
}

MODIFY
gateway/engine/core/src/main/java/io/apiman/gateway/engine/io/ISignalWriteStream.java
gateway/engine/core/src/main/java/io/apiman/gateway/engine/io/ISignalWriteStream.java
*/
package io.apiman.gateway.engine.io;
+import io.apiman.gateway.engine.async.IAsyncHandler;
/**
* As {@link IWriteStream} but with {@link #abort()}.
@@ -22,4 +23,12 @@ package io.apiman.gateway.engine.io;
* @author Marc Savy <msavy@redhat.com>
*/
public interface ISignalWriteStream extends IWriteStream, IAbortable {
+
+ default void drainHandler(IAsyncHandler<Void> drainHandler) {
+ throw new UnsupportedOperationException("Drain handler has not been implemented."); //$NON-NLS-1$
+ }
+
+ default boolean isFull() {
+ return false;
+ }
}

MODIFY
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/connector/HttpConnector.java
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/connector/HttpConnector.java
@@ -82,6 +82,7 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
private ApiResponse apiResponse;
private IAsyncResultHandler<IApiConnectionResponse> resultHandler;
+ private IAsyncHandler<Void> drainHandler;
private IAsyncHandler<IApimanBuffer> bodyHandler;
private IAsyncHandler<Void> endHandler;
private ExceptionHandler exceptionHandler;
@@ -173,10 +174,7 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
apiPort,
apiHost,
endpoint,
- new Handler<HttpClientResponse>() {
-
- @Override
- public void handle(final HttpClientResponse vxClientResponse) {
+ (HttpClientResponse vxClientResponse) -> {
clientResponse = vxClientResponse;
// Pause until we're given permission to xfer the response.
@@ -197,7 +195,6 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
// The response is only ever returned when vxClientResponse is valid.
resultHandler.handle(AsyncResultImpl
.create((IApiConnectionResponse) HttpConnector.this));
- }
});
clientRequest.exceptionHandler(exceptionHandler);
@@ -263,6 +260,10 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
if (chunk.getNativeBuffer() instanceof Buffer) {
clientRequest.write((Buffer) chunk.getNativeBuffer());
+ // When write queue has diminished sufficiently, drain handler will be invoked.
+ if (clientRequest.writeQueueFull() && drainHandler != null) {
+ clientRequest.drainHandler(drainHandler::handle);
+ }
} else {
throw new IllegalArgumentException(Messages.getString("HttpConnector.1"));
}
@@ -279,14 +280,21 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
return inboundFinished && outboundFinished;
}
- /**
- * @see io.apiman.gateway.engine.IApiConnection#isConnected()
- */
@Override
public boolean isConnected() {
return !isFinished();
}
+ @Override
+ public void drainHandler(IAsyncHandler<Void> drainHandler) {
+ this.drainHandler = drainHandler;
+ }
+
+ @Override
+ public boolean isFull() {
+ return clientRequest.writeQueueFull();
+ }
+
private URL parseApiEndpoint(Api api) {
try {
return new URL(api.getEndpoint());
@@ -325,4 +333,5 @@ class HttpConnector implements IApiConnectionResponse, IApiConnection {
.<IApiConnectionResponse> create(error));
}
}
+
}

MODIFY
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/http/HttpPolicyAdapter.java
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/http/HttpPolicyAdapter.java
@@ -80,11 +80,19 @@ public class HttpPolicyAdapter {
// Write data into executor. Called when ready.
executor.streamHandler(writeStream -> {
vertxRequest.handler(bufferChunk -> {
+ // Apply back-pressure.
+ if (writeStream.isFull()) {
+ vertxRequest.pause();
+ }
// Wrap Vert.x buffer into apiman's buffer IR, then write into engine.
writeStream.write(new VertxApimanBuffer(bufferChunk));
});
+ // Called when back-pressure has reduced sufficiently to resume.
+ writeStream.drainHandler(onDrain -> vertxRequest.resume());
+
vertxRequest.endHandler(end -> writeStream.end());
+
// Now safe to resume body transmission.
vertxRequest.resume();
});

feat(apiman-gateway) Simple HTTP backpressure first pass impl; only implemented in Vert.x platform.