Full impl of an ATM machine example introduced in book: C++ Concurrency in Action, chapter 4.

atm.cpp

/**
 * Full impl of an ATM machine example introduced in book: C++ Concurrency in Action, chapter 4.
 */

#include <thread>
#include <iostream>
#include <string>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <memory>

// message utility
namespace msg {

struct Message {
    virtual ~Message() = default;
};

template <typename Msg>
struct MessageWrapper : public Message {
    Msg contents;
    explicit MessageWrapper(const Msg &ctns) : contents(ctns) {}
};

class MessageQueue {
public:
    // add a message into message queue
    template <typename T>
    void push(const T &msg) {
        std::lock_guard<std::mutex> lock(mMutex);
        mMsgQueue.push(std::make_shared<MessageWrapper<T>>(msg));
        mCond.notify_all(); // notify all threads
    }

    // pop out a message from message queue if it contains
    std::shared_ptr<Message> wait_and_pop() {
        std::unique_lock<std::mutex> lock(mMutex); // unique_lock for potential unlock
        mCond.wait(lock, [&]{ return !mMsgQueue.empty(); });
        auto msg = mMsgQueue.front();
        mMsgQueue.pop();
        return msg;
    }

private:
    std::mutex mMutex;
    std::condition_variable mCond;
    std::queue<std::shared_ptr<Message>> mMsgQueue;
};

struct CloseMessageQueue {}; // done message, doing nothing

template <typename Dsp, typename Msg, typename Fn>
class TemplateDispatcher {
public:
    TemplateDispatcher(TemplateDispatcher &&other)
            : pMsgQueue(other.pMsgQueue),
              pPrevDispatcher(other.pPrevDispatcher),
              mFunc(std::move(other.mFunc)),
              mIsChained(other.mIsChained) {
        other.mIsChained = true;
    }

    TemplateDispatcher(MessageQueue *queue, Dsp *prev, Fn &&f)
            : pMsgQueue(queue),
              pPrevDispatcher(prev),
              mFunc(std::forward<Fn>(f)),
              mIsChained(false) {
        pPrevDispatcher->mIsChained = true;
    }

    template <typename OtherMsg, typename OtherFn>
    TemplateDispatcher<TemplateDispatcher, OtherMsg, OtherFn> handle(OtherFn &&of) {
        return TemplateDispatcher<TemplateDispatcher, OtherMsg, OtherFn>{
            pMsgQueue, this, std::forward<OtherFn>(of)};
    }

    virtual ~TemplateDispatcher() noexcept(false) {
        if (!mIsChained) {
            wait_and_dispatch();
        }
    }

private:
    MessageQueue *pMsgQueue;
    Dsp *pPrevDispatcher;
    Fn mFunc;
    bool mIsChained;

    TemplateDispatcher(const TemplateDispatcher &) = delete;
    TemplateDispatcher &operator=(const TemplateDispatcher &) = delete;

    // mark other type specialization as friend claSS
    template <typename OtherDsp, typename OtherMsg, typename OtherFn>
    friend class TemplateDispatcher;

    // wait for message until received and dispatched
    void wait_and_dispatch() {
        for (;;) {
            auto msg = pMsgQueue->wait_and_pop();
            if (dispatch(msg)) {
                break;
            }
        }
    }

    // dispatch/handle message
    bool dispatch(const std::shared_ptr<Message> &msg) {
        if (auto *wrapper = dynamic_cast<MessageWrapper<Msg> *>(msg.get())) {
            mFunc(wrapper->contents);
            return true;
        } else {
            return pPrevDispatcher->dispatch(msg);
        }
    }
};

class Dispatcher {
public:
    Dispatcher(Dispatcher &&other)
            : pMsgQueue(other.pMsgQueue),
              mIsChained(other.mIsChained) {
        other.mIsChained = true;
    }
    explicit Dispatcher(MessageQueue *queue)
            : pMsgQueue(queue), mIsChained(false) {}

    template<typename Msg, typename Fn>
    TemplateDispatcher<Dispatcher, Msg, Fn> handle(Fn &&f) {
        return TemplateDispatcher<Dispatcher, Msg, Fn>(pMsgQueue, this, std::forward<Fn>(f));
    }

    virtual ~Dispatcher() noexcept(false) {
        if (!mIsChained) {
            wait_and_dispatch();
        }
    }

private:
    MessageQueue *pMsgQueue;
    bool mIsChained;

    Dispatcher(const Dispatcher &) = delete;
    Dispatcher &operator=(const Dispatcher &) = delete;

