@VisibleForTesting
void initTestCluster(String clusterAddrList, String host, int port) {
  isTest = true;
  this.serverHost = host;
  this.raftServerPort = port;

  // clear
  clusterNodes.clear();
  raftAddressMap.clear();
  clusterMemberIds.clear();

  String cluster[] = clusterAddrList.split(",");
  for (int i = 0; i < cluster.length; i++) {
    String[] parts = cluster[i].split(":");
    String clusterHost = parts[0];
    int clusterPort = Integer.valueOf(parts[1]);

    String memberId = clusterHost + ":" + clusterPort;
    Address address = Address.from(clusterHost, clusterPort);
    Node node = Node.builder().withId(memberId).withAddress(address).build();
    clusterNodes.add(node);
    raftAddressMap.put(MemberId.from(memberId), address);
    clusterMemberIds.add(MemberId.from(memberId));
  }
}
 

public void broadcastClusterEvent(String topic, String msg) {
  if (LOG.isDebugEnabled()) {
    LOG.debug("send broadcastClusterEvent message {}", msg);
  }
  for (Node node : clusterNodes) {
    if (StringUtils.equals(node.address().host(), serverHost)
        && node.address().port() == raftServerPort) {
      // skip myself
      continue;
    }

    CompletableFuture<byte[]> response = messagingService.sendAndReceive(node.address(),
        topic, msg.getBytes(), Duration.ofSeconds(2));
    response.whenComplete((r, e) -> {
      if (null == e) {
        LOG.error(e.getMessage(), e);
      } else {
        LOG.info("broadcastClusterNoteEvent success! {}", msg);
      }
    });
  }
}
 

@Override
public CompletableFuture<Void> join(BootstrapService bootstrap, Node localNode) {
  if (nodes.putIfAbsent(localNode.id(), localNode) == null) {
    this.bootstrap = bootstrap;
    post(new NodeDiscoveryEvent(NodeDiscoveryEvent.Type.JOIN, localNode));
    bootstrap.getBroadcastService().addListener(DISCOVERY_SUBJECT, broadcastListener);
    broadcastFuture = broadcastScheduler.scheduleAtFixedRate(
        () -> broadcastNode(localNode),
        config.getBroadcastInterval().toMillis(),
        config.getBroadcastInterval().toMillis(),
        TimeUnit.MILLISECONDS);
    broadcastNode(localNode);
    LOGGER.info("Joined");
  }
  return CompletableFuture.completedFuture(null);
}
 

/**
 * Creates an Atomix instance.
 */
protected static AtomixBuilder buildAtomix(int id, List<Integer> memberIds, Properties properties) {
  Collection<Node> nodes = memberIds.stream()
      .map(memberId -> Node.builder()
          .withId(String.valueOf(memberId))
          .withAddress(Address.from("localhost", BASE_PORT + memberId))
          .build())
      .collect(Collectors.toList());

  return Atomix.builder()
      .withClusterId("test")
      .withMemberId(String.valueOf(id))
      .withHost("localhost")
      .withPort(BASE_PORT + id)
      .withProperties(properties)
      .withMulticastEnabled()
      .withMembershipProvider(!nodes.isEmpty() ? new BootstrapDiscoveryProvider(nodes) : new MulticastDiscoveryProvider());
}
 

protected ClusterManager() {
  try {
    this.serverHost = NetworkUtils.findAvailableHostAddress();
    String clusterAddr = sconf.getClusterAddress();
    LOG.info("clusterAddr = {}", clusterAddr);
    if (!StringUtils.isEmpty(clusterAddr)) {
      String cluster[] = clusterAddr.split(",");

      for (int i = 0; i < cluster.length; i++) {
        String[] parts = cluster[i].split(":");
        String clusterHost = parts[0];
        int clusterPort = Integer.valueOf(parts[1]);
        if (this.serverHost.equalsIgnoreCase(clusterHost)) {
          raftServerPort = clusterPort;
        }

        String memberId = clusterHost + ":" + clusterPort;
        Address address = Address.from(clusterHost, clusterPort);
        Node node = Node.builder().withId(memberId).withAddress(address).build();
        clusterNodes.add(node);
        raftAddressMap.put(MemberId.from(memberId), address);
        clusterMemberIds.add(MemberId.from(memberId));
      }
    }
  } catch (UnknownHostException | SocketException e) {
    LOG.error(e.getMessage(), e);
  }
}
 

