content: /*
   * Copyright (C) 2016-2023 ActionTech.
   * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
   */
  
  package com.actiontech.dble.cluster.general.impl.ushard;
  
  import com.actiontech.dble.cluster.ClusterHelper;
  import com.actiontech.dble.cluster.general.AbstractConsulSender;
  import com.actiontech.dble.cluster.general.bean.ClusterAlertBean;
  import com.actiontech.dble.cluster.general.bean.KvBean;
  import com.actiontech.dble.cluster.general.bean.SubscribeRequest;
  import com.actiontech.dble.cluster.general.bean.SubscribeReturnBean;
  import com.actiontech.dble.cluster.general.kVtoXml.ClusterToXml;
  import com.actiontech.dble.cluster.path.ClusterPathUtil;
  import com.actiontech.dble.config.model.ClusterConfig;
  import com.actiontech.dble.config.model.SystemConfig;
  import com.actiontech.dble.singleton.OnlineStatus;
  import com.actiontech.dble.util.exception.DetachedException;
  import io.grpc.Channel;
  import io.grpc.ManagedChannel;
  import io.grpc.ManagedChannelBuilder;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.locks.LockSupport;
  
  import static com.actiontech.dble.cluster.ClusterController.GRPC_SUBTIMEOUT;
  
  /**
   * Created by szf on 2019/3/13.
   */
  public class UshardSender extends AbstractConsulSender {
  
      private volatile DbleClusterGrpc.DbleClusterBlockingStub stub = null;
      private final String sourceComponentType = "dble";
      private String serverId = null;
      private String sourceComponentId = null;
      private volatile boolean connectionDetached = false;
  
      public String getRenewThreadPrefix() {
          return "USHARD_RENEW_";
      }
  
      @Override
      public void initConInfo() {
          Channel channel = ManagedChannelBuilder.forAddress("127.0.0.1",
                  ClusterConfig.getInstance().getClusterPort()).usePlaintext().build();
          setStubIfPossible(DbleClusterGrpc.newBlockingStub(channel).withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS));
      }
  
      @Override
      public void initCluster() {
          serverId = SystemConfig.getInstance().getServerId();
          sourceComponentId = SystemConfig.getInstance().getInstanceName();
          Channel channel = ManagedChannelBuilder.forAddress("127.0.0.1",
                  ClusterConfig.getInstance().getClusterPort()).usePlaintext().build();
          setStubIfPossible(DbleClusterGrpc.newBlockingStub(channel).withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS));
          startUpdateNodes();
          ClusterToXml.loadKVtoFile(this);
      }
  
      @Override
      public String lock(String path, String value) throws Exception {
          UshardInterface.LockOnSessionInput input = UshardInterface.LockOnSessionInput.newBuilder().setKey(path).setValue(value).setTTLSeconds(30).build();
          UshardInterface.LockOnSessionOutput output;
  
          try {
              output = stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).lockOnSession(input);
              if (!"".equals(output.getSessionId())) {
                  final String session = output.getSessionId();
                  Thread renewThread = new Thread(new Runnable() {
                      @Override
                      public void run() {
                          String sessionId = session;
                          while (!Thread.currentThread().isInterrupted()) {
                              try {
                                  LOGGER.debug("renew lock of session  start:" + sessionId + " " + path);
                                  if (!Boolean.TRUE.equals(ClusterHelper.isExist(path))) {
                                      log("renew lock of session  failure:" + sessionId + " " + path + ", the key is missing ", null);
                                      // alert
                                      Thread.currentThread().interrupt();
                                  } else if (!renewLock(sessionId)) {
                                      log("renew lock of session  failure:" + sessionId + " " + path, null);
                                      // alert
                                  } else {
                                      LOGGER.debug("renew lock of session  success:" + sessionId + " " + path);
                                  }
                                  LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10000));
                              } catch (Throwable e) {
                                  log("renew lock of session  failure:" + sessionId + " " + path, e);
                                  LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(5000));
                              }
                          }
                      }
  
                      private void log(String message, Throwable e) {
                          if (!Thread.currentThread().isInterrupted()) {
                              if (e == null) {
                                  LOGGER.warn(message);
                              } else {
                                  LOGGER.warn(message, e);
                              }
                          }
                      }
                  });
                  lockMap.put(path, renewThread);
                  renewThread.setName(getRenewThreadPrefix() + path);
                  renewThread.start();
              }
              return output.getSessionId();
          } catch (Exception e1) {
              throw new IOException("ushard connect failure", e1);
          }
      }
  
      @Override
      public void unlockKey(String path, String sessionId) {
          UshardInterface.UnlockOnSessionInput put = UshardInterface.UnlockOnSessionInput.newBuilder().setKey(path).setSessionId(sessionId).build();
          try {
              Thread renewThread = lockMap.get(path);
              if (renewThread != null) {
                  renewThread.interrupt();
              }
              stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).unlockOnSession(put);
          } catch (Exception e) {
              LOGGER.info(sessionId + " unlockKey " + path + " error ," + stub, e);
          }
      }
  
      @Override
      public void setKV(String path, String value) throws Exception {
          UshardInterface.PutKvInput input = UshardInterface.PutKvInput.newBuilder().setKey(path).setValue(value).build();
          try {
              stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).putKv(input);
          } catch (Exception e1) {
              throw new IOException(ERROR_MSG);
          }
      }
  
      @Override
      public KvBean getKV(String path) {
          UshardInterface.GetKvInput input = UshardInterface.GetKvInput.newBuilder().setKey(path).build();
          UshardInterface.GetKvOutput output;
  
          try {
              output = stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).getKv(input);
          } catch (Exception e1) {
              throw new RuntimeException(ERROR_MSG);
          }
  
          return new KvBean(path, output.getValue(), 0);
      }
  
      @Override
      public List<KvBean> getKVPath(String path) {
          if (!(path.charAt(path.length() - 1) == '/')) {
              path = path + "/";
          }
          List<KvBean> result = new ArrayList<>();
          UshardInterface.GetKvTreeInput input = UshardInterface.GetKvTreeInput.newBuilder().setKey(path).build();
  
          UshardInterface.GetKvTreeOutput output;
  
          try {
              output = stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).getKvTree(input);
          } catch (Exception e1) {
              throw new RuntimeException(ERROR_MSG);
          }
  
          for (int i = 0; i < output.getKeysCount(); i++) {
              KvBean bean = new KvBean(output.getKeys(i), output.getValues(i), output.getIndex());
              result.add(bean);
          }
          return result;
      }
  
      @Override
      public void cleanPath(String path) {
          try {
              if (!(path.charAt(path.length() - 1) == '/')) {
                  path = path + "/";
              }
              UshardInterface.DeleteKvTreeInput input = UshardInterface.DeleteKvTreeInput.newBuilder().setKey(path).build();
              try {
                  stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).deleteKvTree(input);
              } catch (Exception e1) {
                  throw new RuntimeException(ERROR_MSG);
              }
              cleanKV(path.substring(0, path.length() - 1));
          } catch (Exception e) {
              LOGGER.warn(" clean ushard Path failed ", e);
          }
  
      }
  
      @Override
      public void cleanKV(String path) {
          UshardInterface.DeleteKvInput input = UshardInterface.DeleteKvInput.newBuilder().setKey(path).build();
          try {
              stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).deleteKv(input);
          } catch (Exception e1) {
              throw new RuntimeException("cleanKV failure for" + path);
          }
      }
  
      @Override
      public SubscribeReturnBean subscribeKvPrefix(SubscribeRequest request) throws Exception {
          UshardInterface.SubscribeKvPrefixInput input = UshardInterface.SubscribeKvPrefixInput.newBuilder().
                  setIndex(request.getIndex()).setDuration(request.getDuration()).setKeyPrefix(request.getPath()).build();
          try {
              UshardInterface.SubscribeKvPrefixOutput output = stub.withDeadlineAfter(GRPC_SUBTIMEOUT, TimeUnit.SECONDS).subscribeKvPrefix(input);
              return groupSubscribeResult(output);
          } catch (Exception e1) {
              throw new IOException("ushard connect failure");
          }
  
      }
  
      @Override
      public void alert(ClusterAlertBean alert) {
          UshardInterface.AlertInput input = getInput(alert);
          try {
              stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).alert(input);
          } catch (Exception e) {
              LOGGER.info("alert to ushard error ", e);
          }
      }
  
      @Override
      public boolean alertResolve(ClusterAlertBean alert) {
          UshardInterface.AlertInput input = getInput(alert);
          try {
              stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).alertResolve(input);
              return true;
          } catch (Exception e) {
              return false;
          }
      }
  
  
      private boolean renewLock(String sessionId) throws Exception {
          UshardInterface.RenewSessionInput input = UshardInterface.RenewSessionInput.newBuilder().setSessionId(sessionId).build();
          try {
              stub.withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS).renewSession(input);
              return true;
          } catch (Exception e1) {
              LOGGER.info("connect to ushard renew error and will retry");
              return false;
          }
      }
  
      private SubscribeReturnBean groupSubscribeResult(UshardInterface.SubscribeKvPrefixOutput output) {
          SubscribeReturnBean result = new SubscribeReturnBean();
          result.setIndex(output.getIndex());
          if (output.getKeysCount() > 0) {
              List<KvBean> kvList = new ArrayList<>();
              for (int i = 0; i < output.getKeysCount(); i++) {
                  kvList.add(new KvBean(output.getKeys(i), output.getValues(i), 0));
              }
              result.setKvList(kvList);
          }
          return result;
      }
  
  
      private UshardInterface.AlertInput getInput(ClusterAlertBean alert) {
          UshardInterface.AlertInput.Builder builder = UshardInterface.AlertInput.newBuilder().
                  setCode(alert.getCode()).
                  setDesc(alert.getDesc()).
                  setLevel(alert.getLevel()).
                  setSourceComponentType(sourceComponentType).
                  setSourceComponentId(sourceComponentId).
                  setAlertComponentId(alert.getAlertComponentId()).
                  setAlertComponentType(alert.getAlertComponentType()).
                  setResolveTimestampUnix(alert.getResolveTimestampUnix()).
                  setServerId(serverId).
                  setTimestampUnix(alert.getTimestampUnix());
          if (alert.getLabels() != null) {
              builder.putAllLabels(alert.getLabels());
          }
          return builder.build();
      }
  
  
      private void startUpdateNodes() {
          Thread nodes = new Thread(new Runnable() {
              @Override
              public void run() {
                  for (; ; ) {
                      try {
                          if (isDetach()) {
                              LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(2000));
                              continue;
                          }
                          UshardInterface.SubscribeKvPrefixInput input = UshardInterface.SubscribeKvPrefixInput.newBuilder().
                                  setIndex(0).setDuration(60).setKeyPrefix(ClusterPathUtil.BASE_PATH).build();
                          stub.withDeadlineAfter(GRPC_SUBTIMEOUT, TimeUnit.SECONDS).subscribeKvPrefix(input);
                          firstReturnToCluster();
                          return;
                      } catch (DetachedException e) {
                          LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(2000));
                      } catch (Throwable e) {
                          if (isDetach()) {
                              LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(2000));
                              continue;
                          }
                          LOGGER.warn("error in ucore nodes watch,try for another time", e);
                          Channel channel = ManagedChannelBuilder.forAddress("127.0.0.1",
                                  ClusterConfig.getInstance().getClusterPort()).usePlaintext().build();
                          UshardSender.this.setStubIfPossible(DbleClusterGrpc.newBlockingStub(channel).withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS));
                          LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(2000));
                      }
                  }
              }
          });
          nodes.setName("USHARD_RECONNECT_LISTENER");
          nodes.start();
      }
  
  
      @Override
      public void detachCluster() throws Exception {
          OnlineStatus.getInstance().shutdownClear();
          LOGGER.info("cluster detach begin close connection");
          ((ManagedChannel) stub.getChannel()).shutdownNow();
  
  
      }
  
      @Override
      public boolean isDetach() {
  
          return connectionDetached;
      }
  
      @Override
      public void markDetach(boolean isConnectionDetached) {
          this.connectionDetached = isConnectionDetached;
      }
  
      @Override
      public void attachCluster() throws Exception {
          ManagedChannel channel = null;
          try {
              channel = ManagedChannelBuilder.forAddress("127.0.0.1",
                      ClusterConfig.getInstance().getClusterPort()).usePlaintext().build();
              stub = DbleClusterGrpc.newBlockingStub(channel).withDeadlineAfter(ClusterConfig.getInstance().getGrpcTimeout(), TimeUnit.SECONDS);
              //check connection is ready
              ClusterHelper.isExist(ClusterPathUtil.getOnlinePath(SystemConfig.getInstance().getInstanceName()));
          } catch (Exception e) {
              if (channel != null) {
                  channel.shutdownNow();
              }
              throw new IllegalStateException("can't connect to ucore. ");
          }
  
          connectionDetached = false;
          LOGGER.info("cluster attach begin rebuild online information");
          if (!OnlineStatus.getInstance().rebuildOnline()) {
              ((ManagedChannel) stub.getChannel()).shutdownNow();
              throw new IllegalStateException("can't create online information to ucore. ");
          }
      }
  
      public void setStubIfPossible(DbleClusterGrpc.DbleClusterBlockingStub stubTmp) {
          if (connectionDetached) {
              return;
          }
          this.stub = stubTmp;
      }
  }
  