    template <typename Dsp, typename Msg, typename Fn>
    friend class TemplateDispatcher;

    void wait_and_dispatch() {
        for (;;) {
            auto msg = pMsgQueue->wait_and_pop();
            dispatch(msg);
        }
    }

    bool dispatch(const std::shared_ptr<Message> &msg) {
        if (dynamic_cast<MessageWrapper<CloseMessageQueue> *>(msg.get())) {
            throw CloseMessageQueue{};
        }
        return false;
    }
};

// focus on sending message
class Sender {
public:
    Sender() : pMsgQueue(nullptr) {}
    explicit Sender(MessageQueue *queue) : pMsgQueue(queue) {}

    template <typename Msg>
    void send(const Msg &msg) {
        if (pMsgQueue) {
            pMsgQueue->push(msg);
        }
    }

private:
    MessageQueue *pMsgQueue;
};

// focus on receiving message
class Receiver {
public:
    // for cast (get corresponding sender with my message queue)
    operator Sender() {
        return Sender{&mMsgQueue};
    }

    Dispatcher wait() {
        return Dispatcher{&mMsgQueue};
    }

private:
    MessageQueue mMsgQueue;
};

}  // namespace msg

// -------------------------------------------------------------------------------------------------
// messages

// bank card
struct Card {
    std::string account;
};
struct CardInserted {
    std::string account;
    CardInserted(const std::string &acnt) : account(acnt) {}
};
struct EjectCard {};

// withdraw
struct Withdraw {
    std::string account;
    unsigned amount;
    mutable msg::Sender sender;
    Withdraw(const std::string &acnt, unsigned amnt, msg::Sender snd)
            : account(acnt), amount(amnt), sender(snd) {}
};

struct CancelWithdraw {
    std::string account;
    unsigned amount;
    CancelWithdraw(const std::string &acnt, unsigned amnt)
            : account(acnt), amount(amnt) {}
};

struct WithdrawOK {};
struct WithdrawDenied {};
struct WithdrawProcessed {
    std::string account;
    unsigned amount;
    WithdrawProcessed(const std::string &acnt, unsigned amnt)
        : account(acnt), amount(amnt) {}
};
struct WithdrawPressed {
    unsigned amount;
    WithdrawPressed(unsigned amnt) : amount(amnt) {}
};

// PIN code
struct DigitPressed {
    char digit;
    explicit DigitPressed(char d) : digit(d) {}
};
struct ClearLastPressed{};
struct CancelPressed{};
struct VerifyPIN {
    std::string account;
    std::string pin;
    mutable msg::Sender sender;
    VerifyPIN(const std::string &acnt, const std::string &pn, msg::Sender snd)
            : account(acnt), pin(pn), sender(snd) {}
};
struct PINVerified {};
struct PINIncorrect {};

// OK, issue money
struct IssueMoney {
    unsigned amount;
    IssueMoney(unsigned amnt) : amount(amnt) {}
};

// account balance
struct Balance {
    unsigned amount;
    explicit Balance(unsigned amnt) : amount(amnt) {}
};
struct GetBalance {
    std::string account;
    mutable msg::Sender sender;
    GetBalance(const std::string &acnt, msg::Sender snd)
            : account(acnt), sender(snd) {}
};
struct BalancePressed {};

// Display some infos
struct DisplayEnterPIN {};
struct DisplayEnterCard {};
struct DisplayInsufficientFunds {};
struct DisplayWithdrawalCanceled {};
struct DisplayPINIncorrectMessage {};
struct DisplayWithdrawalOptions {};
struct DisplayBalance {
    unsigned amount;
    explicit DisplayBalance(unsigned amnt) : amount(amnt) {}
};

// -------------------------------------------------------------------------------------------------
// machines

// ATM
class ATMMachine {
public:
    ATMMachine(msg::Sender bank, msg::Sender intf)
            : mBank(bank), mIntf(intf) {}
    
    void done() {
        std::cout << "ATM done." << std::endl;
        get_sender().send(msg::CloseMessageQueue{});
    }

    void run() {
        mHook = &ATMMachine::waiting_for_card;
        try {
            for (;;) {
                (this->*mHook)();
            }
        }
        catch(const msg::CloseMessageQueue &) {}
    }

    msg::Sender get_sender() {
        return mReceiver; // will cast
    }

private:
    msg::Receiver mReceiver;
    msg::Sender mBank;
    msg::Sender mIntf;
    void (ATMMachine::*mHook)();
    std::string mAccount;
    unsigned mWithdrawalAmount;
    std::string mPIN;