public static void main(String[] args) {
    List<Node> members = Lists.newArrayList();
    members.add(Member.builder().withId("gateway1").withAddress("127.0.0.1", 6001).build());
    members.add(Member.builder().withId("gateway2").withAddress("127.0.0.1", 6002).build());
    members.add(Member.builder().withId("gateway3").withAddress("127.0.0.1", 6003).build());

    Member member = Member.builder().withId("store").withAddress("localhost", 7001).build();
    Atomix atomix = Atomix.builder()
            .withMemberId(member.id())
            .withAddress(member.address())
            .withMembershipProvider(BootstrapDiscoveryProvider.builder()
                    .withNodes((Collection) members)
                    .build())
            .withProfiles(Profile.client())
            .build();

    atomix.start().join();


    AtomicMap<String, String> a = atomix.getAtomicMap("store-info");

    a.put("a", "test");
    try {
        Thread.sleep(1000 * 60 * 60 * 24);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

}
 

/**
 * Creates an Atomix instance.
 */
private Atomix createAtomix(int id, int... ids) {
  Collection<Node> nodes = IntStream.of(ids)
      .mapToObj(memberId -> Node.builder()
          .withId(String.valueOf(memberId))
          .withAddress(Address.from("localhost", BASE_PORT + memberId))
          .build())
      .collect(Collectors.toList());

  return Atomix.builder()
      .withClusterId("test")
      .withMemberId(String.valueOf(id))
      .withHost("localhost")
      .withPort(BASE_PORT + id)
      .withMembershipProtocol(SwimMembershipProtocol.builder()
          .withBroadcastDisputes(true)
          .withBroadcastUpdates(true)
          .withProbeInterval(Duration.ofMillis(100))
          .withNotifySuspect(true)
          .withFailureTimeout(Duration.ofSeconds(3))
          .build())
      .withMembershipProvider(new BootstrapDiscoveryProvider(nodes))
      .withManagementGroup(RaftPartitionGroup.builder("system")
          .withNumPartitions(1)
          .withPartitionSize(ids.length)
          .withMembers(nodes.stream().map(node -> node.id().id()).collect(Collectors.toSet()))
          .withDataDirectory(new File("target/test-logs/" + id + "/system"))
          .build())
      .withPartitionGroups(RaftPartitionGroup.builder("test")
          .withNumPartitions(3)
          .withPartitionSize(ids.length)
          .withMembers(nodes.stream().map(node -> node.id().id()).collect(Collectors.toSet()))
          .withDataDirectory(new File("target/test-logs/" + id + "/test"))
          .build())
      .build();
}
 

protected Atomix buildAtomix(int memberId) {
  return Atomix.builder()
      .withClusterId("test")
      .withMemberId(String.valueOf(memberId))
      .withHost("localhost")
      .withPort(5000 + memberId)
      .withMulticastEnabled()
      .withMembershipProvider(new BootstrapDiscoveryProvider(
          Node.builder()
              .withId("1")
              .withHost("localhost")
              .withPort(5001)
              .build(),
          Node.builder()
              .withId("2")
              .withHost("localhost")
              .withPort(5002)
              .build(),
          Node.builder()
              .withId("3")
              .withHost("localhost")
              .withPort(5003)
              .build()))
      .withManagementGroup(PrimaryBackupPartitionGroup.builder("system")
          .withNumPartitions(1)
          .build())
      .addPartitionGroup(PrimaryBackupPartitionGroup.builder("data")
          .withNumPartitions(3)
          .build())
      .build();
}
 

/**
 * Creates an Atomix server node.
 */
private Atomix createServer(Member member, List<Node> members) {
  Atomix atomix = Atomix.builder()
      .withMemberId(member.id())
      .withAddress(member.address())
      .withMembershipProvider(BootstrapDiscoveryProvider.builder()
          .withNodes(members)
          .build())
      .withManagementGroup(managementGroup.apply(member))
      .withPartitionGroups(dataGroup.apply(member))
      .build();
  servers.add(atomix);
  return atomix;
}
 

/**
 * Handles a node join event.
 */
private void handleJoinEvent(Node node) {
  GossipMember member = new GossipMember(MemberId.from(node.id().id()), node.address());
  if (!members.containsKey(member.id())) {
    sendHeartbeat(member);
  }
}
 

/**
 * Handles a node join event.
 */
private void handleJoinEvent(Node node) {
  SwimMember member = new SwimMember(MemberId.from(node.id().id()), node.address());
  if (!members.containsKey(member.id())) {
    probe(member.copy());
  }
}
 

/**
 * Handles a node leave event.
 */
private void handleLeaveEvent(Node node) {
  SwimMember member = members.get(MemberId.from(node.id().id()));
  if (member != null && !member.isActive()) {
    members.remove(member.id());
  }
}
 

private void handleBroadcastMessage(byte[] message) {
  Node node = SERIALIZER.decode(message);
  Node oldNode = nodes.put(node.id(), node);
  if (oldNode != null && !oldNode.id().equals(node.id())) {
    post(new NodeDiscoveryEvent(NodeDiscoveryEvent.Type.LEAVE, oldNode));
    post(new NodeDiscoveryEvent(NodeDiscoveryEvent.Type.JOIN, node));
  } else if (oldNode == null) {
    post(new NodeDiscoveryEvent(NodeDiscoveryEvent.Type.JOIN, node));
  }
  updateTimes.put(node.id(), System.currentTimeMillis());
}
 

private void expireNodes() {
  Iterator<Map.Entry<NodeId, Node>> iterator = nodes.entrySet().iterator();
  while (iterator.hasNext()) {
    Map.Entry<NodeId, Node> entry = iterator.next();
    if (System.currentTimeMillis() - updateTimes.get(entry.getKey()) > config.getFailureTimeout().toMillis()) {
      iterator.remove();
      post(new NodeDiscoveryEvent(NodeDiscoveryEvent.Type.LEAVE, entry.getValue()));
    }
  }
}
 

@Override
public CompletableFuture<Void> leave(Node localNode) {
  if (nodes.remove(localNode.id()) != null) {
    post(new NodeDiscoveryEvent(NodeDiscoveryEvent.Type.LEAVE, localNode));
    bootstrap.getBroadcastService().removeListener(DISCOVERY_SUBJECT, broadcastListener);
    ScheduledFuture<?> broadcastFuture = this.broadcastFuture;
    if (broadcastFuture != null) {
      broadcastFuture.cancel(false);
    }
    LOGGER.info("Left");
  }
  return CompletableFuture.completedFuture(null);
}
 

@Override
public CompletableFuture<Void> join(BootstrapService bootstrap, Node localNode) {
  LOGGER.info("Joined");
  resolverScheduler.scheduleAtFixedRate(
      this::resolveNodes, 0, resolutionInterval.toMillis(), TimeUnit.MILLISECONDS);
  return CompletableFuture.completedFuture(null);
}
 

/**
 * Sets the bootstrap nodes.
 *
 * @param nodes the bootstrap nodes
 * @return the location provider builder
 */
public BootstrapDiscoveryBuilder withNodes(Address... nodes) {
  return withNodes(Stream.of(nodes)
      .map(address -> Node.builder()
          .withAddress(address)
          .build())
      .collect(Collectors.toSet()));
}
 

@Override
public CompletableFuture<NodeDiscoveryService> start() {
  if (started.compareAndSet(false, true)) {
    provider.addListener(discoveryEventListener);
    Node node = Node.builder().withId(localNode.id().id()).withAddress(localNode.address()).build();
    return provider.join(bootstrapService, node).thenApply(v -> this);
  }
  return CompletableFuture.completedFuture(this);
}
 

private Collection<Node> buildBootstrapNodes(int nodes) {
  return IntStream.range(1, nodes + 1)
      .mapToObj(id -> Node.builder()
          .withId(String.valueOf(id))
          .withAddress(Address.from("localhost", id))
          .build())
      .collect(Collectors.toList());
}
 

private Collection<Node> buildBootstrapNodes(int nodes) {
  return IntStream.range(1, nodes + 1)
      .mapToObj(id -> Node.builder()
          .withId(String.valueOf(id))
          .withAddress(Address.from("localhost", id))
          .build())
      .collect(Collectors.toList());
}
 

/**
 * Creates a Raft server.
 */
private RaftServer createServer(Member member, List<Node> members) {
  RaftServerProtocol protocol;
  ManagedMessagingService messagingService;
  if (USE_NETTY) {
    messagingService = (ManagedMessagingService) new NettyMessagingService("test", member.address(), new MessagingConfig())
        .start()
        .join();
    messagingServices.add(messagingService);
    protocol = new RaftServerMessagingProtocol(messagingService, PROTOCOL_SERIALIZER, addressMap::get);
  } else {
    protocol = protocolFactory.newServerProtocol(member.id());
  }

  BootstrapService bootstrapService = new BootstrapService() {
    @Override
    public MessagingService getMessagingService() {
      return messagingService;
    }

    @Override
    public UnicastService getUnicastService() {
      return new UnicastServiceAdapter();
    }

    @Override
    public BroadcastService getBroadcastService() {
      return new BroadcastServiceAdapter();
    }
  };

  RaftServer.Builder builder = RaftServer.builder(member.id())
      .withProtocol(protocol)
      .withThreadModel(ThreadModel.SHARED_THREAD_POOL)
      .withMembershipService(new DefaultClusterMembershipService(
          member,
          Version.from("1.0.0"),
          new DefaultNodeDiscoveryService(bootstrapService, member, new BootstrapDiscoveryProvider(members)),
          bootstrapService,
          new HeartbeatMembershipProtocol(new HeartbeatMembershipProtocolConfig())))
      .withStorage(RaftStorage.builder()
          .withStorageLevel(StorageLevel.DISK)
          .withDirectory(new File(String.format("target/perf-logs/%s", member.id())))
          .withNamespace(STORAGE_NAMESPACE)
          .withMaxSegmentSize(1024 * 1024 * 64)
          .withDynamicCompaction()
          .withFlushOnCommit(false)
          .build());

  RaftServer server = builder.build();
  servers.add(server);
  return server;
}
 

/**
 * Handles a node leave event.
 */
private void handleLeaveEvent(Node node) {
  members.compute(MemberId.from(node.id().id()), (id, member) -> member == null || !member.isActive() ? null : member);
}
 

@Override
public Set<Node> getNodes() {
  return ImmutableSet.copyOf(nodes.values());
}
 

private void broadcastNode(Node localNode) {
  bootstrap.getBroadcastService().broadcast(DISCOVERY_SUBJECT, SERIALIZER.encode(localNode));
  expireNodes();
}
 

public BootstrapDiscoveryProvider(Node... bootstrapNodes) {
  this(Arrays.asList(bootstrapNodes));
}
 

public BootstrapDiscoveryProvider(Collection<Node> bootstrapNodes) {
  this(new BootstrapDiscoveryConfig().setNodes(bootstrapNodes.stream()
      .map(node -> new NodeConfig().setId(node.id())
          .setAddress(node.address()))
      .collect(Collectors.toList())));
}
 

BootstrapDiscoveryProvider(BootstrapDiscoveryConfig config) {
  this.config = checkNotNull(config);
  this.bootstrapNodes = ImmutableSet.copyOf(config.getNodes().stream().map(Node::new).collect(Collectors.toList()));
}
 

@Override
public Set<Node> getNodes() {
  return bootstrapNodes;
}
 

@Override
public CompletableFuture<Void> join(BootstrapService bootstrap, Node localNode) {
  LOGGER.info("Joined");
  return CompletableFuture.completedFuture(null);
}
 

@Override
public CompletableFuture<Void> leave(Node localNode) {
  LOGGER.info("Left");
  return CompletableFuture.completedFuture(null);
}