Members and Status

/**
 * Represents the address, current status, and roles of a cluster member node.
 * Note: hashCode and equals are solely based on the underlying Address, not its MemberStatus and roles.
 */
class Member private[cluster] (
  val uniqueAddress: UniqueAddress,
  private[cluster] val upNumber: Int,
  val status: MemberStatus,
  val roles: Set[String]
) extends Serializable {
  /** The network address of this member */
  def address: Address
  
  /** Data center this member belongs to */
  lazy val dataCenter: DataCenter
  
  /** Check if member has specific role */
  def hasRole(role: String): Boolean
  
  /** Java API: get all roles as Java Set */
  def getRoles: java.util.Set[String]
  
  /** 
   * Is this member older, has been part of cluster longer, than another member.
   * Only correct when comparing two existing members in a cluster.
   * Throws IllegalArgumentException if members from different data centers.
   */
  def isOlderThan(other: Member): Boolean
  
  /** Create copy with new status */
  def copy(status: MemberStatus): Member
  
  /** Create copy with Up status and new up number */
  def copyUp(upNumber: Int): Member
}

// Working with member information
cluster.state.members.foreach { member =>
  println(s"Member: ${member.address}")
  println(s"Status: ${member.status}")
  println(s"Roles: ${member.roles.mkString(", ")}")
  println(s"Data Center: ${member.dataCenter}")
  
  // Check for specific roles
  if (member.hasRole("frontend")) {
    println("This is a frontend node")
  }
  
  if (member.hasRole("backend")) {
    println("This is a backend node")
  }
}

// Compare member ages (within same data center)
val members = cluster.state.members.toList
if (members.size >= 2) {
  val member1 = members(0) 
  val member2 = members(1)
  
  if (member1.dataCenter == member2.dataCenter) {
    if (member1.isOlderThan(member2)) {
      println(s"${member1.address} is older than ${member2.address}")
    }
  }
}

/**
 * Defines the current status of a cluster member node.
 * Can be one of: Joining, WeaklyUp, Up, Leaving, Exiting, Down, and Removed.
 */
sealed abstract class MemberStatus

object MemberStatus {
  /** Member is in the process of joining the cluster */
  case object Joining extends MemberStatus
  
  /** 
   * Member is weakly up - joined but not all nodes have seen it yet.
   * This happens when convergence cannot be reached due to unreachable nodes.
   */
  case object WeaklyUp extends MemberStatus
  
  /** Member is fully up and participating in the cluster */
  case object Up extends MemberStatus
  
  /** Member is gracefully leaving the cluster */
  case object Leaving extends MemberStatus
  
  /** Member is in the exiting phase of leaving */
  case object Exiting extends MemberStatus
  
  /** Member has been marked as down (forceful removal) */
  case object Down extends MemberStatus
  
  /** Member has been completely removed from the cluster */
  case object Removed extends MemberStatus
  
  // Java API accessors
  def joining: MemberStatus = Joining
  def weaklyUp: MemberStatus = WeaklyUp  
  def up: MemberStatus = Up
  def leaving: MemberStatus = Leaving
  def exiting: MemberStatus = Exiting
  def down: MemberStatus = Down
  def removed: MemberStatus = Removed
}

import akka.cluster.MemberStatus._

// Check member status
cluster.state.members.foreach { member =>
  member.status match {
    case Joining => println(s"${member.address} is joining")
    case WeaklyUp => println(s"${member.address} is weakly up")
    case Up => println(s"${member.address} is up and ready")
    case Leaving => println(s"${member.address} is leaving")
    case Exiting => println(s"${member.address} is exiting")
    case Down => println(s"${member.address} is down")
    case Removed => println(s"${member.address} was removed")
  }
}

// Filter members by status
val upMembers = cluster.state.members.filter(_.status == Up)
val joiningMembers = cluster.state.members.filter(_.status == Joining)

println(s"Up members: ${upMembers.size}")
println(s"Joining members: ${joiningMembers.size}")

/**
 * Member identifier consisting of address and random uid.
 * The uid is needed to be able to distinguish different
 * incarnations of a member with same hostname and port.
 */
case class UniqueAddress(address: Address, longUid: Long) extends Ordered[UniqueAddress] {
  /** Compare unique addresses for ordering */
  def compare(that: UniqueAddress): Int
}

object UniqueAddress {
  /** Create unique address with address and uid */
  def apply(address: Address, uid: Long): UniqueAddress
}

// Access unique address information
val member = cluster.selfMember
println(s"Address: ${member.uniqueAddress.address}")
println(s"Unique ID: ${member.uniqueAddress.longUid}")

// Unique addresses allow distinguishing restarts
val addr1 = UniqueAddress(Address("akka.tcp", "System", "host", 2551), 123456L)
val addr2 = UniqueAddress(Address("akka.tcp", "System", "host", 2551), 789012L)
// Same network address but different UIDs - different incarnations

object Member {
  /** Empty member set constant */
  val none: Set[Member] = Set.empty[Member]
  
  /**
   * Address ordering type class, sorts addresses by host and port
   */
  implicit val addressOrdering: Ordering[Address]
  
  /**
   * Member ordering type class, sorts members by host and port
   */
  implicit val ordering: Ordering[Member]
  
