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_connection.h"
       7             : 
       8             : #include <boost/date_time/posix_time/posix_time.hpp>
       9             : #include <sstream>
      10             : 
      11             : #include "base/lifetime.h"
      12             : #include "base/task_annotations.h"
      13             : #include "io/event_manager.h"
      14             : #include "xml/xml_base.h"
      15             : #include "xmpp/xmpp_client.h"
      16             : #include "xmpp/xmpp_config.h"
      17             : #include "xmpp/xmpp_factory.h"
      18             : #include "xmpp/xmpp_log.h"
      19             : #include "xmpp/xmpp_server.h"
      20             : #include "xmpp/xmpp_session.h"
      21             : 
      22             : #include "sandesh/common/vns_types.h"
      23             : #include "sandesh/common/vns_constants.h"
      24             : #include "sandesh/xmpp_client_server_sandesh_types.h"
      25             : #include "sandesh/xmpp_message_sandesh_types.h"
      26             : #include "sandesh/xmpp_server_types.h"
      27             : #include "sandesh/xmpp_state_machine_sandesh_types.h"
      28             : #include "sandesh/xmpp_trace_sandesh_types.h"
      29             : #include "sandesh/xmpp_peer_info_types.h"
      30             : 
      31             : using namespace std;
      32             : using boost::system::error_code;
      33             : 
      34             : const char *XmppConnection::kAuthTypeNil = "NIL";
      35             : const char *XmppConnection::kAuthTypeTls = "TLS";
      36             : 
      37             : // Maximum XMPP control message size. Typically this is less then 256 bytes. But
      38             : // in scenarios where host names are quite long, we need a larger buffer size.
      39             : #define XMPP_CONTROL_MESSAGE_MAX_SIZE 1024
      40             : 
      41       11606 : XmppConnection::XmppConnection(TcpServer *server,
      42       11606 :                                const XmppChannelConfig *config)
      43       11606 :     : server_(server),
      44       11606 :       session_(NULL),
      45       11606 :       endpoint_(config->endpoint),
      46       11606 :       local_endpoint_(config->local_endpoint),
      47       11606 :       config_(NULL),
      48       23212 :       keepalive_timer_(TimerManager::CreateTimer(
      49       11606 :                   *server->event_manager()->io_service(),
      50             :                   "Xmpp keepalive timer",
      51             :                   TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
      52       11606 :                   GetTaskInstance(config->ClientOnly()))),
      53       11606 :       is_client_(config->ClientOnly()),
      54       11606 :       log_uve_(config->logUVE),
      55       11606 :       admin_down_(false),
      56       11606 :       disable_read_(false),
      57       11606 :       from_(config->FromAddr),
      58       11606 :       to_(config->ToAddr),
      59       11606 :       auth_enabled_(config->auth_enabled),
      60       11606 :       dscp_value_(config->dscp_value), xmlns_(config->xmlns),
      61       11606 :       state_machine_(XmppStaticObjectFactory::Create<XmppStateMachine>(
      62       11606 :           this, config->ClientOnly(), config->auth_enabled, config->xmpp_hold_time)),
      63       69636 :       mux_(XmppStaticObjectFactory::Create<XmppChannelMux>(this)) {
      64       11606 :     ostringstream oss;
      65       11606 :     oss << FromString() << ":" << endpoint().address().to_string();
      66       11606 :     uve_key_str_ = oss.str();
      67       11606 : }
      68             : 
      69       11606 : XmppConnection::~XmppConnection() {
      70       11606 :     StopKeepAliveTimer();
      71       11606 :     TimerManager::DeleteTimer(keepalive_timer_);
      72       11606 :     XMPP_UTDEBUG(XmppConnectionDelete, ToUVEKey(), XMPP_PEER_DIR_NA,
      73             :                  "XmppConnection destructor", FromString(), ToString());
      74       11606 : }
      75             : 
      76        1046 : std::string XmppConnection::GetXmppAuthenticationType() const {
      77        1046 :     if (auth_enabled_) {
      78          36 :         return (XmppConnection::kAuthTypeTls);
      79             :     } else {
      80        1010 :         return (XmppConnection::kAuthTypeNil);
      81             :     }
      82             : }
      83             : 
      84           0 : void XmppConnection::SetConfig(const XmppChannelConfig *config) {
      85           0 :     config_ = config;
      86           0 : }
      87             : 
      88       19316 : void XmppConnection::set_session(XmppSession *session) {
      89       19316 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
      90       19318 :     assert(session);
      91       19318 :     session_ = session;
      92       19318 :     if (session_ && dscp_value_) {
      93           8 :         session_->SetDscpSocketOption(dscp_value_);
      94             :     }
      95       19318 : }
      96             : 
      97       20828 : void XmppConnection::clear_session() {
      98       20828 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
      99       20829 :     if (!session_)
     100        1618 :         return;
     101       19211 :     session_->ClearConnection();
     102       19210 :     session_ = NULL;
     103       20828 : }
     104             : 
     105          40 : const XmppSession *XmppConnection::session() const {
     106          40 :     return session_;
     107             : }
     108             : 
     109       17463 : XmppSession *XmppConnection::session() {
     110       17463 :     return session_;
     111             : }
     112             : 
     113           1 : void XmppConnection::WriteReady() {
     114           1 :     boost::system::error_code ec;
     115           1 :     mux_->WriteReady(ec);
     116           1 : }
     117             : 
     118         128 : void XmppConnection::Shutdown() {
     119         128 :     ManagedDelete();
     120         128 : }
     121             : 
     122      118165 : bool XmppConnection::IsDeleted() const {
     123      118165 :     return deleter()->IsDeleted();
     124             : }
     125             : 
     126       17659 : bool XmppConnection::MayDelete() const {
     127       17659 :     return !mux_->ReceiverCount() && !mux_->RefererCount();
     128             : }
     129             : 
     130        7746 : XmppSession *XmppConnection::CreateSession() {
     131        7746 :     TcpSession *session = server_->CreateSession();
     132        7746 :     XmppSession *xmpp_session = static_cast<XmppSession *>(session);
     133        7746 :     xmpp_session->SetConnection(this);
     134        7746 :     return xmpp_session;
     135             : }
     136             : 
     137             : //
     138             : // Return the task instance for this XmppConnection.
     139             : // Calculate from the remote IpAddress so that a restarting session uses the
     140             : // same value as before.
     141             : // Do not make this method virtual since it gets called from the constructor.
     142             : //
     143             : 
     144       95750 : int XmppConnection::GetTaskInstance(bool is_client) const {
     145       95750 :     if (is_client)
     146       36757 :         return 0;
     147       58993 :     IpAddress address = endpoint().address();
     148       58988 :     int thread_count = TaskScheduler::GetInstance()->HardwareThreadCount();
     149       58988 :     if (address.is_v4()) {
     150       58983 :         return address.to_v4().to_ulong() % thread_count;
     151             :     } else {
     152           0 :         return 0;
     153             :     }
     154             : }
     155             : 
     156     7877882 : xmsm::XmState XmppConnection::GetStateMcState() const {
     157     7877882 :     const XmppStateMachine *sm = state_machine();
     158     7877848 :     assert(sm);
     159     7877848 :     return sm->StateType();
     160             : }
     161             : 
     162     4346599 : xmsm::XmOpenConfirmState XmppConnection::GetStateMcOpenConfirmState() const {
     163     4346599 :     const XmppStateMachine *sm = state_machine();
     164     4346597 :     assert(sm);
     165     4346597 :     return sm->OpenConfirmStateType();
     166             : }
     167             : 
     168             : 
     169       58993 : boost::asio::ip::tcp::endpoint XmppConnection::endpoint() const {
     170       58993 :     return endpoint_;
     171             : }
     172             : 
     173        7746 : boost::asio::ip::tcp::endpoint XmppConnection::local_endpoint() const {
     174        7746 :     return local_endpoint_;
     175             : }
     176             : 
     177          72 : string XmppConnection::endpoint_string() const {
     178          72 :     ostringstream oss;
     179          72 :     oss << endpoint_;
     180         144 :     return oss.str();
     181          72 : }
     182             : 
     183          72 : string XmppConnection::local_endpoint_string() const {
     184          72 :     ostringstream oss;
     185          72 :     oss << local_endpoint_;
     186         144 :     return oss.str();
     187          72 : }
     188             : 
     189       34244 : const string &XmppConnection::FromString() const {
     190       34244 :     return from_;
     191             : }
     192             : 
     193     1829770 : const string &XmppConnection::ToString() const {
     194     1829770 :     return to_;
     195             : }
     196             : 
     197     6215071 : const std::string &XmppConnection::ToUVEKey() const {
     198     6215071 :     return uve_key_str_;
     199             : }
     200             : 
     201           0 : static void XMPPPeerInfoSend(XmppPeerInfoData &peer_info) {
     202           0 :     assert(!peer_info.get_name().empty());
     203           0 :     XMPPPeerInfo::Send(peer_info);
     204           0 : }
     205             : 
     206       15771 : void XmppConnection::SetTo(const string &to) {
     207       15771 :     if ((to_.size() == 0) && (to.size() != 0)) {
     208        6562 :         to_ = to;
     209        6562 :         if (!logUVE()) return;
     210           0 :         XmppPeerInfoData peer_info;
     211           0 :         peer_info.set_name(ToUVEKey());
     212           0 :         peer_info.set_identifier(to_);
     213           0 :         XMPPPeerInfoSend(peer_info);
     214           0 :     }
     215             : }
     216             : 
     217           0 : void XmppConnection::SetAdminDown(bool toggle) {
     218             :     // TODO: generate state machine event.
     219           0 :     admin_down_ = toggle;
     220           0 : }
     221             : 
     222        6521 : bool XmppConnection::AcceptSession(XmppSession *session) {
     223        6521 :     session->SetConnection(this);
     224        6521 :     return state_machine_->PassiveOpen(session);
     225             : }
     226             : 
     227     1624148 : bool XmppConnection::Send(const uint8_t *data, size_t size,
     228             :     const string *msg_str) {
     229     1624148 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     230     1623957 :     if (session_ == NULL) {
     231          11 :         return false;
     232             :     }
     233             : 
     234     1623946 :     TcpSession::Endpoint endpoint = session_->remote_endpoint();
     235     1623729 :     const string &endpoint_addr_str = session_->remote_addr_string();
     236     1623675 :     string str;
     237     1623676 :     if (!msg_str) {
     238       57159 :         str.append(reinterpret_cast<const char *>(data), size);
     239       57159 :         msg_str = &str;
     240             :     }
     241             : 
     242     1623361 :     if (!(mux_ &&
     243     1623627 :          (mux_->TxMessageTrace(endpoint_addr_str, endpoint.port(),
     244     1623676 :                                size, *msg_str, NULL)))) {
     245     1623364 :         XMPP_MESSAGE_TRACE(XmppTxStream,
     246             :                            endpoint_addr_str, endpoint.port(), size, *msg_str);
     247             :     }
     248             : 
     249     1624257 :     stats_[1].update++;
     250             :     size_t sent;
     251     1624260 :     return session_->Send(data, size, &sent);
     252     1624173 : }
     253             : 
     254          11 : int XmppConnection::SetDscpValue(uint8_t value) {
     255          11 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     256          11 :     dscp_value_ = value;
     257          11 :     if (!session_) {
     258           0 :         return 0;
     259             :     }
     260          11 :     return session_->SetDscpSocketOption(value);
     261          11 : }
     262             : 
     263        8045 : bool XmppConnection::SendOpen(XmppSession *session) {
     264        8045 :     if (!session) return false;
     265        8045 :     XmppProto::XmppStanza::XmppStreamMessage openstream;
     266        8045 :     openstream.strmtype = XmppStanza::XmppStreamMessage::INIT_STREAM_HEADER;
     267             :     uint8_t data[XMPP_CONTROL_MESSAGE_MAX_SIZE];
     268        8045 :     int len = XmppProto::EncodeStream(openstream, to_, from_, xmlns_, data,
     269             :                                       sizeof(data));
     270        8045 :     if (len <= 0) {
     271           0 :         inc_open_fail();
     272           0 :         return false;
     273             :     } else {
     274        8045 :         XMPP_UTDEBUG(XmppOpen, ToUVEKey(), XMPP_PEER_DIR_OUT, len, from_, to_,
     275             :                      xmlns_);
     276        8045 :         session->Send(data, len, NULL);
     277        8045 :         stats_[1].open++;
     278        8045 :         return true;
     279             :     }
     280        8045 : }
     281             : 
     282        7817 : bool XmppConnection::SendOpenConfirm(XmppSession *session) {
     283        7817 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     284        7817 :     if (!session_) return false;
     285        7817 :     XmppStanza::XmppStreamMessage openstream;
     286        7817 :     openstream.strmtype = XmppStanza::XmppStreamMessage::INIT_STREAM_HEADER_RESP;
     287             :     uint8_t data[XMPP_CONTROL_MESSAGE_MAX_SIZE];
     288        7817 :     int len = XmppProto::EncodeStream(openstream, to_, from_, xmlns_, data,
     289             :                                       sizeof(data));
     290        7817 :     if (len <= 0) {
     291           0 :         inc_open_fail();
     292           0 :         return false;
     293             :     } else {
     294        7817 :         XMPP_UTDEBUG(XmppOpenConfirm, ToUVEKey(), XMPP_PEER_DIR_OUT, len,
     295             :                      from_, to_);
     296        7817 :         session_->Send(data, len, NULL);
     297        7817 :         stats_[1].open++;
     298        7817 :         return true;
     299             :     }
     300        7817 : }
     301             : 
     302        1512 : bool XmppConnection::SendStreamFeatureRequest(XmppSession *session) {
     303        1512 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     304        1512 :     if (!session_) return false;
     305        1512 :     XmppStanza::XmppStreamMessage featurestream;
     306        1512 :     featurestream.strmtype = XmppStanza::XmppStreamMessage::FEATURE_TLS;
     307        1512 :     featurestream.strmtlstype = XmppStanza::XmppStreamMessage::TLS_FEATURE_REQUEST;
     308             :     uint8_t data[XMPP_CONTROL_MESSAGE_MAX_SIZE];
     309        1512 :     int len = XmppProto::EncodeStream(featurestream, to_, from_, xmlns_, data,
     310             :                                       sizeof(data));
     311        1512 :     if (len <= 0) {
     312           0 :         inc_stream_feature_fail();
     313           0 :         return false;
     314             :     } else {
     315        1512 :         XMPP_UTDEBUG(XmppControlMessage, ToUVEKey(), XMPP_PEER_DIR_OUT,
     316             :                      "Send Stream Feature Request", len, from_, to_);
     317        1512 :         session_->Send(data, len, NULL);
     318             :         //stats_[1].open++;
     319        1512 :         return true;
     320             :     }
     321        1512 : }
     322             : 
     323        1469 : bool XmppConnection::SendStartTls(XmppSession *session) {
     324        1469 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     325        1469 :     if (!session_) return false;
     326        1469 :     XmppStanza::XmppStreamMessage stream;
     327        1469 :     stream.strmtype = XmppStanza::XmppStreamMessage::FEATURE_TLS;
     328        1469 :     stream.strmtlstype = XmppStanza::XmppStreamMessage::TLS_START;
     329             :     uint8_t data[XMPP_CONTROL_MESSAGE_MAX_SIZE];
     330        1469 :     int len = XmppProto::EncodeStream(stream, to_, from_, xmlns_, data,
     331             :                                       sizeof(data));
     332        1469 :     if (len <= 0) {
     333           0 :         inc_stream_feature_fail();
     334           0 :         return false;
     335             :     } else {
     336        1469 :         XMPP_UTDEBUG(XmppControlMessage, ToUVEKey(), XMPP_PEER_DIR_OUT,
     337             :                      "Send Start Tls", len, from_, to_);
     338        1469 :         session_->Send(data, len, NULL);
     339             :         //stats_[1].open++;
     340        1469 :         return true;
     341             :     }
     342        1469 : }
     343             : 
     344        1464 : bool XmppConnection::SendProceedTls(XmppSession *session) {
     345        1464 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     346        1464 :     if (!session_) return false;
     347        1464 :     XmppStanza::XmppStreamMessage stream;
     348        1464 :     stream.strmtype = XmppStanza::XmppStreamMessage::FEATURE_TLS;
     349        1464 :     stream.strmtlstype = XmppStanza::XmppStreamMessage::TLS_PROCEED;
     350             :     uint8_t data[XMPP_CONTROL_MESSAGE_MAX_SIZE];
     351        1464 :     int len = XmppProto::EncodeStream(stream, to_, from_, xmlns_, data,
     352             :                                       sizeof(data));
     353        1464 :     if (len <= 0) {
     354           0 :         inc_stream_feature_fail();
     355           0 :         return false;
     356             :     } else {
     357        1464 :         XMPP_UTDEBUG(XmppControlMessage, ToUVEKey(), XMPP_PEER_DIR_OUT,
     358             :                      "Send Proceed Tls", len, from_, to_);
     359        1464 :         session_->Send(data, len, NULL);
     360             :         //stats_[1].open++;
     361        1464 :         return true;
     362             :     }
     363        1464 : }
     364             : 
     365           2 : void XmppConnection::SendClose(XmppSession *session) {
     366           2 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     367           2 :     if (!session_) return;
     368           1 :     string str("</stream:stream>");
     369             :     uint8_t data[64];
     370           1 :     memcpy(data, str.data(), str.size());
     371           1 :     XMPP_UTDEBUG(XmppClose, ToUVEKey(), XMPP_PEER_DIR_OUT, str.size(), from_,
     372             :                  to_);
     373           1 :     session_->Send(data, str.size(), NULL);
     374           1 :     stats_[1].close++;
     375           2 : }
     376             : 
     377        2910 : void XmppConnection::ProcessSslHandShakeResponse(SslSessionPtr session,
     378             :     const boost::system::error_code& error) {
     379        2910 :     if (!state_machine())
     380           0 :         return;
     381             : 
     382        2910 :     if (error) {
     383           0 :         inc_handshake_failure();
     384             : 
     385           0 :         if (error.category() == boost::asio::error::get_ssl_category()) {
     386           0 :             string err = error.message();
     387           0 :             err = string(" (")
     388           0 :                          +boost::lexical_cast<string>(ERR_GET_LIB(error.value()))+","
     389           0 :                          +boost::lexical_cast<string>(ERR_GET_REASON(error.value()))+") ";
     390             : 
     391             :              char buf[128];
     392           0 :              ::ERR_error_string_n(error.value(), buf, sizeof(buf));
     393           0 :              err += buf;
     394           0 :              XMPP_ALERT(XmppSslHandShakeFailure, ToUVEKey(), XMPP_PEER_DIR_IN,
     395             :                         "failure", err);
     396           0 :         }
     397             : 
     398           0 :         state_machine()->OnEvent(session.get(), xmsm::EvTLSHANDSHAKE_FAILURE);
     399             : 
     400             :     } else {
     401        2910 :         XMPP_DEBUG(XmppSslHandShakeMessage, session->ToUVEKey(),
     402             :                    XMPP_PEER_DIR_IN, "success", "");
     403        2910 :         state_machine()->OnEvent(session.get(), xmsm::EvTLSHANDSHAKE_SUCCESS);
     404             :     }
     405             : }
     406             : 
     407           0 : void XmppConnection::LogMsg(std::string msg) {
     408           0 :     log4cplus::Logger logger = log4cplus::Logger::getRoot();
     409           0 :     LOG4CPLUS_DEBUG(logger, msg << ToString() << " " <<
     410             :          local_endpoint_.address() << ":" << local_endpoint_.port() << "::" <<
     411             :          endpoint_.address() << ":" << endpoint_.port());
     412           0 : }
     413             : 
     414        6444 : void XmppConnection::LogKeepAliveSend() {
     415             :     static bool init_ = false;
     416             :     static bool log_ = false;
     417             : 
     418        6444 :     if (!init_) {
     419          61 :         char *str = getenv("XMPP_ASSERT_ON_HOLD_TIMEOUT");
     420          61 :         if (str && strtoul(str, NULL, 0) != 0) log_ = true;
     421          61 :         init_ = true;
     422             :     }
     423             : 
     424        6444 :     if (log_) LogMsg("SEND KEEPALIVE: ");
     425        6444 : }
     426             : 
     427        6444 : void XmppConnection::SendKeepAlive() {
     428        6444 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     429        6444 :     if (!session_) return;
     430        6444 :     XmppStanza::XmppMessage msg(XmppStanza::WHITESPACE_MESSAGE_STANZA);
     431             :     uint8_t data[XMPP_CONTROL_MESSAGE_MAX_SIZE];
     432        6444 :     int len = XmppProto::EncodeStream(msg, data, sizeof(data));
     433        6444 :     assert(len > 0);
     434        6444 :     session_->Send(data, len, NULL);
     435        6444 :     stats_[1].keepalive++;
     436        6444 :     LogKeepAliveSend();
     437        6444 : }
     438             : 
     439         144 : bool XmppConnection::KeepAliveTimerExpired() {
     440         144 :     if (state_machine_->get_state() != xmsm::ESTABLISHED)
     441           0 :         return false;
     442             : 
     443             :     // TODO: check timestamp of last received packet.
     444         144 :     SendKeepAlive();
     445             : 
     446             :     //
     447             :     // Start the timer again, by returning true
     448             :     //
     449         144 :     return true;
     450             : }
     451             : 
     452           0 : void XmppConnection::KeepaliveTimerErrorHanlder(string error_name,
     453             :                                                 string error_message) {
     454           0 :     XMPP_WARNING(XmppKeepaliveTimeError, ToUVEKey(), XMPP_PEER_DIR_NA,
     455             :                  error_name, error_message);
     456           0 : }
     457             : 
     458       12628 : void XmppConnection::StartKeepAliveTimer() {
     459       12628 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     460       12629 :     if (!session_)
     461           0 :         return;
     462             : 
     463       12629 :     int holdtime_msecs = state_machine_->hold_time_msecs();
     464       12629 :     if (holdtime_msecs <= 0)
     465           0 :         return;
     466             : 
     467       12629 :     keepalive_timer_->Start(holdtime_msecs / 3,
     468             :         boost::bind(&XmppConnection::KeepAliveTimerExpired, this),
     469             :         boost::bind(&XmppConnection::KeepaliveTimerErrorHanlder, this, _1, _2));
     470       12629 : }
     471             : 
     472       24059 : void XmppConnection::StopKeepAliveTimer() {
     473       24059 :     tbb::spin_mutex::scoped_lock lock(spin_mutex_);
     474       24059 :     keepalive_timer_->Cancel();
     475       24060 : }
     476             : 
     477          24 : void XmppConnection::UpdateKeepAliveTimer(uint8_t time_out) {
     478          24 :     state_machine_->set_hold_time(time_out);
     479          24 :     StopKeepAliveTimer();
     480          24 :     StartKeepAliveTimer();
     481          24 : }
     482             : 
     483       75345 : XmppStateMachine *XmppConnection::state_machine() {
     484       75345 :     return state_machine_.get();
     485             : }
     486             : 
     487    12224373 : const XmppStateMachine *XmppConnection::state_machine() const {
     488    12224373 :     return state_machine_.get();
     489             : }
     490             : 
     491          72 : const XmppChannelMux *XmppConnection::channel_mux() const {
     492          72 :     return mux_.get();
     493             : }
     494             : 
     495     4152413 : void XmppConnection::IncProtoStats(unsigned int type) {
     496     4152413 :     switch (type) {
     497       20076 :         case XmppStanza::STREAM_HEADER:
     498       20076 :             stats_[0].open++;
     499       20078 :             break;
     500     2476952 :         case XmppStanza::WHITESPACE_MESSAGE_STANZA:
     501     2476952 :             stats_[0].keepalive++;
     502     2476955 :             break;
     503       98567 :         case XmppStanza::IQ_STANZA:
     504       98567 :             stats_[0].update++;
     505       98573 :             break;
     506     1556845 :         case XmppStanza::MESSAGE_STANZA:
     507     1556845 :             stats_[0].update++;
     508     1556845 :             break;
     509             :     }
     510     4152424 : }
     511             : 
     512        1159 : void XmppConnection::inc_connect_error() {
     513        1159 :     error_stats_.connect_error++;
     514        1159 : }
     515             : 
     516        6506 : void XmppConnection::inc_session_close() {
     517        6506 :     error_stats_.session_close++;
     518        6506 : }
     519             : 
     520           0 : void XmppConnection::inc_open_fail() {
     521           0 :     error_stats_.open_fail++;
     522           0 : }
     523             : 
     524           0 : void XmppConnection::inc_stream_feature_fail() {
     525           0 :     error_stats_.stream_feature_fail++;
     526           0 : }
     527             : 
     528           0 : void XmppConnection::inc_handshake_failure() {
     529           0 :     error_stats_.handshake_fail++;
     530           0 : }
     531             : 
     532         794 : size_t XmppConnection::get_connect_error() {
     533         794 :     return error_stats_.connect_error;
     534             : }
     535             : 
     536         576 : size_t XmppConnection::get_session_close() {
     537         576 :     return error_stats_.session_close;
     538             : }
     539             : 
     540           0 : size_t XmppConnection::get_open_fail() {
     541           0 :     return error_stats_.open_fail;
     542             : }
     543             : 
     544           0 : size_t XmppConnection::get_stream_feature_fail() {
     545           0 :     return error_stats_.stream_feature_fail;
     546             : }
     547             : 
     548           0 : size_t XmppConnection::get_handshake_failure() {
     549           0 :     return error_stats_.handshake_fail;
     550             : }
     551             : 
     552          72 : size_t XmppConnection::get_sm_connect_attempts() {
     553          72 :     return state_machine_->get_connect_attempts();
     554             : }
     555             : 
     556         267 : size_t XmppConnection::get_sm_keepalive_count() {
     557         267 :     return state_machine_->get_keepalive_count();
     558             : }
     559             : 
     560     4152422 : void XmppConnection::ReceiveMsg(XmppSession *session, const string &msg) {
     561     4152422 :     XmppStanza::XmppMessage *minfo = XmppDecode(msg);
     562             : 
     563     4152393 :     if (minfo) {
     564     4110758 :         session->IncStats((unsigned int)minfo->type, msg.size());
     565     4110758 :         if (minfo->type != XmppStanza::WHITESPACE_MESSAGE_STANZA) {
     566     1633822 :             if (!(mux_ &&
     567     3267637 :                   (mux_->RxMessageTrace(session->
     568     3267634 :                                         remote_endpoint().address().to_string(),
     569     3267637 :                                         session->remote_endpoint().port(),
     570     3267630 :                                         msg.size(), msg, minfo)))) {
     571     1633821 :                 XMPP_MESSAGE_TRACE(XmppRxStream,
     572             :                                    session->
     573             :                                    remote_endpoint().address().to_string(),
     574             :                                    session->
     575             :                                    remote_endpoint().port(), msg.size(), msg);
     576             :             }
     577             :         }
     578     4110761 :         IncProtoStats((unsigned int)minfo->type);
     579     4110778 :         state_machine_->OnMessage(session, minfo);
     580       41635 :     } else if ((minfo = last_msg_.get()) != NULL) {
     581       41665 :         session->IncStats((unsigned int)minfo->type, msg.size());
     582       41665 :         IncProtoStats((unsigned int)minfo->type);
     583             :     } else {
     584           0 :         session->IncStats(XmppStanza::INVALID, msg.size());
     585           0 :         XMPP_MESSAGE_TRACE(XmppRxStreamInvalid,
     586             :              session->remote_endpoint().address().to_string(),
     587             :              session->remote_endpoint().port(), msg.size(), msg);
     588             :     }
     589     4152444 :     return;
     590             : }
     591             : 
     592     4152422 : XmppStanza::XmppMessage *XmppConnection::XmppDecode(const string &msg) {
     593     4152422 :     unique_ptr<XmppStanza::XmppMessage> minfo(XmppProto::Decode(this, msg));
     594     4152412 :     if (minfo.get() == NULL) {
     595           0 :         XMPP_INFO(XmppSessionDelete, ToUVEKey(), XMPP_PEER_DIR_IN, "Server",
     596             :                   FromString(), ToString());
     597           0 :         Clear();
     598           0 :         return NULL;
     599             :     }
     600             : 
     601     4152401 :     if (minfo->type == XmppStanza::IQ_STANZA) {
     602             :         const XmppStanza::XmppMessageIq *iq =
     603       98564 :             static_cast<const XmppStanza::XmppMessageIq *>(minfo.get());
     604             : 
     605             : 
     606       98564 :         if (iq->action.compare("publish") == 0) {
     607       41666 :             last_msg_.reset(minfo.release());
     608       41665 :             return NULL;
     609             :         }
     610             : 
     611       56901 :         if (iq->action.compare("collection") == 0) {
     612       41669 :             if (last_msg_.get() != NULL) {
     613             :                 XmppStanza::XmppMessageIq *last_iq =
     614       41669 :                     static_cast<XmppStanza::XmppMessageIq *>(last_msg_.get());
     615             : 
     616       41669 :                 if (last_iq->node.compare(iq->as_node) == 0) {
     617       41669 :                     XmlBase *impl = last_iq->dom.get();
     618       41669 :                     impl->ReadNode("publish");
     619       41668 :                     impl->ModifyAttribute("node", iq->node);
     620       41668 :                     last_iq->node = impl->ReadAttrib("node");
     621       41670 :                     last_iq->is_as_node = iq->is_as_node;
     622             :                     //Save the complete ass/dissociate node
     623       41670 :                     last_iq->as_node = iq->as_node;
     624             :                 } else {
     625           0 :                     XMPP_WARNING(XmppIqMessageInvalid, ToUVEKey(),
     626             :                                  XMPP_PEER_DIR_IN);
     627           0 :                     goto error;
     628             :                 }
     629             :             } else {
     630           0 :                 XMPP_ERROR(XmppIqCollectionError, ToUVEKey(), XMPP_PEER_DIR_IN);
     631           0 :                 goto error;
     632             :             }
     633             :             // iq message merged with collection info
     634       41670 :             return last_msg_.release();
     635             :         }
     636             :     }
     637     4069060 :     return minfo.release();
     638             : 
     639           0 : error:
     640           0 :     last_msg_.reset();
     641           0 :     return NULL;
     642     4152431 : }
     643             : 
     644     1556755 : int XmppConnection::ProcessXmppChatMessage(
     645             :         const XmppStanza::XmppChatMessage *msg) {
     646     1556755 :     mux_->ProcessXmppMessage(msg);
     647     1556755 :     return 0;
     648             : }
     649             : 
     650       56895 : int XmppConnection::ProcessXmppIqMessage(const XmppStanza::XmppMessage *msg) {
     651       56895 :     mux_->ProcessXmppMessage(msg);
     652       56895 :     return 0;
     653             : }
     654             : 
     655             : class XmppServerConnection::DeleteActor : public LifetimeActor {
     656             : public:
     657        6568 :     DeleteActor(XmppServer *server, XmppServerConnection *parent)
     658        6568 :         : LifetimeActor(server->lifetime_manager()),
     659        6568 :           server_(server), parent_(parent) {
     660        6568 :     }
     661             : 
     662       12645 :     virtual bool MayDelete() const {
     663       12645 :         return (!parent_->on_work_queue() && parent_->MayDelete());
     664             :     }
     665             : 
     666        6568 :     virtual void Shutdown() {
     667        6568 :         CHECK_CONCURRENCY("bgp::Config");
     668             : 
     669             :         // If the connection is still on the WorkQueue, simply add it to the
     670             :         // ConnectionSet.  It won't be on the ConnectionMap.
     671        6568 :         if (parent_->on_work_queue()) {
     672          24 :             server_->InsertDeletedConnection(parent_);
     673             :         }
     674             : 
     675             :         // If the connection was rejected as duplicate, it will already be in
     676             :         // the ConnectionSet. Non-duplicate connections need to be moved from
     677             :         // from the ConnectionMap into the ConnectionSet.  We add it to the
     678             :         // ConnectionSet and then remove it from ConnectionMap to ensure that
     679             :         // the XmppServer connection count doesn't temporarily become 0. This
     680             :         // is friendly to tests that wait for the XmppServer connection count
     681             :         // to become 0.
     682             :         //
     683             :         // Breaking association with the XmppConnectionEndpoint here allows
     684             :         // a new XmppServerConnection with the same Endpoint to come up. We
     685             :         // may end up leaking memory if current XmppServerConnection doesn't
     686             :         // get cleaned up completely, but we at least prevent the other end
     687             :         // from getting stuck forever.
     688        6544 :         else if (!parent_->duplicate()) {
     689        6544 :             server_->InsertDeletedConnection(parent_);
     690        6544 :             server_->RemoveConnection(parent_);
     691        6544 :             server_->ReleaseConnectionEndpoint(parent_);
     692             :         }
     693             : 
     694        6568 :         if (parent_->state_machine()) {
     695        6568 :             parent_->state_machine()->Clear();
     696             :         }
     697             : 
     698        6568 :         XmppSession *session = NULL;
     699        6568 :         if (parent_->state_machine()) {
     700        6568 :             session = parent_->state_machine()->session();
     701        6568 :             parent_->state_machine()->clear_session();
     702             :         }
     703        6568 :         if (session) {
     704        1441 :             server_->DeleteSession(session);
     705             :         }
     706        6568 :     }
     707             : 
     708        6568 :     virtual void Destroy() {
     709        6568 :         delete parent_;
     710        6568 :     }
     711             : 
     712             : private:
     713             :     XmppServer *server_;
     714             :     XmppServerConnection *parent_;
     715             : };
     716             : 
     717        6568 : XmppServerConnection::XmppServerConnection(XmppServer *server,
     718        6568 :     const XmppChannelConfig *config)
     719             :     : XmppConnection(server, config),
     720        6568 :       duplicate_(false),
     721        6568 :       on_work_queue_(false),
     722        6568 :       conn_endpoint_(NULL),
     723        6568 :       deleter_(new DeleteActor(server, this)),
     724       13136 :       server_delete_ref_(this, server->deleter()) {
     725        6568 :     assert(!config->ClientOnly());
     726        6568 :     XMPP_INFO(XmppConnectionCreate, ToUVEKey(), XMPP_PEER_DIR_IN,
     727             :               "Server", FromString(), ToString());
     728        6568 : }
     729             : 
     730       13123 : XmppServerConnection::~XmppServerConnection() {
     731        6568 :     CHECK_CONCURRENCY("bgp::Config");
     732             : 
     733        6568 :     XMPP_INFO(XmppConnectionDelete, ToUVEKey(), XMPP_PEER_DIR_NA, "Server",
     734             :               FromString(), ToString());
     735        6568 :     server()->RemoveDeletedConnection(this);
     736       13123 : }
     737             : 
     738        8243 : void XmppServerConnection::ManagedDelete() {
     739        8243 :     XMPP_UTDEBUG(XmppConnectionDelete, ToUVEKey(), XMPP_PEER_DIR_NA,
     740             :                  "Managed server connection delete", FromString(), ToString());
     741        8243 :     deleter_->Delete();
     742        8243 : }
     743             : 
     744        6190 : void XmppServerConnection::RetryDelete() {
     745        6190 :     if (!deleter()->IsDeleted())
     746          17 :         return;
     747        6173 :     deleter()->RetryDelete();
     748             : }
     749             : 
     750           0 : LifetimeManager *XmppServerConnection::lifetime_manager() {
     751           0 :     return server()->lifetime_manager();
     752             : }
     753             : 
     754        8349 : XmppServer *XmppServerConnection::server() {
     755        8349 :     return static_cast<XmppServer *>(server_);
     756             : }
     757             : 
     758       12411 : LifetimeActor *XmppServerConnection::deleter() {
     759       12411 :     return deleter_.get();
     760             : }
     761             : 
     762      109280 : const LifetimeActor *XmppServerConnection::deleter() const {
     763      109280 :     return deleter_.get();
     764             : }
     765             : 
     766           4 : void XmppServerConnection::set_close_reason(const string &close_reason) {
     767           4 :     if (conn_endpoint_)
     768           0 :         conn_endpoint_->set_close_reason(close_reason);
     769             : 
     770           4 :     if (!logUVE())
     771           4 :         return;
     772             : 
     773           0 :     XmppPeerInfoData peer_info;
     774           0 :     peer_info.set_name(ToUVEKey());
     775           0 :     peer_info.set_close_reason(close_reason);
     776           0 :     XMPPPeerInfoSend(peer_info);
     777           0 : }
     778             : 
     779        1461 : uint32_t XmppServerConnection::flap_count() const {
     780        1461 :     return conn_endpoint_ ? conn_endpoint_->flap_count() : 0;
     781             : }
     782             : 
     783        6299 : void XmppServerConnection::increment_flap_count() {
     784        6299 :     XmppConnectionEndpoint *conn_endpoint = conn_endpoint_;
     785        6299 :     if (!conn_endpoint)
     786        1407 :         conn_endpoint = server()->FindConnectionEndpoint(ToString());
     787        6299 :     if (!conn_endpoint)
     788        6300 :         return;
     789        6299 :     conn_endpoint->increment_flap_count();
     790             : 
     791        6300 :     if (!logUVE())
     792        6300 :         return;
     793             : 
     794           0 :     XmppPeerInfoData peer_info;
     795           0 :     peer_info.set_name(ToUVEKey());
     796           0 :     PeerFlapInfo flap_info;
     797           0 :     flap_info.set_flap_count(conn_endpoint->flap_count());
     798           0 :     flap_info.set_flap_time(conn_endpoint->last_flap());
     799           0 :     peer_info.set_flap_info(flap_info);
     800           0 :     XMPPPeerInfoSend(peer_info);
     801           0 : }
     802             : 
     803        1461 : const std::string XmppServerConnection::last_flap_at() const {
     804        1461 :     return conn_endpoint_ ? conn_endpoint_->last_flap_at() : "";
     805             : }
     806             : 
     807          72 : void XmppServerConnection::FillShowInfo(
     808             :     ShowXmppConnection *show_connection) const {
     809          72 :     show_connection->set_name(ToString());
     810          72 :     show_connection->set_deleted(IsDeleted());
     811          72 :     show_connection->set_remote_endpoint(endpoint_string());
     812          72 :     show_connection->set_local_endpoint(local_endpoint_string());
     813          72 :     show_connection->set_state(StateName());
     814          72 :     show_connection->set_last_event(LastEvent());
     815          72 :     show_connection->set_last_state(LastStateName());
     816          72 :     show_connection->set_last_state_at(LastStateChangeAt());
     817          72 :     show_connection->set_receivers(channel_mux()->GetReceiverList());
     818          72 :     show_connection->set_server_auth_type(GetXmppAuthenticationType());
     819          72 :     show_connection->set_dscp_value(dscp_value());
     820          72 : }
     821             : 
     822             : class XmppClientConnection::DeleteActor : public LifetimeActor {
     823             : public:
     824        5038 :     DeleteActor(XmppClient *client, XmppClientConnection *parent)
     825        5038 :         : LifetimeActor(client->lifetime_manager()),
     826        5038 :           client_(client), parent_(parent) {
     827        5038 :     }
     828             : 
     829        5038 :     virtual bool MayDelete() const {
     830        5038 :         return parent_->MayDelete();
     831             :     }
     832             : 
     833        5038 :     virtual void Shutdown() {
     834        5038 :         if (parent_->session()) {
     835         544 :             client_->NotifyConnectionEvent(parent_->ChannelMux(),
     836             :                 xmps::NOT_READY);
     837             :         }
     838             : 
     839        5038 :         XmppSession *session = NULL;
     840        5038 :         if (parent_->state_machine()) {
     841        5038 :             session = parent_->state_machine()->session();
     842        5038 :             parent_->state_machine()->clear_session();
     843             :         }
     844        5038 :         if (session) {
     845         565 :             client_->DeleteSession(session);
     846             :         }
     847        5038 :     }
     848             : 
     849        5038 :     virtual void Destroy() {
     850        5038 :         delete parent_;
     851        5038 :     }
     852             : 
     853             : private:
     854             :     XmppClient *client_;
     855             :     XmppClientConnection *parent_;
     856             : };
     857             : 
     858        5038 : XmppClientConnection::XmppClientConnection(XmppClient *server,
     859        5038 :     const XmppChannelConfig *config)
     860             :     : XmppConnection(server, config),
     861        5038 :       flap_count_(0),
     862        5038 :       deleter_(new DeleteActor(server, this)),
     863       10076 :       server_delete_ref_(this, server->deleter()) {
     864        5038 :     assert(config->ClientOnly());
     865        5038 :     XMPP_UTDEBUG(XmppConnectionCreate, ToUVEKey(), XMPP_PEER_DIR_NA, "Client",
     866             :                  FromString(), ToString());
     867        5038 : }
     868             : 
     869       10049 : XmppClientConnection::~XmppClientConnection() {
     870        5038 :     CHECK_CONCURRENCY("bgp::Config");
     871             : 
     872        5038 :     XMPP_INFO(XmppConnectionDelete, ToUVEKey(), XMPP_PEER_DIR_NA,
     873             :               "Client", FromString(), ToString());
     874        5038 :     server()->RemoveConnection(this);
     875       10049 : }
     876             : 
     877        5038 : void XmppClientConnection::ManagedDelete() {
     878        5038 :     XMPP_UTDEBUG(XmppConnectionDelete, ToUVEKey(), XMPP_PEER_DIR_NA,
     879             :                  "Managed Client Delete", FromString(), ToString());
     880        5038 :     deleter_->Delete();
     881        5038 : }
     882             : 
     883        3823 : void XmppClientConnection::RetryDelete() {
     884        3823 :     if (!deleter()->IsDeleted())
     885        3823 :         return;
     886           0 :     deleter()->RetryDelete();
     887             : }
     888             : 
     889           0 : LifetimeManager *XmppClientConnection::lifetime_manager() {
     890           0 :     return server()->lifetime_manager();
     891             : }
     892             : 
     893        5038 : XmppClient *XmppClientConnection::server() {
     894        5038 :     return static_cast<XmppClient *>(server_);
     895             : }
     896             : 
     897        3823 : LifetimeActor *XmppClientConnection::deleter() {
     898        3823 :     return deleter_.get();
     899             : }
     900             : 
     901        8888 : const LifetimeActor *XmppClientConnection::deleter() const {
     902        8888 :     return deleter_.get();
     903             : }
     904             : 
     905        3364 : void XmppClientConnection::set_close_reason(const string &close_reason) {
     906        3364 :     close_reason_ = close_reason;
     907        3364 :     if (!logUVE())
     908        3364 :         return;
     909             : 
     910           0 :     XmppPeerInfoData peer_info;
     911           0 :     peer_info.set_name(ToUVEKey());
     912           0 :     peer_info.set_close_reason(close_reason_);
     913           0 :     XMPPPeerInfoSend(peer_info);
     914           0 : }
     915             : 
     916        3475 : uint32_t XmppClientConnection::flap_count() const {
     917        3475 :     return flap_count_;
     918             : }
     919             : 
     920        5761 : void XmppClientConnection::increment_flap_count() {
     921        5761 :     flap_count_++;
     922        5761 :     last_flap_ = UTCTimestampUsec();
     923             : 
     924        5761 :     if (!logUVE())
     925        5761 :         return;
     926             : 
     927           0 :     XmppPeerInfoData peer_info;
     928           0 :     peer_info.set_name(ToUVEKey());
     929           0 :     PeerFlapInfo flap_info;
     930           0 :     flap_info.set_flap_count(flap_count_);
     931           0 :     flap_info.set_flap_time(last_flap_);
     932           0 :     peer_info.set_flap_info(flap_info);
     933           0 :     XMPPPeerInfoSend(peer_info);
     934           0 : }
     935             : 
     936           0 : const std::string XmppClientConnection::last_flap_at() const {
     937           0 :     return last_flap_ ? integerToString(UTCUsecToPTime(last_flap_)) : "";
     938             : }
     939             : 
     940        4876 : XmppConnectionEndpoint::XmppConnectionEndpoint(const string &client)
     941        4876 :     : client_(client), flap_count_(0), last_flap_(0), connection_(NULL) {
     942        4876 : }
     943             : 
     944           0 : void XmppConnectionEndpoint::set_close_reason(const string &close_reason) {
     945           0 :     close_reason_ = close_reason;
     946           0 : }
     947             : 
     948        2479 : uint32_t XmppConnectionEndpoint::flap_count() const {
     949        2479 :     return flap_count_;
     950             : }
     951             : 
     952        6299 : void XmppConnectionEndpoint::increment_flap_count() {
     953        6299 :     flap_count_++;
     954        6299 :     last_flap_ = UTCTimestampUsec();
     955        6300 : }
     956             : 
     957           0 : uint64_t XmppConnectionEndpoint::last_flap() const {
     958           0 :     return last_flap_;
     959             : }
     960             : 
     961        1425 : const std::string XmppConnectionEndpoint::last_flap_at() const {
     962        1425 :     return last_flap_ ? integerToString(UTCUsecToPTime(last_flap_)) : "";
     963             : }
     964             : 
     965       11121 : XmppConnection *XmppConnectionEndpoint::connection() {
     966       11121 :     return connection_;
     967             : }
     968             : 
     969          47 : const XmppConnection *XmppConnectionEndpoint::connection() const {
     970          47 :     return connection_;
     971             : }
     972             : 
     973        6201 : void XmppConnectionEndpoint::set_connection(XmppConnection *connection) {
     974        6201 :     assert(!connection_);
     975        6201 :     connection_ = connection;
     976        6201 : }
     977             : 
     978        6201 : void XmppConnectionEndpoint::reset_connection() {
     979        6201 :     assert(connection_);
     980        6201 :     connection_ = NULL;
     981        6201 : }
     982             : 
     983             : // Swap relavent contents between two XmppConnection objects.
     984         104 : void XmppConnection::SwapContents(XmppConnection *other) {
     985         104 :     assert(!IsClient());
     986         104 :     assert(!other->IsClient());
     987             :     // Update the ConnectionMap in the server as the endpoints are the keys.
     988         104 :     XmppServer *server = dynamic_cast<XmppServerConnection *>(this)->server();
     989         104 :     server->SwapXmppConnectionMapEntries(this, other);
     990             :     // Swap all other connection related information.
     991         104 :     swap(local_endpoint_, other->local_endpoint_);
     992         104 :     stats_[0].swap(other->stats_[0]);
     993         104 :     stats_[1].swap(other->stats_[1]);
     994         104 :     error_stats_.swap(other->error_stats_);
     995         104 :     swap(last_msg_, other->last_msg_);
     996         104 :     swap(to_, other->to_);
     997         104 :     swap(from_, other->from_);
     998         104 :     swap(xmlns_, other->xmlns_);
     999         104 :     swap(dscp_value_, other->dscp_value_);
    1000         104 :     swap(disable_read_, other->disable_read_);
    1001         104 : }