4

I am implementing netty proxy server as follows: A http request comes in,

  • if local cache has data, write to channel and flush
  • if not, fetch the data from remote server, add it to cache and flush

I am having difficulty extracting the byteBuf from response in samehandler as where I write to client.

In the example below, if you see channelRead method of HexDumpProxyFrontendHandler, you will see how I fetch from cache and write. I have added comments in that method below where I am facing difficulty

This code works end to end. so it can be copied and tested locally.

I can see FullHttpResponse object in HexDumpProxyBackendhandler#channelRead. but inside this method, I have no reference to cache nor the id I want to add inside cache.

There are two ways I think this can be solved, I am not clear on how this can be done though.

1) either get cache reference and id in HexdumpProxyBackendHandler, then it becomes easy. but hexDumpBackendhander is instantiated in channelActive of HexDumpFrontendHandler at which point I have not parsed my incoming request

2) get the response bytebuf extracted in HexdumpFrontendHandler#dchannelRead, in which case it is just cache insertion.

HexDumpProxy.java

public final class HexDumpProxy {

static final int LOCAL_PORT = Integer.parseInt(System.getProperty("localPort", "8082"));
static final String REMOTE_HOST = System.getProperty("remoteHost", "api.icndb.com");
static final int REMOTE_PORT = Integer.parseInt(System.getProperty("remotePort", "80"));
static Map<Long,String> localCache = new HashMap<>();
public static void main(String[] args) throws Exception {
    System.err.println("Proxying *:" + LOCAL_PORT + " to " + REMOTE_HOST + ':' + REMOTE_PORT + " ...");
    localCache.put(123L, "profile1");
    localCache.put(234L, "profile2");
    // Configure the bootstrap.
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .handler(new LoggingHandler(LogLevel.INFO))
         .childHandler(new HexDumpProxyInitializer(localCache, REMOTE_HOST, REMOTE_PORT))
         .childOption(ChannelOption.AUTO_READ, false)
         .bind(LOCAL_PORT).sync().channel().closeFuture().sync();
    } finally {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

}

HexDumpProxyInitializer.java

public class HexDumpProxyInitializer extends ChannelInitializer<SocketChannel> {

private final String remoteHost;
private final int remotePort;
private Map<Long, String> cache;

public HexDumpProxyInitializer(Map<Long,String> cache, String remoteHost, int remotePort) {
    this.remoteHost = remoteHost;
    this.remotePort = remotePort;
    this.cache=cache;
}

@Override
public void initChannel(SocketChannel ch) {
    ch.pipeline().addLast(
            new LoggingHandler(LogLevel.INFO),
            new HttpServerCodec(),
            new HttpObjectAggregator(8*1024, true),
            new HexDumpProxyFrontendHandler(cache, remoteHost, remotePort));
}

}

HexDumpProxyFrontendHandler.java

 public class HexDumpProxyFrontendHandler extends ChannelInboundHandlerAdapter {
private final String remoteHost;
private final int remotePort;
private Channel outboundChannel;
private Map<Long, String> cache;

public HexDumpProxyFrontendHandler(Map<Long, String> cache, String remoteHost, int remotePort) {
    this.remoteHost = remoteHost;
    this.remotePort = remotePort;
    this.cache = cache;
}

@Override
public void channelActive(ChannelHandlerContext ctx) {
    final Channel inboundChannel = ctx.channel();

    // Start the connection attempt.
    Bootstrap b = new Bootstrap();
    b.group(inboundChannel.eventLoop())
     .channel(ctx.channel().getClass())
     .handler((new ChannelInitializer() {
         protected void initChannel(Channel ch) {
             ChannelPipeline var2 = ch.pipeline();
             var2.addLast((new HttpClientCodec()));
             var2.addLast(new HttpObjectAggregator(8192, true));
             var2.addLast(new HexDumpProxyBackendHandler(inboundChannel));
         }
     }))
     .option(ChannelOption.AUTO_READ, false);
    ChannelFuture f = b.connect(remoteHost, remotePort);
    outboundChannel = f.channel();
    f.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                // connection complete start to read first data
                inboundChannel.read();
            } else {
                // Close the connection if the connection attempt has failed.
                inboundChannel.close();
            }
        }
    });
}

