<commit_before>/*
 * Copyright 2014 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.kuujo.copycat.cluster;

import net.kuujo.copycat.internal.util.Assert;
import net.kuujo.copycat.util.Copyable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * Immutable cluster configuration.<p>
 *
 * The {@code Cluster} is an immutable cluster configuration that is ultimately based on a mutable configuration. Each
 * {@code Cluster} is related to a specific {@link net.kuujo.copycat.cluster.Member} type and
 * {@link net.kuujo.copycat.spi.protocol.Protocol}. This allows Copycat's communication model to be effectively
 * altered based on the protocol implementation. For instance, for HTTP protocols, an {@code HttpMember} will be required
 * by the {@code HttpCluster} in order to provide the {@code host} and {@code port} required for operating the TCP
 * protocol.<p>
 *
 * All Copycat clusters are modifiable even while the cluster is running. When the underlying
 * {@link net.kuujo.copycat.cluster.ClusterConfig} is changed by the user, the {@code Cluster} membership will be
 * automatically updated. Additionally, when the {@code Cluster} membership changes, Copycat will detect the change
 * via an {@link java.util.Observer} and update the Copycat cluster's internal configuration safely. However, it's
 * important to note that changes to the {@code Cluster} or its underlying {@link net.kuujo.copycat.cluster.ClusterConfig}
 * may not necessarily be represented in the actual Copycat cluster. Copycat directs all cluster membership changes
 * through the cluster leader and does so in a safe manner. This means that cluster configuration changes must occur
 * only on the leader node for the time being (this will be changed in the future), and configuration changes on follower
 * nodes will be essentially ignored unless they occur prior to starting the cluster.<p>
 *
 * Copycat core provides a {@link net.kuujo.copycat.cluster.LocalCluster} for performing inter-thread communication.
 * This cluster type should be used in testing environments only.
 *
 * @author <a href="http://github.com/kuujo"><NAME></a>
 */
public class Cluster<M extends Member> extends Observable implements Observer, Copyable<Cluster<M>> {
  private static final Logger LOGGER = LoggerFactory.getLogger(Cluster.class);
  protected final ClusterConfig<M> config;
  private final M localMember;
  private final Set<M> remoteMembers;
  private final Map<String, M> members;

  public Cluster(ClusterConfig<M> config) {
    this.config = Assert.isNotNull(config, "config");
    this.localMember = config.getLocalMember();
    this.members = new HashMap<>(config.getMembers().size());
    this.members.put(localMember.id(), localMember);
    this.remoteMembers = new HashSet<>(config.getRemoteMembers().size());
    this.config.addObserver(this);
    clusterChanged(config);
  }

  @Override
  @SuppressWarnings("unchecked")
  public Cluster<M> copy() {
    return new Cluster(config.copy());
  }

  @Override
  @SuppressWarnings("unchecked")
  public void update(Observable o, Object arg) {
    LOGGER.info("{} - Configuration change detected, updating membership", this);
    clusterChanged((ClusterConfig<M>) o);
  }

