Telnet client example on netty

ClientAuthHandler.java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.concurrent.BlockingQueue;

/**
 * Created by Vladislav Kisliy
 */
public class ClientAuthHandler extends SimpleChannelInboundHandler<ByteBuf> {

    private enum ClientState {
        AUTHENTICATING,
        AUTHENTICATED,
        ERROR
    }

    private static final Logger LOG = LoggerFactory.getLogger(ClientAuthHandler.class);
    private static final byte req_1[] = {
            (byte) 0xff, (byte) 0xfd, 0x18, (byte) 0xff, (byte) 0xfd, 0x20, (byte) 0xff, (byte) 0xfd,
            0x23, (byte) 0xff, (byte) 0xfd, 0x27};

    private static final byte req_2[] = {
            (byte) 0xff, (byte) 0xfb, 0x03, (byte) 0xff, (byte) 0xfd, 0x1f, (byte) 0xff, (byte) 0xfd,
            0x21, (byte) 0xff, (byte) 0xfe, 0x22, (byte) 0xff, (byte) 0xfb, 0x05, (byte) 0xff,
            (byte) 0xfa, 0x20, 0x01, (byte) 0xff, (byte) 0xf0, (byte) 0xff, (byte) 0xfa, 0x23,
            0x01, (byte) 0xff, (byte) 0xf0, (byte) 0xff, (byte) 0xfa, 0x27, 0x01, (byte) 0xff,
            (byte) 0xf0, (byte) 0xff, (byte) 0xfa, 0x18, 0x01, (byte) 0xff, (byte) 0xf0};

    private static final byte req_3[] = {(byte) 0xff, (byte) 0xfd, 0x01};

    private static final byte repl_1[] = {
            (byte) 0xff, (byte) 0xfd, 0x03, (byte) 0xff, (byte) 0xfb, 0x18, (byte) 0xff, (byte) 0xfb,
            0x1f, (byte) 0xff, (byte) 0xfb, 0x20, (byte) 0xff, (byte) 0xfb, 0x21, (byte) 0xff,
            (byte) 0xfb, 0x22, (byte) 0xff, (byte) 0xfb, 0x27, (byte) 0xff, (byte) 0xfd, 0x05,
            (byte) 0xff, (byte) 0xfb, 0x23};
    private static final byte repl_2[] = {
            (byte) 0xff, (byte) 0xfa, 0x1f, 0x00, 0x7b, 0x00, 0x25, (byte) 0xff,
            (byte) 0xf0, (byte) 0xff, (byte) 0xfa, 0x20, 0x00, 0x33, 0x38, 0x34,
            0x30, 0x30, 0x2c, 0x33, 0x38, 0x34, 0x30, 0x30,
            (byte) 0xff, (byte) 0xf0, (byte) 0xff, (byte) 0xfa, 0x23, 0x00, 0x6c, 0x69,
            0x6e, 0x75, 0x78, 0x2d, 0x65, 0x73, 0x75, 0x33,
            0x3a, 0x30, (byte) 0xff, (byte) 0xf0, (byte) 0xff, (byte) 0xfa, 0x27, 0x00,
            0x03, 0x58, 0x41, 0x55, 0x54, 0x48, 0x4f, 0x52,
            0x49, 0x54, 0x59, 0x01, 0x2f, 0x74, 0x6d, 0x70,
            0x2f, 0x78, 0x61, 0x75, 0x74, 0x68, 0x2d, 0x31,
            0x30, 0x30, 0x30, 0x2d, 0x5f, 0x30, 0x00, 0x44,
            0x49, 0x53, 0x50, 0x4c, 0x41, 0x59, 0x01, 0x6c,
            0x69, 0x6e, 0x75, 0x78, 0x2d, 0x65, 0x73, 0x75,
            0x33, 0x3a, 0x30, (byte) 0xff, (byte) 0xf0, (byte) 0xff, (byte) 0xfa, 0x18,
            0x00, 0x58, 0x54, 0x45, 0x52, 0x4d, (byte) 0xff, (byte) 0xf0};


    private static final byte repl_3[] = {(byte) 0xff, (byte) 0xfc, 0x01};

    private final String user;
    private final String pw;
    private final BlockingQueue<String> queue;
    private ClientState clientState;
    private int handShakeCounter = 0;

