You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: change.rs
    path of file: ./repos/quickwit/quickwit/quickwit-cluster/src
    
    the code of the file until where you have to start completion: // Copyright (C) 2024 Quickwit, Inc.
//
// Quickwit is offered under the AGPL v3.0 and as commercial software.
// For commercial licensing, contact us at hello@quickwit.io.
//
// AGPL:
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

use std::collections::btree_map::Entry;
use std::collections::BTreeMap;
use std::pin::Pin;
use std::task::{Context, Poll};

use chitchat::{ChitchatId, NodeState};
use futures::Stream;
use pin_project::pin_project;
use quickwit_common::sorted_iter::{KeyDiff, SortedByKeyIterator};
use quickwit_common::tower::{make_channel, warmup_channel};
use quickwit_proto::types::NodeId;
use tokio::sync::mpsc;
use tokio_stream::wrappers::UnboundedReceiverStream;
use tonic::transport::Channel;
use tracing::{info, warn};

use crate::member::NodeStateExt;
use crate::ClusterNode;

/// Describes a change in the cluster.
#[derive(Debug, Clone)]
pub enum ClusterChange {
    Add(ClusterNode),
    Update(ClusterNode),
    Remove(ClusterNode),
}

/// A stream of cluster change events.
#[pin_project]
pub struct ClusterChangeStream(#[pin] UnboundedReceiverStream<ClusterChange>);

impl ClusterChangeStream {
    pub fn new_unbounded() -> (Self, mpsc::UnboundedSender<ClusterChange>) {
        let (change_stream_tx, change_stream_rx) = mpsc::unbounded_channel();
        (
            Self(UnboundedReceiverStream::new(change_stream_rx)),
            change_stream_tx,
        )
    }
}

impl Stream for ClusterChangeStream {
    type Item = ClusterChange;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.project().0.poll_next(cx)
    }
}

/// A factory for creating cluster change streams.
pub trait ClusterChangeStreamFactory: Clone + Send + 'static {
    fn create(&self) -> ClusterChangeStream;
}

/// Compares the digests of the previous and new set of lives nodes, identifies the changes that
/// occurred in the cluster, and emits the corresponding events, focusing on ready nodes only.
pub(crate) async fn compute_cluster_change_events(
    cluster_id: &str,
    self_chitchat_id: &ChitchatId,
    previous_nodes: &mut BTreeMap<NodeId, ClusterNode>,
    previous_node_states: &BTreeMap<ChitchatId, NodeState>,
    new_node_states: &BTreeMap<ChitchatId, NodeState>,
) -> Vec<ClusterChange> {
    let mut cluster_events = Vec::new();

    for key_diff in previous_node_states
        .iter()
        .diff_by_key(new_node_states.iter())
    {
        match key_diff {
            // The node has joined the cluster.
            KeyDiff::Added(chitchat_id, node_state) => {
                let node_events = compute_cluster_change_events_on_added(
                    cluster_id,
                    self_chitchat_id,
                    chitchat_id,
                    node_state,
                    previous_nodes,
                )
                .await;

                cluster_events.extend(node_events);
            }
            // The node's state has changed.
            KeyDiff::Unchanged(chitchat_id, previous_node_state, new_node_state)
                if previous_node_state.max_version() != new_node_state.max_version() =>
            {
                let node_event_opt = compute_cluster_change_events_on_updated(
                    cluster_id,
                    self_chitchat_id,
                    chitchat_id,
                    new_node_state,
                    previous_nodes,
                )
                .await;

                if let Some(node_event) = node_event_opt {
                    cluster_events.push(node_event);
                }
            }
            // The node's state has not changed.
            KeyDiff::Unchanged(_chitchat_id, _previous_max_version, _new_max_version) => {}
            // The node has left the cluster, i.e. it is considered dead by the failure detector.
            KeyDiff::Removed(chitchat_id, _node_state) => {
                let node_event_opt =
                    compute_cluster_change_events_on_removed(chitchat_id, previous_nodes);

                if let Some(node_event) = node_event_opt {
                    cluster_events.push(node_event);
                }
            }
        };
    }
    cluster_events
}