@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof HttpRequest) {
        System.out.println("msg is instanceof httpRequest");
        HttpRequest req = (HttpRequest)msg;
        QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());
        String userId = queryStringDecoder.parameters().get("id").get(0);
        Long id = Long.valueOf(userId);
        if (cache.containsKey(id)){
            StringBuilder buf = new StringBuilder();
            buf.append(cache.get(id));
            writeResponse(req, ctx, buf);
            closeOnFlush(ctx.channel());
            return;
        }
    }
    if (outboundChannel.isActive()) {
        outboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                if (future.isSuccess()) {
                    // was able to flush out data, start to read the next chunk
                    ctx.channel().read();
                } else {
                    future.channel().close();
                }
            }
        });
    }

    //get response back from HexDumpProxyBackendHander and write to cache
    //basically I need to do cache.put(id, parse(response));
    //how to get response buf from inboundChannel here is the question I am trying to solve
}

@Override
public void channelInactive(ChannelHandlerContext ctx) {
    if (outboundChannel != null) {
        closeOnFlush(outboundChannel);
    }

}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    closeOnFlush(ctx.channel());
}

/**
 * Closes the specified channel after all queued write requests are flushed.
 */
static void closeOnFlush(Channel ch) {
    if (ch.isActive()) {
        ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }
}

//borrowed from HttpSnoopServerHandler.java in snoop example
private boolean writeResponse(HttpRequest request, ChannelHandlerContext ctx, StringBuilder buf) {
    // Decide whether to close the connection or not.
    boolean keepAlive = HttpUtil.isKeepAlive(request);
    // Build the response object.
    FullHttpResponse response = new DefaultFullHttpResponse(
            HTTP_1_1, request.decoderResult().isSuccess()? OK : BAD_REQUEST,
            Unpooled.copiedBuffer(buf.toString(), CharsetUtil.UTF_8));

    response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");

    if (keepAlive) {
        // Add 'Content-Length' header only for a keep-alive connection.
        response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        // Add keep alive header as per:
        // - http://www.w3.org/Protocols/HTTP/1.1/draft-ietf-http-v11-spec-01.html#Connection
        response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
    }

    // Encode the cookie.
    String cookieString = request.headers().get(HttpHeaderNames.COOKIE);
    if (cookieString != null) {
        Set<Cookie> cookies = ServerCookieDecoder.STRICT.decode(cookieString);
        if (!cookies.isEmpty()) {
            // Reset the cookies if necessary.
            for (io.netty.handler.codec.http.cookie.Cookie cookie: cookies) {
                response.headers().add(HttpHeaderNames.SET_COOKIE, io.netty.handler.codec.http.cookie.ServerCookieEncoder.STRICT.encode(cookie));
            }
        }
    } else {
        // Browser sent no cookie.  Add some.
        response.headers().add(HttpHeaderNames.SET_COOKIE, io.netty.handler.codec.http.cookie.ServerCookieEncoder.STRICT.encode("key1", "value1"));
        response.headers().add(HttpHeaderNames.SET_COOKIE, ServerCookieEncoder.STRICT.encode("key2", "value2"));
    }

    // Write the response.
    ctx.write(response);

    return keepAlive;
}

}

HexDumpProxyBackendHandler.java

public class HexDumpProxyBackendHandler extends ChannelInboundHandlerAdapter {

private final Channel inboundChannel;

public HexDumpProxyBackendHandler(Channel inboundChannel) {
    this.inboundChannel = inboundChannel;
}

@Override
public void channelActive(ChannelHandlerContext ctx) {
    ctx.read();
}

@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof FullHttpResponse) {
        System.out.println("this is fullHttpResponse");
    }
    inboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                ctx.channel().read();
            } else {
                future.channel().close();
            }
        }
    });
}

@Override
public void channelInactive(ChannelHandlerContext ctx) {
    HexDumpProxyFrontendHandler.closeOnFlush(inboundChannel);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    HexDumpProxyFrontendHandler.closeOnFlush(ctx.channel());
}

}

P.S: I have taken most of the code from netty-example project and customized it

EDIT

Per Ferrygig suggestions, I changed the FrontEndChannelHander#channelRead as follows. I have removed channelActive and have write method implemented

