tickerproxy.cc

这是P2P流媒体方案－NICE的实现源码

#include <strstream>
#include <netdb.h>
#include <unistd.h>
#include "TickerProxy.h"

TickerProxy::TickerProxy(int port) {
  if((m_Sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
    cout << "ERROR: Could not create socket" << endl;
    exit(-1);
  }
  memset(&m_MyAddr, 0, sizeof(m_MyAddr));
  memset(&m_ReqAddr, 0, sizeof(m_ReqAddr));
  memset(&m_OutAddr, 0, sizeof(m_OutAddr));
  m_ReqAddrLen = sizeof(m_ReqAddr);
  
  m_MyAddr.sin_family = AF_INET;
  m_MyAddr.sin_addr.s_addr = htonl(INADDR_ANY);
  m_MyAddr.sin_port = htons(port);
  if(bind(m_Sock, (sockaddr*)&m_MyAddr, sizeof(m_MyAddr)) < 0) {
    cout << "ERROR: Could not bind socket" << endl;
    exit(-1);
  }
  if(listen(m_Sock, 1) < 0) {
    cout << "ERROR: Could not listen" << endl;
    exit(-1);
  }
}

TickerProxy::~TickerProxy() {

}

void TickerProxy::run() {
  while(1) {
    m_ReqSock = accept(m_Sock, (sockaddr*)&m_ReqAddr, &m_ReqAddrLen);
    if(m_ReqSock < 0) {
      cout << "ERROR: Could not accept connection" << endl;
      exit(-1);
    }
    HandleConnection();
  }
}

int TickerProxy::GetContentLength(string s) {
  istrstream in(s.c_str(), s.size());
  string token;

  while(1) {
    if(!in)
      break;;
    in >> token;
    if(token == "Content-Length:") {
      in >> token;
      return atoi(token.c_str());
    }
  }
  return -1;
}

int TickerProxy::HandleConnection() {
  string s;
  int len;
  HostPort_t dest;
  
  s = ReadMsg(m_ReqSock);
  cout << s << endl;
  dest = GetDestination(s);
  cout << "destination: " << dest.h << ":" << dest.p << endl;
  OpenConnection(dest);
  WriteMsg(s, m_OutSock);
  cout << "Message sent" << endl;
  s = ReadMsg(m_OutSock);
  cout << s << endl;
  len = GetContentLength(s);
  s += ReadMsg(m_OutSock, len);
  //cout << s << endl;
  WriteMsg(s, m_ReqSock);
  close(m_ReqSock);
  close(m_OutSock);
  //cout << s << endl;
  return 0;
}

string TickerProxy::ReadMsg(int sock) {
  string ret, line;

  while(1) {
    line = ReadLine(sock);
    ret += line;
    if(line == "\n" || line == "\r\n")
      break;
  }
  return ret;
}

string TickerProxy::ReadMsg(int sock, int len) {
  char c;
  string ret = "";
  int size = 0, recv_len;

  while(size < len && (recv_len = recv(sock, &c, 1, 0)) == 1) {
    ret += c;
    size++;
    if(len == size)
      break;
  }
  return ret;
}

int TickerProxy::WriteMsg(string Msg, int sock) {
  if(send(sock, Msg.c_str(), Msg.size(), 0) < 0) {
    cout << "ERROR: Could not send packet" << endl;
    return -1;
  }
  return 0;
}

int TickerProxy::OpenConnection(const HostPort_t& dest) {
  if((m_OutSock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
    cout << "ERROR: Could not create outgoing socket" << endl;
  }
  hostent* h;
  h = gethostbyname(dest.h.c_str());
  if(h == NULL) {
    cout << "ERROR: Could not resolve outgoing host" << endl;
    return -1;
  }
  memset(&m_OutAddr, 0, sizeof(m_OutAddr));
  m_OutAddr.sin_family = AF_INET;
  m_OutAddr.sin_addr.s_addr = *((unsigned long*)h->h_addr_list[0]);
  m_OutAddr.sin_port = htons(dest.p);
  if(connect(m_OutSock, (sockaddr*)&m_OutAddr, sizeof(m_OutAddr)) < 0) {
    cout << "ERROR: Could not connect to outgoing host" << endl;
    return -2;
  }
  return 0;
}

string TickerProxy::ReadLine(int sock) {
  string ret = "";
  char c;
  int recv_len;
  
  while((recv_len = recv(sock, &c, 1, 0)) == 1) {
    ret += c;
    if(c == '\n')
      break;
  }
  return ret;
}

HostPort_t TickerProxy::GetDestination(string Msg) {
  HostPort_t ret;
  istrstream in(Msg.c_str(), Msg.size());
  string token;
  bool found = false;

  ret.h = "";
  ret.p = 80;

  while(!found) {
    if(!in)
      return ret;
    in >> token;
    if(token.find("http://", 0, 7) == 0) {
      cout << "Token: " << token << endl;
      found = true;
      ret.h = token.substr(7, token.find_first_of('/', 7)-7);
    }
  }
  return ret;
}