消息传递框架与完整的ATM示例

    1回到第4章，我举了一个使用消息传递框架在线程间发送信息的例子。这里就会使用这个实现来完成ATM功能。下面完整代码就是功能的实现，包括消息传递框架。

    清单C.1实现了一个消息队列。其可以将消息以指针(指向基类)的方式存储在列表中；指定消息类型会由基类派生模板进行处理。推送包装类的构造实例，以及存储指向这个实例的指针；弹出实例的时候，将会返回指向其的指针。因为message_base类没有任何成员函数，在访问存储消息之前，弹出线程就需要将指针转为wrapped_message指针。

    清单C.1 简单的消息队列

    发送通过sender类(见清单C.2)实例处理过的消息。只能对已推送到队列中的消息进行包装。对sender实例的拷贝，只是拷贝了指向队列的指针，而非队列本身。

    清单C.2 sender类

    2. {
    3.   class sender
    4.   {
    5.     queue*q;  // sender是一个队列指针的包装类
    6.   public:
    7.     sender():  // sender无队列(默认构造函数)
    8.       q(nullptr)
    9.     {}
    11.     explicit sender(queue*q_):  // 从指向队列的指针进行构造
    12.       q(q_)
    13.     {}
    15.     template<typename Message>
    16.     void send(Message const& msg)
    17.     {
    18.       if(q)
    19.       {
    20.         q->push(msg);  // 将发送信息推送给队列
    21.       }
    22.     }
    23.   };
    24. }

    接收信息部分有些麻烦。不仅要等待队列中的消息，还要检查消息类型是否与所等待的消息类型匹配，并调用处理函数进行处理。那么就从receiver类的实现开始吧。

    1. namespace messaging
    2. {
    3.   class receiver
    4.   {
    5.     queue q;  // 接受者拥有对应队列
    6.   public:
    7.     operator sender()  // 允许将类中队列隐式转化为一个sender队列
    8.     {
    9.       return sender(&q);
    10.     }
    11.     dispatcher wait()  // 等待对队列进行调度
    12.     {
    13.       return dispatcher(&q);
    14.     }
    15.   };
    16. }

    sender只是引用一个消息队列，而receiver是拥有一个队列。可以使用隐式转换的方式获取sender引用的类。难点在于wait()中的调度。这里创建了一个dispatcher对象引用receiver中的队列。dispatcher类实现会在下一个清单中看到；如你所见，任务是在析构函数中完成的。在这个例子中，所要做的工作是对消息进行等待，以及对其进行调度。

    清单C.4 dispatcher类

    从wait()返回的dispatcher实例将马上被销毁，因为是临时变量，也向前文提到的，析构函数在这里做真正的工作。析构函数调用wait_and_dispatch()函数，这个函数中有一个循环①，等待消息的传入(这样才能进行弹出操作)，然后将消息传递给dispatch()函数。dispatch()函数本身②很简单；会检查小时是否是一个close_queue消息，当是close_queue消息时，抛出一个异常；如果不是，函数将会返回false来表明消息没有被处理。因为会抛出close_queue异常，所以析构函数会标示为noexcept(false)；在没有任何标识的情况下，一般情况下析构函数都noexcept(true)④型，这表示没有任何异常抛出，并且close_queue异常将会使程序终止。

    虽然，不会经常的去调用wait()函数，不过，在大多数时间里，你都希望对一条消息进行处理。这时就需要handle()成员函数③的加入。这个函数是一个模板，并且消息类型不可推断，所以你需要指定需要处理的消息类型，并且传入函数(或可调用对象)进行处理，并将队列传入当前dispatcher对象的handle()函数。这将在清单C.5中展示。这就是为什么，在测试析构函数中的chained值前，要等待消息耳朵原因；不仅是避免“移动”类型的对象对消息进行等待，而且允许将等待状态转移到新的TemplateDispatcher实例中。

    清单C.5 TemplateDispatcher类模板

    1. namespace messaging
    2. {
    3.   template<typename PreviousDispatcher,typename Msg,typename Func>
    4.   class TemplateDispatcher
    5.   {
    6.     queue* q;
    7.     PreviousDispatcher* prev;
    8.     Func f;
    9.     bool chained;
    11.     TemplateDispatcher(TemplateDispatcher const&)=delete;
    12.     TemplateDispatcher& operator=(TemplateDispatcher const&)=delete;
    14.     template<typename Dispatcher,typename OtherMsg,typename OtherFunc>
    15.     friend class TemplateDispatcher;  // 所有特化的TemplateDispatcher类型实例都是友元类
    17.     void wait_and_dispatch()
    18.     {
    19.       for(;;)
    20.       {
    21.         auto msg=q->wait_and_pop();
    22.         if(dispatch(msg))  // 1 如果消息处理过后，会跳出循环
    23.           break;
    24.       }
    25.     }
    27.     bool dispatch(std::shared_ptr<message_base> const& msg)
    28.     {
    29.       if(wrapped_message<Msg>* wrapper=
    30.          dynamic_cast<wrapped_message<Msg>*>(msg.get()))  // 2 检查消息类型，并且调用函数
    31.       {
    32.         f(wrapper->contents);
    33.         return true;
    34.       }
    35.       else
    36.       {
    37.         return prev->dispatch(msg);  // 3 链接到之前的调度器上
    38.       }
    39.     }
    40.   public:
    41.     TemplateDispatcher(TemplateDispatcher&& other):
    42.         q(other.q),prev(other.prev),f(std::move(other.f)),
    43.         chained(other.chained)
    44.     {
    45.       other.chained=true;
    46.     }
    47.     TemplateDispatcher(queue* q_,PreviousDispatcher* prev_,Func&& f_):
    48.         q(q_),prev(prev_),f(std::forward<Func>(f_)),chained(false)
    49.     {
    50.       prev_->chained=true;
    51.     }
    53.     template<typename OtherMsg,typename OtherFunc>
    54.     TemplateDispatcher<TemplateDispatcher,OtherMsg,OtherFunc>
    55.     handle(OtherFunc&& of)  // 4 可以链接其他处理器
    56.     {
    57.       return TemplateDispatcher<
    58.           TemplateDispatcher,OtherMsg,OtherFunc>(
    59.           q,this,std::forward<OtherFunc>(of));
    60.     }
    62.     ~TemplateDispatcher() noexcept(false)  // 5 这个析构函数也是noexcept(false)的
    63.     {
    64.       if(!chained)
    65.       {
    66.         wait_and_dispatch();
    67.       }
    68.     }
    69. }

    TemplateDispatcher<>类模板仿照了dispatcher类，二者几乎相同。特别是在析构函数上，都是调用wait_and_dispatch()等待处理消息。

    这种简单的架构允许你想队列推送任何类型的消息，并且调度器有选择的与接收端的消息进行匹配。同样，也允许为了推送消息，将消息队列的引用进行传递的同时，保持接收端的私有性。

    为了完成第4章的例子，消息的组成将在清单C.6中给出，各种状态机将在清单C.7,C.8和C.9中给出。最后，驱动代码将在C.10给出。

    清单C.6 ATM消息

    1. struct withdraw
    2. {
    3.   std::string account;
    4.   unsigned amount;
    5.   mutable messaging::sender atm_queue;
    7.   withdraw(std::string const& account_,
    8.            unsigned amount_,
    9.            messaging::sender atm_queue_):
    10.     account(account_),amount(amount_),
    11.     atm_queue(atm_queue_)
    12.   {}
    13. };
    15. struct withdraw_ok
    16. {};
    18. struct withdraw_denied
    19. {};
    21. struct cancel_withdrawal
    22. {
    23.   std::string account;
    24.   unsigned amount;
    25.   cancel_withdrawal(std::string const& account_,
    26.                     unsigned amount_):
    27.     account(account_),amount(amount_)
    28.   {}
    29. };
    31. struct withdrawal_processed
    32. {
    33.   std::string account;
    34.   unsigned amount;
    35.   withdrawal_processed(std::string const& account_,
    36.                        unsigned amount_):
    37.     account(account_),amount(amount_)
    38.   {}
    39. };
    41. struct card_inserted
    42. {
    43.   std::string account;
    44.   explicit card_inserted(std::string const& account_):
    45.     account(account_)
    46.   {}
    47. };
    49. struct digit_pressed
    50. {
    51.   char digit;
    52.   explicit digit_pressed(char digit_):
    53.     digit(digit_)
    54.   {}
    55. };
    57. struct clear_last_pressed
    58. {};
    60. struct eject_card
    61. {};
    63. struct withdraw_pressed
    64. {
    65.   unsigned amount;
    66.   explicit withdraw_pressed(unsigned amount_):
    67.     amount(amount_)
    68.   {}
    69. };
    71. struct cancel_pressed
    72. {};
    74. struct issue_money
    75. {
    76.   unsigned amount;
    77.   issue_money(unsigned amount_):
    78.     amount(amount_)
    79.   {}
    80. };
    82. struct verify_pin
    83. {
    84.   std::string account;
    85.   std::string pin;
    86.   mutable messaging::sender atm_queue;
    88.   verify_pin(std::string const& account_,std::string const& pin_,
    89.              messaging::sender atm_queue_):
    90.     account(account_),pin(pin_),atm_queue(atm_queue_)
    91.   {}
    92. };
    94. struct pin_verified
    95. {};
    97. struct pin_incorrect
    98. {};
    100. struct display_enter_pin
    101. {};
    103. struct display_enter_card
    104. {};
    106. struct display_insufficient_funds
    107. {};
    108. struct display_withdrawal_cancelled
    109. {};
    111. struct display_pin_incorrect_message
    112. {};
    114. struct display_withdrawal_options
    115. {};
    117. struct get_balance
    118. {
    119.   std::string account;
    120.   mutable messaging::sender atm_queue;
    122.   get_balance(std::string const& account_,messaging::sender atm_queue_):
    123.     account(account_),atm_queue(atm_queue_)
    124.   {} 
    125. };
    127. struct balance
    128. {
    129.   unsigned amount;
    130.   explicit balance(unsigned amount_):
    131.     amount(amount_)
    132.   {}
    133. };
    135. struct display_balance
    136. {
    137.   unsigned amount;
    138.   explicit display_balance(unsigned amount_):
    139.     amount(amount_)
    140.   {}
    141. };
    143. struct balance_pressed
    144. {};

    清单C.7 ATM状态机

    清单C.8 银行状态机

    1. class bank_machine
    2. {
    3.   messaging::receiver incoming;
    4.   unsigned balance;
    5. public:
    6.   bank_machine():
    8.   balance(199)
    9.   {}
    11.   void done()
    12.   {
    13.     get_sender().send(messaging::close_queue());
    14.   }
    16.   void run()
    17.   {
    18.     try
    19.     {
    20.       for(;;)
    21.       {
    22.         incoming.wait()
    23.           .handle<verify_pin>(
    24.            [&](verify_pin const& msg)
    25.            {
    26.              if(msg.pin=="1937")
    27.              {
    28.                msg.atm_queue.send(pin_verified());
    29.              }
    30.              else
    31.              {
    32.                msg.atm_queue.send(pin_incorrect());
    33.              }
    34.            })
    35.           .handle<withdraw>(
    36.            [&](withdraw const& msg)
    37.            {
    38.              if(balance>=msg.amount)
    39.              {
    40.                msg.atm_queue.send(withdraw_ok());
    41.                balance-=msg.amount;
    42.              }
    43.              else
    44.              {
    45.                msg.atm_queue.send(withdraw_denied());
    46.              }
    47.            })
    48.           .handle<get_balance>(
    49.            [&](get_balance const& msg)
    50.            {
    51.              msg.atm_queue.send(::balance(balance));
    52.            })
    53.           .handle<withdrawal_processed>(
    54.            [&](withdrawal_processed const& msg)
    55.            {
    56.            })
    57.           .handle<cancel_withdrawal>(
    58.            [&](cancel_withdrawal const& msg)
    59.            {
    60.            });
    61.       }
    62.     }
    63.     catch(messaging::close_queue const&)
    64.     {
    65.     }
    66.   }
    68.   messaging::sender get_sender()
    69.   {
    70.   return incoming;
    71.   }

    清单C.9 用户状态机