    void process_withdrawal() {
        mReceiver.wait()
            .handle<WithdrawOK>(
                [&](const WithdrawOK &msg) {
                    mIntf.send(IssueMoney(mWithdrawalAmount));
                    mBank.send(WithdrawProcessed(mAccount, mWithdrawalAmount));
                    mHook = &ATMMachine::done_processing;
                }
            )
            .handle<WithdrawDenied>(
                [&](const WithdrawDenied &msg) {
                    mIntf.send(DisplayInsufficientFunds{});
                    mHook = &ATMMachine::done_processing;
                }
            )
            .handle<CancelPressed>(
                [&](const CancelPressed &msg) {
                    mBank.send(CancelWithdraw(mAccount, mWithdrawalAmount));
                    mIntf.send(DisplayWithdrawalCanceled{});
                    mHook = &ATMMachine::done_processing;
                }
            );
    }

    void process_balance() {
        mReceiver.wait()
            .handle<Balance>(
                [&](const Balance &msg) {
                    mIntf.send(DisplayBalance{msg.amount});
                    mHook = &ATMMachine::wait_for_action;
                }
            )
            .handle<CancelPressed>(
                [&](const CancelPressed &msg) {
                    mHook = &ATMMachine::done_processing;
                }
            );
    }

    void wait_for_action() {
        mIntf.send(DisplayWithdrawalOptions{});
        mReceiver.wait()
            .handle<WithdrawPressed>(
                [&](const WithdrawPressed &msg) {
                    mWithdrawalAmount = msg.amount;
                    mBank.send(Withdraw{mAccount, msg.amount, mReceiver});
                    mHook = &ATMMachine::process_withdrawal;
                }
            )
            .handle<BalancePressed>(
                [&](const BalancePressed &msg) {
                    mBank.send(GetBalance{mAccount, mReceiver});
                    mHook = &ATMMachine::process_balance;
                }
            )
            .handle<CancelPressed>(
                [&](const CancelPressed &msg) {
                    mHook = &ATMMachine::done_processing;
                }
            );
    }

    void verifying_pin() {
        mReceiver.wait()
            .handle<PINVerified>(
                [&](const PINVerified &msg) {
                    mHook = &ATMMachine::wait_for_action;
                }
            )
            .handle<PINIncorrect>(
                [&](const PINIncorrect &msg) {
                    mIntf.send(DisplayPINIncorrectMessage{});
                    mHook = &ATMMachine::done_processing;
                }
            )
            .handle<CancelPressed>(
                [&](const CancelPressed &msg) {
                    mHook = &ATMMachine::done_processing;
                }
            );
    }

    void getting_pin() {
        mReceiver.wait()
            .handle<DigitPressed>(
                [&](const DigitPressed &msg) {
                    const unsigned pinLen = 4;
                    mPIN += msg.digit;
                    if (mPIN.length() == pinLen) {
                        mBank.send(VerifyPIN{mAccount, mPIN, mReceiver});
                        mHook = &ATMMachine::verifying_pin;
                    }
                }
            )
            .handle<ClearLastPressed>(
                [&](const ClearLastPressed &msg) {
                    if (!mPIN.empty()) {
                        mPIN.pop_back();
                    }
                }
            )
            .handle<CancelPressed>(
                [&](const CancelPressed &msg) {
                    mHook = &ATMMachine::done_processing;
                }
            );
    }

    void waiting_for_card() {
        mIntf.send(DisplayEnterCard{});
        mReceiver.wait()
            .handle<CardInserted>(
                [&](const CardInserted &msg) {
                    mAccount = msg.account;
                    mPIN = "";
                    mIntf.send(DisplayEnterPIN{});
                    mHook = &ATMMachine::getting_pin;
                }
            );
    }

    void done_processing() {
        mIntf.send(EjectCard{});
        mHook = &ATMMachine::waiting_for_card;
    }

    ATMMachine(const ATMMachine &) = delete;
    ATMMachine &operator=(const ATMMachine &) = delete;
};

// bank
class BankMachine {
public:
    BankMachine() : mBalance(199) {}
    void done() {
        std::cout << "Bank done." << std::endl;
        get_sender().send(msg::CloseMessageQueue{});
    }