  /**
   * Sort members by age, i.e. using Member#isOlderThan.
   * Only makes sense to compare members of same data center.
   */
  val ageOrdering: Ordering[Member]
  
  /**
   * Picks the Member with the highest "priority" MemberStatus.
   */
  def highestPriorityOf(m1: Member, m2: Member): Member
}

// Sort members by address
val sortedByAddress = cluster.state.members.toList.sorted(Member.ordering)

// Sort members by age (oldest first)
val membersList = cluster.state.members.toList
val sortedByAge = membersList.sorted(Member.ageOrdering)

// Find member with highest priority status
val members = List(member1, member2)
val highestPriority = members.reduceLeft(Member.highestPriorityOf)

// Work with empty member set
if (cluster.state.members == Member.none) {
  println("No members in cluster")
}

// Member role methods
def hasRole(role: String): Boolean
def roles: Set[String]
def getRoles: java.util.Set[String] // Java API

// Cluster-level role information
def selfRoles: Set[String] // From Cluster class
def getSelfRoles: java.util.Set[String] // Java API

// Configure roles in application.conf
// akka.cluster.roles = ["frontend", "backend", "database"]

// Check self roles
println(s"My roles: ${cluster.selfRoles}")

// Find members by role
val frontendMembers = cluster.state.members.filter(_.hasRole("frontend"))
val backendMembers = cluster.state.members.filter(_.hasRole("backend"))
val dbMembers = cluster.state.members.filter(_.hasRole("database"))

println(s"Frontend nodes: ${frontendMembers.size}")
println(s"Backend nodes: ${backendMembers.size}")
println(s"Database nodes: ${dbMembers.size}")

// Route work based on roles
def routeToBackend(message: Any): Unit = {
  val backendAddresses = cluster.state.members
    .filter(_.hasRole("backend"))
    .filter(_.status == Up)
    .map(_.address)
    
  if (backendAddresses.nonEmpty) {
    // Send to backend nodes
    backendAddresses.foreach { addr =>
      // Route message to backend at addr
    }
  }
}

// Status transition validation is handled internally
// Valid transitions:
// Joining -> WeaklyUp, Up, Leaving, Down, Removed
// WeaklyUp -> Up, Leaving, Down, Removed  
// Up -> Leaving, Down, Removed
// Leaving -> Exiting, Down, Removed
// Down -> Removed
// Exiting -> Removed, Down
// Removed -> (no transitions - terminal state)

// Monitor member lifecycle in event handler
class MemberLifecycleMonitor extends Actor with ActorLogging {
  var memberHistory = Map.empty[Address, List[MemberStatus]]
  
  def receive = {
    case event: MemberEvent =>
      val member = event.member
      val history = memberHistory.getOrElse(member.address, List.empty)
      val newHistory = member.status :: history
      memberHistory = memberHistory + (member.address -> newHistory)
      
      log.info("Member {} status: {} (history: {})", 
        member.address, member.status, newHistory.reverse.mkString(" -> "))
        
      // Detect problematic transitions
      if (member.status == Down && history.headOption.contains(Up)) {
        log.warning("Member {} went directly from Up to Down - possible network failure", 
          member.address)
      }
  }
}

// Data center information
def dataCenter: DataCenter // From Member
def selfDataCenter: DataCenter // From Cluster
type DataCenter = String

// Data center discovery from current state
def allDataCenters: Set[String] // From CurrentClusterState
def getAllDataCenters: java.util.Set[String] // Java API

// Work with data centers
val state = cluster.state
println(s"All data centers: ${state.allDataCenters}")
println(s"My data center: ${cluster.selfDataCenter}")

// Organize members by data center
val membersByDc = cluster.state.members.groupBy(_.dataCenter)
membersByDc.foreach { case (dc, members) =>
  println(s"Data center '$dc': ${members.size} members")
  members.foreach(m => println(s"  ${m.address} (${m.status})"))
}

// Find members in same data center
val sameDcMembers = cluster.state.members.filter(_.dataCenter == cluster.selfDataCenter)
val otherDcMembers = cluster.state.members.filter(_.dataCenter != cluster.selfDataCenter)

// Age comparison only valid within same data center
val myDcMembers = cluster.state.members.filter(_.dataCenter == cluster.selfDataCenter).toList
if (myDcMembers.size >= 2) {
  try {
    val oldest = myDcMembers.minBy(_.upNumber)
    println(s"Oldest member in my DC: ${oldest.address}")
  } catch {
    case _: IllegalArgumentException => 
      println("Cannot compare members from different data centers")
  }
}

// Core member types
class Member(uniqueAddress: UniqueAddress, status: MemberStatus, roles: Set[String])
case class UniqueAddress(address: Address, longUid: Long)
type DataCenter = String

// Member status enumeration
sealed abstract class MemberStatus
case object Joining extends MemberStatus
case object WeaklyUp extends MemberStatus
case object Up extends MemberStatus
case object Leaving extends MemberStatus
case object Exiting extends MemberStatus
case object Down extends MemberStatus
case object Removed extends MemberStatus

// Ordering types
implicit val addressOrdering: Ordering[Address]
implicit val ordering: Ordering[Member]
val ageOrdering: Ordering[Member]