session.cc

本人收集整理的一份c/c++跨平台网络库

}

void Session::DestroyChannel(TransportChannel* channel) {
  ChannelMap::iterator iter = channels_.find(channel->name());
  ASSERT(iter != channels_.end());
  ASSERT(channel == iter->second);
  channels_.erase(iter);
  channel->SignalDestroyed(channel);
  delete channel;
}

TransportChannelImpl* Session::GetImplementation(TransportChannel* channel) {
  ChannelMap::iterator iter = channels_.find(channel->name());
  return (iter != channels_.end()) ? iter->second->impl() : NULL;
}

void Session::CreateTransports() {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  ASSERT((state_ == STATE_INIT)
      || (state_ == STATE_RECEIVEDINITIATE));
  if (potential_transports_.empty()) {
    if (gDefaultTransports == NULL) {
      gNumDefaultTransports = 1;
      gDefaultTransports = new std::string[1];
      gDefaultTransports[0] = kNsP2pTransport;
    }
    SetPotentialTransports(gDefaultTransports, gNumDefaultTransports);
  }
}

bool Session::ChooseTransport(const buzz::XmlElement* stanza) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  ASSERT(state_ == STATE_RECEIVEDINITIATE);
  ASSERT(transport_ == NULL);

  // Make sure we have decided on our own transports.
  CreateTransports();

  // Retrieve the session message.
  const buzz::XmlElement* session = stanza->FirstNamed(QN_SESSION);
  ASSERT(session != NULL);

  // Try the offered transports until we find one that we support.
  bool found_offer = false;
  const buzz::XmlElement* elem = session->FirstElement();
  while (elem) {
    if (elem->Name().LocalPart() == "transport") {
      found_offer = true;
      Transport* transport = GetTransport(elem->Name().Namespace());
      if (transport && transport->OnTransportOffer(elem)) {
        SetTransport(transport);
        break;
      }
    }
    elem = elem->NextElement();
  }

  // If the offer did not include any transports, then we are talking to an
  // old client.  In that case, we turn on compatibility mode, and we assume
  // an offer containing just P2P, which is all that old clients support.
  if (!found_offer) {
    compatibility_mode_ = true;

    Transport* transport = GetTransport(kNsP2pTransport);
    ASSERT(transport != NULL);

    scoped_ptr<buzz::XmlElement> transport_offer(
      new buzz::XmlElement(kQnP2pTransport, true));
    bool valid = transport->OnTransportOffer(transport_offer.get());
    ASSERT(valid);
    if (valid)
      SetTransport(transport);
  }

  if (!transport_) {
    SignalErrorMessage(this, stanza, buzz::QN_STANZA_NOT_ACCEPTABLE, "modify",
                       "no supported transport in offer", NULL);
    return false;
  }

  // Get the description of the transport we picked.
  buzz::XmlElement* answer = transport_->CreateTransportAnswer();
  ASSERT(answer->Name() == buzz::QName(transport_->name(), "transport"));

  // Send a transport-accept message telling the other side our decision,
  // unless this is an old client that is not expecting one.
  if (!compatibility_mode_) {
    XmlElements elems;
    elems.push_back(answer);
    SendSessionMessage("transport-accept", elems);
  }

  // If the user wants to accept, allow that now
  if (description_) {
    Accept(description_);
  }

  return true;
}

void Session::SetTransport(Transport* transport) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  ASSERT(transport_ == NULL);
  transport_ = transport;

  // Drop the transports that were not selected.
  bool found = false;
  for (TransportList::iterator iter = potential_transports_.begin();
       iter != potential_transports_.end();
       ++iter) {
    if (*iter == transport_) {
      found = true;
    } else {
      delete *iter;
    }
  }
  potential_transports_.clear();

  // We require the selected transport to be one of the potential transports
  ASSERT(found);

  // Create implementations for all of the channels if they don't exist.
  for (ChannelMap::iterator iter = channels_.begin();
       iter != channels_.end();
       ++iter) {
    TransportChannelProxy* channel = iter->second;
    TransportChannelImpl* impl = transport_->GetChannel(channel->name());
    if (impl == NULL)
      impl = transport_->CreateChannel(channel->name(), session_type());
    ASSERT(impl != NULL);
    channel->SetImplementation(impl);
  }

  // Have this transport start connecting if it is not already.
  // (We speculatively connect the most common transport right away.)
  transport_->ConnectChannels();
}

