message.py

from typing import Dict
from paxos_utils import Address, BallotValuePair, BallotNumber
from typing import List


class Message:
    def __init__(self, addr: Address) -> None:
        self.addr: Address = addr  # address of sender

    def __str__(self) -> str:
        return f"addr: {str(self.addr)}"


class LeaderHeartbeat(Message):
    def __init__(self, addr: Address, log: Dict[int, BallotValuePair], ballot_num: BallotNumber, leader_slot: List[bool]) -> None:
        super().__init__(addr)
        self.log: Dict[int, BallotValuePair] = log
        self.ballot_num: BallotNumber = ballot_num
        self.leader_slot: List[bool] = leader_slot # for example, [False, True, False] means that we believe we are the leader for 2nd slot only. etc.

    def __str__(self) -> str:
        return f"LeaderHeartbeat({super().__str__()}, {str(self.ballot_num)}, log: {self.log}, leader_slot: {str(self.leader_slot)})"

class P1A(Message):
    def __init__(self, addr: Address, ballot_num: BallotNumber, leader_slot: int) -> None:
        super().__init__(addr)
        # ballot_num is a BallotNumber
        # leader_slot is an int. It is the slot we are trying to become the leader of. 
        # Let n be the number of paxos nodes, then 0 <= leader_slot < n. 
        # To clarify, If leader_slot is 1, then we give this new leader slot 1, 1+n, 1+2n, ... if their revoke attempt is successful
        self.ballot_num: BallotNumber = ballot_num
        self.leader_slot: int = leader_slot

    def __str__(self) -> str:
        return f"P1A({super().__str__()}, {str(self.ballot_num)}, leader_slot: {str(self.leader_slot)})"


class P1B(Message):
    def __init__(self, addr: Address, accepted_ballot: BallotNumber, accepted: Dict[int, BallotValuePair], leader_slot: int) -> None:
        super().__init__(addr)
        # accepted_ballot is a BallotNumber
        # accepted is a dict from int to BallotValuePair
        self.accepted_ballot: BallotNumber = accepted_ballot
        self.accepted: Dict[int, BallotValuePair] = accepted
        self.leader_slot: int = leader_slot

    def __str__(self) -> str:
        return f"P1B({super().__str__()}, {str(self.accepted_ballot)}, {str(self.accepted)}, leader_slot: {str(self.leader_slot)})"

class Skip(Message):
    # Special propose for no-op, For now this is not used since Skip is using P2A with value = None
    def __init__(self, addr: Address, ballot_num: BallotNumber, slot_num: int) -> None:
        super().__init__(addr)
        # ballot is BallotNumber
        # slot_num is an int
        self.ballot_num: BallotNumber = ballot_num
        self.slot_num: int = slot_num

    def __str__(self) -> str:
        return f"Skip({super().__str__()}, {str(self.ballot_num)}, slot_num: {str(self.slot_num)})"

class P2A(Message):
    def __init__(self, addr: Address, ballot_num: BallotNumber, slot_num: int, value) -> None:
        super().__init__(addr)
        # ballot is BallotNumber
        # slot_num is an int
        # value is a PaxosRequest
        self.ballot_num: BallotNumber = ballot_num
        self.slot_num: int = slot_num
        self.value: PaxosRequest = value

    def __str__(self) -> str:
        return f"P2A({super().__str__()}, {str(self.ballot_num)}, slot_num: {str(self.slot_num)}, value: {str(self.value)})"


class P2B(Message):
    def __init__(self, addr: Address, ballot: BallotNumber, slot_num: int) -> None:
        super().__init__(addr)
        # ballot is BallotNumber
        # slot_num is an int
        self.ballot_num: BallotNumber = ballot
        self.slot_num: int = slot_num

    def __str__(self) -> str:
        return f"P2B({super().__str__()}, {str(self.ballot_num)}, slot_num: {str(self.slot_num)})"


class PaxosRequest(Message):
    def __init__(self, addr: Address, lock_command) -> None:
        super().__init__(addr)
        # lock_command is of type LockCommand
        self.lock_command: LockCommand = lock_command

    def __str__(self) -> str:
        return f"PaxosRequest({super().__str__()}, {str(self.lock_command)})"


class PaxosResult(Message):
    def __init__(self, addr: Address, value: bool, lock_command) -> None:
        super().__init__(addr)
        self.value: bool = value
        self.cmd: LockCommand = lock_command

    def __str__(self) -> str:
        return f"PaxosResult({super().__str__()}, value: {str(self.value)}, cmd: {str(self.cmd)})"


class LockCommand:
    def __init__(self, op, value, req_num) -> None:
        # op is either "lock" or "unlock"
        # value is value to lock (int)
        self.op = op
        self.value = value
        self.req_num = req_num

    def __str__(self) -> str:
        return f"LockCommand(op: {str(self.op)}, value: {str(self.value)})"