    void run() {
        try {
            for (;;) {
                mReceiver.wait()
                    .handle<VerifyPIN>(
                        [&](const VerifyPIN &msg) {
                            if (msg.pin == "1937") {
                                msg.sender.send(PINVerified{});
                            } else {
                                msg.sender.send(PINIncorrect{});
                            }
                        }
                    )
                    .handle<Withdraw>(
                        [&](const Withdraw &msg) {
                            if (mBalance >= msg.amount) {
                                msg.sender.send(WithdrawOK{});
                                mBalance -= msg.amount;
                            } else {
                                msg.sender.send(WithdrawDenied{});
                            }
                        }
                    )
                    .handle<GetBalance>(
                        [&](const GetBalance &msg) {
                            msg.sender.send(Balance{mBalance});
                        }
                    )
                    .handle<WithdrawProcessed>(
                        [&](const WithdrawProcessed &msg) {}
                    )
                    .handle<CancelWithdraw>(
                        [&](const CancelWithdraw &msg) {}
                    );
            }
        } catch (const msg::CloseMessageQueue &) {}
    }

    msg::Sender get_sender() {
        return mReceiver;
    }

private:
    msg::Receiver mReceiver;
    unsigned mBalance;
};

// interface
class InterfaceMachine {
public:
    void done() {
        std::cout << "Intf done." << std::endl;
        get_sender().send(msg::CloseMessageQueue{});
    }

    void run() {
        try {
            for (;;) {
                mReceiver.wait()
                    .handle<IssueMoney>(
                        [&](const IssueMoney &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << "-- Issuing money: " << msg.amount << std::endl;
                            }
                        }
                    )
                    .handle<DisplayInsufficientFunds>(
                        [&](const DisplayInsufficientFunds &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << "** Insufficient Founds!!" << std::endl;
                            }
                        }
                    )
                    .handle<DisplayEnterPIN>(
                        [&](const DisplayEnterPIN &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << ">> Please Enter Your PIN(0-9)?" << std::endl;
                            }
                        }
                    )
                    .handle<DisplayEnterCard>(
                        [&](const DisplayEnterCard &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << ">> Please Enter Your Card(i)?" << std::endl;
                            }
                        }
                    )
                    .handle<DisplayBalance>(
                        [&](const DisplayBalance &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << "-- Balance of Your Account is :" << msg.amount << std::endl;
                            }
                        }
                    )
                    .handle<DisplayWithdrawalOptions>(
                        [&](const DisplayWithdrawalOptions &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << ">> Withdraw 50? (w)" << std::endl;
                                std::cout << ">> Display Balance? (b)" << std::endl;
                                std::cout << ">> Cencel? (c)" << std::endl;
                            }
                        }
                    )
                    .handle<DisplayWithdrawalCanceled>(
                        [&](const DisplayWithdrawalCanceled &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << "-- Withdraw Canceled." << std::endl;
                            }
                        }
                    )
                    .handle<DisplayPINIncorrectMessage>(
                        [&](const DisplayPINIncorrectMessage &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << "** PIN Incorrect." << std::endl;
                            }
                        }
                    )
                    .handle<EjectCard>(
                        [&](const EjectCard &msg) {
                            {
                                std::lock_guard<std::mutex> lk(mMutex);
                                std::cout << "-- Ejecting Card." << std::endl;
                            }
                        }
                    );
            }
        } catch (const msg::CloseMessageQueue &) {}
    }

    msg::Sender get_sender() {
        return mReceiver;
    }

private:
    msg::Receiver mReceiver;
    std::mutex mMutex;
};


int main() {
    BankMachine bank;
    InterfaceMachine intf;
    ATMMachine atm{bank.get_sender(), intf.get_sender()};

    std::thread bank_thread{&BankMachine::run, &bank};
    std::thread intf_thread{&InterfaceMachine::run, &intf};
    std::thread atm_thread{&ATMMachine::run, &atm};

    msg::Sender sender{atm.get_sender()};
    bool quit = false;

    while (!quit) {
        char input = getchar();
        switch(input) {
            case '0':
            case '1':
            case '2':
            case '3':
            case '4':
            case '5':
            case '6':
            case '7':
            case '8':
            case '9':
                sender.send(DigitPressed{input});
                break;
            case 'b':
                sender.send(BalancePressed{});
                break;
            case 'w':
                sender.send(WithdrawPressed{50});
                break;
            case 'c':
                sender.send(CancelPressed{});
                break;
            case 'i':
                sender.send(CardInserted{"acc1234"});
                break;
            case 'q':
                quit = true;
                break;
            default:
                break;
        }
    }

    bank.done();
    atm.done();
    intf.done();

    atm_thread.join();
    bank_thread.join();
    intf_thread.join();
}