session_unittest.cc

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

#include <iostream>
#include <sstream>
#include <deque>
#include <map>

#include "talk/base/common.h"
#include "talk/base/logging.h"
#include "talk/base/host.h"
#include "talk/base/natserver.h"
#include "talk/base/natsocketfactory.h"
#include "talk/base/helpers.h"
#include "talk/xmpp/constants.h"
#include "talk/p2p/base/constants.h"
#include "talk/p2p/base/sessionmanager.h"
#include "talk/p2p/base/sessionclient.h"
#include "talk/p2p/base/session.h"
#include "talk/p2p/base/portallocator.h"
#include "talk/p2p/base/transportchannel.h"
#include "talk/p2p/base/udpport.h"
#include "talk/p2p/base/stunport.h"
#include "talk/p2p/base/relayport.h"
#include "talk/p2p/base/p2ptransport.h"
#include "talk/p2p/base/rawtransport.h"
#include "talk/p2p/base/stunserver.h"
#include "talk/p2p/base/relayserver.h"

using namespace cricket;
using namespace buzz;

const std::string kSessionType = "http://oink.splat/session";

const talk_base::SocketAddress kStunServerAddress("127.0.0.1", 7000);
const talk_base::SocketAddress kStunServerAddress2("127.0.0.1", 7001);

const talk_base::SocketAddress kRelayServerIntAddress("127.0.0.1", 7002);
const talk_base::SocketAddress kRelayServerExtAddress("127.0.0.1", 7003);

const int kNumPorts = 2;

int gPort = 28653;
int GetNextPort() {
  int p = gPort;
  gPort += 5;
  return p;
}

int gID = 0;
std::string GetNextID() {
  std::ostringstream ost;
  ost << gID++;
  return ost.str();
}

class TestPortAllocatorSession : public PortAllocatorSession {
public:
  TestPortAllocatorSession(talk_base::Thread* worker_thread, talk_base::SocketFactory* factory,
                           const std::string& name, const std::string& session_type)
      : PortAllocatorSession(0), worker_thread_(worker_thread),
        factory_(factory), name_(name), ports_(kNumPorts),
        address_("127.0.0.1", 0), network_("network", address_.ip()),
        running_(false) {
  }

  ~TestPortAllocatorSession() {
    for (int i = 0; i < ports_.size(); i++)
      delete ports_[i];
  }

  virtual void GetInitialPorts() {
    // These are the flags set by the raw transport.
    uint32 raw_flags = PORTALLOCATOR_DISABLE_UDP | PORTALLOCATOR_DISABLE_TCP;

    // If the client doesn't care, just give them two UDP ports.
    if (flags() == 0) {
      for (int i = 0; i < kNumPorts; i++) {
        ports_[i] = new UDPPort(worker_thread_, factory_, &network_,
                                GetAddress());
        AddPort(ports_[i]);
      }

    // If the client requested just stun and relay, we have to oblidge.
    } else if (flags() == raw_flags) {
      StunPort* sport = new StunPort(worker_thread_, factory_, &network_,
                                     GetAddress(), kStunServerAddress);
      sport->set_server_addr2(kStunServerAddress2);
      ports_[0] = sport;
      AddPort(sport);

      std::string username = CreateRandomString(16);
      std::string password = CreateRandomString(16);
      RelayPort* rport = new RelayPort(worker_thread_, factory_, &network_,
                                       GetAddress(), username, password, "");
      rport->AddServerAddress(
          ProtocolAddress(kRelayServerIntAddress, PROTO_UDP));
      ports_[1] = rport;
      AddPort(rport);
    } else {
      ASSERT(false);
    }
  }

  virtual void StartGetAllPorts() { running_ = true; }
  virtual void StopGetAllPorts() { running_ = false; }
  virtual bool IsGettingAllPorts() { return running_; }

  talk_base::SocketAddress GetAddress() const {
    talk_base::SocketAddress addr(address_);
    addr.SetPort(GetNextPort());
    return addr;
  }

  void AddPort(Port* port) {
    port->set_name(name_);
    port->set_preference(1.0);
    port->set_generation(0);
    port->SignalDestroyed.connect(
        this, &TestPortAllocatorSession::OnPortDestroyed);
    port->SignalAddressReady.connect(
        this, &TestPortAllocatorSession::OnAddressReady);
    port->PrepareAddress();
    SignalPortReady(this, port);
  }

  void OnPortDestroyed(Port* port) {
    for (int i = 0; i < ports_.size(); i++) {
      if (ports_[i] == port)
        ports_[i] = NULL;
    }
  }

  void OnAddressReady(Port* port) {
    SignalCandidatesReady(this, port->candidates());
  }

private:
  talk_base::Thread* worker_thread_;
  talk_base::SocketFactory* factory_;
  std::string name_;
  std::vector<Port*> ports_;
  talk_base::SocketAddress address_;
  talk_base::Network network_;
  bool running_;
};

class TestPortAllocator : public PortAllocator {
public:
  TestPortAllocator(talk_base::Thread* worker_thread, talk_base::SocketFactory* factory)
    : worker_thread_(worker_thread), factory_(factory) {
    if (factory_ == NULL)
      factory_ = worker_thread_->socketserver();
  }

  virtual PortAllocatorSession *CreateSession(const std::string &name, const std::string &session_type) {
    return new TestPortAllocatorSession(worker_thread_, factory_, name, session_type);
  }

private:
  talk_base::Thread* worker_thread_;
  talk_base::SocketFactory* factory_;
};