void Session::SetState(State state) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  if (state != state_) {
    state_ = state;
    SignalState(this, state_);
    session_manager_->signaling_thread()->Post(this, MSG_STATE);
  }
}

void Session::SetError(Error error) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  if (error != error_) {
    error_ = error;
    SignalError(this, error);
    if (error_ != ERROR_NONE)
      session_manager_->signaling_thread()->Post(this, MSG_ERROR);
  }
}

void Session::OnSignalingReady() {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());

  // Forward this to every transport.  Those that did not request it should
  // ignore this call.
  if (transport_ != NULL) {
    transport_->OnSignalingReady();
  } else {
    for (TransportList::iterator iter = potential_transports_.begin();
        iter != potential_transports_.end();
        ++iter) {
      (*iter)->OnSignalingReady();
    }
  }
}

void Session::OnTransportConnecting(Transport* transport) {
  // This is an indication that we should begin watching the writability
  // state of the transport.
  OnTransportWritable(transport);
}

void Session::OnTransportWritable(Transport* transport) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  ASSERT((NULL == transport_) || (transport == transport_));

  // If the transport is not writable, start a timer to make sure that it
  // becomes writable within a reasonable amount of time.  If it does not, we
  // terminate since we can't actually send data.  If the transport is writable,
  // cancel the timer.  Note that writability transitions may occur repeatedly
  // during the lifetime of the session.

  session_manager_->signaling_thread()->Clear(this, MSG_TIMEOUT);
  if (transport->HasChannels() && !transport->writable()) {
    session_manager_->signaling_thread()->PostDelayed(
        session_manager_->session_timeout() * 1000, this, MSG_TIMEOUT);
  }
}

void Session::OnTransportRequestSignaling(Transport* transport) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  SignalRequestSignaling(this);
}

void Session::OnTransportSendMessage(Transport* transport,
                                     const XmlElements& elems) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  for (size_t i = 0; i < elems.size(); ++i)
    ASSERT(elems[i]->Name() == buzz::QName(transport->name(), "transport"));

  if (compatibility_mode_) {
    // In backward compatibility mode, we send a candidates message.
    XmlElements candidates;
    for (size_t i = 0; i < elems.size(); ++i) {
      for (const buzz::XmlElement* elem = elems[i]->FirstElement();
           elem != NULL;
           elem = elem->NextElement()) {
        ASSERT(elem->Name() == kQnP2pCandidate);

        // Convert this candidate to an old style candidate (namespace change)
        buzz::XmlElement* legacy_candidate = new buzz::XmlElement(*elem);        
        legacy_candidate->SetName(kQnLegacyCandidate);
        candidates.push_back(legacy_candidate);
      }
      delete elems[i];
    }

    SendSessionMessage("candidates", candidates);
  } else {
    // If we haven't finished negotiation, then we may later discover that we
    // need compatibility mode, in which case, we will need to re-send these.
    if ((transport_ == NULL) && (transport->name() == kNsP2pTransport)) {
      for (size_t i = 0; i < elems.size(); ++i)
        candidates_.push_back(new buzz::XmlElement(*elems[i]));
    }

    SendSessionMessage("transport-info", elems);
  }
}

void Session::OnTransportSendError(Transport* transport,
                                   const buzz::XmlElement* stanza,
                                   const buzz::QName& name,
                                   const std::string& type,
                                   const std::string& text,
                                   const buzz::XmlElement* extra_info) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  SignalErrorMessage(this, stanza, name, type, text, extra_info);
}

void Session::OnTransportChannelGone(Transport* transport,
                                     const std::string& name) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  SignalChannelGone(this, name);
}

