src/third_party/libjingle/source/talk/p2p/base/session.cc

   1 /*
   2  * libjingle
   3  * Copyright 2004--2005, Google Inc.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are met:
   7  *
   8  *  1. Redistributions of source code must retain the above copyright notice,
   9  *     this list of conditions and the following disclaimer.
  10  *  2. Redistributions in binary form must reproduce the above copyright notice,
  11  *     this list of conditions and the following disclaimer in the documentation
  12  *     and/or other materials provided with the distribution.
  13  *  3. The name of the author may not be used to endorse or promote products
  14  *     derived from this software without specific prior written permission.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
  17  * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  18  * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
  19  * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  20  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  21  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  22  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  23  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
  24  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  25  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26  */
  27
  28 #include "talk/p2p/base/session.h"
  29
  30 #include "talk/base/bind.h"
  31 #include "talk/base/common.h"
  32 #include "talk/base/logging.h"
  33 #include "talk/base/helpers.h"
  34 #include "talk/base/scoped_ptr.h"
  35 #include "talk/base/sslstreamadapter.h"
  36 #include "talk/xmpp/constants.h"
  37 #include "talk/xmpp/jid.h"
  38 #include "talk/p2p/base/dtlstransport.h"
  39 #include "talk/p2p/base/p2ptransport.h"
  40 #include "talk/p2p/base/sessionclient.h"
  41 #include "talk/p2p/base/transport.h"
  42 #include "talk/p2p/base/transportchannelproxy.h"
  43 #include "talk/p2p/base/transportinfo.h"
  44
  45 #include "talk/p2p/base/constants.h"
  46
  47 namespace cricket {
  48
  49 using talk_base::Bind;
  50
  51 bool BadMessage(const buzz::QName type,
  52                 const std::string& text,
  53                 MessageError* err) {
  54   err->SetType(type);
  55   err->SetText(text);
  56   return false;
  57 }
  58
  59 TransportProxy::~TransportProxy() {
  60   for (ChannelMap::iterator iter = channels_.begin();
  61        iter != channels_.end(); ++iter) {
  62     iter->second->SignalDestroyed(iter->second);
  63     delete iter->second;
  64   }
  65 }
  66
  67 std::string TransportProxy::type() const {
  68   return transport_->get()->type();
  69 }
  70
  71 TransportChannel* TransportProxy::GetChannel(int component) {
  72   ASSERT(talk_base::Thread::Current() == worker_thread_);
  73   return GetChannelProxy(component);
  74 }
  75
  76 TransportChannel* TransportProxy::CreateChannel(
  77     const std::string& name, int component) {
  78   ASSERT(talk_base::Thread::Current() == worker_thread_);
  79   ASSERT(GetChannel(component) == NULL);
  80   ASSERT(!transport_->get()->HasChannel(component));
  81
  82   // We always create a proxy in case we need to change out the transport later.
  83   TransportChannelProxy* channel =
  84       new TransportChannelProxy(content_name(), name, component);
  85   channels_[component] = channel;
  86
  87   // If we're already negotiated, create an impl and hook it up to the proxy
  88   // channel. If we're connecting, create an impl but don't hook it up yet.
  89   if (negotiated_) {
  90     SetupChannelProxy_w(component, channel);
  91   } else if (connecting_) {
  92     GetOrCreateChannelProxyImpl_w(component);
  93   }
  94   return channel;
  95 }
  96
  97 bool TransportProxy::HasChannel(int component) {
  98   return transport_->get()->HasChannel(component);
  99 }
 100
 101 void TransportProxy::DestroyChannel(int component) {
 102   ASSERT(talk_base::Thread::Current() == worker_thread_);
 103   TransportChannel* channel = GetChannel(component);
 104   if (channel) {
 105     // If the state of TransportProxy is not NEGOTIATED
 106     // then TransportChannelProxy and its impl are not
 107     // connected. Both must be connected before
 108     // deletion.
 109     if (!negotiated_) {
 110       SetupChannelProxy_w(component, GetChannelProxy(component));
 111     }
 112
 113     channels_.erase(component);
 114     channel->SignalDestroyed(channel);
 115     delete channel;
 116   }
 117 }
 118
 119 void TransportProxy::ConnectChannels() {
 120   if (!connecting_) {
 121     if (!negotiated_) {
 122       for (ChannelMap::iterator iter = channels_.begin();
 123            iter != channels_.end(); ++iter) {
 124         GetOrCreateChannelProxyImpl(iter->first);
 125       }
 126     }
 127     connecting_ = true;
 128   }
 129   // TODO(juberti): Right now Transport::ConnectChannels doesn't work if we
 130   // don't have any channels yet, so we need to allow this method to be called
 131   // multiple times. Once we fix Transport, we can move this call inside the
 132   // if (!connecting_) block.
 133   transport_->get()->ConnectChannels();
 134 }
 135
 136 void TransportProxy::CompleteNegotiation() {
 137   if (!negotiated_) {
 138     for (ChannelMap::iterator iter = channels_.begin();
 139          iter != channels_.end(); ++iter) {
 140       SetupChannelProxy(iter->first, iter->second);
 141     }
 142     negotiated_ = true;
 143   }
 144 }
 145
 146 void TransportProxy::AddSentCandidates(const Candidates& candidates) {
 147   for (Candidates::const_iterator cand = candidates.begin();
 148        cand != candidates.end(); ++cand) {
 149     sent_candidates_.push_back(*cand);
 150   }
 151 }
 152
 153 void TransportProxy::AddUnsentCandidates(const Candidates& candidates) {
 154   for (Candidates::const_iterator cand = candidates.begin();
 155        cand != candidates.end(); ++cand) {
 156     unsent_candidates_.push_back(*cand);
 157   }
 158 }
 159
 160 bool TransportProxy::GetChannelNameFromComponent(
 161     int component, std::string* channel_name) const {
 162   const TransportChannelProxy* channel = GetChannelProxy(component);
 163   if (channel == NULL) {
 164     return false;
 165   }
 166
 167   *channel_name = channel->name();
 168   return true;
 169 }
 170
 171 bool TransportProxy::GetComponentFromChannelName(
 172     const std::string& channel_name, int* component) const {
 173   const TransportChannelProxy* channel = GetChannelProxyByName(channel_name);
 174   if (channel == NULL) {
 175     return false;
 176   }
 177
 178   *component = channel->component();
 179   return true;
 180 }
 181
 182 TransportChannelProxy* TransportProxy::GetChannelProxy(int component) const {
 183   ChannelMap::const_iterator iter = channels_.find(component);
 184   return (iter != channels_.end()) ? iter->second : NULL;
 185 }
 186
 187 TransportChannelProxy* TransportProxy::GetChannelProxyByName(
 188     const std::string& name) const {
 189   for (ChannelMap::const_iterator iter = channels_.begin();
 190        iter != channels_.end();
 191        ++iter) {
 192     if (iter->second->name() == name) {
 193       return iter->second;
 194     }
 195   }
 196   return NULL;
 197 }
 198
 199 TransportChannelImpl* TransportProxy::GetOrCreateChannelProxyImpl(
 200     int component) {
 201   return worker_thread_->Invoke<TransportChannelImpl*>(Bind(
 202       &TransportProxy::GetOrCreateChannelProxyImpl_w, this, component));
 203 }
 204
 205 TransportChannelImpl* TransportProxy::GetOrCreateChannelProxyImpl_w(
 206     int component) {
 207   ASSERT(talk_base::Thread::Current() == worker_thread_);
 208   TransportChannelImpl* impl = transport_->get()->GetChannel(component);
 209   if (impl == NULL) {
 210     impl = transport_->get()->CreateChannel(component);
 211   }
 212   return impl;
 213 }
 214
 215 void TransportProxy::SetupChannelProxy(
 216     int component, TransportChannelProxy* transproxy) {
 217   worker_thread_->Invoke<void>(Bind(
 218       &TransportProxy::SetupChannelProxy_w, this, component, transproxy));
 219 }
 220
 221 void TransportProxy::SetupChannelProxy_w(
 222     int component, TransportChannelProxy* transproxy) {
 223   ASSERT(talk_base::Thread::Current() == worker_thread_);
 224   TransportChannelImpl* impl = GetOrCreateChannelProxyImpl(component);
 225   ASSERT(impl != NULL);
 226   transproxy->SetImplementation(impl);
 227 }
 228
 229 void TransportProxy::ReplaceChannelProxyImpl(TransportChannelProxy* proxy,
 230                                              TransportChannelImpl* impl) {
 231   worker_thread_->Invoke<void>(Bind(
 232       &TransportProxy::ReplaceChannelProxyImpl_w, this, proxy, impl));
 233 }
 234
 235 void TransportProxy::ReplaceChannelProxyImpl_w(TransportChannelProxy* proxy,
 236                                                TransportChannelImpl* impl) {
 237   ASSERT(talk_base::Thread::Current() == worker_thread_);
 238   ASSERT(proxy != NULL);
 239   proxy->SetImplementation(impl);
 240 }
 241
 242 // This function muxes |this| onto |target| by repointing |this| at
 243 // |target|'s transport and setting our TransportChannelProxies
 244 // to point to |target|'s underlying implementations.
 245 bool TransportProxy::SetupMux(TransportProxy* target) {
 246   // Bail out if there's nothing to do.
 247   if (transport_ == target->transport_) {
 248     return true;
 249   }
 250
 251   // Run through all channels and remove any non-rtp transport channels before
 252   // setting target transport channels.
 253   for (ChannelMap::const_iterator iter = channels_.begin();
 254        iter != channels_.end(); ++iter) {
 255     if (!target->transport_->get()->HasChannel(iter->first)) {
 256       // Remove if channel doesn't exist in |transport_|.
 257       ReplaceChannelProxyImpl(iter->second, NULL);
 258     } else {
 259       // Replace the impl for all the TransportProxyChannels with the channels
 260       // from |target|'s transport. Fail if there's not an exact match.
 261       ReplaceChannelProxyImpl(
 262           iter->second, target->transport_->get()->CreateChannel(iter->first));
 263     }
 264   }
 265
 266   // Now replace our transport. Must happen afterwards because
 267   // it deletes all impls as a side effect.
 268   transport_ = target->transport_;
 269   transport_->get()->SignalCandidatesReady.connect(
 270       this, &TransportProxy::OnTransportCandidatesReady);
 271   set_candidates_allocated(target->candidates_allocated());
 272   return true;
 273 }
 274
 275 void TransportProxy::SetIceRole(IceRole role) {
 276   transport_->get()->SetIceRole(role);
 277 }
 278
 279 bool TransportProxy::SetLocalTransportDescription(
 280     const TransportDescription& description,
 281     ContentAction action,
 282     std::string* error_desc) {
 283   // If this is an answer, finalize the negotiation.
 284   if (action == CA_ANSWER) {
 285     CompleteNegotiation();
 286   }
 287   return transport_->get()->SetLocalTransportDescription(description,
 288                                                          action,
 289                                                          error_desc);
 290 }
 291
 292 bool TransportProxy::SetRemoteTransportDescription(
 293     const TransportDescription& description,
 294     ContentAction action,
 295     std::string* error_desc) {
 296   // If this is an answer, finalize the negotiation.
 297   if (action == CA_ANSWER) {
 298     CompleteNegotiation();
 299   }
 300   return transport_->get()->SetRemoteTransportDescription(description,
 301                                                           action,
 302                                                           error_desc);
 303 }
 304
 305 void TransportProxy::OnSignalingReady() {
 306   // If we're starting a new allocation sequence, reset our state.
 307   set_candidates_allocated(false);
 308   transport_->get()->OnSignalingReady();
 309 }
 310
 311 bool TransportProxy::OnRemoteCandidates(const Candidates& candidates,
 312                                         std::string* error) {
 313   // Ensure the transport is negotiated before handling candidates.
 314   // TODO(juberti): Remove this once everybody calls SetLocalTD.
 315   CompleteNegotiation();
 316
 317   // Verify each candidate before passing down to transport layer.
 318   for (Candidates::const_iterator cand = candidates.begin();
 319        cand != candidates.end(); ++cand) {
 320     if (!transport_->get()->VerifyCandidate(*cand, error))
 321       return false;
 322     if (!HasChannel(cand->component())) {
 323       *error = "Candidate has unknown component: " + cand->ToString() +
 324                " for content: " + content_name_;
 325       return false;
 326     }
 327   }
 328   transport_->get()->OnRemoteCandidates(candidates);
 329   return true;
 330 }
 331
 332 void TransportProxy::SetIdentity(
 333     talk_base::SSLIdentity* identity) {
 334   transport_->get()->SetIdentity(identity);
 335 }
 336
 337 std::string BaseSession::StateToString(State state) {
 338   switch (state) {
 339     case Session::STATE_INIT:
 340       return "STATE_INIT";
 341     case Session::STATE_SENTINITIATE:
 342       return "STATE_SENTINITIATE";
 343     case Session::STATE_RECEIVEDINITIATE:
 344       return "STATE_RECEIVEDINITIATE";
 345     case Session::STATE_SENTPRACCEPT:
 346       return "STATE_SENTPRACCEPT";
 347     case Session::STATE_SENTACCEPT:
 348       return "STATE_SENTACCEPT";
 349     case Session::STATE_RECEIVEDPRACCEPT:
 350       return "STATE_RECEIVEDPRACCEPT";
 351     case Session::STATE_RECEIVEDACCEPT:
 352       return "STATE_RECEIVEDACCEPT";
 353     case Session::STATE_SENTMODIFY:
 354       return "STATE_SENTMODIFY";
 355     case Session::STATE_RECEIVEDMODIFY:
 356       return "STATE_RECEIVEDMODIFY";
 357     case Session::STATE_SENTREJECT:
 358       return "STATE_SENTREJECT";
 359     case Session::STATE_RECEIVEDREJECT:
 360       return "STATE_RECEIVEDREJECT";
 361     case Session::STATE_SENTREDIRECT:
 362       return "STATE_SENTREDIRECT";
 363     case Session::STATE_SENTTERMINATE:
 364       return "STATE_SENTTERMINATE";
 365     case Session::STATE_RECEIVEDTERMINATE:
 366       return "STATE_RECEIVEDTERMINATE";
 367     case Session::STATE_INPROGRESS:
 368       return "STATE_INPROGRESS";
 369     case Session::STATE_DEINIT:
 370       return "STATE_DEINIT";
 371     default:
 372       break;
 373   }
 374   return "STATE_" + talk_base::ToString(state);
 375 }
 376
 377 BaseSession::BaseSession(talk_base::Thread* signaling_thread,
 378                          talk_base::Thread* worker_thread,
 379                          PortAllocator* port_allocator,
 380                          const std::string& sid,
 381                          const std::string& content_type,
 382                          bool initiator)
 383     : state_(STATE_INIT),
 384       error_(ERROR_NONE),
 385       signaling_thread_(signaling_thread),
 386       worker_thread_(worker_thread),
 387       port_allocator_(port_allocator),
 388       sid_(sid),
 389       content_type_(content_type),
 390       transport_type_(NS_GINGLE_P2P),
 391       initiator_(initiator),
 392       identity_(NULL),
 393       local_description_(NULL),
 394       remote_description_(NULL),
 395       ice_tiebreaker_(talk_base::CreateRandomId64()),
 396       role_switch_(false) {
 397   ASSERT(signaling_thread->IsCurrent());
 398 }
 399
 400 BaseSession::~BaseSession() {
 401   ASSERT(signaling_thread()->IsCurrent());
 402
 403   ASSERT(state_ != STATE_DEINIT);
 404   LogState(state_, STATE_DEINIT);
 405   state_ = STATE_DEINIT;
 406   SignalState(this, state_);
 407
 408   for (TransportMap::iterator iter = transports_.begin();
 409        iter != transports_.end(); ++iter) {
 410     delete iter->second;
 411   }
 412
 413   delete remote_description_;
 414   delete local_description_;
 415 }
 416
 417 bool BaseSession::SetIdentity(talk_base::SSLIdentity* identity) {
 418   if (identity_)
 419     return false;
 420   identity_ = identity;
 421   for (TransportMap::iterator iter = transports_.begin();
 422        iter != transports_.end(); ++iter) {
 423     iter->second->SetIdentity(identity_);
 424   }
 425   return true;
 426 }
 427
 428 bool BaseSession::PushdownTransportDescription(ContentSource source,
 429                                                ContentAction action,
 430                                                std::string* error_desc) {
 431   if (source == CS_LOCAL) {
 432     return PushdownLocalTransportDescription(local_description_,
 433                                              action,
 434                                              error_desc);
 435   }
 436   return PushdownRemoteTransportDescription(remote_description_,
 437                                             action,
 438                                             error_desc);
 439 }
 440
 441 bool BaseSession::PushdownLocalTransportDescription(
 442     const SessionDescription* sdesc,
 443     ContentAction action,
 444     std::string* error_desc) {
 445   // Update the Transports with the right information, and trigger them to
 446   // start connecting.
 447   for (TransportMap::iterator iter = transports_.begin();
 448        iter != transports_.end(); ++iter) {
 449     // If no transport info was in this session description, ret == false
 450     // and we just skip this one.
 451     TransportDescription tdesc;
 452     bool ret = GetTransportDescription(
 453         sdesc, iter->second->content_name(), &tdesc);
 454     if (ret) {
 455       if (!iter->second->SetLocalTransportDescription(tdesc, action,
 456                                                       error_desc)) {
 457         return false;
 458       }
 459
 460       iter->second->ConnectChannels();
 461     }
 462   }
 463
 464   return true;
 465 }
 466
 467 bool BaseSession::PushdownRemoteTransportDescription(
 468     const SessionDescription* sdesc,
 469     ContentAction action,
 470     std::string* error_desc) {
 471   // Update the Transports with the right information.
 472   for (TransportMap::iterator iter = transports_.begin();
 473        iter != transports_.end(); ++iter) {
 474     TransportDescription tdesc;
 475
 476     // If no transport info was in this session description, ret == false
 477     // and we just skip this one.
 478     bool ret = GetTransportDescription(
 479         sdesc, iter->second->content_name(), &tdesc);
 480     if (ret) {
 481       if (!iter->second->SetRemoteTransportDescription(tdesc, action,
 482                                                        error_desc)) {
 483         return false;
 484       }
 485     }
 486   }
 487
 488   return true;
 489 }
 490
 491 TransportChannel* BaseSession::CreateChannel(const std::string& content_name,
 492                                              const std::string& channel_name,
 493                                              int component) {
 494   // We create the proxy "on demand" here because we need to support
 495   // creating channels at any time, even before we send or receive
 496   // initiate messages, which is before we create the transports.
 497   TransportProxy* transproxy = GetOrCreateTransportProxy(content_name);
 498   return transproxy->CreateChannel(channel_name, component);
 499 }
 500
 501 TransportChannel* BaseSession::GetChannel(const std::string& content_name,
 502                                           int component) {
 503   TransportProxy* transproxy = GetTransportProxy(content_name);
 504   if (transproxy == NULL)
 505     return NULL;
 506   else
 507     return transproxy->GetChannel(component);
 508 }
 509
 510 void BaseSession::DestroyChannel(const std::string& content_name,
 511                                  int component) {
 512   TransportProxy* transproxy = GetTransportProxy(content_name);
 513   ASSERT(transproxy != NULL);
 514   transproxy->DestroyChannel(component);
 515 }
 516
 517 TransportProxy* BaseSession::GetOrCreateTransportProxy(
 518     const std::string& content_name) {
 519   TransportProxy* transproxy = GetTransportProxy(content_name);
 520   if (transproxy)
 521     return transproxy;
 522
 523   Transport* transport = CreateTransport(content_name);
 524   transport->SetIceRole(initiator_ ? ICEROLE_CONTROLLING : ICEROLE_CONTROLLED);
 525   transport->SetIceTiebreaker(ice_tiebreaker_);
 526   // TODO: Connect all the Transport signals to TransportProxy
 527   // then to the BaseSession.
 528   transport->SignalConnecting.connect(
 529       this, &BaseSession::OnTransportConnecting);
 530   transport->SignalWritableState.connect(
 531       this, &BaseSession::OnTransportWritable);
 532   transport->SignalRequestSignaling.connect(
 533       this, &BaseSession::OnTransportRequestSignaling);
 534   transport->SignalTransportError.connect(
 535       this, &BaseSession::OnTransportSendError);
 536   transport->SignalRouteChange.connect(
 537       this, &BaseSession::OnTransportRouteChange);
 538   transport->SignalCandidatesAllocationDone.connect(
 539       this, &BaseSession::OnTransportCandidatesAllocationDone);
 540   transport->SignalRoleConflict.connect(
 541       this, &BaseSession::OnRoleConflict);
 542
 543   transproxy = new TransportProxy(worker_thread_, sid_, content_name,
 544                                   new TransportWrapper(transport));
 545   transproxy->SignalCandidatesReady.connect(
 546       this, &BaseSession::OnTransportProxyCandidatesReady);
 547   transports_[content_name] = transproxy;
 548
 549   return transproxy;
 550 }
 551
 552 Transport* BaseSession::GetTransport(const std::string& content_name) {
 553   TransportProxy* transproxy = GetTransportProxy(content_name);
 554   if (transproxy == NULL)
 555     return NULL;
 556   return transproxy->impl();
 557 }
 558
 559 TransportProxy* BaseSession::GetTransportProxy(
 560     const std::string& content_name) {
 561   TransportMap::iterator iter = transports_.find(content_name);
 562   return (iter != transports_.end()) ? iter->second : NULL;
 563 }
 564
 565 TransportProxy* BaseSession::GetTransportProxy(const Transport* transport) {
 566   for (TransportMap::iterator iter = transports_.begin();
 567        iter != transports_.end(); ++iter) {
 568     TransportProxy* transproxy = iter->second;
 569     if (transproxy->impl() == transport) {
 570       return transproxy;
 571     }
 572   }
 573   return NULL;
 574 }
 575
 576 TransportProxy* BaseSession::GetFirstTransportProxy() {
 577   if (transports_.empty())
 578     return NULL;
 579   return transports_.begin()->second;
 580 }
 581
 582 void BaseSession::DestroyTransportProxy(
 583     const std::string& content_name) {
 584   TransportMap::iterator iter = transports_.find(content_name);
 585   if (iter != transports_.end()) {
 586     delete iter->second;
 587     transports_.erase(content_name);
 588   }
 589 }
 590
 591 cricket::Transport* BaseSession::CreateTransport(
 592     const std::string& content_name) {
 593   ASSERT(transport_type_ == NS_GINGLE_P2P);
 594   return new cricket::DtlsTransport<P2PTransport>(
 595       signaling_thread(), worker_thread(), content_name,
 596       port_allocator(), identity_);
 597 }
 598
 599 bool BaseSession::GetStats(SessionStats* stats) {
 600   for (TransportMap::iterator iter = transports_.begin();
 601        iter != transports_.end(); ++iter) {
 602     std::string proxy_id = iter->second->content_name();
 603     // We are ignoring not-yet-instantiated transports.
 604     if (iter->second->impl()) {
 605       std::string transport_id = iter->second->impl()->content_name();
 606       stats->proxy_to_transport[proxy_id] = transport_id;
 607       if (stats->transport_stats.find(transport_id)
 608           == stats->transport_stats.end()) {
 609         TransportStats subinfos;
 610         if (!iter->second->impl()->GetStats(&subinfos)) {
 611           return false;
 612         }
 613         stats->transport_stats[transport_id] = subinfos;
 614       }
 615     }
 616   }
 617   return true;
 618 }
 619
 620 void BaseSession::SetState(State state) {
 621   ASSERT(signaling_thread_->IsCurrent());
 622   if (state != state_) {
 623     LogState(state_, state);
 624     state_ = state;
 625     SignalState(this, state_);
 626     signaling_thread_->Post(this, MSG_STATE);
 627   }
 628   SignalNewDescription();
 629 }
 630
 631 void BaseSession::SetError(Error error, const std::string& error_desc) {
 632   ASSERT(signaling_thread_->IsCurrent());
 633   if (error != error_) {
 634     error_ = error;
 635     error_desc_ = error_desc;
 636     SignalError(this, error);
 637   }
 638 }
 639
 640 void BaseSession::OnSignalingReady() {
 641   ASSERT(signaling_thread()->IsCurrent());
 642   for (TransportMap::iterator iter = transports_.begin();
 643        iter != transports_.end(); ++iter) {
 644     iter->second->OnSignalingReady();
 645   }
 646 }
 647
 648 // TODO(juberti): Since PushdownLocalTD now triggers the connection process to
 649 // start, remove this method once everyone calls PushdownLocalTD.
 650 void BaseSession::SpeculativelyConnectAllTransportChannels() {
 651   // Put all transports into the connecting state.
 652   for (TransportMap::iterator iter = transports_.begin();
 653        iter != transports_.end(); ++iter) {
 654     iter->second->ConnectChannels();
 655   }
 656 }
 657
 658 bool BaseSession::OnRemoteCandidates(const std::string& content_name,
 659                                      const Candidates& candidates,
 660                                      std::string* error) {
 661   // Give candidates to the appropriate transport, and tell that transport
 662   // to start connecting, if it's not already doing so.
 663   TransportProxy* transproxy = GetTransportProxy(content_name);
 664   if (!transproxy) {
 665     *error = "Unknown content name " + content_name;
 666     return false;
 667   }
 668   if (!transproxy->OnRemoteCandidates(candidates, error)) {
 669     return false;
 670   }
 671   // TODO(juberti): Remove this call once we can be sure that we always have
 672   // a local transport description (which will trigger the connection).
 673   transproxy->ConnectChannels();
 674   return true;
 675 }
 676
 677 bool BaseSession::MaybeEnableMuxingSupport() {
 678   // We need both a local and remote description to decide if we should mux.
 679   if ((state_ == STATE_SENTINITIATE ||
 680       state_ == STATE_RECEIVEDINITIATE) &&
 681       ((local_description_ == NULL) ||
 682       (remote_description_ == NULL))) {
 683     return false;
 684   }
 685
 686   // In order to perform the multiplexing, we need all proxies to be in the
 687   // negotiated state, i.e. to have implementations underneath.
 688   // Ensure that this is the case, regardless of whether we are going to mux.
 689   for (TransportMap::iterator iter = transports_.begin();
 690        iter != transports_.end(); ++iter) {
 691     ASSERT(iter->second->negotiated());
 692     if (!iter->second->negotiated())
 693       return false;
 694   }
 695
 696   // If both sides agree to BUNDLE, mux all the specified contents onto the
 697   // transport belonging to the first content name in the BUNDLE group.
 698   // If the contents are already muxed, this will be a no-op.
 699   // TODO(juberti): Should this check that local and remote have configured
 700   // BUNDLE the same way?
 701   bool candidates_allocated = IsCandidateAllocationDone();
 702   const ContentGroup* local_bundle_group =
 703       local_description()->GetGroupByName(GROUP_TYPE_BUNDLE);
 704   const ContentGroup* remote_bundle_group =
 705       remote_description()->GetGroupByName(GROUP_TYPE_BUNDLE);
 706   if (local_bundle_group && remote_bundle_group &&
 707       local_bundle_group->FirstContentName()) {
 708     const std::string* content_name = local_bundle_group->FirstContentName();
 709     const ContentInfo* content =
 710         local_description_->GetContentByName(*content_name);
 711     ASSERT(content != NULL);
 712     if (!SetSelectedProxy(content->name, local_bundle_group)) {
 713       LOG(LS_WARNING) << "Failed to set up BUNDLE";
 714       return false;
 715     }
 716
 717     // If we weren't done gathering before, we might be done now, as a result
 718     // of enabling mux.
 719     LOG(LS_INFO) << "Enabling BUNDLE, bundling onto transport: "
 720                  << *content_name;
 721     if (!candidates_allocated) {
 722       MaybeCandidateAllocationDone();
 723     }
 724   } else {
 725     LOG(LS_INFO) << "No BUNDLE information, not bundling.";
 726   }
 727   return true;
 728 }
 729
 730 bool BaseSession::SetSelectedProxy(const std::string& content_name,
 731                                    const ContentGroup* muxed_group) {
 732   TransportProxy* selected_proxy = GetTransportProxy(content_name);
 733   if (!selected_proxy) {
 734     return false;
 735   }
 736
 737   ASSERT(selected_proxy->negotiated());
 738   for (TransportMap::iterator iter = transports_.begin();
 739        iter != transports_.end(); ++iter) {
 740     // If content is part of the mux group, then repoint its proxy at the
 741     // transport object that we have chosen to mux onto. If the proxy
 742     // is already pointing at the right object, it will be a no-op.
 743     if (muxed_group->HasContentName(iter->first) &&
 744         !iter->second->SetupMux(selected_proxy)) {
 745       return false;
 746     }
 747   }
 748   return true;
 749 }
 750
 751 void BaseSession::OnTransportCandidatesAllocationDone(Transport* transport) {
 752   // TODO(juberti): This is a clunky way of processing the done signal. Instead,
 753   // TransportProxy should receive the done signal directly, set its allocated
 754   // flag internally, and then reissue the done signal to Session.
 755   // Overall we should make TransportProxy receive *all* the signals from
 756   // Transport, since this removes the need to manually iterate over all
 757   // the transports, as is needed to make sure signals are handled properly
 758   // when BUNDLEing.
 759 #if 0
 760   ASSERT(!IsCandidateAllocationDone());
 761 #endif
 762   for (TransportMap::iterator iter = transports_.begin();
 763        iter != transports_.end(); ++iter) {
 764     if (iter->second->impl() == transport) {
 765       iter->second->set_candidates_allocated(true);
 766     }
 767   }
 768   MaybeCandidateAllocationDone();
 769 }
 770
 771 bool BaseSession::IsCandidateAllocationDone() const {
 772   for (TransportMap::const_iterator iter = transports_.begin();
 773        iter != transports_.end(); ++iter) {
 774     if (!iter->second->candidates_allocated())
 775       return false;
 776   }
 777   return true;
 778 }
 779
 780 void BaseSession::MaybeCandidateAllocationDone() {
 781   if (IsCandidateAllocationDone()) {
 782     LOG(LS_INFO) << "Candidate gathering is complete.";
 783     OnCandidatesAllocationDone();
 784   }
 785 }
 786
 787 void BaseSession::OnRoleConflict() {
 788   if (role_switch_) {
 789     LOG(LS_WARNING) << "Repeat of role conflict signal from Transport.";
 790     return;
 791   }
 792
 793   role_switch_ = true;
 794   for (TransportMap::iterator iter = transports_.begin();
 795        iter != transports_.end(); ++iter) {
 796     // Role will be reverse of initial role setting.
 797     IceRole role = initiator_ ? ICEROLE_CONTROLLED : ICEROLE_CONTROLLING;
 798     iter->second->SetIceRole(role);
 799   }
 800 }
 801
 802 void BaseSession::LogState(State old_state, State new_state) {
 803   LOG(LS_INFO) << "Session:" << id()
 804                << " Old state:" << StateToString(old_state)
 805                << " New state:" << StateToString(new_state)
 806                << " Type:" << content_type()
 807                << " Transport:" << transport_type();
 808 }
 809
 810 bool BaseSession::GetTransportDescription(const SessionDescription* description,
 811                                           const std::string& content_name,
 812                                           TransportDescription* tdesc) {
 813   if (!description || !tdesc) {
 814     return false;
 815   }
 816   const TransportInfo* transport_info =
 817       description->GetTransportInfoByName(content_name);
 818   if (!transport_info) {
 819     return false;
 820   }
 821   *tdesc = transport_info->description;
 822   return true;
 823 }
 824
 825 void BaseSession::SignalNewDescription() {
 826   ContentAction action;
 827   ContentSource source;
 828   if (!GetContentAction(&action, &source)) {
 829     return;
 830   }
 831   if (source == CS_LOCAL) {
 832     SignalNewLocalDescription(this, action);
 833   } else {
 834     SignalNewRemoteDescription(this, action);
 835   }
 836 }
 837
 838 bool BaseSession::GetContentAction(ContentAction* action,
 839                                    ContentSource* source) {
 840   switch (state_) {
 841     // new local description
 842     case STATE_SENTINITIATE:
 843       *action = CA_OFFER;
 844       *source = CS_LOCAL;
 845       break;
 846     case STATE_SENTPRACCEPT:
 847       *action = CA_PRANSWER;
 848       *source = CS_LOCAL;
 849       break;
 850     case STATE_SENTACCEPT:
 851       *action = CA_ANSWER;
 852       *source = CS_LOCAL;
 853       break;
 854     // new remote description
 855     case STATE_RECEIVEDINITIATE:
 856       *action = CA_OFFER;
 857       *source = CS_REMOTE;
 858       break;
 859     case STATE_RECEIVEDPRACCEPT:
 860       *action = CA_PRANSWER;
 861       *source = CS_REMOTE;
 862       break;
 863     case STATE_RECEIVEDACCEPT:
 864       *action = CA_ANSWER;
 865       *source = CS_REMOTE;
 866       break;
 867     default:
 868       return false;
 869   }
 870   return true;
 871 }
 872
 873 void BaseSession::OnMessage(talk_base::Message *pmsg) {
 874   switch (pmsg->message_id) {
 875   case MSG_TIMEOUT:
 876     // Session timeout has occured.
 877     SetError(ERROR_TIME, "Session timeout has occured.");
 878     break;
 879
 880   case MSG_STATE:
 881     switch (state_) {
 882     case STATE_SENTACCEPT:
 883     case STATE_RECEIVEDACCEPT:
 884       SetState(STATE_INPROGRESS);
 885       break;
 886
 887     default:
 888       // Explicitly ignoring some states here.
 889       break;
 890     }
 891     break;
 892   }
 893 }
 894
 895 Session::Session(SessionManager* session_manager,
 896                  const std::string& local_name,
 897                  const std::string& initiator_name,
 898                  const std::string& sid,
 899                  const std::string& content_type,
 900                  SessionClient* client)
 901     : BaseSession(session_manager->signaling_thread(),
 902                   session_manager->worker_thread(),
 903                   session_manager->port_allocator(),
 904                   sid, content_type, initiator_name == local_name) {
 905   ASSERT(client != NULL);
 906   session_manager_ = session_manager;
 907   local_name_ = local_name;
 908   initiator_name_ = initiator_name;
 909   transport_parser_ = new P2PTransportParser();
 910   client_ = client;
 911   initiate_acked_ = false;
 912   current_protocol_ = PROTOCOL_HYBRID;
 913 }
 914
 915 Session::~Session() {
 916   delete transport_parser_;
 917 }
 918
 919 bool Session::Initiate(const std::string &to,
 920                        const SessionDescription* sdesc) {
 921   ASSERT(signaling_thread()->IsCurrent());
 922   SessionError error;
 923
 924   // Only from STATE_INIT
 925   if (state() != STATE_INIT)
 926     return false;
 927
 928   // Setup for signaling.
 929   set_remote_name(to);
 930   set_local_description(sdesc);
 931   if (!CreateTransportProxies(GetEmptyTransportInfos(sdesc->contents()),
 932                               &error)) {
 933     LOG(LS_ERROR) << "Could not create transports: " << error.text;
 934     return false;
 935   }
 936
 937   if (!SendInitiateMessage(sdesc, &error)) {
 938     LOG(LS_ERROR) << "Could not send initiate message: " << error.text;
 939     return false;
 940   }
 941
 942   // We need to connect transport proxy and impl here so that we can process
 943   // the TransportDescriptions.
 944   SpeculativelyConnectAllTransportChannels();
 945
 946   PushdownTransportDescription(CS_LOCAL, CA_OFFER, NULL);
 947   SetState(Session::STATE_SENTINITIATE);
 948   return true;
 949 }
 950
 951 bool Session::Accept(const SessionDescription* sdesc) {
 952   ASSERT(signaling_thread()->IsCurrent());
 953
 954   // Only if just received initiate
 955   if (state() != STATE_RECEIVEDINITIATE)
 956     return false;
 957
 958   // Setup for signaling.
 959   set_local_description(sdesc);
 960
 961   SessionError error;
 962   if (!SendAcceptMessage(sdesc, &error)) {
 963     LOG(LS_ERROR) << "Could not send accept message: " << error.text;
 964     return false;
 965   }
 966   // TODO(juberti): Add BUNDLE support to transport-info messages.
 967   PushdownTransportDescription(CS_LOCAL, CA_ANSWER, NULL);
 968   MaybeEnableMuxingSupport();  // Enable transport channel mux if supported.
 969   SetState(Session::STATE_SENTACCEPT);
 970   return true;
 971 }
 972
 973 bool Session::Reject(const std::string& reason) {
 974   ASSERT(signaling_thread()->IsCurrent());
 975
 976   // Reject is sent in response to an initiate or modify, to reject the
 977   // request
 978   if (state() != STATE_RECEIVEDINITIATE && state() != STATE_RECEIVEDMODIFY)
 979     return false;
 980
 981   SessionError error;
 982   if (!SendRejectMessage(reason, &error)) {
 983     LOG(LS_ERROR) << "Could not send reject message: " << error.text;
 984     return false;
 985   }
 986
 987   SetState(STATE_SENTREJECT);
 988   return true;
 989 }
 990
 991 bool Session::TerminateWithReason(const std::string& reason) {
 992   ASSERT(signaling_thread()->IsCurrent());
 993
 994   // Either side can terminate, at any time.
 995   switch (state()) {
 996     case STATE_SENTTERMINATE:
 997     case STATE_RECEIVEDTERMINATE:
 998       return false;
 999
1000     case STATE_SENTREJECT:
1001     case STATE_RECEIVEDREJECT:
1002       // We don't need to send terminate if we sent or received a reject...
1003       // it's implicit.
1004       break;
1005
1006     default:
1007       SessionError error;
1008       if (!SendTerminateMessage(reason, &error)) {
1009         LOG(LS_ERROR) << "Could not send terminate message: " << error.text;
1010         return false;
1011       }
1012       break;
1013   }
1014
1015   SetState(STATE_SENTTERMINATE);
1016   return true;
1017 }
1018
1019 bool Session::SendInfoMessage(const XmlElements& elems,
1020                               const std::string& remote_name) {
1021   ASSERT(signaling_thread()->IsCurrent());
1022   SessionError error;
1023   if (!SendMessage(ACTION_SESSION_INFO, elems, remote_name, &error)) {
1024     LOG(LS_ERROR) << "Could not send info message " << error.text;
1025     return false;
1026   }
1027   return true;
1028 }
1029
1030 bool Session::SendDescriptionInfoMessage(const ContentInfos& contents) {
1031   XmlElements elems;
1032   WriteError write_error;
1033   if (!WriteDescriptionInfo(current_protocol_,
1034                             contents,
1035                             GetContentParsers(),
1036                             &elems, &write_error)) {
1037     LOG(LS_ERROR) << "Could not write description info message: "
1038                   << write_error.text;
1039     return false;
1040   }
1041   SessionError error;
1042   if (!SendMessage(ACTION_DESCRIPTION_INFO, elems, &error)) {
1043     LOG(LS_ERROR) << "Could not send description info message: "
1044                   << error.text;
1045     return false;
1046   }
1047   return true;
1048 }
1049
1050 TransportInfos Session::GetEmptyTransportInfos(
1051     const ContentInfos& contents) const {
1052   TransportInfos tinfos;
1053   for (ContentInfos::const_iterator content = contents.begin();
1054        content != contents.end(); ++content) {
1055     tinfos.push_back(TransportInfo(content->name,
1056                                    TransportDescription(transport_type(),
1057                                                         std::string(),
1058                                                         std::string())));
1059   }
1060   return tinfos;
1061 }
1062
1063 bool Session::OnRemoteCandidates(
1064     const TransportInfos& tinfos, ParseError* error) {
1065   for (TransportInfos::const_iterator tinfo = tinfos.begin();
1066        tinfo != tinfos.end(); ++tinfo) {
1067     std::string str_error;
1068     if (!BaseSession::OnRemoteCandidates(
1069         tinfo->content_name, tinfo->description.candidates, &str_error)) {
1070       return BadParse(str_error, error);
1071     }
1072   }
1073   return true;
1074 }
1075
1076 bool Session::CreateTransportProxies(const TransportInfos& tinfos,
1077                                      SessionError* error) {
1078   for (TransportInfos::const_iterator tinfo = tinfos.begin();
1079        tinfo != tinfos.end(); ++tinfo) {
1080     if (tinfo->description.transport_type != transport_type()) {
1081       error->SetText("No supported transport in offer.");
1082       return false;
1083     }
1084
1085     GetOrCreateTransportProxy(tinfo->content_name);
1086   }
1087   return true;
1088 }
1089
1090 TransportParserMap Session::GetTransportParsers() {
1091   TransportParserMap parsers;
1092   parsers[transport_type()] = transport_parser_;
1093   return parsers;
1094 }
1095
1096 CandidateTranslatorMap Session::GetCandidateTranslators() {
1097   CandidateTranslatorMap translators;
1098   // NOTE: This technique makes it impossible to parse G-ICE
1099   // candidates in session-initiate messages because the channels
1100   // aren't yet created at that point.  Since we don't use candidates
1101   // in session-initiate messages, we should be OK.  Once we switch to
1102   // ICE, this translation shouldn't be necessary.
1103   for (TransportMap::const_iterator iter = transport_proxies().begin();
1104        iter != transport_proxies().end(); ++iter) {
1105     translators[iter->first] = iter->second;
1106   }
1107   return translators;
1108 }
1109
1110 ContentParserMap Session::GetContentParsers() {
1111   ContentParserMap parsers;
1112   parsers[content_type()] = client_;
1113   // We need to be able parse both RTP-based and SCTP-based Jingle
1114   // with the same client.
1115   if (content_type() == NS_JINGLE_RTP) {
1116     parsers[NS_JINGLE_DRAFT_SCTP] = client_;
1117   }
1118   return parsers;
1119 }
1120
1121 void Session::OnTransportRequestSignaling(Transport* transport) {
1122   ASSERT(signaling_thread()->IsCurrent());
1123   TransportProxy* transproxy = GetTransportProxy(transport);
1124   ASSERT(transproxy != NULL);
1125   if (transproxy) {
1126     // Reset candidate allocation status for the transport proxy.
1127     transproxy->set_candidates_allocated(false);
1128   }
1129   SignalRequestSignaling(this);
1130 }
1131
1132 void Session::OnTransportConnecting(Transport* transport) {
1133   // This is an indication that we should begin watching the writability
1134   // state of the transport.
1135   OnTransportWritable(transport);
1136 }
1137
1138 void Session::OnTransportWritable(Transport* transport) {
1139   ASSERT(signaling_thread()->IsCurrent());
1140
1141   // If the transport is not writable, start a timer to make sure that it
1142   // becomes writable within a reasonable amount of time.  If it does not, we
1143   // terminate since we can't actually send data.  If the transport is writable,
1144   // cancel the timer.  Note that writability transitions may occur repeatedly
1145   // during the lifetime of the session.
1146   signaling_thread()->Clear(this, MSG_TIMEOUT);
1147   if (transport->HasChannels() && !transport->writable()) {
1148     signaling_thread()->PostDelayed(
1149         session_manager_->session_timeout() * 1000, this, MSG_TIMEOUT);
1150   }
1151 }
1152
1153 void Session::OnTransportProxyCandidatesReady(TransportProxy* transproxy,
1154                                               const Candidates& candidates) {
1155   ASSERT(signaling_thread()->IsCurrent());
1156   if (transproxy != NULL) {
1157     if (initiator() && !initiate_acked_) {
1158       // TODO: This is to work around server re-ordering
1159       // messages.  We send the candidates once the session-initiate
1160       // is acked.  Once we have fixed the server to guarantee message
1161       // order, we can remove this case.
1162       transproxy->AddUnsentCandidates(candidates);
1163     } else {
1164       if (!transproxy->negotiated()) {
1165         transproxy->AddSentCandidates(candidates);
1166       }
1167       SessionError error;
1168       if (!SendTransportInfoMessage(transproxy, candidates, &error)) {
1169         LOG(LS_ERROR) << "Could not send transport info message: "
1170                       << error.text;
1171         return;
1172       }
1173     }
1174   }
1175 }
1176
1177 void Session::OnTransportSendError(Transport* transport,
1178                                    const buzz::XmlElement* stanza,
1179                                    const buzz::QName& name,
1180                                    const std::string& type,
1181                                    const std::string& text,
1182                                    const buzz::XmlElement* extra_info) {
1183   ASSERT(signaling_thread()->IsCurrent());
1184   SignalErrorMessage(this, stanza, name, type, text, extra_info);
1185 }
1186
1187 void Session::OnIncomingMessage(const SessionMessage& msg) {
1188   ASSERT(signaling_thread()->IsCurrent());
1189   ASSERT(state() == STATE_INIT || msg.from == remote_name());
1190
1191   if (current_protocol_== PROTOCOL_HYBRID) {
1192     if (msg.protocol == PROTOCOL_GINGLE) {
1193       current_protocol_ = PROTOCOL_GINGLE;
1194     } else {
1195       current_protocol_ = PROTOCOL_JINGLE;
1196     }
1197   }
1198
1199   bool valid = false;
1200   MessageError error;
1201   switch (msg.type) {
1202     case ACTION_SESSION_INITIATE:
1203       valid = OnInitiateMessage(msg, &error);
1204       break;
1205     case ACTION_SESSION_INFO:
1206       valid = OnInfoMessage(msg);
1207       break;
1208     case ACTION_SESSION_ACCEPT:
1209       valid = OnAcceptMessage(msg, &error);
1210       break;
1211     case ACTION_SESSION_REJECT:
1212       valid = OnRejectMessage(msg, &error);
1213       break;
1214     case ACTION_SESSION_TERMINATE:
1215       valid = OnTerminateMessage(msg, &error);
1216       break;
1217     case ACTION_TRANSPORT_INFO:
1218       valid = OnTransportInfoMessage(msg, &error);
1219       break;
1220     case ACTION_TRANSPORT_ACCEPT:
1221       valid = OnTransportAcceptMessage(msg, &error);
1222       break;
1223     case ACTION_DESCRIPTION_INFO:
1224       valid = OnDescriptionInfoMessage(msg, &error);
1225       break;
1226     default:
1227       valid = BadMessage(buzz::QN_STANZA_BAD_REQUEST,
1228                          "unknown session message type",
1229                          &error);
1230   }
1231
1232   if (valid) {
1233     SendAcknowledgementMessage(msg.stanza);
1234   } else {
1235     SignalErrorMessage(this, msg.stanza, error.type,
1236                        "modify", error.text, NULL);
1237   }
1238 }
1239
1240 void Session::OnIncomingResponse(const buzz::XmlElement* orig_stanza,
1241                                  const buzz::XmlElement* response_stanza,
1242                                  const SessionMessage& msg) {
1243   ASSERT(signaling_thread()->IsCurrent());
1244
1245   if (msg.type == ACTION_SESSION_INITIATE) {
1246     OnInitiateAcked();
1247   }
1248 }
1249
1250 void Session::OnInitiateAcked() {
1251     // TODO: This is to work around server re-ordering
1252     // messages.  We send the candidates once the session-initiate
1253     // is acked.  Once we have fixed the server to guarantee message
1254     // order, we can remove this case.
1255   if (!initiate_acked_) {
1256     initiate_acked_ = true;
1257     SessionError error;
1258     SendAllUnsentTransportInfoMessages(&error);
1259   }
1260 }
1261
1262 void Session::OnFailedSend(const buzz::XmlElement* orig_stanza,
1263                            const buzz::XmlElement* error_stanza) {
1264   ASSERT(signaling_thread()->IsCurrent());
1265
1266   SessionMessage msg;
1267   ParseError parse_error;
1268   if (!ParseSessionMessage(orig_stanza, &msg, &parse_error)) {
1269     LOG(LS_ERROR) << "Error parsing failed send: " << parse_error.text
1270                   << ":" << orig_stanza;
1271     return;
1272   }
1273
1274   // If the error is a session redirect, call OnRedirectError, which will
1275   // continue the session with a new remote JID.
1276   SessionRedirect redirect;
1277   if (FindSessionRedirect(error_stanza, &redirect)) {
1278     SessionError error;
1279     if (!OnRedirectError(redirect, &error)) {
1280       // TODO: Should we send a message back?  The standard
1281       // says nothing about it.
1282       std::ostringstream desc;
1283       desc << "Failed to redirect: " << error.text;
1284       LOG(LS_ERROR) << desc.str();
1285       SetError(ERROR_RESPONSE, desc.str());
1286     }
1287     return;
1288   }
1289
1290   std::string error_type = "cancel";
1291
1292   const buzz::XmlElement* error = error_stanza->FirstNamed(buzz::QN_ERROR);
1293   if (error) {
1294     error_type = error->Attr(buzz::QN_TYPE);
1295
1296     LOG(LS_ERROR) << "Session error:\n" << error->Str() << "\n"
1297                   << "in response to:\n" << orig_stanza->Str();
1298   } else {
1299     // don't crash if <error> is missing
1300     LOG(LS_ERROR) << "Session error without <error/> element, ignoring";
1301     return;
1302   }
1303
1304   if (msg.type == ACTION_TRANSPORT_INFO) {
1305     // Transport messages frequently generate errors because they are sent right
1306     // when we detect a network failure.  For that reason, we ignore such
1307     // errors, because if we do not establish writability again, we will
1308     // terminate anyway.  The exceptions are transport-specific error tags,
1309     // which we pass on to the respective transport.
1310   } else if ((error_type != "continue") && (error_type != "wait")) {
1311     // We do not set an error if the other side said it is okay to continue
1312     // (possibly after waiting).  These errors can be ignored.
1313     SetError(ERROR_RESPONSE, "");
1314   }
1315 }
1316
1317 bool Session::OnInitiateMessage(const SessionMessage& msg,
1318                                 MessageError* error) {
1319   if (!CheckState(STATE_INIT, error))
1320     return false;
1321
1322   SessionInitiate init;
1323   if (!ParseSessionInitiate(msg.protocol, msg.action_elem,
1324                             GetContentParsers(), GetTransportParsers(),
1325                             GetCandidateTranslators(),
1326                             &init, error))
1327     return false;
1328
1329   SessionError session_error;
1330   if (!CreateTransportProxies(init.transports, &session_error)) {
1331     return BadMessage(buzz::QN_STANZA_NOT_ACCEPTABLE,
1332                       session_error.text, error);
1333   }
1334
1335   set_remote_name(msg.from);
1336   set_initiator_name(msg.initiator);
1337   set_remote_description(new SessionDescription(init.ClearContents(),
1338                                                 init.transports,
1339                                                 init.groups));
1340   // Updating transport with TransportDescription.
1341   PushdownTransportDescription(CS_REMOTE, CA_OFFER, NULL);
1342   SetState(STATE_RECEIVEDINITIATE);
1343
1344   // Users of Session may listen to state change and call Reject().
1345   if (state() != STATE_SENTREJECT) {
1346     if (!OnRemoteCandidates(init.transports, error))
1347       return false;
1348
1349     // TODO(juberti): Auto-generate and push down the local transport answer.
1350     // This is necessary for trickling to work with RFC 5245 ICE.
1351   }
1352   return true;
1353 }
1354
1355 bool Session::OnAcceptMessage(const SessionMessage& msg, MessageError* error) {
1356   if (!CheckState(STATE_SENTINITIATE, error))
1357     return false;
1358
1359   SessionAccept accept;
1360   if (!ParseSessionAccept(msg.protocol, msg.action_elem,
1361                           GetContentParsers(), GetTransportParsers(),
1362                           GetCandidateTranslators(),
1363                           &accept, error)) {
1364     return false;
1365   }
1366
1367   // If we get an accept, we can assume the initiate has been
1368   // received, even if we haven't gotten an IQ response.
1369   OnInitiateAcked();
1370
1371   set_remote_description(new SessionDescription(accept.ClearContents(),
1372                                                 accept.transports,
1373                                                 accept.groups));
1374   // Updating transport with TransportDescription.
1375   PushdownTransportDescription(CS_REMOTE, CA_ANSWER, NULL);
1376   MaybeEnableMuxingSupport();  // Enable transport channel mux if supported.
1377   SetState(STATE_RECEIVEDACCEPT);
1378
1379   if (!OnRemoteCandidates(accept.transports, error))
1380     return false;
1381
1382   return true;
1383 }
1384
1385 bool Session::OnRejectMessage(const SessionMessage& msg, MessageError* error) {
1386   if (!CheckState(STATE_SENTINITIATE, error))
1387     return false;
1388
1389   SetState(STATE_RECEIVEDREJECT);
1390   return true;
1391 }
1392
1393 bool Session::OnInfoMessage(const SessionMessage& msg) {
1394   SignalInfoMessage(this, msg.action_elem);
1395   return true;
1396 }
1397
1398 bool Session::OnTerminateMessage(const SessionMessage& msg,
1399                                  MessageError* error) {
1400   SessionTerminate term;
1401   if (!ParseSessionTerminate(msg.protocol, msg.action_elem, &term, error))
1402     return false;
1403
1404   SignalReceivedTerminateReason(this, term.reason);
1405   if (term.debug_reason != buzz::STR_EMPTY) {
1406     LOG(LS_VERBOSE) << "Received error on call: " << term.debug_reason;
1407   }
1408
1409   SetState(STATE_RECEIVEDTERMINATE);
1410   return true;
1411 }
1412
1413 bool Session::OnTransportInfoMessage(const SessionMessage& msg,
1414                                      MessageError* error) {
1415   TransportInfos tinfos;
1416   if (!ParseTransportInfos(msg.protocol, msg.action_elem,
1417                            initiator_description()->contents(),
1418                            GetTransportParsers(), GetCandidateTranslators(),
1419                            &tinfos, error))
1420     return false;
1421
1422   if (!OnRemoteCandidates(tinfos, error))
1423     return false;
1424
1425   return true;
1426 }
1427
1428 bool Session::OnTransportAcceptMessage(const SessionMessage& msg,
1429                                        MessageError* error) {
1430   // TODO: Currently here only for compatibility with
1431   // Gingle 1.1 clients (notably, Google Voice).
1432   return true;
1433 }
1434
1435 bool Session::OnDescriptionInfoMessage(const SessionMessage& msg,
1436                               MessageError* error) {
1437   if (!CheckState(STATE_INPROGRESS, error))
1438     return false;
1439
1440   DescriptionInfo description_info;
1441   if (!ParseDescriptionInfo(msg.protocol, msg.action_elem,
1442                             GetContentParsers(), GetTransportParsers(),
1443                             GetCandidateTranslators(),
1444                             &description_info, error)) {
1445     return false;
1446   }
1447
1448   ContentInfos& updated_contents = description_info.contents;
1449
1450   // TODO: Currently, reflector sends back
1451   // video stream updates even for an audio-only call, which causes
1452   // this to fail.  Put this back once reflector is fixed.
1453   //
1454   // ContentInfos::iterator it;
1455   // First, ensure all updates are valid before modifying remote_description_.
1456   // for (it = updated_contents.begin(); it != updated_contents.end(); ++it) {
1457   //   if (remote_description()->GetContentByName(it->name) == NULL) {
1458   //     return false;
1459   //   }
1460   // }
1461
1462   // TODO: We used to replace contents from an update, but
1463   // that no longer works with partial updates.  We need to figure out
1464   // a way to merge patial updates into contents.  For now, users of
1465   // Session should listen to SignalRemoteDescriptionUpdate and handle
1466   // updates.  They should not expect remote_description to be the
1467   // latest value.
1468   //
1469   // for (it = updated_contents.begin(); it != updated_contents.end(); ++it) {
1470   //     remote_description()->RemoveContentByName(it->name);
1471   //     remote_description()->AddContent(it->name, it->type, it->description);
1472   //   }
1473   // }
1474
1475   SignalRemoteDescriptionUpdate(this, updated_contents);
1476   return true;
1477 }
1478
1479 bool BareJidsEqual(const std::string& name1,
1480                    const std::string& name2) {
1481   buzz::Jid jid1(name1);
1482   buzz::Jid jid2(name2);
1483
1484   return jid1.IsValid() && jid2.IsValid() && jid1.BareEquals(jid2);
1485 }
1486
1487 bool Session::OnRedirectError(const SessionRedirect& redirect,
1488                               SessionError* error) {
1489   MessageError message_error;
1490   if (!CheckState(STATE_SENTINITIATE, &message_error)) {
1491     return BadWrite(message_error.text, error);
1492   }
1493
1494   if (!BareJidsEqual(remote_name(), redirect.target))
1495     return BadWrite("Redirection not allowed: must be the same bare jid.",
1496                     error);
1497
1498   // When we receive a redirect, we point the session at the new JID
1499   // and resend the candidates.
1500   set_remote_name(redirect.target);
1501   return (SendInitiateMessage(local_description(), error) &&
1502           ResendAllTransportInfoMessages(error));
1503 }
1504
1505 bool Session::CheckState(State expected, MessageError* error) {
1506   if (state() != expected) {
1507     // The server can deliver messages out of order/repeated for various
1508     // reasons. For example, if the server does not recive our iq response,
1509     // it could assume that the iq it sent was lost, and will then send
1510     // it again. Ideally, we should implement reliable messaging with
1511     // duplicate elimination.
1512     return BadMessage(buzz::QN_STANZA_NOT_ALLOWED,
1513                       "message not allowed in current state",
1514                       error);
1515   }
1516   return true;
1517 }
1518
1519 void Session::SetError(Error error, const std::string& error_desc) {
1520   BaseSession::SetError(error, error_desc);
1521   if (error != ERROR_NONE)
1522     signaling_thread()->Post(this, MSG_ERROR);
1523 }
1524
1525 void Session::OnMessage(talk_base::Message* pmsg) {
1526   // preserve this because BaseSession::OnMessage may modify it
1527   State orig_state = state();
1528
1529   BaseSession::OnMessage(pmsg);
1530
1531   switch (pmsg->message_id) {
1532   case MSG_ERROR:
1533     TerminateWithReason(STR_TERMINATE_ERROR);
1534     break;
1535
1536   case MSG_STATE:
1537     switch (orig_state) {
1538     case STATE_SENTREJECT:
1539     case STATE_RECEIVEDREJECT:
1540       // Assume clean termination.
1541       Terminate();
1542       break;
1543
1544     case STATE_SENTTERMINATE:
1545     case STATE_RECEIVEDTERMINATE:
1546       session_manager_->DestroySession(this);
1547       break;
1548
1549     default:
1550       // Explicitly ignoring some states here.
1551       break;
1552     }
1553     break;
1554   }
1555 }
1556
1557 bool Session::SendInitiateMessage(const SessionDescription* sdesc,
1558                                   SessionError* error) {
1559   SessionInitiate init;
1560   init.contents = sdesc->contents();
1561   init.transports = GetEmptyTransportInfos(init.contents);
1562   init.groups = sdesc->groups();
1563   return SendMessage(ACTION_SESSION_INITIATE, init, error);
1564 }
1565
1566 bool Session::WriteSessionAction(
1567     SignalingProtocol protocol, const SessionInitiate& init,
1568     XmlElements* elems, WriteError* error) {
1569   return WriteSessionInitiate(protocol, init.contents, init.transports,
1570                               GetContentParsers(), GetTransportParsers(),
1571                               GetCandidateTranslators(), init.groups,
1572                               elems, error);
1573 }
1574
1575 bool Session::SendAcceptMessage(const SessionDescription* sdesc,
1576                                 SessionError* error) {
1577   XmlElements elems;
1578   if (!WriteSessionAccept(current_protocol_,
1579                           sdesc->contents(),
1580                           GetEmptyTransportInfos(sdesc->contents()),
1581                           GetContentParsers(), GetTransportParsers(),
1582                           GetCandidateTranslators(), sdesc->groups(),
1583                           &elems, error)) {
1584     return false;
1585   }
1586   return SendMessage(ACTION_SESSION_ACCEPT, elems, error);
1587 }
1588
1589 bool Session::SendRejectMessage(const std::string& reason,
1590                                 SessionError* error) {
1591   SessionTerminate term(reason);
1592   return SendMessage(ACTION_SESSION_REJECT, term, error);
1593 }
1594
1595 bool Session::SendTerminateMessage(const std::string& reason,
1596                                    SessionError* error) {
1597   SessionTerminate term(reason);
1598   return SendMessage(ACTION_SESSION_TERMINATE, term, error);
1599 }
1600
1601 bool Session::WriteSessionAction(SignalingProtocol protocol,
1602                                  const SessionTerminate& term,
1603                                  XmlElements* elems, WriteError* error) {
1604   WriteSessionTerminate(protocol, term, elems);
1605   return true;
1606 }
1607
1608 bool Session::SendTransportInfoMessage(const TransportInfo& tinfo,
1609                                        SessionError* error) {
1610   return SendMessage(ACTION_TRANSPORT_INFO, tinfo, error);
1611 }
1612
1613 bool Session::SendTransportInfoMessage(const TransportProxy* transproxy,
1614                                        const Candidates& candidates,
1615                                        SessionError* error) {
1616   return SendTransportInfoMessage(TransportInfo(transproxy->content_name(),
1617       TransportDescription(transproxy->type(), std::vector<std::string>(),
1618                            std::string(), std::string(), ICEMODE_FULL,
1619                            CONNECTIONROLE_NONE, NULL, candidates)), error);
1620 }
1621
1622 bool Session::WriteSessionAction(SignalingProtocol protocol,
1623                                  const TransportInfo& tinfo,
1624                                  XmlElements* elems, WriteError* error) {
1625   TransportInfos tinfos;
1626   tinfos.push_back(tinfo);
1627   return WriteTransportInfos(protocol, tinfos,
1628                              GetTransportParsers(), GetCandidateTranslators(),
1629                              elems, error);
1630 }
1631
1632 bool Session::ResendAllTransportInfoMessages(SessionError* error) {
1633   for (TransportMap::const_iterator iter = transport_proxies().begin();
1634        iter != transport_proxies().end(); ++iter) {
1635     TransportProxy* transproxy = iter->second;
1636     if (transproxy->sent_candidates().size() > 0) {
1637       if (!SendTransportInfoMessage(
1638               transproxy, transproxy->sent_candidates(), error)) {
1639         LOG(LS_ERROR) << "Could not resend transport info messages: "
1640                       << error->text;
1641         return false;
1642       }
1643       transproxy->ClearSentCandidates();
1644     }
1645   }
1646   return true;
1647 }
1648
1649 bool Session::SendAllUnsentTransportInfoMessages(SessionError* error) {
1650   for (TransportMap::const_iterator iter = transport_proxies().begin();
1651        iter != transport_proxies().end(); ++iter) {
1652     TransportProxy* transproxy = iter->second;
1653     if (transproxy->unsent_candidates().size() > 0) {
1654       if (!SendTransportInfoMessage(
1655               transproxy, transproxy->unsent_candidates(), error)) {
1656         LOG(LS_ERROR) << "Could not send unsent transport info messages: "
1657                       << error->text;
1658         return false;
1659       }
1660       transproxy->ClearUnsentCandidates();
1661     }
1662   }
1663   return true;
1664 }
1665
1666 bool Session::SendMessage(ActionType type, const XmlElements& action_elems,
1667                           SessionError* error) {
1668     return SendMessage(type, action_elems, remote_name(), error);
1669 }
1670
1671 bool Session::SendMessage(ActionType type, const XmlElements& action_elems,
1672                           const std::string& remote_name, SessionError* error) {
1673   talk_base::scoped_ptr<buzz::XmlElement> stanza(
1674       new buzz::XmlElement(buzz::QN_IQ));
1675
1676   SessionMessage msg(current_protocol_, type, id(), initiator_name());
1677   msg.to = remote_name;
1678   WriteSessionMessage(msg, action_elems, stanza.get());
1679
1680   SignalOutgoingMessage(this, stanza.get());
1681   return true;
1682 }
1683
1684 template <typename Action>
1685 bool Session::SendMessage(ActionType type, const Action& action,
1686                           SessionError* error) {
1687   talk_base::scoped_ptr<buzz::XmlElement> stanza(
1688       new buzz::XmlElement(buzz::QN_IQ));
1689   if (!WriteActionMessage(type, action, stanza.get(), error))
1690     return false;
1691
1692   SignalOutgoingMessage(this, stanza.get());
1693   return true;
1694 }
1695
1696 template <typename Action>
1697 bool Session::WriteActionMessage(ActionType type, const Action& action,
1698                                  buzz::XmlElement* stanza,
1699                                  WriteError* error) {
1700   if (current_protocol_ == PROTOCOL_HYBRID) {
1701     if (!WriteActionMessage(PROTOCOL_JINGLE, type, action, stanza, error))
1702       return false;
1703     if (!WriteActionMessage(PROTOCOL_GINGLE, type, action, stanza, error))
1704       return false;
1705   } else {
1706     if (!WriteActionMessage(current_protocol_, type, action, stanza, error))
1707       return false;
1708   }
1709   return true;
1710 }
1711
1712 template <typename Action>
1713 bool Session::WriteActionMessage(SignalingProtocol protocol,
1714                                  ActionType type, const Action& action,
1715                                  buzz::XmlElement* stanza, WriteError* error) {
1716   XmlElements action_elems;
1717   if (!WriteSessionAction(protocol, action, &action_elems, error))
1718     return false;
1719
1720   SessionMessage msg(protocol, type, id(), initiator_name());
1721   msg.to = remote_name();
1722
1723   WriteSessionMessage(msg, action_elems, stanza);
1724   return true;
1725 }
1726
1727 void Session::SendAcknowledgementMessage(const buzz::XmlElement* stanza) {
1728   talk_base::scoped_ptr<buzz::XmlElement> ack(
1729       new buzz::XmlElement(buzz::QN_IQ));
1730   ack->SetAttr(buzz::QN_TO, remote_name());
1731   ack->SetAttr(buzz::QN_ID, stanza->Attr(buzz::QN_ID));
1732   ack->SetAttr(buzz::QN_TYPE, "result");
1733
1734   SignalOutgoingMessage(this, ack.get());
1735 }
1736
1737 }  // namespace cricket