struct SessionManagerHandler : sigslot::has_slots<> {
  SessionManagerHandler(SessionManager* m, const std::string& u)
      : manager(m), username(u), create_count(0), destroy_count(0) {
    manager->SignalSessionCreate.connect(
        this, &SessionManagerHandler::OnSessionCreate);
    manager->SignalSessionDestroy.connect(
        this, &SessionManagerHandler::OnSessionDestroy);
    manager->SignalOutgoingMessage.connect(
        this, &SessionManagerHandler::OnOutgoingMessage);
    manager->SignalRequestSignaling.connect(
        this, &SessionManagerHandler::OnRequestSignaling);
  }

  void OnSessionCreate(Session *session, bool initiate) {
    create_count += 1;
    last_id = session->id();
  }

  void OnSessionDestroy(Session *session) {
    destroy_count += 1;
    last_id = session->id();
  }

  void OnOutgoingMessage(const XmlElement* stanza) {
    XmlElement* elem = new XmlElement(*stanza);
    ASSERT(elem->Name() == QN_IQ);
    ASSERT(elem->HasAttr(QN_TO));
    ASSERT(!elem->HasAttr(QN_FROM));
    ASSERT(elem->HasAttr(QN_TYPE));
    ASSERT((elem->Attr(QN_TYPE) == "set") ||
           (elem->Attr(QN_TYPE) == "result") ||
           (elem->Attr(QN_TYPE) == "error"));

    // Add in the appropriate "from".
    elem->SetAttr(QN_FROM, username);

    // Add in the appropriate IQ ID.
    if (elem->Attr(QN_TYPE) == "set") {
      ASSERT(!elem->HasAttr(QN_ID));
      elem->SetAttr(QN_ID, GetNextID());
    }

    stanzas_.push_back(elem);
  }

  void OnRequestSignaling() {
    manager->OnSignalingReady();
  }


  XmlElement* CheckNextStanza(const std::string& expected) {
    // Get the next stanza, which should exist.
    ASSERT(stanzas_.size() > 0);
    XmlElement* stanza = stanzas_.front();
    stanzas_.pop_front();

    // Make sure the stanza is correct.
    std::string actual = stanza->Str();
    if (actual != expected) {
      LOG(LERROR) << "Incorrect stanza: expected=\"" << expected
                  << "\" actual=\"" << actual << "\"";
      ASSERT(actual == expected);
    }

    return stanza;
  }

  void CheckNoStanza() {
    ASSERT(stanzas_.size() == 0);
  }

  void PrintNextStanza() {
    ASSERT(stanzas_.size() > 0);
    printf("Stanza: %s\n", stanzas_.front()->Str().c_str());
  }

  SessionManager* manager;
  std::string username;
  SessionID last_id;
  uint32 create_count;
  uint32 destroy_count;
  std::deque<XmlElement*> stanzas_;
};

struct SessionHandler : sigslot::has_slots<> {
  SessionHandler(Session* s) : session(s) {
    session->SignalState.connect(this, &SessionHandler::OnState);
    session->SignalError.connect(this, &SessionHandler::OnError);
  }

  void PrepareTransport() {
    Transport* transport = session->GetTransport(kNsP2pTransport);
    if (transport != NULL)
      transport->set_allow_local_ips(true);
  }

  void OnState(Session* session, Session::State state) {
    ASSERT(session == this->session);
    last_state = state;
  }

  void OnError(Session* session, Session::Error error) {
    ASSERT(session == this->session);
    ASSERT(false); // errors are bad!
  }

  Session* session;
  Session::State last_state;
};

struct MySessionClient: public SessionClient, public sigslot::has_slots<> {
  MySessionClient() : create_count(0), a(NULL), b(NULL) { }

  void AddManager(SessionManager* manager) {
    manager->AddClient(kSessionType, this);
    ASSERT(manager->GetClient(kSessionType) == this);
    manager->SignalSessionCreate.connect(
        this, &MySessionClient::OnSessionCreate);
  }

  const SessionDescription* CreateSessionDescription(
      const XmlElement* element) {
    return new SessionDescription();
  }

  XmlElement* TranslateSessionDescription(
      const SessionDescription* description) {
    return new XmlElement(QName(kSessionType, "description"));
  }

  void OnSessionCreate(Session *session, bool initiate) {
    create_count += 1;
    a = session->CreateChannel("a");
    b = session->CreateChannel("b");

    if (transport_name.size() > 0)
      session->SetPotentialTransports(&transport_name, 1);
  }

  void OnSessionDestroy(Session *session)
  {
  }
  
  void SetTransports(bool p2p, bool raw) {
    if (p2p && raw)
      return;  // this is the default

    if (p2p) {
      transport_name = kNsP2pTransport;
    }
  }

  int create_count;
  TransportChannel* a;
  TransportChannel* b;
  std::string transport_name;
};

struct ChannelHandler : sigslot::has_slots<> {
  ChannelHandler(TransportChannel* p)
    : channel(p), last_readable(false), last_writable(false), data_count(0),
      last_size(0) {
    p->SignalReadableState.connect(this, &ChannelHandler::OnReadableState);
    p->SignalWritableState.connect(this, &ChannelHandler::OnWritableState);
    p->SignalReadPacket.connect(this, &ChannelHandler::OnReadPacket);
  }

  void OnReadableState(TransportChannel* p) {
    ASSERT(p == channel);
    last_readable = channel->readable();
  }

  void OnWritableState(TransportChannel* p) {
    ASSERT(p == channel);
    last_writable = channel->writable();
  }

  void OnReadPacket(TransportChannel* p, const char* buf, size_t size) {
    ASSERT(p == channel);
    ASSERT(size <= sizeof(last_data));
    data_count += 1;
    last_size = size;
    std::memcpy(last_data, buf, size);
  }

  void Send(const char* data, size_t size) {