What you are missing in your code is the *Relay, *the part passing bytes to
and from *Client-A *and *Server-C*. The HexDumpProxy's implementation of
the relay is in two classes: HexDumpProxyBackendHandler
<https://github.com/netty/netty/blob/4.1/example/src/main/java/io/netty/example/proxy/HexDumpProxyBackendHandler.java>
and HexDumpProxyFrontendHandler
<https://github.com/netty/netty/blob/4.1/example/src/main/java/io/netty/example/proxy/HexDumpProxyFrontendHandler.java>
.
After the line logger.info("Server B successfully connected to Server C") you
would want to do what HexDumpProxy does (adding the relay handlers):
HexDumpProxyInitializer.java
- Line 34
<https://github.com/netty/netty/blob/4.1/example/src/main/java/io/netty/example/proxy/HexDumpProxyInitializer.java#L34>
The HexDump uses two *ChannelInboundHandlerAdapter *handlers, This is
because it does not make sense to use a duplex handler. You have 2
channels, and you want to pass (write) the bytes just *read* into the other
channel. Nothing to be done in case of a *write* event (and thus the write
method of duplex handler never used).
If you need to do any processing on the data both read and write in the
client's channel, there is a better way: forget the fact that there is a
*channelFrom_B_to_C.
*This second channel, managing it's state, handing exceptions or anything
else is responsibility of Relay, and a relay is a simple passive handler,
it does NOT do any processing.
Now, in the *channelFrom_A_to_B *add a handler, let's call it MitmHandler
(Man in the middle handler).
MitmHandler is a duplex handler, something like this:
@Slf4jpublic class MitmHandler extends ChannelDuplexHandler {
@Override
public void channelRead(final ChannelHandlerContext ctx,
final Object msg) throws Exception {
if (msg instanceof ByteBuf) {
final ByteBuf b = (ByteBuf) msg;
for (int i = b.readerIndex(); i < b.readableBytes(); i++)
b.setByte(i, (byte) b.getByte(i) + 1);
}
super.channelRead(ctx, msg);
}
@Override
public void write(final ChannelHandlerContext ctx,
final Object msg,
final ChannelPromise promise) throws Exception {
if (msg instanceof ByteBuf) {
final ByteBuf b = (ByteBuf) msg;
for (int i = b.readerIndex(); i < b.readableBytes(); i++)
b.setByte(i, (byte) b.getByte(i) - 1);
}
super.write(ctx, msg, promise);
}
}
Just a dummy (crazy) handler, subtracts 1 from each byte in write direction
(from C to A / Outbound), add 1 to each byte in read direction (from A to C
/ Inbound). You could add it to *channelFrom_B_to_C *but it just makes
things complex, *channelFrom_B_to_C *is a dependency (?) of
*channelFrom_A_to_B,
*I suggest to call them master and slave, and do things in the master and
forget the slave.
One thing about auto read: *DO NOT FORGET* to set auto read of *BOTH *channels
to false, and manually initiate a read in your relay. The HexDumpProxy
shows how to do it.
These were source of confusion for me for a long while. hope it helps.
On Tuesday, September 6, 2016 at 9:29:22 PM UTC+4:30, L_DisplayName wrote:
>
> Greetings All,
> I wanted to know how I can use ChannelDuplexHandler from a server to
> connect to another server? Also I would like to know if my understanding of
> channelDuplex handler is correct and what's the difference between
> bootstrapping a client and having it connect to a server verses having the
> ChannelDuplexHandler connect to the server?
>
> For example: If I have a *client A*, a *Server B *and a *Server C*.
> and I want *client A* to connect to *Server B* and *Server B *connect to
> *Server
> C. *where pictorially* I have:*
> * A-->B-->C *
>
> *Below I show Server's B Handler code, as soon as client A connects to
> Server B (The code for this is below B's Server Handler code), I want
> Server B to connect to Server C, can you tell me if the code below is
> correct or if any thing needs to be changed and if so what and how? Thank
> you!*
>
> *Attempting to connect Server B to Server C using ChannelDuplexHandler
> (Note I am showing Server B's handler class)*
> public class Server_B_Handler extends ChannelDuplexHandler {
>
> private String ipAddressToConnectTo, localIpAddress;
> private int portToConnectTo, localPort;
> private Logger logger;
> private ChannelHandlerContext ctx;
> private Channel channelFrom_A_to_B;
> private Channel channelFrom_B_to_C;
>
> public Server_B_Handler(String ipAddress, int port, localIpAddress,
> localPort) {
> ipAddressToConnectTo = ipAddress;
> portToConnectTo = port
> this.localIpAddress = localIpAddress;
> this.localPort = localPort;
> logger = Logger.getLogger(this.getClass().getName());
> this.ctx = null;
> channelFrom_A_to_B = null; //Inbound channel
> channelFrom_B_to_C = null; //Outbound Channel
>
> }
>
> @Override
> public void channelActive(ChannelHandlerContext ctx) throws Exception{
> //THIS CTX (Channel context) is for the channel connecting Client A to
> this Server (Server B)
> //Please see below for code connecting client A to Server B
> this.ctx = ctx;
> channelFrom_A_to_B = this.ctx().channel();
>
> //QUESTION HOW CAN I CONNECT TO SERVER C using this server's CONNECT
> METHOD?
>
> final Promise<Channel> myPromise = this.ctx.executor().newPromise();
> ctx.connect(new InetSocketAddress(InetAddress.getByName(
> ipAddressToConnectTo), portToConnectTo),new InetSocketAddress(this.
> localIpAddress,this.localPort),myPromise);
> myPromise.addListener(new FutureListener<Channel>(){
> @override
> public void operationComplete(final Future<Channel> future) throws
> Exception {
> if (future.isSuccess()){
> //Connection attempt successful
> channelFrom_B_to_C = future.get();
> logger.info("Server B successfully connected to Server C")
> }
> else {
> //Connection attempt was not successful
> channelFrom_B_to_C = future.get().close();
> logger.info("Server B did not successfully connect to Server C.
> closing...")
> }
> });
>
> }
>
>
> @Override
> public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws
> Exception {
> //What ever the inbound channel (channelFrom_A_to_B) reads write it to
> the outbound channel
> if ( channelFrom_B_to_C.isActive()){
> channelFrom_B_to_C.write(msg);
> }
> }
>
> @Override
> public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
> cause.printStackTrace();
>
> if (ctx.channel().isActive()) {
> ctx.writeAndFlush("ERR: " +
> cause.getClass().getSimpleName() + ": " +
> cause.getMessage() + '\n').addListener(ChannelFutureListener.CLOSE);
> }
> }
> }
>
>
>
> *I know I can connect Client A to Client B by the following code*
>
> public final class Client_A {
>
> static final boolean SSL = System.getProperty("ssl") != null;
> static final int SIZE = Integer.parseInt(System.getProperty("size", "256"));
> static final String HOST = System.getProperty("host", "129.456.7.9");
> static final int PORT = Integer.parseInt(System.getProperty("port","4959"));
>
> public static void main(String[] args) throws Exception {
> final SslContext sslCtx;
> if (SSL) {
> sslCtx = SslContextBuilder.forClient()
> .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
> } else {
> sslCtx = null;
> }
>
> // Configure the client.
> EventLoopGroup group = new NioEventLoopGroup();
> try {
> Bootstrap b = new Bootstrap();
> b.group(group)
> .channel(NioSocketChannel.class)
> .option(ChannelOption.TCP_NODELAY, true)
> .handler(new ChannelInitializer<SocketChannel>() {
> @Override
> public void initChannel(SocketChannel ch) throws Exception {
> ChannelPipeline p = ch.pipeline();
> if (sslCtx != null) {
> p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
> }
> //p.addLast(new LoggingHandler(LogLevel.INFO));
> p.addLast(
> new LengthFieldBasedFrameDecoder(1024*1024*128, 0, 8, 0, 8),
> new Client_A_Handler());
> }
> });
>
> // Start the client.
> ChannelFuture f = b.connect(HOST, PORT).sync();
>
> // Wait until the connection is closed.
> f.channel().closeFuture().sync();
> } finally {
> // Shut down the event loop to terminate all threads.
> group.shutdownGracefully();
> }
> }
> }
>
>
>
>
--
You received this message because you are subscribed to the Google Groups
"Netty discussions" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to netty+unsubscr...@googlegroups.com.
To view this discussion on the web visit
https://groups.google.com/d/msgid/netty/83af8c56-03e7-4c3c-852c-748a1952341e%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
