This repository was archived by the owner on Feb 24, 2026. It is now read-only.
Async message processing logic #59
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,6 +3,8 @@ | |
| import io.netty.buffer.ByteBuf; | ||
| import io.netty.buffer.Unpooled; | ||
| import io.netty.channel.Channel; | ||
| import io.netty.channel.ChannelHandlerContext; | ||
| import io.netty.channel.ChannelInboundHandlerAdapter; | ||
| import io.netty.channel.ChannelPipeline; | ||
| import io.netty.handler.codec.http.DefaultHttpRequest; | ||
| import io.netty.handler.codec.http.FullHttpRequest; | ||
|
|
@@ -21,13 +23,16 @@ | |
| import io.netty.handler.timeout.IdleStateHandler; | ||
| import io.netty.handler.traffic.GlobalTrafficShapingHandler; | ||
| import io.netty.util.ReferenceCounted; | ||
| import io.netty.util.concurrent.EventExecutorGroup; | ||
| import io.netty.util.concurrent.Future; | ||
| import io.netty.util.concurrent.GenericFutureListener; | ||
|
|
||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.littleshoot.proxy.ActivityTracker; | ||
| import org.littleshoot.proxy.DefaultFailureHttpResponseComposer; | ||
| import org.littleshoot.proxy.ExceptionHandler; | ||
| import org.littleshoot.proxy.FailureHttpResponseComposer; | ||
| import org.littleshoot.proxy.GlobalStateHandler; | ||
| import org.littleshoot.proxy.ratelimit.RateLimiter; | ||
| import org.littleshoot.proxy.FlowContext; | ||
| import org.littleshoot.proxy.FullFlowContext; | ||
|
|
@@ -44,6 +49,7 @@ | |
| import java.util.List; | ||
| import java.util.Locale; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.RejectedExecutionException; | ||
|
|
@@ -143,15 +149,17 @@ public class ClientToProxyConnection extends ProxyConnection<HttpRequest> { | |
| final DefaultHttpProxyServer proxyServer, | ||
| SslEngineSource sslEngineSource, | ||
| boolean authenticateClients, | ||
| GlobalTrafficShapingHandler globalTrafficShapingHandler, | ||
| Channel channel) { | ||
|
osklyarenko marked this conversation as resolved.
|
||
| super(AWAITING_INITIAL, proxyServer, false); | ||
|
|
||
| this.channel = channel; | ||
|
Collaborator
Author
There was a problem hiding this comment. initial the channel is assigned on |
||
|
|
||
| initChannelPipeline(channel.pipeline()); | ||
|
|
||
| if (sslEngineSource != null) { | ||
| LOG.debug("Enabling encryption of traffic from client to proxy"); | ||
| encrypt(channel.pipeline(), sslEngineSource.newSslEngine(), | ||
| authenticateClients) | ||
| .addListener( | ||
| new GenericFutureListener<Future<? super Channel>>() { | ||
|
|
@@ -177,7 +185,8 @@ public void operationComplete( | |
| **************************************************************************/ | ||
|
|
||
| @Override | ||
| protected ConnectionState readHTTPInitial(ChannelHandlerContext ctx, Object httpRequestObj) { | ||
|
There was a problem hiding this comment. why did we change
Collaborator
Author
There was a problem hiding this comment. yeah, I had some weird behavior with generic here.. |
||
| HttpRequest httpRequest = (HttpRequest) httpRequestObj; | ||
| LOG.debug("Received raw request: {}", httpRequest); | ||
|
|
||
| // if we cannot parse the request, immediately return a 400 and close the connection, since we do not know what state | ||
|
|
@@ -195,14 +204,9 @@ protected ConnectionState readHTTPInitial(HttpRequest httpRequest) { | |
| return DISCONNECT_REQUESTED; | ||
| } | ||
|
|
||
| ctx.fireChannelRead(httpRequest); | ||
|
|
||
| return getCurrentState(); | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -223,34 +227,11 @@ protected ConnectionState readHTTPInitial(HttpRequest httpRequest) { | |
| * @param httpRequest | ||
| * @return | ||
| */ | ||
| public ConnectionState setupUpstreamConnection(HttpResponse shortCircuitResponse, HttpRequest httpRequest) { | ||
| if (shortCircuitResponse != null) { | ||
| LOG.debug("Responding to client with short-circuit response from filter: {}", shortCircuitResponse); | ||
|
|
||
| boolean keepAlive = respondWithShortCircuitResponse(shortCircuitResponse); | ||
| if (keepAlive) { | ||
| return AWAITING_INITIAL; | ||
| } else { | ||
|
|
@@ -366,6 +347,96 @@ private ConnectionState doReadHTTPInitial(HttpRequest httpRequest) { | |
| } | ||
| } | ||
|
|
||
| @Sharable | ||
| protected class ClientToProxyMessageProcessor extends ChannelInboundHandlerAdapter { | ||
|
Collaborator
Author
There was a problem hiding this comment. this handler contains authentication and payload transformation logic which is executed in a separate executor service There was a problem hiding this comment. could we move it to a separate class / java file?
Collaborator
Author
There was a problem hiding this comment. this class encapsulate some code of client to proxy connection.. it is easier to have it here.. |
||
|
|
||
| @Override | ||
| public void channelRead(ChannelHandlerContext ctx, Object msg) { | ||
| final HttpRequest httpRequest = (HttpRequest) msg; | ||
|
|
||
| if (ProxyUtils.isChunked(httpRequest)) { | ||
| process(ctx, httpRequest); | ||
|
Collaborator
Author
There was a problem hiding this comment. we can not do async call when chunks are used. It is due to https://github.com/netty/netty/blob/6fdd7fcddbe964b2f30d7492a926f4f0bf0f083f/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java#L321. The app does not use chunks as they are aggregated by https://github.com/verygoodsecurity/LittleProxy/blob/vgs-edition/src/main/java/org/littleshoot/proxy/impl/ClientToProxyConnection.java#L817 |
||
| } else { | ||
| if (httpRequest instanceof ReferenceCounted) { | ||
| LOG.debug("Retaining reference counted message"); | ||
| ((ReferenceCounted) msg).retain(); | ||
| } | ||
|
|
||
| proxyServer.getMessageProcessingExecutor() | ||
| .execute(wrapTask(() -> { | ||
| try { | ||
| process(ctx, httpRequest); | ||
| } catch (Exception e) { | ||
| ctx.fireExceptionCaught(e); | ||
| } finally { | ||
| if (httpRequest instanceof ReferenceCounted) { | ||
| LOG.debug("Retaining reference counted message"); | ||
| ((ReferenceCounted) httpRequest).release(); | ||
| } | ||
| } | ||
| })); | ||
| } | ||
| } | ||
|
|
||
| private void process(ChannelHandlerContext ctx, HttpRequest httpRequest) { | ||
|
|
||
| boolean authenticationRequired = false; | ||
| HttpResponse shortCircuitResponse = null; | ||
| authenticationRequired = authenticationRequired(httpRequest); | ||
|
|
||
| if (authenticationRequired) { | ||
| LOG.debug("Not authenticated!!"); | ||
| become(AWAITING_PROXY_AUTHENTICATION); | ||
| } else { | ||
| // Make a copy of the original request | ||
| final HttpRequest currentRequest = copy(httpRequest); | ||
|
|
||
| // Set up our filters based on the original request. If the HttpFiltersSource returns null (meaning the request/response | ||
| // should not be filtered), fall back to the default no-op filter source. | ||
| HttpFilters filterInstance; | ||
| try { | ||
| filterInstance = proxyServer.getFiltersSource().filterRequest(currentRequest, ctx); | ||
| } finally { | ||
| // releasing a copied http request | ||
| if (currentRequest instanceof ReferenceCounted) { | ||
| ((ReferenceCounted) currentRequest).release(); | ||
| } | ||
| } | ||
| if (filterInstance != null) { | ||
| currentFilters = filterInstance; | ||
| } else { | ||
| currentFilters = HttpFiltersAdapter.NOOP_FILTER; | ||
| } | ||
|
|
||
| // Send the request through the clientToProxyRequest filter, and respond with the short-circuit response if required | ||
| shortCircuitResponse = currentFilters.clientToProxyRequest(httpRequest); | ||
|
|
||
| } | ||
|
|
||
| if (!authenticationRequired) { | ||
| if (httpRequest instanceof ReferenceCounted) { | ||
| LOG.debug("Retaining reference counted message"); | ||
| ((ReferenceCounted) httpRequest).retain(); | ||
| } | ||
|
|
||
| ctx.fireChannelRead(new UpstreamConnectionHandler.Request(httpRequest, shortCircuitResponse)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| Runnable wrapTask(Runnable task) { | ||
| return () -> { | ||
| final Optional<GlobalStateHandler> globalStateHandler = | ||
| Optional.ofNullable(proxyServer.getGlobalStateHandler()); | ||
| try { | ||
| globalStateHandler.ifPresent(it -> it.restoreFromChannel(channel)); | ||
| task.run(); | ||
|
osklyarenko marked this conversation as resolved.
|
||
| } finally { | ||
| globalStateHandler.ifPresent(GlobalStateHandler::clear); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Returns true if the specified request is a request to an origin server, rather than to a proxy server. If this | ||
| * request is being MITM'd, this method always returns false. The format of requests to a proxy server are defined | ||
|
|
@@ -800,8 +871,10 @@ private void initChannelPipeline(ChannelPipeline pipeline) { | |
| pipeline.addLast("inboundGlobalStateHandler", new InboundGlobalStateHandler(this)); | ||
| } | ||
|
|
||
| EventExecutorGroup globalStateWrapperEvenLoop = new GlobalStateWrapperEvenLoop(this); | ||
|
Collaborator
Author
There was a problem hiding this comment. this event loop will not always be used. Only when a call to The reason is that in the same loop |
||
|
|
||
| pipeline.addLast(globalStateWrapperEvenLoop, "bytesReadMonitor", bytesReadMonitor); | ||
| pipeline.addLast(globalStateWrapperEvenLoop, "bytesWrittenMonitor", bytesWrittenMonitor); | ||
|
|
||
| pipeline.addLast("encoder", new HttpResponseEncoder()); | ||
| // We want to allow longer request lines, headers, and chunks | ||
|
|
@@ -818,8 +891,8 @@ private void initChannelPipeline(ChannelPipeline pipeline) { | |
| aggregateContentForFiltering(pipeline, numberOfBytesToBuffer); | ||
| } | ||
|
|
||
| pipeline.addLast(globalStateWrapperEvenLoop, "requestReadMonitor", requestReadMonitor); | ||
| pipeline.addLast(globalStateWrapperEvenLoop, "responseWrittenMonitor", responseWrittenMonitor); | ||
|
|
||
| pipeline.addLast( | ||
| "idle", | ||
|
|
@@ -830,7 +903,10 @@ private void initChannelPipeline(ChannelPipeline pipeline) { | |
| pipeline.addLast("outboundGlobalStateHandler", new OutboundGlobalStateHandler(this)); | ||
| } | ||
|
|
||
| pipeline.addLast(globalStateWrapperEvenLoop, "router", this); | ||
| pipeline.addLast(globalStateWrapperEvenLoop, "httpInitialHandler", new HttpInitialHandler<>(this)); | ||
| pipeline.addLast(globalStateWrapperEvenLoop, "clientToProxyMessageProcessor", new ClientToProxyMessageProcessor()); | ||
| pipeline.addLast(globalStateWrapperEvenLoop, "upstreamConnectionHandler", new UpstreamConnectionHandler(this)); | ||
|
|
||
| } | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
will be provided from outside, possibly in a wrapper with added metrics
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
i like the idea of a specialized wrapper.