  /**
   * Updates the cluster when the configuration changes.
   */
  private void clusterChanged(ClusterConfig<M> config) {
    // Add any remote members that don't already exist in the cluster.
    config.getRemoteMembers().forEach(member -> {
      if (!members.containsKey(member.id())) {
        remoteMembers.add(member);
        members.put(member.id(), member);
      }
    });

    Iterator<M> iterator = remoteMembers.iterator();
    while (iterator.hasNext()) {
      M member = iterator.next();
      boolean exists = false;
      for (M m : config.getRemoteMembers()) {
        if (m.equals(member)) {
          exists = true;
          break;
        }
      }
      if (!exists) {
        iterator.remove();
        members.remove(member.id());
      }
    }<commit_msg>Fix bug in cluster configuration changes removing members from local configurations.<commit_after>/*
 * Copyright 2014 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.kuujo.copycat.cluster;

import net.kuujo.copycat.internal.util.Assert;
import net.kuujo.copycat.util.Copyable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * Immutable cluster configuration.<p>
 *
 * The {@code Cluster} is an immutable cluster configuration that is ultimately based on a mutable configuration. Each
 * {@code Cluster} is related to a specific {@link net.kuujo.copycat.cluster.Member} type and
 * {@link net.kuujo.copycat.spi.protocol.Protocol}. This allows Copycat's communication model to be effectively
 * altered based on the protocol implementation. For instance, for HTTP protocols, an {@code HttpMember} will be required
 * by the {@code HttpCluster} in order to provide the {@code host} and {@code port} required for operating the TCP
 * protocol.<p>
 *
 * All Copycat clusters are modifiable even while the cluster is running. When the underlying
 * {@link net.kuujo.copycat.cluster.ClusterConfig} is changed by the user, the {@code Cluster} membership will be
 * automatically updated. Additionally, when the {@code Cluster} membership changes, Copycat will detect the change
 * via an {@link java.util.Observer} and update the Copycat cluster's internal configuration safely. However, it's
 * important to note that changes to the {@code Cluster} or its underlying {@link net.kuujo.copycat.cluster.ClusterConfig}
 * may not necessarily be represented in the actual Copycat cluster. Copycat directs all cluster membership changes
 * through the cluster leader and does so in a safe manner. This means that cluster configuration changes must occur
 * only on the leader node for the time being (this will be changed in the future), and configuration changes on follower
 * nodes will be essentially ignored unless they occur prior to starting the cluster.<p>
 *
 * Copycat core provides a {@link net.kuujo.copycat.cluster.LocalCluster} for performing inter-thread communication.
 * This cluster type should be used in testing environments only.
 *
 * @author <a href="http://github.com/kuujo"><NAME></a>
 */
public class Cluster<M extends Member> extends Observable implements Observer, Copyable<Cluster<M>> {
  private static final Logger LOGGER = LoggerFactory.getLogger(Cluster.class);
  protected final ClusterConfig<M> config;
  private final M localMember;
  private final Set<M> remoteMembers;
  private final Map<String, M> members;

  public Cluster(ClusterConfig<M> config) {
    this.config = Assert.isNotNull(config, "config");
    this.localMember = config.getLocalMember();
    this.members = new HashMap<>(config.getMembers().size());
    this.members.put(localMember.id(), localMember);
    this.remoteMembers = new HashSet<>(config.getRemoteMembers().size());
    this.config.addObserver(this);
    clusterChanged(config);
  }

  @Override
  @SuppressWarnings("unchecked")
  public Cluster<M> copy() {
    return new Cluster(config.copy());
  }

  @Override
  @SuppressWarnings("unchecked")
  public void update(Observable o, Object arg) {
    LOGGER.info("{} - Configuration change detected, updating membership", this);
    clusterChanged((ClusterConfig<M>) o);
  }

  /**
   * Updates the cluster when the configuration changes.
   */
  private void clusterChanged(ClusterConfig<M> config) {
    // Add any remote members that don't already exist in the cluster.
    config.getMembers().forEach(member -> {
      if (!member.equals(localMember) && !members.containsKey(member.id())) {
        remoteMembers.add(member);
        members.put(member.id(), member);
      }
    });

    Iterator<M> iterator = remoteMembers.iterator();
    while (iterator.hasNext()) {
      M member = iterator.next();
      boolean exists = false;
      for (M m : config.getMembers()) {
        if (m.equals(member)) {
          exists = true;
          break;
        }
      }
      if (!exists) {
        iterator.remove();
        members.remove(member.id());
      }
    }
{'fraction_non_alphanumeric': 0.09911178509532062, 'fraction_numerical': 0.002166377816291161, 'mean_word_length': 3.966648735879505, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6984357', 'n_tokens_mistral': 2430, 'n_tokens_neox': 2384, 'n_words': 941}