    public ClientAuthHandler(String user, String pw, BlockingQueue<String> queue) {
        this.user = user;
        this.pw = pw;
        this.queue = queue;
        this.clientState = ClientState.AUTHENTICATING;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf byteBuf) throws Exception {
        byte[] networkBytes = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(networkBytes);
        LOG.debug("Got message bytes {} and line {}", networkBytes, byteBuf);
        // First step of handshake
        if (handShakeCounter == 0 && Arrays.equals(networkBytes, req_1)) {
            System.out.println("send repl_1");
            ctx.channel().writeAndFlush(Unpooled.wrappedBuffer(repl_1));

            handShakeCounter++;
        } else if (handShakeCounter == 1 && Arrays.equals(networkBytes, req_2)) {
            System.out.println("send repl_2");
            ctx.channel().writeAndFlush(Unpooled.wrappedBuffer(repl_2));
            handShakeCounter++;
        } else if (handShakeCounter == 2 && Arrays.equals(networkBytes, req_3)) {
            System.out.println("send repl_3");
            ctx.channel().writeAndFlush(Unpooled.wrappedBuffer(repl_3));
            handShakeCounter++;
        } else if (handShakeCounter == 3) {
            String loginMessage = new String(networkBytes).trim().toLowerCase();
            if (loginMessage.contains("login")) {
                System.out.println("send login");
                ctx.channel().writeAndFlush(Unpooled.wrappedBuffer((user + "\r\n").getBytes()));
                handShakeCounter++;
            }
        } else if (handShakeCounter == 4) {
            String loginMessage = new String(networkBytes).trim().toLowerCase();
            if (loginMessage.contains("password")) {
                System.out.println("send password");
                ctx.channel().writeAndFlush(Unpooled.wrappedBuffer((pw + "\r\n").getBytes()));
                handShakeCounter++;
            }
        }
        if (handShakeCounter == 5) {
            System.out.println("remove yourself");
            ctx.pipeline().remove(this);
            queue.put("SUCCESS");
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOG.error("Error {}", cause);
        ctx.close();
    }

    // Simple authorization procedure
    private boolean auth(ChannelHandlerContext ctx, String msg) {
        String response = null;
        boolean result = false;
        LOG.info("Auth steps. got msg {}", msg);
        if ("login".equals(msg)) {
            response = user + "\r\n";
        } else if ("password".equals(msg)) {
            response = pw + "\r\n";
            result = true;
        }
        if (response != null) {
            LOG.debug("Send to server {}", response);
            ctx.writeAndFlush(response);
        }
        return result;
    }
}

TelnetClient.java

import com.google.common.base.Preconditions;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Simplistic telnet client.
 */
public final class TelnetClient {

    public static final int DEFAULT_COMMAND_TIMEOUT = 5000;
    private static final Logger LOG = LoggerFactory.getLogger(TelnetClient.class);
    private final String host;
    private final int port;
    private final String user;
    private final String pw;
    private EventLoopGroup group;

    /**
     * To create instance use Builder class.
     *
     * @param host
     * @param port
     * @param user
     * @param pw
     */
    private TelnetClient(String host, int port, String user, String pw) {
        this.host = host;
        this.port = port;
        this.user = user;
        this.pw = pw;
    }

    /**
     * Runs only one command on remote server. Returns output in responseContext.
     *
     * @param command
     * @param responseContext
     */
    public void executeCommand(String command, Map<String, Object> responseContext) {
        Preconditions.checkNotNull(command);
        Preconditions.checkNotNull(responseContext);
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();
        try {
            final Channel channel = initConnection(queue);
            ChannelFuture lastWriteFuture = channel.writeAndFlush(command + "\r\n");
            // Get command output
            String commandOutput = queue.poll(DEFAULT_COMMAND_TIMEOUT, TimeUnit.MILLISECONDS);
            responseContext.put(command, commandOutput);
            lastWriteFuture.sync();
        } catch (InterruptedException ex) {
            LOG.error("Exception was caught {}", ex.getMessage());
        } finally {
            closeConnecton();
        }
    }