async fn compute_cluster_change_events_on_added(
    cluster_id: &str,
    self_chitchat_id: &ChitchatId,
    new_chitchat_id: &ChitchatId,
    new_node_state: &NodeState,
    previous_nodes: &mut BTreeMap<NodeId, ClusterNode>,
) -> Vec<ClusterChange> {
    let is_self_node = self_chitchat_id == new_chitchat_id;
    let new_node_id: NodeId = new_chitchat_id.node_id.clone().into();
    let maybe_previous_node_entry = previous_nodes.entry(new_node_id);

    let mut events = Vec::new();
    let mut verb = "joined";

    if let Entry::Occupied(previous_node_entry) = maybe_previous_node_entry {
        let previous_node_ref = previous_node_entry.get();

        if previous_node_ref.chitchat_id().generation_id > new_chitchat_id.generation_id {
            warn!(
                node_id=%new_chitchat_id.node_id,
                generation_id=%new_chitchat_id.generation_id,
                "ignoring node `{}` rejoining the cluster with a lower generation ID",
                new_chitchat_id.node_id
            );
            return events;
        }
        let previous_node = previous_node_entry.remove();
        verb = "rejoined";

        if previous_node.is_ready
    
    async fn test_compute_cluster_change_events_on_added() {
        let cluster_id = "test-cluster".to_string();
        let self_port = 1234;
        let self_chitchat_id = ChitchatId::for_local_test(self_port);
        {
            // New node joins the cluster with an invalid gRPC advertise address.
            let port = 1235;
            let new_chitchat_id = ChitchatId::for_local_test(port);
            let mut new_node_state = NodeStateBuilder::default().build();
            new_node_state.set(GRPC_ADVERTISE_ADDR_KEY, "bogus-grpc-advertise-addr");
            let mut previous_nodes = BTreeMap::new();

            let events = compute_cluster_change_events_on_added(
                &cluster_id,
                &self_chitchat_id,
                &new_chitchat_id,
                &new_node_state,
                &mut previous_nodes,
            )
            .await;
            assert!(events.is_empty());
            assert!(previous_nodes.is_empty());
        }
        {
            // New node joins the cluster but is not ready.
            let port = 1235;
            let grpc_advertise_addr: SocketAddr = ([127, 0, 0, 1], port + 1).into();
            let new_chitchat_id = ChitchatId::for_local_test(port);
            let new_node_state = NodeStateBuilder::default()
                .with_grpc_advertise_addr(grpc_advertise_addr)
                .with_readiness(false)
                .build();
            let mut previous_nodes = BTreeMap::new();

            let events = compute_cluster_change_events_on_added(
                &cluster_id,
                &self_chitchat_id,
                &new_chitchat_id,
                &new_node_state,
                &mut previous_nodes,
            )
            .await;
            assert!(events.is_empty());

            let node = previous_nodes.get(&new_chitchat_id.node_id).unwrap();

            assert_eq!(node.chitchat_id(), &new_chitchat_id);
            assert_eq!(node.grpc_advertise_addr(), grpc_advertise_addr);
            assert!(!node.is_self_node());
            assert!(!node.is_ready());
        }
        {
            // New node joins the cluster and is ready.
            let port = 1235;
            let grpc_advertise_addr: SocketAddr = ([127, 0, 0, 1], port + 1).into();
            let new_chitchat_id = ChitchatId::for_local_test(port);
            let new_node_state = NodeStateBuilder::default()
                .with_grpc_advertise_addr(grpc_advertise_addr)
                .with_readiness(true)
                .build();
            let mut previous_nodes = BTreeMap::new();

            let events = compute_cluster_change_events_on_added(
                &cluster_id,
                &self_chitchat_id,
                &new_chitchat_id,
                &new_node_state,
                &mut previous_nodes,
            )
            .await;

            let ClusterChange::Add(node) = &events[0] else {
                panic!("expected `ClusterChange::Add` event, got `{:?}`", events[0]);
            };
            assert_eq!(node.chitchat_id(), &new_chitchat_id);
            assert_eq!(node.grpc_advertise_addr(), grpc_advertise_addr);
            assert!(!node.is_self_node());
            assert!(node.is_ready());
            assert_eq!(previous_nodes.get(&new_chitchat_id.node_id).unwrap(), node);

            // Node rejoins with same node ID but newer generation ID.
            let mut rejoined_chitchat_id = ChitchatId::for_local_test(port);
            rejoined_chitchat_id.generation_id += 1;

            let events = compute_cluster_change_events_on_added(
                &cluster_id,
                &self_chitchat_id,
                &rejoined_chitchat_id,
                &new_node_state,
                &mut previous_nodes,
            )
            .await;
            assert_eq!(events.len(), 2);

            let ClusterChange::Remove(removed_node) = &events[0] else {
                panic!(
                    "expected `ClusterChange::Remove` event, got `{:?}`",
                    events[0]
                );
            };
            assert_eq!(removed_node.chitchat_id(), &new_chitchat_id);

            let ClusterChange::Add(rejoined_node) = &events[1] else {
                panic!("expected `ClusterChange::Add` event, got `{:?}`", events[1]);
            };
            assert_eq!(rejoined_node.chitchat_id(), &rejoined_chitchat_id);
            assert_eq!(
                previous_nodes.get(&rejoined_chitchat_id.node_id).unwrap(),
                rejoined_node
            );

            // Node comes back from the dead with an older generation ID.
            let events = compute_cluster_change_events_on_added(
                &cluster_id,
                &self_chitchat_id,
                &new_chitchat_id,
                &new_node_state,
                &mut previous_nodes,
            )
            .await;
            assert!(events.is_empty());
            assert_eq!(
                previous_nodes.get(&rejoined_chitchat_id.node_id).unwrap(),
                rejoined_node
            );
        }
        {
            // Self node joins the cluster and is ready.
            let grpc_advertise_addr: SocketAddr = ([127, 0, 0, 1], self_port + 1).into();
            let new_chitchat_id = self_chitchat_id.clone();
            let new_node_state = NodeStateBuilder::default()
                .with_grpc_advertise_addr(grpc_advertise_addr)
                .with_readiness(true)
                .build();
            let mut previous_nodes = BTreeMap::new();

            let events = compute_cluster_change_events_on_added(
                &cluster_id,
                &self_chitchat_id,
                &new_chitchat_id,
                &new_node_state,
                &mut previous_nodes,
            )
            .await;
            assert_eq!(events.len(), 1);

            let ClusterChange::Add(node) = &events[0] else {
                panic!("expected `ClusterChange::Add` event, got `{:?}`", events[0]);
            };
            assert_eq!(node.chitchat_id(), &new_chitchat_id);
            assert_eq!(node.grpc_advertise_addr(), grpc_advertise_addr);
            assert!(node.is_self_node());
            assert!(node.is_ready());
            assert_eq!(previous_nodes.get(&new_chitchat_id.node_id).unwrap(), node);
        }
    }