Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include "xmpp/xmpp_channel_mux.h"
       7             : 
       8             : #include <boost/foreach.hpp>
       9             : 
      10             : #include "base/task_annotations.h"
      11             : #include "xmpp/xmpp_init.h"
      12             : #include "xmpp/xmpp_connection.h"
      13             : 
      14             : using namespace std;
      15             : using namespace xmsm;
      16             : 
      17       11607 : XmppChannelMux::XmppChannelMux(XmppConnection *connection)
      18       11607 :     : connection_(connection), rx_message_trace_cb_(NULL),
      19       23214 :       tx_message_trace_cb_(NULL) {
      20       11607 :         last_received_ = 0;
      21       11607 :         last_sent_ = 0;
      22       11607 : }
      23             : 
      24       11706 : XmppChannelMux::~XmppChannelMux() {
      25       11607 :     assert(map_.empty());
      26       11706 : }
      27             : 
      28         543 : void XmppChannelMux::Close() {
      29         543 :     connection_->Clear();
      30         543 : }
      31             : 
      32        9598 : bool XmppChannelMux::LastReceived(time_t duration) const {
      33        9598 :     return (UTCTimestamp() - last_received_) <= duration;
      34             : }
      35             : 
      36        3376 : bool XmppChannelMux::LastSent(time_t duration) const {
      37        3376 :     return (UTCTimestamp() - last_sent_) <= duration;
      38             : }
      39             : 
      40     3531477 : xmps::PeerState XmppChannelMux::GetPeerState() const {
      41     3531477 :     xmsm::XmState st = connection_->GetStateMcState();
      42     3531363 :     return (st == xmsm::ESTABLISHED) ? xmps::READY :
      43     3531363 :                                        xmps::NOT_READY;
      44             : }
      45             : 
      46           2 : void XmppChannelMux::WriteReady(const boost::system::error_code &ec) {
      47           2 :     std::scoped_lock lock(mutex_);
      48             : 
      49           2 :     WriteReadyCbMap::iterator iter = map_.begin();
      50           2 :     WriteReadyCbMap::iterator next = iter;
      51           4 :     for (; iter != map_.end(); iter = next) {
      52           2 :         ++next;
      53           2 :         SendReadyCb cb = iter->second;
      54           2 :         cb(ec);
      55           2 :         map_.erase(iter);
      56           2 :     }
      57           2 : }
      58             : 
      59     1623399 : bool XmppChannelMux::Send(const uint8_t *msg, size_t msgsize,
      60             :                           const string *msg_str, xmps::PeerId id,
      61             :                           SendReadyCb cb) {
      62     1623399 :     if (!connection_) return false;
      63             : 
      64     1623399 :     std::scoped_lock lock(mutex_);
      65     1623936 :     last_sent_ = UTCTimestamp();
      66     1624023 :     bool res = connection_->Send(msg, msgsize, msg_str);
      67     1624000 :     if (res == false) {
      68          87 :         RegisterWriteReady(id, cb);
      69             :     }
      70     1624000 :     return res;
      71     1624000 : }
      72             : 
      73       22355 : int XmppChannelMux::GetTaskInstance() const {
      74       22355 :     return connection_->GetTaskInstance();
      75             : }
      76             : 
      77        9995 : void XmppChannelMux::RegisterReferer(xmps::PeerId id) {
      78        9995 :     referers_.insert(id);
      79        9995 : }
      80             : 
      81        9995 : void XmppChannelMux::UnRegisterReferer(xmps::PeerId id) {
      82        9995 :     referers_.erase(id);
      83        9995 : }
      84             : 
      85       13946 : void XmppChannelMux::RegisterReceive(xmps::PeerId id, ReceiveCb cb) {
      86       13946 :     rxmap_.insert(make_pair(id, cb));
      87       13946 : }
      88             : 
      89       14066 : void XmppChannelMux::UnRegisterReceive(xmps::PeerId id) {
      90       14066 :     ReceiveCbMap::iterator it = rxmap_.find(id);
      91       14066 :     if (it != rxmap_.end()) {
      92       13942 :         rxmap_.erase(it);
      93             :     }
      94             : 
      95       14066 :     if (ReceiverCount())
      96        4077 :         return;
      97             : 
      98             :     XmppServerConnection *server_connection =
      99       10242 :         dynamic_cast<XmppServerConnection *>(connection_);
     100             : 
     101             :     // If GracefulRestart helper mode close process is complete, restart the
     102             :     // state machine to form new session with the client.
     103       10512 :     if (!connection_->IsDeleted() && server_connection &&
     104         270 :             server_connection->server()->IsGRHelperModeEnabled()) {
     105         253 :         server_connection->state_machine()->Initialize();
     106         253 :         return;
     107             :     }
     108             : 
     109        9989 :     connection_->RetryDelete();
     110             : }
     111             : 
     112       11668 : size_t XmppChannelMux::RefererCount() const {
     113       11668 :     return referers_.size();
     114             : }
     115             : 
     116       31899 : size_t XmppChannelMux::ReceiverCount() const {
     117       31899 :     return rxmap_.size();
     118             : }
     119             : 
     120          72 : vector<string> XmppChannelMux::GetReceiverList() const {
     121          72 :     vector<string> receivers;
     122          96 :     for (const auto& value : rxmap_) {
     123          24 :         receivers.push_back(xmps::PeerIdToName(value.first));
     124             :     }
     125          72 :     return receivers;
     126           0 : }
     127             : 
     128             : //
     129             : // To be called after acquiring mutex
     130             : //
     131          88 : void XmppChannelMux::RegisterWriteReady(xmps::PeerId id, SendReadyCb cb) {
     132          88 :     map_.insert(make_pair(id, cb));
     133          88 : }
     134             : 
     135             : //
     136             : // To be called after acquiring mutex
     137             : //
     138       13842 : void XmppChannelMux::UnRegisterWriteReady(xmps::PeerId id) {
     139       13842 :     map_.erase(id);
     140       13842 : }
     141             : 
     142     1679415 : const std::string &XmppChannelMux::ToString() const {
     143     1679415 :     return connection_->ToString();
     144             : }
     145             : 
     146           0 : const std::string &XmppChannelMux::FromString() const {
     147           0 :     return connection_->FromString();
     148             : }
     149             : 
     150         974 : std::string XmppChannelMux::StateName() const {
     151         974 :     return connection_->StateName();
     152             : }
     153             : 
     154           0 : std::string XmppChannelMux::AuthType() const {
     155           0 :     return connection_->GetXmppAuthenticationType();
     156             : }
     157             : 
     158           0 : std::string XmppChannelMux::PeerAddress() const {
     159           0 :     return connection_->endpoint_string();
     160             : }
     161             : 
     162     3170014 : inline bool MatchCallback(string to, xmps::PeerId peer) {
     163     3170014 :     if ((to.find(XmppInit::kBgpPeer) != string::npos) &&
     164             :         (peer == xmps::BGP)) {
     165     1613435 :         return true;
     166             :     }
     167     1556581 :     if ((to.find(XmppInit::kConfigPeer) != string::npos) &&
     168             :         (peer == xmps::CONFIG)) {
     169           0 :         return true;
     170             :     }
     171     1556582 :     if ((to.find(XmppInit::kDnsPeer) != string::npos) &&
     172             :         (peer == xmps::DNS)) {
     173           0 :         return true;
     174             :     }
     175     1556582 :     if ((to.find(XmppInit::kOtherPeer) != string::npos) &&
     176             :         (peer == xmps::OTHER)) {
     177           4 :         return true;
     178             :     }
     179     1556578 :     return false;
     180             : }
     181             : 
     182     1613648 : void XmppChannelMux::ProcessXmppMessage(const XmppStanza::XmppMessage *msg) {
     183     1613648 :     last_received_ = UTCTimestamp();
     184     1613650 :     ReceiveCbMap::iterator iter = rxmap_.begin();
     185     4783662 :     for (; iter != rxmap_.end(); ++iter) {
     186     3170013 :         if (MatchCallback(msg->to, iter->first)) {
     187     1613438 :             ReceiveCb cb = iter->second;
     188     1613438 :             cb(msg, GetPeerState());
     189     1613439 :         }
     190             :     }
     191     1613649 : }
     192             : 
     193       25244 : void XmppChannelMux::HandleStateEvent(xmsm::XmState state) {
     194       25244 :     CHECK_CONCURRENCY("xmpp::StateMachine");
     195       25240 :     xmps::PeerState st = xmps::NOT_READY;
     196       25240 :     if (state == xmsm::ESTABLISHED) {
     197       12605 :         st = xmps::READY;
     198       12635 :     } else if (state == xmsm::ACTIVE) {
     199         285 :         st = xmps::TIMEDOUT;
     200             :     }
     201             : 
     202       25240 :     if (connection_->IsClient()) {
     203       12347 :         XmppClient *client = static_cast<XmppClient *>(connection_->server());
     204       12347 :         client->NotifyConnectionEvent(this, st);
     205             :     } else {
     206             :         // Event to create the peer on server
     207       12893 :         XmppServer *server = static_cast<XmppServer *>(connection_->server());
     208       12898 :         server->NotifyConnectionEvent(this, st);
     209             :     }
     210       25248 : }
     211             : 
     212         493 : std::string XmppChannelMux::LastStateName() const {
     213         493 :     return connection_->LastStateName();
     214             : }
     215         487 : std::string XmppChannelMux::LastStateChangeAt() const {
     216         487 :     return connection_->LastStateChangeAt();
     217             : }
     218         487 : std::string XmppChannelMux::LastEvent() const {
     219         487 :     return connection_->LastEvent();
     220             : }
     221         561 : uint32_t XmppChannelMux::rx_open() const {
     222         561 :     return connection_->rx_open();
     223             : }
     224         561 : uint32_t XmppChannelMux::rx_close() const {
     225         561 :     return connection_->rx_close();
     226             : }
     227         561 : uint32_t XmppChannelMux::rx_update() const {
     228         561 :     return connection_->rx_update();
     229             : }
     230         561 : uint32_t XmppChannelMux::rx_keepalive() const {
     231         561 :     return connection_->rx_keepalive();
     232             : }
     233         561 : uint32_t XmppChannelMux::tx_open() const {
     234         561 :     return connection_->tx_open();
     235             : }
     236         561 : uint32_t XmppChannelMux::tx_close() const {
     237         561 :     return connection_->tx_close();
     238             : }
     239         561 : uint32_t XmppChannelMux::tx_update() const {
     240         561 :     return connection_->tx_update();
     241             : }
     242         561 : uint32_t XmppChannelMux::tx_keepalive() const {
     243         561 :     return connection_->tx_keepalive();
     244             : }
     245        1461 : uint32_t XmppChannelMux::FlapCount() const {
     246        1461 :     return connection_->flap_count();
     247             : }
     248        1461 : std::string XmppChannelMux::LastFlap() const {
     249        1461 :     return connection_->last_flap_at();
     250             : }
     251             : 
     252           2 : void XmppChannelMux::RegisterRxMessageTraceCallback(RxMessageTraceCb cb) {
     253           2 :     rx_message_trace_cb_ = cb;
     254           2 : }
     255           2 : void XmppChannelMux::RegisterTxMessageTraceCallback(TxMessageTraceCb cb) {
     256           2 :     tx_message_trace_cb_ = cb;
     257           2 : }
     258             : 
     259     1633821 : bool XmppChannelMux::RxMessageTrace(const std::string &to_address,
     260             :                                     int port,
     261             :                                     int msg_size,
     262             :                                     const std::string &msg,
     263             :                                     const XmppStanza::XmppMessage *xmpp_msg) {
     264     1633821 :     if (rx_message_trace_cb_) {
     265           0 :         return rx_message_trace_cb_(to_address, port, msg_size, msg, xmpp_msg);
     266             :     }
     267     1633822 :     return false;
     268             : }
     269             : 
     270     1623355 : bool XmppChannelMux::TxMessageTrace(const std::string &to_address,
     271             :                                     int port,
     272             :                                     int msg_size,
     273             :                                     const std::string &msg,
     274             :                                     const XmppStanza::XmppMessage *xmpp_msg) {
     275     1623355 :     if (tx_message_trace_cb_) {
     276           0 :         return tx_message_trace_cb_(to_address, port, msg_size, msg, xmpp_msg);
     277             :     }
     278     1623372 :     return false;
     279             : }