    public void executeCommands(Iterable<String> commands, Map<String, Object> responseContext) {
        Preconditions.checkNotNull(commands);
        Preconditions.checkNotNull(responseContext);
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();
        try {
            final Channel channel = initConnection(queue);
            ChannelFuture lastWriteFuture = null;
            for (String command : commands) {
                lastWriteFuture = channel.writeAndFlush(command + "\r\n");
                // Get command output
                String commandOutput = queue.poll(DEFAULT_COMMAND_TIMEOUT, TimeUnit.MILLISECONDS);
                responseContext.put(command, commandOutput);
            }
            // Wait until all messages are flushed before closing the channel.
            if (lastWriteFuture != null) {
                lastWriteFuture.sync();
            }
        } catch (InterruptedException ex) {
            LOG.error("Exception was caught {}", ex.getMessage());
        } finally {
            closeConnecton();
        }
    }

    /**
     * @param queue
     * @return
     */
    private Channel initConnection(BlockingQueue<String> queue) {
        group = new NioEventLoopGroup();
        Channel result = null;
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new TelnetClientInitializer(user, pw, queue));
            // Start the connection attempt.
            result = b.connect(host, port).sync().channel();
            if (!"SUCCESS".equals(queue.poll(DEFAULT_COMMAND_TIMEOUT, TimeUnit.MILLISECONDS))) {
                LOG.error("Authorization issue");
                result = null;
                throw new RuntimeException("Authorization issue");
            }
        } catch (InterruptedException ex) {
            LOG.error("Exception was caught {}", ex.getMessage());
        }
        return result;
    }

    private void closeConnecton() {
        group.shutdownGracefully();
    }

    public static void main(String[] args) throws Exception {
        final TelnetClient telnetClient = new TelnetClient("127.0.0.1", 23,
                "test", "test");
        Map<String, Object> map = new HashMap<>();
        String[] commands = new String[]{"ls -lah", "w", "ls -lah /"};
        long start = System.currentTimeMillis();
        telnetClient.executeCommands(Arrays.asList(commands), map);
        long end = System.currentTimeMillis();
        NumberFormat formatter = new DecimalFormat("#0.00000");
        System.out.println("Batch execution time is " + formatter.format((end - start) / 1000d) + " seconds");
        System.out.println("Result =" + map);
    }
}

TelnetClientHandler.java

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.TimerTask;
import java.util.concurrent.*;

/**
 * Handles a client-side channel.
 */
@Sharable
public class TelnetClientHandler extends SimpleChannelInboundHandler<String> {

    private static final Logger LOG = LoggerFactory.getLogger(TelnetClientHandler.class);
    private static final int DEFAULT_REFRESH_TIMER = 300;
    private final ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(2);
    private final BlockingQueue<String> queue;
    private ScheduledFuture<?> scheduleTask;
    private StringBuilder outputBuilder = new StringBuilder();

    public TelnetClientHandler(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        outputBuilder.append(msg);
        outputBuilder.append("\r\n");
        updateTimer();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOG.error("Error ", cause);
        ctx.close();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        scheduledExecutor.shutdown();
        scheduledExecutor.awaitTermination(1000, TimeUnit.MILLISECONDS);
        super.channelInactive(ctx);
        scheduledExecutor.shutdownNow();
        LOG.debug("Deactivated channel successfully");
    }

    public void updateTimer() {
        if (scheduleTask != null) {
            scheduleTask.cancel(true);
        }
        scheduleTask = scheduledExecutor.schedule(new TimerTask() {
            @Override
            public void run() {
                try {
                    queue.offer(outputBuilder.toString(), 1, TimeUnit.SECONDS);
                    outputBuilder = new StringBuilder();
                } catch (InterruptedException ex) {
                    LOG.warn("Interrupted task. Exception {}", ex.getMessage());
                }
            }
        }, DEFAULT_REFRESH_TIMER, TimeUnit.MILLISECONDS);
    }
}

TelnetClientInitializer.java

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.concurrent.BlockingQueue;

/**
 * Creates a newly configured {@link ChannelPipeline} for a new channel.
 */
public class TelnetClientInitializer extends ChannelInitializer<SocketChannel> {

    private static final StringDecoder DECODER = new StringDecoder();
    private static final StringEncoder ENCODER = new StringEncoder();

    private final String user;
    private final String pw;
    private final BlockingQueue<String> queue;

    public TelnetClientInitializer(String user, String pw, BlockingQueue<String> queue) {
        this.user = user;
        this.pw = pw;
        this.queue = queue;
    }

    @Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new ClientAuthHandler(user, pw, queue));
        pipeline.addLast(DECODER);
        pipeline.addLast(ENCODER);
        // and then business logic.
        pipeline.addLast(new TelnetClientHandler(queue));
    }
}