@Override public void channelRead(final ChannelHandlerContext ctx, Object msg) {

if (msg instanceof HttpRequest) {
    System.out.println("msg is instanceof httpRequest");
    HttpRequest req = (HttpRequest)msg;
    QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());
    String userId = queryStringDecoder.parameters().get("id").get(0);
    id = Long.valueOf(userId);
    if (cache.containsKey(id)){
        StringBuilder buf = new StringBuilder();
        buf.append(cache.get(id));
        writeResponse(req, ctx, buf);
        closeOnFlush(ctx.channel());
        return;
    }

    final Channel inboundChannel = ctx.channel();

    //copied from channelActive method

    // Start the connection attempt.
    Bootstrap b = new Bootstrap();
    b.group(inboundChannel.eventLoop())
            .channel(ctx.channel().getClass())
            .handler((new ChannelInitializer() {
                protected void initChannel(Channel ch) {
                    ChannelPipeline var2 = ch.pipeline();
                    var2.addLast((new HttpClientCodec()));
                    var2.addLast(new HttpObjectAggregator(8192, true));
                    var2.addLast(new HexDumpProxyBackendHandler(inboundChannel, cache));
                }
            }));
            //.option(ChannelOption.AUTO_READ, false);
    ChannelFuture f = b.connect(remoteHost, remotePort);
    outboundChannel = f.channel();
    f.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                // connection complete start to read first data
                inboundChannel.read();
            } else {
                // Close the connection if the connection attempt has failed.
                inboundChannel.close();
            }
        }
    });
}
if (outboundChannel.isActive()) {
    outboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                // was able to flush out data, start to read the next chunk
                ctx.channel().read();
            } else {
                future.channel().close();
            }
        }
    });
}
3
  • Would something like stackoverflow.com/questions/23128232/… work for you to associate an outgoing response with an incoming one (Disclaimer: this is 1 of my older answers)
    – Ferrybig
    Apr 5, 2019 at 7:34
  • I looked at this and couldn’t get it to work in my case. If you could get it work, please do post it! Apr 5, 2019 at 8:18
  • @Ferrybig any thoughts? Apr 5, 2019 at 16:59

2 Answers 2

1
+100

There are multiple ways to approach this problem, and the way to go differs for your ultimate end goal.

At the moment, you are using a topology of 1 connection inbound is 1 connection outbound, this makes the system design slightly easier as you don't have to worry about syncing multiple requests to same outbound stream.

At the moment, you frontend handler extends ChannelInboundHandlerAdapter, this only intercepts "packets" coming into your application, if we make it extends ChannelDuplexHandler, we can also handle "packets" going out of the applications.

To approach this path, we need to update the HexDumpProxyFrontendHandler class to extend ChannelDuplexHandler (Lets call it CDH for now).

The next step in the process, is to override the write method coming from the CDH, so we can intercept when the backend sends us the response back.

After we created the write method, we need to update our (non-threadsafe) map by calling the put method.

public class HexDumpProxyFrontendHandler extends ChannelDuplexHandler {
    Long lastId;
    // ...
    @Override
    public void channelRead(final ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof HttpRequest) {
            System.out.println("msg is instanceof httpRequest");
            HttpRequest req = (HttpRequest)msg;
            QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());
            String userId = queryStringDecoder.parameters().get("id").get(0);
            Long id = Long.valueOf(userId);
            lastId = id; // Store ID of last request
            // ...
        }
        // ...
    }
    // ...
    public void write(
        ChannelHandlerContext ctx,
        java.lang.Object msg,
        ChannelPromise promise
    ) throws java.lang.Exception {

        if (msg instanceof FullHttpResponse) {
            System.out.println("this is fullHttpResponse");
            FullHttpResponse full = (FullHttpResponse)msg;
            cache.put(lastId, parse(full)); // TODO: Include a system here to convert the request to a string
        }
        super.write(ctx, msg, promise);
    }
    // ...
}

We are not done here yet, while we have the code in place, we still need to fix a few bugs in other places in the code.

Non thread-safe map (critical bug)

One of those bugs in that you are using a normal hash map to handle your cache. The problem with this is that this is not thread safe, if multiple people connect to your app at the same time, weird things may happen, including full map corruption as the internal structure of the map updates.

