content: package com.alibaba.otter.canal.admin.connector;
  
  import java.io.IOException;
  import java.net.InetSocketAddress;
  import java.net.SocketAddress;
  import java.nio.ByteBuffer;
  import java.nio.ByteOrder;
  import java.nio.channels.Channel;
  import java.nio.channels.Channels;
  import java.nio.channels.ReadableByteChannel;
  import java.nio.channels.SocketChannel;
  import java.nio.channels.WritableByteChannel;
  import java.security.NoSuchAlgorithmException;
  
  import org.apache.commons.lang.BooleanUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.otter.canal.admin.common.exception.ServiceException;
  import com.alibaba.otter.canal.protocol.AdminPacket.Ack;
  import com.alibaba.otter.canal.protocol.AdminPacket.ClientAuth;
  import com.alibaba.otter.canal.protocol.AdminPacket.Handshake;
  import com.alibaba.otter.canal.protocol.AdminPacket.InstanceAdmin;
  import com.alibaba.otter.canal.protocol.AdminPacket.LogAdmin;
  import com.alibaba.otter.canal.protocol.AdminPacket.Packet;
  import com.alibaba.otter.canal.protocol.AdminPacket.PacketType;
  import com.alibaba.otter.canal.protocol.AdminPacket.ServerAdmin;
  import com.alibaba.otter.canal.protocol.SecurityUtil;
  import com.alibaba.otter.canal.protocol.exception.CanalClientException;
  import com.google.protobuf.ByteString;
  
  /**
   * 基于netty实现的admin控制
   *
   * @author agapple 2019年8月26日 上午10:23:44
   * @since 1.1.4
   */
  public class SimpleAdminConnector implements AdminConnector {
  
      private static final Logger logger      = LoggerFactory.getLogger(SimpleAdminConnector.class);
      private String              user;
      private String              passwd;
      private SocketAddress       address;
      private int                 soTimeout   = 60000;                                              // milliseconds
      private int                 idleTimeout = 60 * 60 * 1000;                                     // client和server之间的空闲链接超时的时间,默认为1小时
      private final ByteBuffer    readHeader  = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
      private final ByteBuffer    writeHeader = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
      private SocketChannel       channel;
      private ReadableByteChannel readableChannel;
      private WritableByteChannel writableChannel;
      private volatile boolean    connected   = false;
  
      public SimpleAdminConnector(String ip, int port, String user, String passwd){
          this.address = new InetSocketAddress(ip, port);
          this.user = user;
          this.passwd = passwd;
      }
  
      @Override
      public void connect() throws ServiceException {
          try {
              if (connected) {
                  return;
              }
  
              channel = SocketChannel.open();
              channel.socket().setSoTimeout(soTimeout);
              channel.connect(address);
              readableChannel = Channels.newChannel(channel.socket().getInputStream());
              writableChannel = Channels.newChannel(channel.socket().getOutputStream());
              Packet p = Packet.parseFrom(readNextPacket());
              if (p.getVersion() != 1) {
                  throw new CanalClientException("unsupported version at this client.");
              }
  
              if (p.getType() != PacketType.HANDSHAKE) {
                  throw new CanalClientException("expect handshake but found other type.");
              }
  
              Handshake handshake = Handshake.parseFrom(p.getBody());
              ByteString seed = handshake.getSeeds(); // seed for auth
              String newPasswd = passwd;
              if (passwd != null) {
                  // encode passwd
                  newPasswd = SecurityUtil.byte2HexStr(SecurityUtil.scramble411(passwd.getBytes(), seed.toByteArray()));
              }
  
              ClientAuth ca = ClientAuth.newBuilder()
                  .setUsername(user != null ? user : "")
                  .setPassword(ByteString.copyFromUtf8(newPasswd != null ? newPasswd : ""))
                  .setNetReadTimeout(idleTimeout)
                  .setNetWriteTimeout(idleTimeout)
                  .build();
              writeWithHeader(Packet.newBuilder()
                  .setType(PacketType.CLIENTAUTHENTICATION)
                  .setBody(ca.toByteString())
                  .build()
                  .toByteArray());
              //
              Packet ack = Packet.parseFrom(readNextPacket());
              if (ack.getType() != PacketType.ACK) {
                  throw new CanalClientException("unexpected packet type when ack is expected");
              }
  
              Ack ackBody = Ack.parseFrom(ack.getBody());
              if (ackBody.getCode() > 0) {
                  throw new ServiceException("something goes wrong when doing authentication: " + ackBody.getMessage());
              }
  
              connected = true;
          } catch (IOException | NoSuchAlgorithmException e) {
              throw new ServiceException(e);
          }
      }
  
      @Override
      public void disconnect() throws ServiceException {
          if (!connected) {
              return;
          }
  
          connected = false;
          if (readableChannel != null) {
              quietlyClose(readableChannel);
              readableChannel = null;
          }
          if (writableChannel != null) {
              quietlyClose(writableChannel);
              writableChannel = null;
          }
          if (channel != null) {
              quietlyClose(channel);
              channel = null;
          }
      }
  
      @Override
      public boolean check() {
          return BooleanUtils.toBoolean(Integer.parseInt(doServerAdmin("check")));
      }
  
      @Override
      public boolean start() {
          return BooleanUtils.toBoolean(Integer.parseInt(doServerAdmin("start")));
      }
  
      @Override
      public boolean stop() {
          return BooleanUtils.toBoolean(Integer.parseInt(doServerAdmin("stop")));
      }
  
      @Override
      public boolean restart() {
          return BooleanUtils.toBoolean(Integer.parseInt(doServerAdmin("restart")));
      }
  
      @Override
      public String getRunningInstances() {
          return doServerAdmin("list");
      }
  
      @Override
      public boolean checkInstance(String destination) {
          return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "check")));
      }
  
      @Override
      public boolean startInstance(String destination) {
          return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "start")));
      }
  
      @Override
      public boolean stopInstance(String destination) {
          return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "stop")));
      }
  
      @Override
      public boolean releaseInstance(String destination) {
          return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "release")));
      }
  
      @Override
      public boolean restartInstance(String destination) {
          return BooleanUtils.toBoolean(Integer.parseInt(doInstanceAdmin(destination, "restart")));
      }
  
      @Override
      public String listCanalLog() {
          return doLogAdmin("server", "list", null, null, 0);
      }
  
      @Override
      public String canalLog(int lines) {
          return doLogAdmin("server", "file", null, null, lines);
      }
  
      @Override
      public String listInstanceLog(String destination) {
          return doLogAdmin("instance", "list", destination, null, 0);
      }
  
      @Override
      public String instanceLog(String destination, String fileName, int lines) {
          return doLogAdmin("instance", "file", destination, fileName, lines);
      }
  
      // ==================== helper method ====================
  
      private String doServerAdmin(String action) {
          try {
              writeWithHeader(Packet.newBuilder()
                  .setType(PacketType.SERVER)
                  .setBody(ServerAdmin.newBuilder().setAction(action).build().toByteString())
                  .build()
                  .toByteArray());
  
              Packet p = Packet.parseFrom(readNextPacket());
              Ack ack = Ack.parseFrom(p.getBody());
              if (ack.getCode() > 0) {
                  throw new ServiceException("failed to subscribe with reason: " + ack.getMessage());
              }
  
              return ack.getMessage();
          } catch (IOException e) {
              throw new ServiceException(e);
          }
      }
  
      private String doInstanceAdmin(String destination, String action) {
          try {
              writeWithHeader(Packet.newBuilder()
                  .setType(PacketType.INSTANCE)
                  .setBody(InstanceAdmin.newBuilder()
                      .setDestination(destination)
                      .setAction(action)
                      .build()
                      .toByteString())
                  .build()
                  .toByteArray());
  
              Packet p = Packet.parseFrom(readNextPacket());
              Ack ack = Ack.parseFrom(p.getBody());
              if (ack.getCode() > 0) {
                  throw new ServiceException("failed to subscribe with reason: " + ack.getMessage());
              }
  
              return ack.getMessage();
          } catch (IOException e) {
              throw new ServiceException(e);
          }
      }
  
      private String doLogAdmin(String type, String action, String destination, String file, int count) {
          try {
              writeWithHeader(Packet.newBuilder()
                  .setType(PacketType.LOG)
                  .setBody(LogAdmin.newBuilder()
                      .setType(type)
                      .setAction(action)
                      .setDestination(destination == null ? "" : destination)
                      .setFile(file == null ? "" : file)
                      .setCount(count)
                      .build()
                      .toByteString())
                  .build()
                  .toByteArray());
  
              Packet p = Packet.parseFrom(readNextPacket());
              Ack ack = Ack.parseFrom(p.getBody());
              if (ack.getCode() > 0) {
                  throw new ServiceException("failed to subscribe with reason: " + ack.getMessage());
              }
  
              return ack.getMessage();
          } catch (IOException e) {
              throw new ServiceException(e);
          }
      }
  
      private void writeWithHeader(byte[] body) throws IOException {
          writeWithHeader(writableChannel, body);
      }
  
      private byte[] readNextPacket() throws IOException {
          return readNextPacket(readableChannel);
      }
  
      private void writeWithHeader(WritableByteChannel channel, byte[] body) throws IOException {
          writeHeader.clear();
          writeHeader.putInt(body.length);
          writeHeader.flip();
          channel.write(writeHeader);
          channel.write(ByteBuffer.wrap(body));
      }
  
      private byte[] readNextPacket(ReadableByteChannel channel) throws IOException {
          readHeader.clear();
          read(channel, readHeader);
          int bodyLen = readHeader.getInt(0);
          ByteBuffer bodyBuf = ByteBuffer.allocate(bodyLen).order(ByteOrder.BIG_ENDIAN);
          read(channel, bodyBuf);
          return bodyBuf.array();
      }
  
      private void read(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
          while (buffer.hasRemaining()) {
              int r = channel.read(buffer);
              if (r == -1) {
                  throw new IOException("end of stream when reading header");
              }
          }
      }
  
      private void quietlyClose(Channel channel) {
          try {
              channel.close();
          } catch (IOException e) {
              logger.warn("exception on closing channel:{} \n {}", channel, e);
          }
      }
  }
  