87 auto listener_adapter = [listener, self_attr](
88 const std::shared_ptr<std::string>& msg_str,
90 auto msg = std::make_shared<MessageT>();
93 uint64_t send_time = msg_info.send_time();
94 if (send_time > recv_time) {
95 AWARN <<
"The message is received (" << recv_time
96 <<
") earlier than the message is sent (" << send_time <<
")";
98 uint64_t diff = recv_time - send_time;
100 statistics::Statistics::Instance()->SamplingTranLatency<uint64_t>(
103 statistics::Statistics::Instance()->SetProcStatus(self_attr, recv_time);
104 listener(msg, msg_info);
107 Dispatcher::AddListener<std::string>(self_attr, listener_adapter);
108 AddSubscriber(self_attr);
115 auto listener_adapter = [listener, self_attr](
116 const std::shared_ptr<std::string>& msg_str,
118 auto msg = std::make_shared<MessageT>();
121 uint64_t send_time = msg_info.send_time();
122 if (send_time > recv_time) {
123 AWARN <<
"The message is received (" << recv_time
124 <<
") earlier than the message is sent (" << send_time <<
")";
126 uint64_t diff = recv_time - send_time;
128 statistics::Statistics::Instance()->SamplingTranLatency<uint64_t>(
131 statistics::Statistics::Instance()->SetProcStatus(self_attr, recv_time);
132 listener(msg, msg_info);
135 Dispatcher::AddListener<std::string>(self_attr, opposite_attr,
137 AddSubscriber(self_attr);