void Session::SendSessionMessage(
    const std::string& type, const std::vector<buzz::XmlElement*>& elems) {
  scoped_ptr<buzz::XmlElement> iq(new buzz::XmlElement(buzz::QN_IQ));
  iq->SetAttr(buzz::QN_TO, remote_name_);
  iq->SetAttr(buzz::QN_TYPE, buzz::STR_SET);

  buzz::XmlElement *session = new buzz::XmlElement(QN_SESSION, true);
  session->AddAttr(buzz::QN_TYPE, type);
  session->AddAttr(buzz::QN_ID, id_.id_str());
  session->AddAttr(QN_INITIATOR, id_.initiator());

  for (size_t i = 0; i < elems.size(); ++i)
    session->AddElement(elems[i]);

  iq->AddElement(session);
  SignalOutgoingMessage(this, iq.get());
}

void Session::SendAcknowledgementMessage(const buzz::XmlElement* stanza) {
  scoped_ptr<buzz::XmlElement> ack(new buzz::XmlElement(buzz::QN_IQ));
  ack->SetAttr(buzz::QN_TO, remote_name_);
  ack->SetAttr(buzz::QN_ID, stanza->Attr(buzz::QN_ID));
  ack->SetAttr(buzz::QN_TYPE, "result");

  SignalOutgoingMessage(this, ack.get());
}

void Session::OnIncomingMessage(const buzz::XmlElement* stanza) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());
  ASSERT(stanza->Name() == buzz::QN_IQ);
  buzz::Jid remote(remote_name_);
  buzz::Jid from(stanza->Attr(buzz::QN_FROM));
  ASSERT(state_ == STATE_INIT || from == remote);

  const buzz::XmlElement* session = stanza->FirstNamed(QN_SESSION);
  ASSERT(session != NULL);

  if (stanza->Attr(buzz::QN_TYPE) != buzz::STR_SET) {
    ASSERT(false);
    return;
  }

  ASSERT(session->HasAttr(buzz::QN_TYPE));
  std::string type = session->Attr(buzz::QN_TYPE);

  bool valid = false;

  if (type == "initiate") {
    valid = OnInitiateMessage(stanza, session);
  } else if (type == "accept") {
    valid = OnAcceptMessage(stanza, session);
  } else if (type == "reject") {
    valid = OnRejectMessage(stanza, session);
  } else if (type == "redirect") {
    valid = OnRedirectMessage(stanza, session);
  } else if (type == "info") {
    valid = OnInfoMessage(stanza, session);
  } else if (type == "transport-accept") {
    valid = OnTransportAcceptMessage(stanza, session);
  } else if (type == "transport-info") {
    valid = OnTransportInfoMessage(stanza, session);
  } else if (type == "terminate") {
    valid = OnTerminateMessage(stanza, session);
  } else if (type == "candidates") {
    // This is provided for backward compatibility.
    // TODO: Remove this once old candidates are gone.
    valid = OnCandidatesMessage(stanza, session);
  } else {
    SignalErrorMessage(this, stanza, buzz::QN_STANZA_BAD_REQUEST, "modify",
                       "unknown session message type", NULL);
  }

  // If the message was not valid, we should have sent back an error above.
  // If it was valid, then we send an acknowledgement here.
  if (valid)
    SendAcknowledgementMessage(stanza);
}

void Session::OnFailedSend(const buzz::XmlElement* orig_stanza,
                           const buzz::XmlElement* error_stanza) {
  ASSERT(session_manager_->signaling_thread()->IsCurrent());

  const buzz::XmlElement* orig_session = orig_stanza->FirstNamed(QN_SESSION);
  ASSERT(orig_session != NULL);

  std::string error_type = "cancel";

  const buzz::XmlElement* error = error_stanza->FirstNamed(buzz::QN_ERROR);
  ASSERT(error != NULL);
  if (error) {
    ASSERT(error->HasAttr(buzz::QN_TYPE));