add fail-resend support.

lichao

2021-04-12 1b52f1cb8c47dd2c0195d2fd65d7b6a4c2f10704

 src/socket.h

@@ -77,7 +77,7 @@
   template <class Body>
   bool Send(const void *valid_remote, const BHMsgHead &head, const Body &body, const int timeout_ms, const RecvCB &cb)
   {
      auto DoSend = [&](MsgI &msg) { return mq().Send(*static_cast<const MQId *>(valid_remote), msg, timeout_ms, [&]() { async_cbs_->Add(head.msg_id(), cb); }); };
      auto DoSend = [&](MsgI &msg) { return mq().Send(*static_cast<const MQId *>(valid_remote), msg, timeout_ms, [&]() { per_msg_cbs_->Add(head.msg_id(), cb); }); };
      MsgI msg;
      return msg.Make(shm(), head, body) && SendImpl(msg, timeout_ms, DoSend);
   }
@@ -109,12 +109,15 @@
               reply.swap(msg);
               reply_head.Swap(&head);
               st->cv.notify_one();
            } else {
            } else { // ignore
            }
         };

         std::unique_lock<std::mutex> lk(st->mutex);
         bool sendok = Send(remote, head, body, timeout_ms, OnRecv);
         if (!sendok) {
            printf("send timeout\n");
         }
         if (sendok && st->cv.wait_until(lk, endtime) == std::cv_status::no_timeout) {
            return true;
         } else {
@@ -161,7 +164,7 @@
      }
   };

   Synced<AsyncCBs> async_cbs_;
   Synced<AsyncCBs> per_msg_cbs_;
};

#endif // end of include guard: SOCKET_GWTJHBPO