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   transport->SignalCompleted.connect(
 543       this, &BaseSession::OnTransportCompleted);
 544   transport->SignalFailed.connect(
 545       this, &BaseSession::OnTransportFailed);
 546
 547   transproxy = new TransportProxy(worker_thread_, sid_, content_name,
 548                                   new TransportWrapper(transport));
 549   transproxy->SignalCandidatesReady.connect(
 550       this, &BaseSession::OnTransportProxyCandidatesReady);
 551   transports_[content_name] = transproxy;
 552
 553   return transproxy;
 554 }
 555
 556 Transport* BaseSession::GetTransport(const std::string& content_name) {
 557   TransportProxy* transproxy = GetTransportProxy(content_name);
 558   if (transproxy == NULL)
 559     return NULL;
 560   return transproxy->impl();
 561 }
 562
 563 TransportProxy* BaseSession::GetTransportProxy(
 564     const std::string& content_name) {
 565   TransportMap::iterator iter = transports_.find(content_name);
 566   return (iter != transports_.end()) ? iter->second : NULL;
 567 }
 568
 569 TransportProxy* BaseSession::GetTransportProxy(const Transport* transport) {
 570   for (TransportMap::iterator iter = transports_.begin();
 571        iter != transports_.end(); ++iter) {
 572     TransportProxy* transproxy = iter->second;
 573     if (transproxy->impl() == transport) {
 574       return transproxy;
 575     }
 576   }
 577   return NULL;
 578 }
 579
 580 TransportProxy* BaseSession::GetFirstTransportProxy() {
 581   if (transports_.empty())
 582     return NULL;
 583   return transports_.begin()->second;
 584 }
 585
 586 void BaseSession::DestroyTransportProxy(
 587     const std::string& content_name) {
 588   TransportMap::iterator iter = transports_.find(content_name);
 589   if (iter != transports_.end()) {
 590     delete iter->second;
 591     transports_.erase(content_name);
 592   }
 593 }
 594
 595 cricket::Transport* BaseSession::CreateTransport(
 596     const std::string& content_name) {
 597   ASSERT(transport_type_ == NS_GINGLE_P2P);
 598   return new cricket::DtlsTransport<P2PTransport>(
 599       signaling_thread(), worker_thread(), content_name,
 600       port_allocator(), identity_);
 601 }
 602
 603 bool BaseSession::GetStats(SessionStats* stats) {
 604   for (TransportMap::iterator iter = transports_.begin();
 605        iter != transports_.end(); ++iter) {
 606     std::string proxy_id = iter->second->content_name();
 607     // We are ignoring not-yet-instantiated transports.
 608     if (iter->second->impl()) {
 609       std::string transport_id = iter->second->impl()->content_name();
 610       stats->proxy_to_transport[proxy_id] = transport_id;
 611       if (stats->transport_stats.find(transport_id)
 612           == stats->transport_stats.end()) {
 613         TransportStats subinfos;
 614         if (!iter->second->impl()->GetStats(&subinfos)) {
 615           return false;
 616         }
 617         stats->transport_stats[transport_id] = subinfos;
 618       }
 619     }
 620   }
 621   return true;
 622 }
 623
 624 void BaseSession::SetState(State state) {
 625   ASSERT(signaling_thread_->IsCurrent());
 626   if (state != state_) {
 627     LogState(state_, state);
 628     state_ = state;
 629     SignalState(this, state_);
 630     signaling_thread_->Post(this, MSG_STATE);
 631   }
 632   SignalNewDescription();
 633 }
 634
 635 void BaseSession::SetError(Error error, const std::string& error_desc) {
 636   ASSERT(signaling_thread_->IsCurrent());
 637   if (error != error_) {
 638     error_ = error;
 639     error_desc_ = error_desc;
 640     SignalError(this, error);
 641   }
 642 }
 643
 644 void BaseSession::OnSignalingReady() {
 645   ASSERT(signaling_thread()->IsCurrent());
 646   for (TransportMap::iterator iter = transports_.begin();
 647        iter != transports_.end(); ++iter) {
 648     iter->second->OnSignalingReady();
 649   }
 650 }
 651
 652 // TODO(juberti): Since PushdownLocalTD now triggers the connection process to
 653 // start, remove this method once everyone calls PushdownLocalTD.
 654 void BaseSession::SpeculativelyConnectAllTransportChannels() {
 655   // Put all transports into the connecting state.
 656   for (TransportMap::iterator iter = transports_.begin();
 657        iter != transports_.end(); ++iter) {
 658     iter->second->ConnectChannels();
 659   }
 660 }
 661
 662 bool BaseSession::OnRemoteCandidates(const std::string& content_name,
 663                                      const Candidates& candidates,
 664                                      std::string* error) {
 665   // Give candidates to the appropriate transport, and tell that transport
 666   // to start connecting, if it's not already doing so.
 667   TransportProxy* transproxy = GetTransportProxy(content_name);
 668   if (!transproxy) {
 669     *error = "Unknown content name " + content_name;
 670     return false;
 671   }
 672   if (!transproxy->OnRemoteCandidates(candidates, error)) {
 673     return false;
 674   }
 675   // TODO(juberti): Remove this call once we can be sure that we always have
 676   // a local transport description (which will trigger the connection).
 677   transproxy->ConnectChannels();
 678   return true;
 679 }
 680
 681 bool BaseSession::MaybeEnableMuxingSupport() {
 682   // We need both a local and remote description to decide if we should mux.
 683   if ((state_ == STATE_SENTINITIATE ||
 684       state_ == STATE_RECEIVEDINITIATE) &&
 685       ((local_description_ == NULL) ||
 686       (remote_description_ == NULL))) {
 687     return false;
 688   }
 689
 690   // In order to perform the multiplexing, we need all proxies to be in the
 691   // negotiated state, i.e. to have implementations underneath.
 692   // Ensure that this is the case, regardless of whether we are going to mux.
 693   for (TransportMap::iterator iter = transports_.begin();
 694        iter != transports_.end(); ++iter) {
 695     ASSERT(iter->second->negotiated());
 696     if (!iter->second->negotiated())
 697       return false;
 698   }
 699
 700   // If both sides agree to BUNDLE, mux all the specified contents onto the
 701   // transport belonging to the first content name in the BUNDLE group.
 702   // If the contents are already muxed, this will be a no-op.
 703   // TODO(juberti): Should this check that local and remote have configured
 704   // BUNDLE the same way?
 705   bool candidates_allocated = IsCandidateAllocationDone();
 706   const ContentGroup* local_bundle_group =
 707       local_description()->GetGroupByName(GROUP_TYPE_BUNDLE);
 708   const ContentGroup* remote_bundle_group =
 709       remote_description()->GetGroupByName(GROUP_TYPE_BUNDLE);
 710   if (local_bundle_group && remote_bundle_group &&
 711       local_bundle_group->FirstContentName()) {
 712     const std::string* content_name = local_bundle_group->FirstContentName();
 713     const ContentInfo* content =
 714         local_description_->GetContentByName(*content_name);
 715     ASSERT(content != NULL);
 716     if (!SetSelectedProxy(content->name, local_bundle_group)) {
 717       LOG(LS_WARNING) << "Failed to set up BUNDLE";
 718       return false;
 719     }
 720
 721     // If we weren't done gathering before, we might be done now, as a result
 722     // of enabling mux.
 723     LOG(LS_INFO) << "Enabling BUNDLE, bundling onto transport: "
 724                  << *content_name;
 725     if (!candidates_allocated) {
 726       MaybeCandidateAllocationDone();
 727     }
 728   } else {
 729     LOG(LS_INFO) << "No BUNDLE information, not bundling.";
 730   }
 731   return true;
 732 }
 733
 734 bool BaseSession::SetSelectedProxy(const std::string& content_name,
 735                                    const ContentGroup* muxed_group) {
 736   TransportProxy* selected_proxy = GetTransportProxy(content_name);
 737   if (!selected_proxy) {
 738     return false;
 739   }
 740
 741   ASSERT(selected_proxy->negotiated());
 742   for (TransportMap::iterator iter = transports_.begin();
 743        iter != transports_.end(); ++iter) {
 744     // If content is part of the mux group, then repoint its proxy at the
 745     // transport object that we have chosen to mux onto. If the proxy
 746     // is already pointing at the right object, it will be a no-op.
 747     if (muxed_group->HasContentName(iter->first) &&
 748         !iter->second->SetupMux(selected_proxy)) {
 749       return false;
 750     }
 751   }
 752   return true;
 753 }
 754
 755 void BaseSession::OnTransportCandidatesAllocationDone(Transport* transport) {
 756   // TODO(juberti): This is a clunky way of processing the done signal. Instead,
 757   // TransportProxy should receive the done signal directly, set its allocated
 758   // flag internally, and then reissue the done signal to Session.
 759   // Overall we should make TransportProxy receive *all* the signals from
 760   // Transport, since this removes the need to manually iterate over all
 761   // the transports, as is needed to make sure signals are handled properly
 762   // when BUNDLEing.
 763 #if 0
 764   ASSERT(!IsCandidateAllocationDone());
 765 #endif
 766   for (TransportMap::iterator iter = transports_.begin();
 767        iter != transports_.end(); ++iter) {
 768     if (iter->second->impl() == transport) {
 769       iter->second->set_candidates_allocated(true);
 770     }
 771   }
 772   MaybeCandidateAllocationDone();
 773 }
 774
 775 bool BaseSession::IsCandidateAllocationDone() const {
 776   for (TransportMap::const_iterator iter = transports_.begin();
 777        iter != transports_.end(); ++iter) {
 778     if (!iter->second->candidates_allocated())
 779       return false;
 780   }
 781   return true;
 782 }
 783
 784 void BaseSession::MaybeCandidateAllocationDone() {
 785   if (IsCandidateAllocationDone()) {
 786     LOG(LS_INFO) << "Candidate gathering is complete.";
 787     OnCandidatesAllocationDone();
 788   }
 789 }
 790
 791 void BaseSession::OnRoleConflict() {
 792   if (role_switch_) {
 793     LOG(LS_WARNING) << "Repeat of role conflict signal from Transport.";
 794     return;
 795   }
 796
 797   role_switch_ = true;
 798   for (TransportMap::iterator iter = transports_.begin();
 799        iter != transports_.end(); ++iter) {
 800     // Role will be reverse of initial role setting.
 801     IceRole role = initiator_ ? ICEROLE_CONTROLLED : ICEROLE_CONTROLLING;
 802     iter->second->SetIceRole(role);
 803   }
 804 }
 805
 806 void BaseSession::LogState(State old_state, State new_state) {
 807   LOG(LS_INFO) << "Session:" << id()
 808                << " Old state:" << StateToString(old_state)
 809                << " New state:" << StateToString(new_state)
 810                << " Type:" << content_type()
 811                << " Transport:" << transport_type();
 812 }
 813
 814 bool BaseSession::GetTransportDescription(const SessionDescription* description,
 815                                           const std::string& content_name,
 816                                           TransportDescription* tdesc) {
 817   if (!description || !tdesc) {
 818     return false;
 819   }
 820   const TransportInfo* transport_info =
 821       description->GetTransportInfoByName(content_name);
 822   if (!transport_info) {
 823     return false;
 824   }
 825   *tdesc = transport_info->description;
 826   return true;
 827 }
 828
 829 void BaseSession::SignalNewDescription() {
 830   ContentAction action;
 831   ContentSource source;
 832   if (!GetContentAction(&action, &source)) {
 833     return;
 834   }
 835   if (source == CS_LOCAL) {
 836     SignalNewLocalDescription(this, action);
 837   } else {
 838     SignalNewRemoteDescription(this, action);
 839   }
 840 }
 841
 842 bool BaseSession::GetContentAction(ContentAction* action,
 843                                    ContentSource* source) {
 844   switch (state_) {
 845     // new local description
 846     case STATE_SENTINITIATE:
 847       *action = CA_OFFER;
 848       *source = CS_LOCAL;
 849       break;
 850     case STATE_SENTPRACCEPT:
 851       *action = CA_PRANSWER;
 852       *source = CS_LOCAL;
 853       break;
 854     case STATE_SENTACCEPT:
 855       *action = CA_ANSWER;
 856       *source = CS_LOCAL;
 857       break;
 858     // new remote description
 859     case STATE_RECEIVEDINITIATE:
 860       *action = CA_OFFER;
 861       *source = CS_REMOTE;
 862       break;
 863     case STATE_RECEIVEDPRACCEPT:
 864       *action = CA_PRANSWER;
 865       *source = CS_REMOTE;
 866       break;
 867     case STATE_RECEIVEDACCEPT:
 868       *action = CA_ANSWER;
 869       *source = CS_REMOTE;
 870       break;
 871     default:
 872       return false;
 873   }
 874   return true;
 875 }
 876
 877 void BaseSession::OnMessage(talk_base::Message *pmsg) {
 878   switch (pmsg->message_id) {
 879   case MSG_TIMEOUT:
 880     // Session timeout has occured.
 881     SetError(ERROR_TIME, "Session timeout has occured.");
 882     break;
 883
 884   case MSG_STATE:
 885     switch (state_) {
 886     case STATE_SENTACCEPT:
 887     case STATE_RECEIVEDACCEPT:
 888       SetState(STATE_INPROGRESS);
 889       break;
 890
 891     default:
 892       // Explicitly ignoring some states here.
 893       break;
 894     }
 895     break;
 896   }
 897 }
 898
 899 Session::Session(SessionManager* session_manager,
 900                  const std::string& local_name,
 901                  const std::string& initiator_name,
 902                  const std::string& sid,
 903                  const std::string& content_type,
 904                  SessionClient* client)
 905     : BaseSession(session_manager->signaling_thread(),
 906                   session_manager->worker_thread(),
 907                   session_manager->port_allocator(),
 908                   sid, content_type, initiator_name == local_name) {
 909   ASSERT(client != NULL);
 910   session_manager_ = session_manager;
 911   local_name_ = local_name;
 912   initiator_name_ = initiator_name;
 913   transport_parser_ = new P2PTransportParser();
 914   client_ = client;
 915   initiate_acked_ = false;
 916   current_protocol_ = PROTOCOL_HYBRID;
 917 }
 918
 919 Session::~Session() {
 920   delete transport_parser_;
 921 }
 922
 923 bool Session::Initiate(const std::string &to,
 924                        const SessionDescription* sdesc) {
 925   ASSERT(signaling_thread()->IsCurrent());
 926   SessionError error;
 927
 928   // Only from STATE_INIT
 929   if (state() != STATE_INIT)
 930     return false;
 931
 932   // Setup for signaling.
 933   set_remote_name(to);
 934   set_local_description(sdesc);
 935   if (!CreateTransportProxies(GetEmptyTransportInfos(sdesc->contents()),
 936                               &error)) {
 937     LOG(LS_ERROR) << "Could not create transports: " << error.text;
 938     return false;
 939   }
 940
 941   if (!SendInitiateMessage(sdesc, &error)) {
 942     LOG(LS_ERROR) << "Could not send initiate message: " << error.text;
 943     return false;
 944   }
 945
 946   // We need to connect transport proxy and impl here so that we can process
 947   // the TransportDescriptions.
 948   SpeculativelyConnectAllTransportChannels();
 949
 950   PushdownTransportDescription(CS_LOCAL, CA_OFFER, NULL);
 951   SetState(Session::STATE_SENTINITIATE);
 952   return true;
 953 }
 954
 955 bool Session::Accept(const SessionDescription* sdesc) {
 956   ASSERT(signaling_thread()->IsCurrent());
 957
 958   // Only if just received initiate
 959   if (state() != STATE_RECEIVEDINITIATE)
 960     return false;
 961
 962   // Setup for signaling.
 963   set_local_description(sdesc);
 964
 965   SessionError error;
 966   if (!SendAcceptMessage(sdesc, &error)) {
 967     LOG(LS_ERROR) << "Could not send accept message: " << error.text;
 968     return false;
 969   }
 970   // TODO(juberti): Add BUNDLE support to transport-info messages.
 971   PushdownTransportDescription(CS_LOCAL, CA_ANSWER, NULL);
 972   MaybeEnableMuxingSupport();  // Enable transport channel mux if supported.
 973   SetState(Session::STATE_SENTACCEPT);
 974   return true;
 975 }
 976
 977 bool Session::Reject(const std::string& reason) {
 978   ASSERT(signaling_thread()->IsCurrent());
 979
 980   // Reject is sent in response to an initiate or modify, to reject the
 981   // request
 982   if (state() != STATE_RECEIVEDINITIATE && state() != STATE_RECEIVEDMODIFY)
 983     return false;
 984
 985   SessionError error;
 986   if (!SendRejectMessage(reason, &error)) {
 987     LOG(LS_ERROR) << "Could not send reject message: " << error.text;
 988     return false;
 989   }
 990
 991   SetState(STATE_SENTREJECT);
 992   return true;
 993 }
 994
 995 bool Session::TerminateWithReason(const std::string& reason) {
 996   ASSERT(signaling_thread()->IsCurrent());
 997
 998   // Either side can terminate, at any time.
 999   switch (state()) {
1000     case STATE_SENTTERMINATE:
1001     case STATE_RECEIVEDTERMINATE:
1002       return false;
1003
1004     case STATE_SENTREJECT:
1005     case STATE_RECEIVEDREJECT:
1006       // We don't need to send terminate if we sent or received a reject...
1007       // it's implicit.
1008       break;
1009
1010     default:
1011       SessionError error;
1012       if (!SendTerminateMessage(reason, &error)) {
1013         LOG(LS_ERROR) << "Could not send terminate message: " << error.text;
1014         return false;
1015       }
1016       break;
1017   }
1018
1019   SetState(STATE_SENTTERMINATE);
1020   return true;
1021 }
1022
1023 bool Session::SendInfoMessage(const XmlElements& elems,
1024                               const std::string& remote_name) {
1025   ASSERT(signaling_thread()->IsCurrent());
1026   SessionError error;
1027   if (!SendMessage(ACTION_SESSION_INFO, elems, remote_name, &error)) {
1028     LOG(LS_ERROR) << "Could not send info message " << error.text;
1029     return false;
1030   }
1031   return true;
1032 }
1033
1034 bool Session::SendDescriptionInfoMessage(const ContentInfos& contents) {
1035   XmlElements elems;
1036   WriteError write_error;
1037   if (!WriteDescriptionInfo(current_protocol_,
1038                             contents,
1039                             GetContentParsers(),
1040                             &elems, &write_error)) {
1041     LOG(LS_ERROR) << "Could not write description info message: "
1042                   << write_error.text;
1043     return false;
1044   }
1045   SessionError error;
1046   if (!SendMessage(ACTION_DESCRIPTION_INFO, elems, &error)) {
1047     LOG(LS_ERROR) << "Could not send description info message: "
1048                   << error.text;
1049     return false;
1050   }
1051   return true;
1052 }
1053
1054 TransportInfos Session::GetEmptyTransportInfos(
1055     const ContentInfos& contents) const {
1056   TransportInfos tinfos;
1057   for (ContentInfos::const_iterator content = contents.begin();
1058        content != contents.end(); ++content) {
1059     tinfos.push_back(TransportInfo(content->name,
1060                                    TransportDescription(transport_type(),
1061                                                         std::string(),
1062                                                         std::string())));
1063   }
1064   return tinfos;
1065 }
1066
1067 bool Session::OnRemoteCandidates(
1068     const TransportInfos& tinfos, ParseError* error) {
1069   for (TransportInfos::const_iterator tinfo = tinfos.begin();
1070        tinfo != tinfos.end(); ++tinfo) {
1071     std::string str_error;
1072     if (!BaseSession::OnRemoteCandidates(
1073         tinfo->content_name, tinfo->description.candidates, &str_error)) {
1074       return BadParse(str_error, error);
1075     }
1076   }
1077   return true;
1078 }
1079
1080 bool Session::CreateTransportProxies(const TransportInfos& tinfos,
1081                                      SessionError* error) {
1082   for (TransportInfos::const_iterator tinfo = tinfos.begin();
1083        tinfo != tinfos.end(); ++tinfo) {
1084     if (tinfo->description.transport_type != transport_type()) {
1085       error->SetText("No supported transport in offer.");
1086       return false;
1087     }
1088
1089     GetOrCreateTransportProxy(tinfo->content_name);
1090   }
1091   return true;
1092 }
1093
1094 TransportParserMap Session::GetTransportParsers() {
1095   TransportParserMap parsers;
1096   parsers[transport_type()] = transport_parser_;
1097   return parsers;
1098 }
1099
1100 CandidateTranslatorMap Session::GetCandidateTranslators() {
1101   CandidateTranslatorMap translators;
1102   // NOTE: This technique makes it impossible to parse G-ICE
1103   // candidates in session-initiate messages because the channels
1104   // aren't yet created at that point.  Since we don't use candidates
1105   // in session-initiate messages, we should be OK.  Once we switch to
1106   // ICE, this translation shouldn't be necessary.
1107   for (TransportMap::const_iterator iter = transport_proxies().begin();
1108        iter != transport_proxies().end(); ++iter) {
1109     translators[iter->first] = iter->second;
1110   }
1111   return translators;
1112 }
1113
1114 ContentParserMap Session::GetContentParsers() {
1115   ContentParserMap parsers;
1116   parsers[content_type()] = client_;
1117   // We need to be able parse both RTP-based and SCTP-based Jingle
1118   // with the same client.
1119   if (content_type() == NS_JINGLE_RTP) {
1120     parsers[NS_JINGLE_DRAFT_SCTP] = client_;
1121   }
1122   return parsers;
1123 }
1124
1125 void Session::OnTransportRequestSignaling(Transport* transport) {
1126   ASSERT(signaling_thread()->IsCurrent());
1127   TransportProxy* transproxy = GetTransportProxy(transport);
1128   ASSERT(transproxy != NULL);
1129   if (transproxy) {
1130     // Reset candidate allocation status for the transport proxy.
1131     transproxy->set_candidates_allocated(false);
1132   }
1133   SignalRequestSignaling(this);
1134 }
1135
1136 void Session::OnTransportConnecting(Transport* transport) {
1137   // This is an indication that we should begin watching the writability
1138   // state of the transport.
1139   OnTransportWritable(transport);
1140 }
1141
1142 void Session::OnTransportWritable(Transport* transport) {
1143   ASSERT(signaling_thread()->IsCurrent());
1144
1145   // If the transport is not writable, start a timer to make sure that it
1146   // becomes writable within a reasonable amount of time.  If it does not, we
1147   // terminate since we can't actually send data.  If the transport is writable,
1148   // cancel the timer.  Note that writability transitions may occur repeatedly
1149   // during the lifetime of the session.
1150   signaling_thread()->Clear(this, MSG_TIMEOUT);
1151   if (transport->HasChannels() && !transport->writable()) {
1152     signaling_thread()->PostDelayed(
1153         session_manager_->session_timeout() * 1000, this, MSG_TIMEOUT);
1154   }
1155 }
1156
1157 void Session::OnTransportProxyCandidatesReady(TransportProxy* transproxy,
1158                                               const Candidates& candidates) {
1159   ASSERT(signaling_thread()->IsCurrent());
1160   if (transproxy != NULL) {
1161     if (initiator() && !initiate_acked_) {
1162       // TODO: This is to work around server re-ordering
1163       // messages.  We send the candidates once the session-initiate
1164       // is acked.  Once we have fixed the server to guarantee message
1165       // order, we can remove this case.
1166       transproxy->AddUnsentCandidates(candidates);
1167     } else {
1168       if (!transproxy->negotiated()) {
1169         transproxy->AddSentCandidates(candidates);
1170       }
1171       SessionError error;
1172       if (!SendTransportInfoMessage(transproxy, candidates, &error)) {
1173         LOG(LS_ERROR) << "Could not send transport info message: "
1174                       << error.text;
1175         return;
1176       }
1177     }
1178   }
1179 }
1180
1181 void Session::OnTransportSendError(Transport* transport,
1182                                    const buzz::XmlElement* stanza,
1183                                    const buzz::QName& name,
1184                                    const std::string& type,
1185                                    const std::string& text,
1186                                    const buzz::XmlElement* extra_info) {
1187   ASSERT(signaling_thread()->IsCurrent());
1188   SignalErrorMessage(this, stanza, name, type, text, extra_info);
1189 }
1190
1191 void Session::OnIncomingMessage(const SessionMessage& msg) {
1192   ASSERT(signaling_thread()->IsCurrent());
1193   ASSERT(state() == STATE_INIT || msg.from == remote_name());
1194
1195   if (current_protocol_== PROTOCOL_HYBRID) {
1196     if (msg.protocol == PROTOCOL_GINGLE) {
1197       current_protocol_ = PROTOCOL_GINGLE;
1198     } else {
1199       current_protocol_ = PROTOCOL_JINGLE;
1200     }
1201   }
1202
1203   bool valid = false;
1204   MessageError error;
1205   switch (msg.type) {
1206     case ACTION_SESSION_INITIATE:
1207       valid = OnInitiateMessage(msg, &error);
1208       break;
1209     case ACTION_SESSION_INFO:
1210       valid = OnInfoMessage(msg);
1211       break;
1212     case ACTION_SESSION_ACCEPT:
1213       valid = OnAcceptMessage(msg, &error);
1214       break;
1215     case ACTION_SESSION_REJECT:
1216       valid = OnRejectMessage(msg, &error);
1217       break;
1218     case ACTION_SESSION_TERMINATE:
1219       valid = OnTerminateMessage(msg, &error);
1220       break;
1221     case ACTION_TRANSPORT_INFO:
1222       valid = OnTransportInfoMessage(msg, &error);
1223       break;
1224     case ACTION_TRANSPORT_ACCEPT:
1225       valid = OnTransportAcceptMessage(msg, &error);
1226       break;
1227     case ACTION_DESCRIPTION_INFO:
1228       valid = OnDescriptionInfoMessage(msg, &error);
1229       break;
1230     default:
1231       valid = BadMessage(buzz::QN_STANZA_BAD_REQUEST,
1232                          "unknown session message type",
1233                          &error);
1234   }
1235
1236   if (valid) {
1237     SendAcknowledgementMessage(msg.stanza);
1238   } else {
1239     SignalErrorMessage(this, msg.stanza, error.type,
1240                        "modify", error.text, NULL);
1241   }
1242 }
1243
1244 void Session::OnIncomingResponse(const buzz::XmlElement* orig_stanza,
1245                                  const buzz::XmlElement* response_stanza,
1246                                  const SessionMessage& msg) {
1247   ASSERT(signaling_thread()->IsCurrent());
1248
1249   if (msg.type == ACTION_SESSION_INITIATE) {
1250     OnInitiateAcked();
1251   }
1252 }
1253
1254 void Session::OnInitiateAcked() {
1255     // TODO: This is to work around server re-ordering
1256     // messages.  We send the candidates once the session-initiate
1257     // is acked.  Once we have fixed the server to guarantee message
1258     // order, we can remove this case.
1259   if (!initiate_acked_) {
1260     initiate_acked_ = true;
1261     SessionError error;
1262     SendAllUnsentTransportInfoMessages(&error);
1263   }
1264 }
1265
1266 void Session::OnFailedSend(const buzz::XmlElement* orig_stanza,
1267                            const buzz::XmlElement* error_stanza) {
1268   ASSERT(signaling_thread()->IsCurrent());
1269
1270   SessionMessage msg;
1271   ParseError parse_error;
1272   if (!ParseSessionMessage(orig_stanza, &msg, &parse_error)) {
1273     LOG(LS_ERROR) << "Error parsing failed send: " << parse_error.text
1274                   << ":" << orig_stanza;
1275     return;
1276   }
1277
1278   // If the error is a session redirect, call OnRedirectError, which will
1279   // continue the session with a new remote JID.
1280   SessionRedirect redirect;
1281   if (FindSessionRedirect(error_stanza, &redirect)) {
1282     SessionError error;
1283     if (!OnRedirectError(redirect, &error)) {
1284       // TODO: Should we send a message back?  The standard
1285       // says nothing about it.
1286       std::ostringstream desc;
1287       desc << "Failed to redirect: " << error.text;
1288       LOG(LS_ERROR) << desc.str();
1289       SetError(ERROR_RESPONSE, desc.str());
1290     }
1291     return;
1292   }
1293
1294   std::string error_type = "cancel";
1295
1296   const buzz::XmlElement* error = error_stanza->FirstNamed(buzz::QN_ERROR);
1297   if (error) {
1298     error_type = error->Attr(buzz::QN_TYPE);
1299
1300     LOG(LS_ERROR) << "Session error:\n" << error->Str() << "\n"
1301                   << "in response to:\n" << orig_stanza->Str();
1302   } else {
1303     // don't crash if <error> is missing
1304     LOG(LS_ERROR) << "Session error without <error/> element, ignoring";
1305     return;
1306   }
1307
1308   if (msg.type == ACTION_TRANSPORT_INFO) {
1309     // Transport messages frequently generate errors because they are sent right
1310     // when we detect a network failure.  For that reason, we ignore such
1311     // errors, because if we do not establish writability again, we will
1312     // terminate anyway.  The exceptions are transport-specific error tags,
1313     // which we pass on to the respective transport.
1314   } else if ((error_type != "continue") && (error_type != "wait")) {
1315     // We do not set an error if the other side said it is okay to continue
1316     // (possibly after waiting).  These errors can be ignored.
1317     SetError(ERROR_RESPONSE, "");
1318   }
1319 }
1320
1321 bool Session::OnInitiateMessage(const SessionMessage& msg,
1322                                 MessageError* error) {
1323   if (!CheckState(STATE_INIT, error))
1324     return false;
1325
1326   SessionInitiate init;
1327   if (!ParseSessionInitiate(msg.protocol, msg.action_elem,
1328                             GetContentParsers(), GetTransportParsers(),
1329                             GetCandidateTranslators(),
1330                             &init, error))
1331     return false;
1332
1333   SessionError session_error;
1334   if (!CreateTransportProxies(init.transports, &session_error)) {
1335     return BadMessage(buzz::QN_STANZA_NOT_ACCEPTABLE,
1336                       session_error.text, error);
1337   }
1338
1339   set_remote_name(msg.from);
1340   set_initiator_name(msg.initiator);
1341   set_remote_description(new SessionDescription(init.ClearContents(),
1342                                                 init.transports,
1343                                                 init.groups));
1344   // Updating transport with TransportDescription.
1345   PushdownTransportDescription(CS_REMOTE, CA_OFFER, NULL);
1346   SetState(STATE_RECEIVEDINITIATE);
1347
1348   // Users of Session may listen to state change and call Reject().
1349   if (state() != STATE_SENTREJECT) {
1350     if (!OnRemoteCandidates(init.transports, error))
1351       return false;
1352
1353     // TODO(juberti): Auto-generate and push down the local transport answer.
1354     // This is necessary for trickling to work with RFC 5245 ICE.
1355   }
1356   return true;
1357 }
1358
1359 bool Session::OnAcceptMessage(const SessionMessage& msg, MessageError* error) {
1360   if (!CheckState(STATE_SENTINITIATE, error))
1361     return false;
1362
1363   SessionAccept accept;
1364   if (!ParseSessionAccept(msg.protocol, msg.action_elem,
1365                           GetContentParsers(), GetTransportParsers(),
1366                           GetCandidateTranslators(),
1367                           &accept, error)) {
1368     return false;
1369   }
1370
1371   // If we get an accept, we can assume the initiate has been
1372   // received, even if we haven't gotten an IQ response.
1373   OnInitiateAcked();
1374
1375   set_remote_description(new SessionDescription(accept.ClearContents(),
1376                                                 accept.transports,
1377                                                 accept.groups));
1378   // Updating transport with TransportDescription.
1379   PushdownTransportDescription(CS_REMOTE, CA_ANSWER, NULL);
1380   MaybeEnableMuxingSupport();  // Enable transport channel mux if supported.
1381   SetState(STATE_RECEIVEDACCEPT);
1382
1383   if (!OnRemoteCandidates(accept.transports, error))
1384     return false;
1385
1386   return true;
1387 }
1388
1389 bool Session::OnRejectMessage(const SessionMessage& msg, MessageError* error) {
1390   if (!CheckState(STATE_SENTINITIATE, error))
1391     return false;
1392
1393   SetState(STATE_RECEIVEDREJECT);
1394   return true;
1395 }
1396
1397 bool Session::OnInfoMessage(const SessionMessage& msg) {
1398   SignalInfoMessage(this, msg.action_elem);
1399   return true;
1400 }
1401
1402 bool Session::OnTerminateMessage(const SessionMessage& msg,
1403                                  MessageError* error) {
1404   SessionTerminate term;
1405   if (!ParseSessionTerminate(msg.protocol, msg.action_elem, &term, error))
1406     return false;
1407
1408   SignalReceivedTerminateReason(this, term.reason);
1409   if (term.debug_reason != buzz::STR_EMPTY) {
1410     LOG(LS_VERBOSE) << "Received error on call: " << term.debug_reason;
1411   }
1412
1413   SetState(STATE_RECEIVEDTERMINATE);
1414   return true;
1415 }
1416
1417 bool Session::OnTransportInfoMessage(const SessionMessage& msg,
1418                                      MessageError* error) {
1419   TransportInfos tinfos;
1420   if (!ParseTransportInfos(msg.protocol, msg.action_elem,
1421                            initiator_description()->contents(),
1422                            GetTransportParsers(), GetCandidateTranslators(),
1423                            &tinfos, error))
1424     return false;
1425
1426   if (!OnRemoteCandidates(tinfos, error))
1427     return false;
1428
1429   return true;
1430 }
1431
1432 bool Session::OnTransportAcceptMessage(const SessionMessage& msg,
1433                                        MessageError* error) {
1434   // TODO: Currently here only for compatibility with
1435   // Gingle 1.1 clients (notably, Google Voice).
1436   return true;
1437 }
1438
1439 bool Session::OnDescriptionInfoMessage(const SessionMessage& msg,
1440                               MessageError* error) {
1441   if (!CheckState(STATE_INPROGRESS, error))
1442     return false;
1443
1444   DescriptionInfo description_info;
1445   if (!ParseDescriptionInfo(msg.protocol, msg.action_elem,
1446                             GetContentParsers(), GetTransportParsers(),
1447                             GetCandidateTranslators(),
1448                             &description_info, error)) {
1449     return false;
1450   }
1451
1452   ContentInfos& updated_contents = description_info.contents;
1453
1454   // TODO: Currently, reflector sends back
1455   // video stream updates even for an audio-only call, which causes
1456   // this to fail.  Put this back once reflector is fixed.
1457   //
1458   // ContentInfos::iterator it;
1459   // First, ensure all updates are valid before modifying remote_description_.
1460   // for (it = updated_contents.begin(); it != updated_contents.end(); ++it) {
1461   //   if (remote_description()->GetContentByName(it->name) == NULL) {
1462   //     return false;
1463   //   }
1464   // }
1465
1466   // TODO: We used to replace contents from an update, but
1467   // that no longer works with partial updates.  We need to figure out
1468   // a way to merge patial updates into contents.  For now, users of
1469   // Session should listen to SignalRemoteDescriptionUpdate and handle
1470   // updates.  They should not expect remote_description to be the
1471   // latest value.
1472   //
1473   // for (it = updated_contents.begin(); it != updated_contents.end(); ++it) {
1474   //     remote_description()->RemoveContentByName(it->name);
1475   //     remote_description()->AddContent(it->name, it->type, it->description);
1476   //   }
1477   // }
1478
1479   SignalRemoteDescriptionUpdate(this, updated_contents);
1480   return true;
1481 }
1482
1483 bool BareJidsEqual(const std::string& name1,
1484                    const std::string& name2) {
1485   buzz::Jid jid1(name1);
1486   buzz::Jid jid2(name2);
1487
1488   return jid1.IsValid() && jid2.IsValid() && jid1.BareEquals(jid2);
1489 }
1490
1491 bool Session::OnRedirectError(const SessionRedirect& redirect,
1492                               SessionError* error) {
1493   MessageError message_error;
1494   if (!CheckState(STATE_SENTINITIATE, &message_error)) {
1495     return BadWrite(message_error.text, error);
1496   }
1497
1498   if (!BareJidsEqual(remote_name(), redirect.target))
1499     return BadWrite("Redirection not allowed: must be the same bare jid.",
1500                     error);
1501
1502   // When we receive a redirect, we point the session at the new JID
1503   // and resend the candidates.
1504   set_remote_name(redirect.target);
1505   return (SendInitiateMessage(local_description(), error) &&
1506           ResendAllTransportInfoMessages(error));
1507 }
1508
1509 bool Session::CheckState(State expected, MessageError* error) {
1510   if (state() != expected) {
1511     // The server can deliver messages out of order/repeated for various
1512     // reasons. For example, if the server does not recive our iq response,
1513     // it could assume that the iq it sent was lost, and will then send
1514     // it again. Ideally, we should implement reliable messaging with
1515     // duplicate elimination.
1516     return BadMessage(buzz::QN_STANZA_NOT_ALLOWED,
1517                       "message not allowed in current state",
1518                       error);
1519   }
1520   return true;
1521 }
1522
1523 void Session::SetError(Error error, const std::string& error_desc) {
1524   BaseSession::SetError(error, error_desc);
1525   if (error != ERROR_NONE)
1526     signaling_thread()->Post(this, MSG_ERROR);
1527 }
1528
1529 void Session::OnMessage(talk_base::Message* pmsg) {
1530   // preserve this because BaseSession::OnMessage may modify it
1531   State orig_state = state();
1532
1533   BaseSession::OnMessage(pmsg);
1534
1535   switch (pmsg->message_id) {
1536   case MSG_ERROR:
1537     TerminateWithReason(STR_TERMINATE_ERROR);
1538     break;
1539
1540   case MSG_STATE:
1541     switch (orig_state) {
1542     case STATE_SENTREJECT:
1543     case STATE_RECEIVEDREJECT:
1544       // Assume clean termination.
1545       Terminate();
1546       break;
1547
1548     case STATE_SENTTERMINATE:
1549     case STATE_RECEIVEDTERMINATE:
1550       session_manager_->DestroySession(this);
1551       break;
1552
1553     default:
1554       // Explicitly ignoring some states here.
1555       break;
1556     }
1557     break;
1558   }
1559 }
1560
1561 bool Session::SendInitiateMessage(const SessionDescription* sdesc,
1562                                   SessionError* error) {
1563   SessionInitiate init;
1564   init.contents = sdesc->contents();
1565   init.transports = GetEmptyTransportInfos(init.contents);
1566   init.groups = sdesc->groups();
1567   return SendMessage(ACTION_SESSION_INITIATE, init, error);
1568 }
1569
1570 bool Session::WriteSessionAction(
1571     SignalingProtocol protocol, const SessionInitiate& init,
1572     XmlElements* elems, WriteError* error) {
1573   return WriteSessionInitiate(protocol, init.contents, init.transports,
1574                               GetContentParsers(), GetTransportParsers(),
1575                               GetCandidateTranslators(), init.groups,
1576                               elems, error);
1577 }
1578
1579 bool Session::SendAcceptMessage(const SessionDescription* sdesc,
1580                                 SessionError* error) {
1581   XmlElements elems;
1582   if (!WriteSessionAccept(current_protocol_,
1583                           sdesc->contents(),
1584                           GetEmptyTransportInfos(sdesc->contents()),
1585                           GetContentParsers(), GetTransportParsers(),
1586                           GetCandidateTranslators(), sdesc->groups(),
1587                           &elems, error)) {
1588     return false;
1589   }
1590   return SendMessage(ACTION_SESSION_ACCEPT, elems, error);
1591 }
1592
1593 bool Session::SendRejectMessage(const std::string& reason,
1594                                 SessionError* error) {
1595   SessionTerminate term(reason);
1596   return SendMessage(ACTION_SESSION_REJECT, term, error);
1597 }
1598
1599 bool Session::SendTerminateMessage(const std::string& reason,
1600                                    SessionError* error) {
1601   SessionTerminate term(reason);
1602   return SendMessage(ACTION_SESSION_TERMINATE, term, error);
1603 }
1604
1605 bool Session::WriteSessionAction(SignalingProtocol protocol,
1606                                  const SessionTerminate& term,
1607                                  XmlElements* elems, WriteError* error) {
1608   WriteSessionTerminate(protocol, term, elems);
1609   return true;
1610 }
1611
1612 bool Session::SendTransportInfoMessage(const TransportInfo& tinfo,
1613                                        SessionError* error) {
1614   return SendMessage(ACTION_TRANSPORT_INFO, tinfo, error);
1615 }
1616
1617 bool Session::SendTransportInfoMessage(const TransportProxy* transproxy,
1618                                        const Candidates& candidates,
1619                                        SessionError* error) {
1620   return SendTransportInfoMessage(TransportInfo(transproxy->content_name(),
1621       TransportDescription(transproxy->type(), std::vector<std::string>(),
1622                            std::string(), std::string(), ICEMODE_FULL,
1623                            CONNECTIONROLE_NONE, NULL, candidates)), error);
1624 }
1625
1626 bool Session::WriteSessionAction(SignalingProtocol protocol,
1627                                  const TransportInfo& tinfo,
1628                                  XmlElements* elems, WriteError* error) {
1629   TransportInfos tinfos;
1630   tinfos.push_back(tinfo);
1631   return WriteTransportInfos(protocol, tinfos,
1632                              GetTransportParsers(), GetCandidateTranslators(),
1633                              elems, error);
1634 }
1635
1636 bool Session::ResendAllTransportInfoMessages(SessionError* error) {
1637   for (TransportMap::const_iterator iter = transport_proxies().begin();
1638        iter != transport_proxies().end(); ++iter) {
1639     TransportProxy* transproxy = iter->second;
1640     if (transproxy->sent_candidates().size() > 0) {
1641       if (!SendTransportInfoMessage(
1642               transproxy, transproxy->sent_candidates(), error)) {
1643         LOG(LS_ERROR) << "Could not resend transport info messages: "
1644                       << error->text;
1645         return false;
1646       }
1647       transproxy->ClearSentCandidates();
1648     }
1649   }
1650   return true;
1651 }
1652
1653 bool Session::SendAllUnsentTransportInfoMessages(SessionError* error) {
1654   for (TransportMap::const_iterator iter = transport_proxies().begin();
1655        iter != transport_proxies().end(); ++iter) {
1656     TransportProxy* transproxy = iter->second;
1657     if (transproxy->unsent_candidates().size() > 0) {
1658       if (!SendTransportInfoMessage(
1659               transproxy, transproxy->unsent_candidates(), error)) {
1660         LOG(LS_ERROR) << "Could not send unsent transport info messages: "
1661                       << error->text;
1662         return false;
1663       }
1664       transproxy->ClearUnsentCandidates();
1665     }
1666   }
1667   return true;
1668 }
1669
1670 bool Session::SendMessage(ActionType type, const XmlElements& action_elems,
1671                           SessionError* error) {
1672     return SendMessage(type, action_elems, remote_name(), error);
1673 }
1674
1675 bool Session::SendMessage(ActionType type, const XmlElements& action_elems,
1676                           const std::string& remote_name, SessionError* error) {
1677   talk_base::scoped_ptr<buzz::XmlElement> stanza(
1678       new buzz::XmlElement(buzz::QN_IQ));
1679
1680   SessionMessage msg(current_protocol_, type, id(), initiator_name());
1681   msg.to = remote_name;
1682   WriteSessionMessage(msg, action_elems, stanza.get());
1683
1684   SignalOutgoingMessage(this, stanza.get());
1685   return true;
1686 }
1687
1688 template <typename Action>
1689 bool Session::SendMessage(ActionType type, const Action& action,
1690                           SessionError* error) {
1691   talk_base::scoped_ptr<buzz::XmlElement> stanza(
1692       new buzz::XmlElement(buzz::QN_IQ));
1693   if (!WriteActionMessage(type, action, stanza.get(), error))
1694     return false;
1695
1696   SignalOutgoingMessage(this, stanza.get());
1697   return true;
1698 }
1699
1700 template <typename Action>
1701 bool Session::WriteActionMessage(ActionType type, const Action& action,
1702                                  buzz::XmlElement* stanza,
1703                                  WriteError* error) {
1704   if (current_protocol_ == PROTOCOL_HYBRID) {
1705     if (!WriteActionMessage(PROTOCOL_JINGLE, type, action, stanza, error))
1706       return false;
1707     if (!WriteActionMessage(PROTOCOL_GINGLE, type, action, stanza, error))
1708       return false;
1709   } else {
1710     if (!WriteActionMessage(current_protocol_, type, action, stanza, error))
1711       return false;
1712   }
1713   return true;
1714 }
1715
1716 template <typename Action>
1717 bool Session::WriteActionMessage(SignalingProtocol protocol,
1718                                  ActionType type, const Action& action,
1719                                  buzz::XmlElement* stanza, WriteError* error) {
1720   XmlElements action_elems;
1721   if (!WriteSessionAction(protocol, action, &action_elems, error))
1722     return false;
1723
1724   SessionMessage msg(protocol, type, id(), initiator_name());
1725   msg.to = remote_name();
1726
1727   WriteSessionMessage(msg, action_elems, stanza);
1728   return true;
1729 }
1730
1731 void Session::SendAcknowledgementMessage(const buzz::XmlElement* stanza) {
1732   talk_base::scoped_ptr<buzz::XmlElement> ack(
1733       new buzz::XmlElement(buzz::QN_IQ));
1734   ack->SetAttr(buzz::QN_TO, remote_name());
1735   ack->SetAttr(buzz::QN_ID, stanza->Attr(buzz::QN_ID));
1736   ack->SetAttr(buzz::QN_TYPE, "result");
1737
1738   SignalOutgoingMessage(this, ack.get());
1739 }
1740
1741 }  // namespace cricket