To counter this issue, we are going to "upgrade" the map to a ConcurrentHashMap, this map has special structures in place to deal with multiple threads requesting and storing data at the same time, without a huge loss in performance. (if performance is a main concern, you might get higher performance by using an per-thread hash map instead of a global cache, but this means that every resource can be cached up to the amount of threads.

No cache removal rules (major bug)

At the moment, there is no code in place to remove outdated resources, this means the cache is going to fill up, until the program has no memory left, and it will then crash spectacularly.

This can be solved by either using a map implementation that provides both thread-safe access and so called removal rules, or using already pre-made caching solutions like Gnuava caches.

Failure to handle HTTP Pipelining properly (minor-major bug)

One of the lesser known features of HTTP is pipelining, this basically means that the client can send another request to the server, without waiting for a response on the previous request. Bugs of this type include servers that swap the content of both requests around, or even mangle them completely.

While pipelined requests are rare these days with more and more HTTP2 support and the knowledge of that there are broken servers out there, it still happens with certain CLI tools that use it.

To solve this issue, ONLY read a request AFTER you send the previous response, one of the ways to do is keeping a list of requests, or go for more advanced pre-make solutions

9
  • Thanks for this workaround and tips on bugs. I am aware of first two, since this is just starting stage. pipelining is something I learned new today! one question I have here, is that in the frontendHandler#channelActive() I create s bootStrapserver. This is useless, if my data is in cache already. how can I avoid that? Apr 5, 2019 at 20:19
  • 1
    With your current architecture, its not easy to solve properly, but you should start by removing the read logic from your code, and make it read automatically (Remove the AUTO_READ, false from the Initializer, and all the calls to channel.read, and comment out the connection code in channelActive, then inside your channelRead, instead of piping directly the request to the backend, check if channel is null, and if it is, make a new connection and send the data, or send the data directly if it isn't null
    – Ferrybig
    Apr 5, 2019 at 20:57
  • I don't do any channel.read in frontEndHandler except in outboundChannel listener. do you want me to remove that? which channel should I check for null? outboundChannel will be null always if I comment out connection code in channelActive. since they handlers are created one per request, any connection made in one request, is not availabble for other rite.. Apr 5, 2019 at 21:24
  • Netty has 2 differend systems of receiving data, either it is done automatically, or it is done manually (autoread true/false), We want to convert the code from autoread: false to autoread: true, and since reading isdone automatically now, all calls to read can be removed now. Since you want to only connect to the server once an uncached request comes in, you now have to check if an connection is already open, because you cannot send to a non-open connection
    – Ferrybig
    Apr 5, 2019 at 22:01
  • why do you think a connection will be open? If an uncached request comes in, I create connection, fetch data, cache it, send response and close connection. When new request for uncached data comes in, I will create a new connection. Does it sound rite? Apr 5, 2019 at 22:06
1

storm

I may be wrong, when I read this part of your HexDumpProxyFrontendHandler, I feel like something is probably incorrect (I put my comments a litlle bit ahead according to correct style to make them visible):

 // Not incorrect but better to have only one bootstrap and reusing it
    Bootstrap b = new Bootstrap(); 
    b.group(inboundChannel.eventLoop())
            .channel(ctx.channel().getClass())
            .handler(new HexDumpProxyBackendHandler(inboundChannel))
 // I know what AUTO_READ false is, but my question is why you need it?
            .option(ChannelOption.AUTO_READ, false);
    ChannelFuture f = b.connect(remoteHost, remotePort);
 // Strange to me to try to get the channel while you did not test yet it is linked
    outboundChannel = f.channel();
    f.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
 // Maybe you should start to send there, therefore getting the outboundChannel right there?
 // add a log in order to see if you come there
 // probably you have to send first, before asking to read anything?
 // position (1)
                inboundChannel.read();
            } else {
                inboundChannel.close();
            }
        }
    });
 // I suggest to move this in position named (1)
    if (outboundChannel.isActive()) {
 // maybe a log to see if anything will be written?
        outboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                if (future.isSuccess()) {
                    System.out.println("success!! - FrontEndHandler");
                    ctx.channel().read();
                } else {
                    future.channel().close();
                }
            }
        });
    }

For me, it's look like you didn't wait for the channel to be opened. You lack of some log when you send to wire, in order to ensure that you really send something (in the logs, we can only see that the connection is opened, then closing mainly, with nothing in between).

Maybe some more logs could help us and you?

4
  • pretty much of this code is from netty example repo.I need only few minor changes. github.com/netty/netty/tree/4.1/example/src/main/java/io/netty/… Apr 3, 2019 at 23:44
  • I have opened bounty on this. I am looking for a working solution for this problem since it will be useful for wider audience Apr 3, 2019 at 23:45
  • Note that it is a bit mixed from the repo example. Maybe the various positions are not the right place, as I suggested? It is in the channelRead override method, as yours. But the opening of the outboundChannel was in the channelActive, not in the same method. From my perspective, it seems those 2 steps should be separated (opening and starting reading in one place, writing in another place). I may be wrong, of course. Apr 4, 2019 at 7:41
  • Thanks for looking into this. I edited the complete post with a clear problem what i am facing. It was a bit vague before. my apologies. This entire code pasted above works. However, if you see the code, you will see where my real problem is. Thank you for the help Apr 5, 2019 at 5:31

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.