开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > P2P编程 > 查看源码
servent.cpp
资源名称：P2PSource-PeerCast.rar　[点击查看]
上传用户：chn_coc
上传日期：2007-12-20
资源大小：563k
文件大小：58k
源码类别：

P2P编程

开发平台：

Windows_Unix

servent.cpp：源码内容
  1. // ------------------------------------------------ // File : servent.cpp // Date: 4-apr-2002 // Author: giles // Desc:  // Servents are the actual connections between clients. They do the handshaking, // transfering of data and processing of GnuPackets. Each servent has one socket allocated // to it on connect, it uses this to transfer all of its data. // // (c) 2002 peercast.org // ------------------------------------------------ // This program is free software; you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation; either version 2 of the License, or // (at your option) any later version. // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the // GNU General Public License for more details. // ------------------------------------------------ // todo: make lan->yp not check firewall #include <stdlib.h> #include "servent.h" #include "sys.h" #include "gnutella.h" #include "xml.h" #include "html.h" #include "http.h" #include "stats.h" #include "servmgr.h" #include "peercast.h" #include "atom.h"
  2. #include "pcp.h"
  3. #include "version2.h"
  6. const int DIRECT_WRITE_TIMEOUT = 60; // ----------------------------------- char *Servent::statusMsgs[]= {         "NONE", "CONNECTING",         "PROTOCOL",         "HANDSHAKE",         "CONNECTED",         "CLOSING", "LISTENING", "TIMEOUT", "REFUSED", "VERIFIED", "ERROR", "WAIT",
  7. "FREE" }; // ----------------------------------- char *Servent::typeMsgs[]= {
  8. "NONE",         "INCOMING",
  9.         "SERVER", "RELAY", "DIRECT",
  10. "COUT",
  11. "CIN",
  12. "PGNU"
  13. }; // ----------------------------------- bool Servent::isPrivate()  { Host h = getHost(); return servMgr->isFiltered(ServFilter::F_PRIVATE,h) || h.isLocalhost(); } // ----------------------------------- bool Servent::isAllowed(int a)  { Host h = getHost(); if (servMgr->isFiltered(ServFilter::F_BAN,h)) return false; return (allow&a)!=0; }
  14. // -----------------------------------
  15. bool Servent::isFiltered(int f) 
  16. {
  17. Host h = getHost();
  18. return servMgr->isFiltered(f,h);
  19. }
  20. // ----------------------------------- Servent::Servent(int index) :outPacketsPri(MAX_OUTPACKETS) ,outPacketsNorm(MAX_OUTPACKETS) ,seenIDs(MAX_HASH) ,serventIndex(index) ,sock(NULL) ,next(NULL) { reset(); } // ----------------------------------- Servent::~Servent() { } // ----------------------------------- void Servent::kill()  {
  21. thread.active = false;
  23. setStatus(S_CLOSING);
  25. if (pcpStream)
  26. {
  27. PCPStream *pcp = pcpStream;
  28. pcpStream = NULL;
  29. pcp->kill();
  30. delete pcp;
  31. }
  34. if (sock)
  35. {
  36. sock->close();
  37. delete sock;
  38. sock = NULL;
  39. }
  41. if (pushSock)
  42. {
  43. pushSock->close();
  44. delete pushSock;
  45. pushSock = NULL;
  46. }
  47. // thread.unlock();
  49. if (type != T_SERVER) { reset(); setStatus(S_FREE); }
  50. } // ----------------------------------- void Servent::abort()  { thread.active = false; if (sock)
  51. { sock->close();
  52. } } // ----------------------------------- void Servent::reset() {
  54. remoteID.clear();
  56. servPort = 0;
  58. pcpStream = NULL;
  60. flowControl = false; networkID.clear();
  61. chanID.clear();
  63. outputProtocol = ChanInfo::SP_UNKNOWN;
  64. agent.clear(); sock = NULL; allow = ALLOW_ALL; syncPos = 0; addMetadata = false;
  65. nsSwitchNum = 0;
  66. pack.func = 255; lastConnect = lastPing = lastPacket = 0; loginPassword[0] = 0; loginMount[0] = 0; bytesPerSecond = 0; priorityConnect = false;
  67. pushSock = NULL;
  68. sendHeader = true; outPacketsNorm.reset(); outPacketsPri.reset(); seenIDs.clear(); status = S_NONE; type = T_NONE; }
  69. // -----------------------------------
  70. bool Servent::sendPacket(ChanPacket &pack,GnuID &cid,GnuID &sid,GnuID &did,Servent::TYPE t)
  71. {
  73. if  (    (type == t) 
  74. && (isConnected())
  75. && (!cid.isSet() || chanID.isSame(cid))
  76. && (!sid.isSet() || !sid.isSame(remoteID))
  77. && (pcpStream != NULL)
  78. )
  79. {
  80. return pcpStream->sendPacket(pack,did);
  81. }
  82. return false;
  83. }
  86. // -----------------------------------
  87. bool Servent::acceptGIV(ClientSocket *givSock)
  88. {
  89. if (!pushSock)
  90. {
  91. pushSock = givSock;
  92. return true;
  93. }else
  94. return false;
  95. }
  96. // ----------------------------------- Host Servent::getHost() { Host h(0,0); if (sock) h = sock->host; return h; } // ----------------------------------- bool Servent::outputPacket(GnuPacket &p, bool pri) { lock.on(); bool r=false; if (pri) r = outPacketsPri.write(p); else { if (servMgr->useFlowControl) { int per = outPacketsNorm.percentFull(); if (per > 50) flowControl = true; else if (per < 25) flowControl = false; } bool send=true; if (flowControl) { // if in flowcontrol, only allow packets with less of a hop count than already in queue if (p.hops >= outPacketsNorm.findMinHop()) send = false; } if (send) r = outPacketsNorm.write(p); } lock.off(); return r; } // ----------------------------------- bool Servent::initServer(Host &h) { try { checkFree(); status = S_WAIT; createSocket(); sock->bind(h); thread.data = this; thread.func = serverProc;
  98. type = T_SERVER; if (!sys->startThread(&thread)) throw StreamException("Can`t start thread"); }catch(StreamException &e) { LOG_ERROR("Bad server: %s",e.msg); kill(); return false; } return true; } // ----------------------------------- void Servent::checkFree() { if (sock) throw StreamException("Socket already set"); if (thread.active) throw StreamException("Thread already active"); } // ----------------------------------- void Servent::initIncoming(ClientSocket *s, unsigned int a) { try{ checkFree(); type = T_INCOMING; sock = s; allow = a; thread.data = this; thread.func = incomingProc; setStatus(S_PROTOCOL);
  100. char ipStr[64];
  101. sock->host.toStr(ipStr);
  102. LOG_DEBUG("Incoming from %s",ipStr);
  103. if (!sys->startThread(&thread)) throw StreamException("Can`t start thread"); }catch(StreamException &e) { //LOG_ERROR("!!FATAL!! Incoming error: %s",e.msg); //servMgr->shutdownTimer = 1;  
  104. kill();
  106. LOG_ERROR("INCOMING FAILED: %s",e.msg);
  107. } } // ----------------------------------- void Servent::initOutgoing(TYPE ty) { try  { checkFree(); type = ty; thread.data = this; thread.func = outgoingProc; if (!sys->startThread(&thread)) throw StreamException("Can`t start thread"); }catch(StreamException &e) { LOG_ERROR("Unable to start outgoing: %s",e.msg); kill(); } }
  108. // -----------------------------------
  109. void Servent::initPCP(Host &rh)
  110. {
  111. char ipStr[64];
  112. rh.toStr(ipStr);
  113. try 
  114. {
  115. checkFree();
  117.     createSocket();
  119. type = T_COUT;
  121. sock->open(rh);
  123. if (!isAllowed(ALLOW_NETWORK))
  124. throw StreamException("Servent not allowed");
  126. thread.data = this;
  127. thread.func = outgoingProc;
  129. LOG_DEBUG("Outgoing to %s",ipStr);
  131. if (!sys->startThread(&thread))
  132. throw StreamException("Can`t start thread");
  134. }catch(StreamException &e)
  135. {
  136. LOG_ERROR("Unable to open connection to %s - %s",ipStr,e.msg);
  137. kill();
  138. }
  139. }
  141. #if 0
  142. // -----------------------------------
  143. void Servent::initChannelFetch(Host &host)
  144. {
  145. type = T_STREAM;
  147. char ipStr[64];
  148. host.toStr(ipStr);
  150. checkFree();
  151.  
  152. createSocket();
  154. sock->open(host);
  157. if (!isAllowed(ALLOW_DATA))
  158. throw StreamException("Servent not allowed");
  160. sock->connect();
  161. }
  162. #endif
  164. // ----------------------------------- void Servent::initGIV(Host &h, GnuID &id) { char ipStr[64]; h.toStr(ipStr); try  { checkFree(); givID = id;     createSocket(); sock->open(h); if (!isAllowed(ALLOW_NETWORK)) throw StreamException("Servent not allowed"); sock->connect(); thread.data = this; thread.func = givProc; type = T_RELAY; if (!sys->startThread(&thread)) throw StreamException("Can`t start thread"); }catch(StreamException &e) { LOG_ERROR("GIV error to %s: %s",ipStr,e.msg); kill(); } } // ----------------------------------- void Servent::createSocket() { if (sock) LOG_ERROR("Servent::createSocket attempt made while active"); sock = sys->createSocket(); } // ----------------------------------- void Servent::setStatus(STATUS s) { if (s != status) { status = s; if ((s == S_HANDSHAKE) || (s == S_CONNECTED) || (s == S_LISTENING)) lastConnect = sys->getTime(); } }
  165. // ----------------------------------- void Servent::handshakeOut() {     sock->writeLine(GNU_PEERCONN); char str[64];      sock->writeLineF("%s %s",HTTP_HS_AGENT,PCX_AGENT);     sock->writeLineF("%s %d",PCX_HS_PCP,1);
  166. if (priorityConnect)     sock->writeLineF("%s %d",PCX_HS_PRIORITY,1); if (networkID.isSet()) { networkID.toStr(str); sock->writeLineF("%s %s",PCX_HS_NETWORKID,str); } servMgr->sessionID.toStr(str); sock->writeLineF("%s %s",PCX_HS_ID,str);     sock->writeLineF("%s %s",PCX_HS_OS,peercastApp->getClientTypeOS()); sock->writeLine(""); HTTP http(*sock);
  167. int r = http.readResponse();
  169. if (r != 200)
  170. {
  171. LOG_ERROR("Expected 200, got %d",r);
  172. throw StreamException("Unexpected HTTP response");
  173. }
  174. bool versionValid = false; GnuID clientID; clientID.clear();     while (http.nextHeader())     { LOG_DEBUG(http.cmdLine); char *arg = http.getArgStr(); if (!arg) continue; if (http.isHeader(HTTP_HS_AGENT)) { agent.set(arg); if (strnicmp(arg,"PeerCast/",9)==0) versionValid = (stricmp(arg+9,MIN_CONNECTVER)>=0); }else if (http.isHeader(PCX_HS_NETWORKID)) clientID.fromStr(arg);     }
  175. if (!clientID.isSame(networkID)) throw HTTPException(HTTP_SC_UNAVAILABLE,503); if (!versionValid) throw HTTPException(HTTP_SC_UNAUTHORIZED,401);     sock->writeLine(GNU_OK);     sock->writeLine("");
  177. }
  179. // -----------------------------------
  180. void Servent::processOutChannel()
  181. {
  182. }
  185. // ----------------------------------- void Servent::handshakeIn() { int osType=0; HTTP http(*sock); bool versionValid = false; bool diffRootVer = false;
  186. GnuID clientID; clientID.clear();     while (http.nextHeader())     { LOG_DEBUG(http.cmdLine); char *arg = http.getArgStr(); if (!arg) continue; if (http.isHeader(HTTP_HS_AGENT)) { agent.set(arg); if (strnicmp(arg,"PeerCast/",9)==0) { versionValid = (stricmp(arg+9,MIN_CONNECTVER)>=0); diffRootVer = stricmp(arg+9,MIN_ROOTVER)<0; } }else if (http.isHeader(PCX_HS_NETWORKID)) { clientID.fromStr(arg); }else if (http.isHeader(PCX_HS_PRIORITY)) { priorityConnect = atoi(arg)!=0;
  187. }else if (http.isHeader(PCX_HS_ID)) { GnuID id; id.fromStr(arg); if (id.isSame(servMgr->sessionID)) throw StreamException("Servent loopback"); }else if (http.isHeader(PCX_HS_OS)) { if (stricmp(arg,PCX_OS_LINUX)==0) osType = 1; else if (stricmp(arg,PCX_OS_WIN32)==0) osType = 2; else if (stricmp(arg,PCX_OS_MACOSX)==0) osType = 3; else if (stricmp(arg,PCX_OS_WINAMP2)==0) osType = 4; }     }
  189. if (!clientID.isSame(networkID)) throw HTTPException(HTTP_SC_UNAVAILABLE,503); // if this is a priority connection and all incoming connections  // are full then kill an old connection to make room. Otherwise reject connection. //if (!priorityConnect) { if (!isPrivate()) if (servMgr->pubInFull()) throw HTTPException(HTTP_SC_UNAVAILABLE,503); } if (!versionValid) throw HTTPException(HTTP_SC_FORBIDDEN,403);     sock->writeLine(GNU_OK);     sock->writeLineF("%s %s",HTTP_HS_AGENT,PCX_OLDAGENT); if (networkID.isSet()) { char idStr[64]; networkID.toStr(idStr); sock->writeLineF("%s %s",PCX_HS_NETWORKID,idStr); } if (servMgr->isRoot) { sock->writeLineF("%s %d",PCX_HS_FLOWCTL,servMgr->useFlowControl?1:0); sock->writeLineF("%s %d",PCX_HS_MINBCTTL,chanMgr->minBroadcastTTL); sock->writeLineF("%s %d",PCX_HS_MAXBCTTL,chanMgr->maxBroadcastTTL); sock->writeLineF("%s %d",PCX_HS_RELAYBC,servMgr->relayBroadcast); //sock->writeLine("%s %d",PCX_HS_FULLHIT,2); if (diffRootVer) { sock->writeString(PCX_HS_DL); sock->writeLine(PCX_DL_URL); } sock->writeLineF("%s %s",PCX_HS_MSG,servMgr->rootMsg.cstr()); } char hostIP[64]; Host h = sock->host; h.IPtoStr(hostIP);     sock->writeLineF("%s %s",PCX_HS_REMOTEIP,hostIP);     sock->writeLine(""); while (http.nextHeader());
  190. }
  191. // -----------------------------------
  192. bool Servent::pingHost(Host &rhost,GnuID &rsid)
  193. {
  194. char ipstr[64];
  195. rhost.toStr(ipstr);
  196. LOG_DEBUG("Ping host %s: trying..",ipstr);
  197. ClientSocket *s=NULL;
  198. bool hostOK=false;
  199. try
  200. {
  201. s = sys->createSocket();
  202. if (!s)
  203. return false;
  204. else
  205. {
  207. s->setReadTimeout(15000);
  208. s->setWriteTimeout(15000);
  209. s->open(rhost);
  210. s->connect();
  212. AtomStream atom(*s);
  214. atom.writeInt(PCP_CONNECT,1);
  215. atom.writeParent(PCP_HELO,1);
  216. atom.writeBytes(PCP_HELO_SESSIONID,servMgr->sessionID.id,16);
  218. GnuID sid;
  219. sid.clear();
  221. int numc,numd;
  222. ID4 id = atom.read(numc,numd);
  223. if (id == PCP_OLEH)
  224. {
  225. for(int i=0; i<numc; i++)
  226. {
  227. int c,d;
  228. ID4 pid = atom.read(c,d);
  229. if (pid == PCP_SESSIONID)
  230. atom.readBytes(sid.id,16,d);
  231. else
  232. atom.skip(c,d);
  233. }
  234. }else
  235. {
  236. LOG_DEBUG("Ping response: %s",id.getString().str());
  237. throw StreamException("Bad ping response");
  238. }
  240. if (!sid.isSame(rsid))
  241. throw StreamException("SIDs don`t match");
  243. hostOK = true;
  244. LOG_DEBUG("Ping host %s: OK",ipstr);
  245. atom.writeInt(PCP_QUIT,PCP_ERROR_QUIT);
  248. }
  249. }catch(StreamException &e)
  250. {
  251. LOG_DEBUG("Ping host %s: %s",ipstr,e.msg);
  252. }
  253. if (s)
  254. {
  255. s->close();
  256. delete s;
  257. }
  259. if (!hostOK)
  260. rhost.port = 0;
  262. return true;
  263. }
  266. // ----------------------------------- bool Servent::handshakeStream(ChanInfo &chanInfo) {
  268. HTTP http(*sock);
  270. bool gotPCP=false;
  271. unsigned int reqPos=0;
  273. nsSwitchNum=0;
  275. while (http.nextHeader()) { char *arg = http.getArgStr(); if (!arg) continue; if (http.isHeader(PCX_HS_PCP))
  276. gotPCP = atoi(arg)!=0;
  277. else if (http.isHeader(PCX_HS_POS))
  278. reqPos = atoi(arg); else if (http.isHeader("icy-metadata")) addMetadata = atoi(arg) > 0;
  279. else if (http.isHeader(HTTP_HS_AGENT))
  280. agent = arg;
  281. else if (http.isHeader("Pragma"))
  282. {
  283. char *ssc = stristr(arg,"stream-switch-count=");
  284. char *so = stristr(arg,"stream-offset");
  286. if (ssc || so)
  287. {
  288. nsSwitchNum=1;
  289. //nsSwitchNum = atoi(ssc+20);
  290. }
  291. }
  292. LOG_DEBUG("Stream: %s",http.cmdLine); }
  295. if ((!gotPCP) && (outputProtocol == ChanInfo::SP_PCP))
  296. outputProtocol = ChanInfo::SP_PEERCAST;
  298. if (outputProtocol == ChanInfo::SP_HTTP)
  299. {
  300. if  ( (chanInfo.srcProtocol == ChanInfo::SP_MMS)
  301.   || (chanInfo.contentType == ChanInfo::T_WMA)
  302.   || (chanInfo.contentType == ChanInfo::T_WMV)
  303.   || (chanInfo.contentType == ChanInfo::T_ASX)
  304. )
  305. outputProtocol = ChanInfo::SP_MMS;
  306. }
  309. bool chanFound=false;
  310. bool chanReady=false;
  312. Channel *ch = chanMgr->findChannelByID(chanInfo.id);
  313. if (ch)
  314. {
  315. sendHeader = true;
  316. if (reqPos)
  317. {
  318. streamPos = ch->rawData.findOldestPos(reqPos);
  319. }else
  320. {
  321. streamPos = ch->rawData.getLatestPos();
  322. }
  324. chanReady = canStream(ch);
  325. }
  327. ChanHitList *chl = chanMgr->findHitList(chanInfo);
  328. if (chl)
  329. {
  330. chanFound = true;
  331. }
  334. bool result = false;
  335. char idStr[64];
  336. chanInfo.id.toStr(idStr);
  338. char sidStr[64];
  339. servMgr->sessionID.toStr(sidStr);
  341. Host rhost = sock->host;
  346. AtomStream atom(*sock);
  350. if (!chanFound)
  351. {
  352. sock->writeLine(HTTP_SC_NOTFOUND);
  353.     sock->writeLine("");
  354. LOG_DEBUG("Sending channel not found");
  355. return false;
  356. }
  359. if (!chanReady)
  360. {
  361. if (outputProtocol == ChanInfo::SP_PCP)
  362. {
  364. sock->writeLine(HTTP_SC_UNAVAILABLE);
  365. sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_XPCP);
  366. sock->writeLine("");
  368. handshakeIncomingPCP(atom,rhost,remoteID,agent);
  370. char ripStr[64];
  371. rhost.toStr(ripStr);
  373. LOG_DEBUG("Sending channel unavailable");
  375. ChanHitSearch chs;
  377. int error = PCP_ERROR_QUIT+PCP_ERROR_UNAVAILABLE;
  379. if (chl)
  380. {
  381. ChanHit best;
  383. // search for up to 8 other hits
  384. int cnt=0;
  385. for(int i=0; i<8; i++)
  386. {
  387. best.init();
  390. // find best hit this network if local IP
  391. if (!rhost.globalIP())
  392. {
  393. chs.init();
  394. chs.matchHost = servMgr->serverHost;
  395. chs.waitDelay = 2;
  396. chs.excludeID = remoteID;
  397. if (chl->pickHits(chs))
  398. best = chs.best[0];
  399. }
  401. // find best hit on same network
  402. if (!best.host.ip)
  403. {
  404. chs.init();
  405. chs.matchHost = rhost;
  406. chs.waitDelay = 2;
  407. chs.excludeID = remoteID;
  408. if (chl->pickHits(chs))
  409. best = chs.best[0];
  411. }
  413. // find best hit on other networks
  414. if (!best.host.ip)
  415. {
  416. chs.init();
  417. chs.waitDelay = 2;
  418. chs.excludeID = remoteID;
  419. if (chl->pickHits(chs))
  420. best = chs.best[0];
  422. }
  424. if (!best.host.ip)
  425. break;
  427. best.writeAtoms(atom,chanInfo.id);
  428. cnt++;
  429. }
  431. if (cnt)
  432. {
  433. LOG_DEBUG("Sent %d channel hit(s) to %s",cnt,ripStr);
  435. }
  436. else if (rhost.port)
  437. {
  438. // find firewalled host
  439. chs.init();
  440. chs.waitDelay = 30;
  441. chs.useFirewalled = true;
  442. chs.excludeID = remoteID;
  443. if (chl->pickHits(chs))
  444. {
  445. best = chs.best[0];
  446. int cnt = servMgr->broadcastPushRequest(best,rhost,chl->info.id,Servent::T_RELAY);
  447. LOG_DEBUG("Broadcasted channel push request to %d clients for %s",cnt,ripStr);
  448. }
  449. } 
  451. // if all else fails, use tracker
  452. if (!best.host.ip)
  453. {
  454. // find best tracker on this network if local IP
  455. if (!rhost.globalIP())
  456. {
  457. chs.init();
  458. chs.matchHost = servMgr->serverHost;
  459. chs.trackersOnly = true;
  460. chs.excludeID = remoteID;
  461. if (chl->pickHits(chs))
  462. best = chs.best[0];
  464. }
  466. // find local tracker
  467. if (!best.host.ip)
  468. {
  469. chs.init();
  470. chs.matchHost = rhost;
  471. chs.trackersOnly = true;
  472. chs.excludeID = remoteID;
  473. if (chl->pickHits(chs))
  474. best = chs.best[0];
  475. }
  477. // find global tracker
  478. if (!best.host.ip)
  479. {
  480. chs.init();
  481. chs.trackersOnly = true;
  482. chs.excludeID = remoteID;
  483. if (chl->pickHits(chs))
  484. best = chs.best[0];
  485. }
  487. if (best.host.ip)
  488. {
  489. best.writeAtoms(atom,chanInfo.id);
  490. LOG_DEBUG("Sent 1 tracker hit to %s",ripStr);
  491. }else if (rhost.port)
  492. {
  493. // find firewalled tracker
  494. chs.init();
  495. chs.useFirewalled = true;
  496. chs.trackersOnly = true;
  497. chs.excludeID = remoteID;
  498. chs.waitDelay = 30;
  499. if (chl->pickHits(chs))
  500. {
  501. best = chs.best[0];
  502. int cnt = servMgr->broadcastPushRequest(best,rhost,chl->info.id,Servent::T_CIN);
  503. LOG_DEBUG("Broadcasted tracker push request to %d clients for %s",cnt,ripStr);
  504. }
  505. }
  507. }
  510. }
  511. // return not available yet code
  512. atom.writeInt(PCP_QUIT,error);
  513. result = false;
  515. }else
  516. {
  517. LOG_DEBUG("Sending channel unavailable");
  518. sock->writeLine(HTTP_SC_UNAVAILABLE);
  519. sock->writeLine("");
  520. result = false;
  521. }
  523. } else {
  525. if (chanInfo.contentType != ChanInfo::T_MP3)
  526. addMetadata=false;
  527. if (addMetadata && (outputProtocol == ChanInfo::SP_HTTP)) // winamp mp3 metadata check {
  528. sock->writeLine(ICY_OK); sock->writeLineF("%s %s",HTTP_HS_SERVER,PCX_AGENT); sock->writeLineF("icy-name:%s",chanInfo.name.cstr()); sock->writeLineF("icy-br:%d",chanInfo.bitrate); sock->writeLineF("icy-genre:%s",chanInfo.genre.cstr()); sock->writeLineF("icy-url:%s",chanInfo.url.cstr()); sock->writeLineF("icy-metaint:%d",chanMgr->icyMetaInterval); sock->writeLineF("%s %s",PCX_HS_CHANNELID,idStr); sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_MP3); }else { sock->writeLine(HTTP_SC_OK); if ((chanInfo.contentType != ChanInfo::T_ASX) && (chanInfo.contentType != ChanInfo::T_WMV) && (chanInfo.contentType != ChanInfo::T_WMA)) { sock->writeLineF("%s %s",HTTP_HS_SERVER,PCX_AGENT); sock->writeLine("Accept-Ranges: none"); sock->writeLineF("x-audiocast-name: %s",chanInfo.name.cstr()); sock->writeLineF("x-audiocast-bitrate: %d",chanInfo.bitrate); sock->writeLineF("x-audiocast-genre: %s",chanInfo.genre.cstr()); sock->writeLineF("x-audiocast-description: %s",chanInfo.desc.cstr()); sock->writeLineF("x-audiocast-url: %s",chanInfo.url.cstr()); sock->writeLineF("%s %s",PCX_HS_CHANNELID,idStr); }
  529. if (outputProtocol == ChanInfo::SP_HTTP) { switch (chanInfo.contentType) { case ChanInfo::T_OGG: sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_XOGG); break; case ChanInfo::T_MP3: sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_MP3); break; case ChanInfo::T_MOV: sock->writeLine("Connection: close"); sock->writeLine("Content-Length: 10000000"); sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_MOV); break; case ChanInfo::T_MPG: sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_MPG); break; case ChanInfo::T_NSV: sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_NSV); break; case ChanInfo::T_ASX: sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_ASX);
  530. break;
  531. case ChanInfo::T_WMA:
  532. sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_WMA);
  533. break;
  534. case ChanInfo::T_WMV: sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_WMV); break; } } else if (outputProtocol == ChanInfo::SP_MMS)
  535. {
  536. sock->writeLine("Server: Rex/9.0.0.2980");
  537. sock->writeLine("Cache-Control: no-cache");
  538. sock->writeLine("Pragma: no-cache");
  539. sock->writeLine("Pragma: client-id=3587303426");
  540. sock->writeLine("Pragma: features="broadcast,playlist"");
  542. if (nsSwitchNum)
  543. {
  544. sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_MMS);
  545. }else
  546. {
  547. sock->writeLine("Content-Type: application/vnd.ms.wms-hdr.asfv1");
  548. if (ch)
  549. sock->writeLineF("Content-Length: %d",ch->headPack.len);
  550. sock->writeLine("Connection: Keep-Alive");
  551. }
  553. } else if (outputProtocol == ChanInfo::SP_PCP)
  554. {
  555. sock->writeLineF("%s %d",PCX_HS_POS,streamPos);
  556. sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_XPCP);
  557. }else if (outputProtocol == ChanInfo::SP_PEERCAST) { sock->writeLineF("%s %s",HTTP_HS_CONTENT,MIME_XPEERCAST); } }
  558. sock->writeLine("");
  559. result = true;
  561. if (gotPCP)
  562. {
  563. handshakeIncomingPCP(atom,rhost,remoteID,agent);
  564. atom.writeInt(PCP_OK,0);
  565. }
  567. }
  570. return result; } // ----------------------------------- void Servent::handshakeGiv(GnuID &id) { if (id.isSet())
  571. { char idstr[64]; id.toStr(idstr); sock->writeLineF("GIV /%s",idstr);
  572. }else sock->writeLine("GIV");
  574. sock->writeLine("");
  575. } // ----------------------------------- void Servent::processGnutella() {
  576. type = T_PGNU;
  578. //if (servMgr->isRoot && !servMgr->needConnections())
  579. if (servMgr->isRoot)
  580. {
  581. processRoot();
  582. return;
  583. }
  585. gnuStream.init(sock); setStatus(S_CONNECTED); if (!servMgr->isRoot)
  586. {
  587. chanMgr->broadcastRelays(this, 1, 1);
  588. GnuPacket *p;
  590. if ((p=outPacketsNorm.curr())) 
  591. gnuStream.sendPacket(*p);
  592. return;
  593. } gnuStream.ping(2); // if (type != T_LOOKUP) // chanMgr->broadcastRelays(this,chanMgr->minBroadcastTTL,2); lastPacket = lastPing = sys->getTime(); bool doneBigPing=false; const unsigned int abortTimeoutSecs = 60; // abort connection after 60 secs of no activitiy const unsigned int packetTimeoutSecs = 30; // ping connection after 30 secs of no activity unsigned int currBytes=0; unsigned int lastWait=0; unsigned int lastTotalIn=0,lastTotalOut=0; while (thread.active && sock->active()) { if (sock->readReady()) { lastPacket = sys->getTime(); if (gnuStream.readPacket(pack)) { char ipstr[64]; sock->host.toStr(ipstr); GnuID routeID; GnuStream::R_TYPE ret = GnuStream::R_PROCESS; if (pack.func != GNU_FUNC_PONG) if (servMgr->seenPacket(pack)) ret = GnuStream::R_DUPLICATE; seenIDs.add(pack.id); if (ret == GnuStream::R_PROCESS) { GnuID routeID; ret = gnuStream.processPacket(pack,this,routeID); if (flowControl && (ret == GnuStream::R_BROADCAST)) ret = GnuStream::R_DROP; } switch(ret) { case GnuStream::R_BROADCAST: if (servMgr->broadcast(pack,this)) stats.add(Stats::NUMBROADCASTED); else stats.add(Stats::NUMDROPPED); break; case GnuStream::R_ROUTE: if (servMgr->route(pack,routeID,NULL)) stats.add(Stats::NUMROUTED); else stats.add(Stats::NUMDROPPED); break; case GnuStream::R_ACCEPTED: stats.add(Stats::NUMACCEPTED); break; case GnuStream::R_DUPLICATE: stats.add(Stats::NUMDUP); break; case GnuStream::R_DEAD: stats.add(Stats::NUMDEAD); break; case GnuStream::R_DISCARD: stats.add(Stats::NUMDISCARDED); break; case GnuStream::R_BADVERSION: stats.add(Stats::NUMOLD); break; case GnuStream::R_DROP: stats.add(Stats::NUMDROPPED); break; } LOG_NETWORK("packet in: %s-%s, %d bytes, %d hops, %d ttl, from %s",GNU_FUNC_STR(pack.func),GnuStream::getRouteStr(ret),pack.len,pack.hops,pack.ttl,ipstr); }else{ LOG_ERROR("Bad packet"); } } GnuPacket *p; if ((p=outPacketsPri.curr())) // priority packet { gnuStream.sendPacket(*p); seenIDs.add(p->id); outPacketsPri.next(); } else if ((p=outPacketsNorm.curr()))  // or.. normal packet { gnuStream.sendPacket(*p); seenIDs.add(p->id); outPacketsNorm.next(); } int lpt =  sys->getTime()-lastPacket; if (!doneBigPing) { if ((sys->getTime()-lastPing) > 15) { gnuStream.ping(7); lastPing = sys->getTime(); doneBigPing = true; } }else{ if (lpt > packetTimeoutSecs) { if ((sys->getTime()-lastPing) > packetTimeoutSecs) { gnuStream.ping(1); lastPing = sys->getTime(); } } } if (lpt > abortTimeoutSecs) throw TimeoutException(); unsigned int totIn = sock->totalBytesIn-lastTotalIn; unsigned int totOut = sock->totalBytesOut-lastTotalOut; unsigned int bytes = totIn+totOut; lastTotalIn = sock->totalBytesIn; lastTotalOut = sock->totalBytesOut; const int serventBandwidth = 1000;
  594. int delay = sys->idleSleepTime; if ((bytes) && (serventBandwidth >= 8)) delay = (bytes*1000)/(serventBandwidth/8); // set delay relative packetsize if (delay < (int)sys->idleSleepTime) delay = sys->idleSleepTime; //LOG("delay %d, in %d, out %d",delay,totIn,totOut); sys->sleep(delay); } }
  597. // ----------------------------------- void Servent::processRoot() { try  { gnuStream.init(sock); setStatus(S_CONNECTED); gnuStream.ping(2);
  599. unsigned int lastConnect = sys->getTime();
  600. while (thread.active && sock->active()) { if (gnuStream.readPacket(pack)) { char ipstr[64]; sock->host.toStr(ipstr); LOG_NETWORK("packet in: %d from %s",pack.func,ipstr); if (pack.func == GNU_FUNC_PING) // if ping then pong back some hosts and close { Host hl[32]; int cnt = servMgr->getNewestServents(hl,32,sock->host); if (cnt) { int start = sys->rnd() % cnt;
  601. int max = cnt>8?8:cnt;
  602. for(int i=0; i<max; i++) { GnuPacket pong; pack.hops = 1; pong.initPong(hl[start],false,pack); gnuStream.sendPacket(pong); char ipstr[64]; hl[start].toStr(ipstr); //LOG_NETWORK("Pong %d: %s",start+1,ipstr); start = (start+1) % cnt; } char str[64]; sock->host.toStr(str); LOG_NETWORK("Sent %d pong(s) to %s",max,str); }else { LOG_NETWORK("No Pongs to send"); //return; } }else if (pack.func == GNU_FUNC_PONG) // pong? { MemoryStream pong(pack.data,pack.len); int ip,port; port = pong.readShort(); ip = pong.readLong(); ip = SWAP4(ip); Host h(ip,port);
  603. if ((ip) && (port) && (h.globalIP())) {
  605. LOG_NETWORK("added pong: %d.%d.%d.%d:%d",ip>>24&0xff,ip>>16&0xff,ip>>8&0xff,ip&0xff,port); servMgr->addHost(h,ServHost::T_SERVENT,sys->getTime()); } //return; } else if (pack.func == GNU_FUNC_HIT)
  606. {
  607. MemoryStream data(pack.data,pack.len);
  608. ChanHit hit;
  609. gnuStream.readHit(data,hit,pack.hops,pack.id);
  610. } //if (gnuStream.packetsIn > 5) // die if we get too many packets // return; }
  612. if((sys->getTime()-lastConnect > 60))
  613. break; } }catch(StreamException &e) { LOG_ERROR("Relay: %s",e.msg); } } // ----------------------------------- int Servent::givProc(ThreadInfo *thread) { // thread->lock(); Servent *sv = (Servent*)thread->data; try  {
  614. sv->handshakeGiv(sv->givID);
  615. sv->handshakeIncoming();
  616. }catch(StreamException &e) { LOG_ERROR("GIV: %s",e.msg); } sv->kill(); sys->endThread(thread);
  617. return 0; }
  619. // -----------------------------------
  620. void Servent::handshakeOutgoingPCP(AtomStream &atom, Host &rhost, GnuID &rid, String &agent, bool isTrusted)
  621. {
  623. bool nonFW = (servMgr->getFirewall() != ServMgr::FW_ON);
  624. bool testFW = (servMgr->getFirewall() == ServMgr::FW_UNKNOWN);
  626. bool sendBCID = isTrusted && chanMgr->isBroadcasting();
  628. atom.writeParent(PCP_HELO,3 + (testFW?1:0) + (nonFW?1:0) + (sendBCID?1:0));
  629. atom.writeString(PCP_HELO_AGENT,PCX_AGENT);
  630. atom.writeInt(PCP_HELO_VERSION,PCP_CLIENT_VERSION);
  631. atom.writeBytes(PCP_HELO_SESSIONID,servMgr->sessionID.id,16);
  632. if (nonFW)
  633. atom.writeShort(PCP_HELO_PORT,servMgr->serverHost.port);
  634. if (testFW)
  635. atom.writeShort(PCP_HELO_PING,servMgr->serverHost.port);
  636. if (sendBCID)
  637. atom.writeBytes(PCP_HELO_BCID,chanMgr->broadcastID.id,16);
  640. LOG_DEBUG("PCP outgoing waiting for OLEH..");
  644. int numc,numd;
  645. ID4 id = atom.read(numc,numd);
  646. if (id != PCP_OLEH)
  647. {
  648. LOG_DEBUG("PCP outgoing reply: %s",id.getString().str());
  649. atom.writeInt(PCP_QUIT,PCP_ERROR_QUIT+PCP_ERROR_BADRESPONSE);
  650. throw StreamException("Got unexpected PCP response");
  651. }
  655. char arg[64];
  657. GnuID clientID;
  658. clientID.clear();
  659. rid.clear();
  660. int version=0;
  661. int disable=0;
  663. Host thisHost;
  665. // read OLEH response
  666. for(int i=0; i<numc; i++)
  667. {
  668. int c,dlen;
  669. ID4 id = atom.read(c,dlen);
  671. if (id == PCP_HELO_AGENT)
  672. {
  673. atom.readString(arg,sizeof(arg),dlen);
  674. agent.set(arg);
  676. }else if (id == PCP_HELO_REMOTEIP)
  677. {
  678. thisHost.ip = atom.readInt();
  680. }else if (id == PCP_HELO_PORT)
  681. {
  682. thisHost.port = atom.readShort();
  684. }else if (id == PCP_HELO_VERSION)
  685. {
  686. version = atom.readInt();
  688. }else if (id == PCP_HELO_DISABLE)
  689. {
  690. disable = atom.readInt();
  692. }else if (id == PCP_HELO_SESSIONID)
  693. {
  694. atom.readBytes(rid.id,16);
  695. if (rid.isSame(servMgr->sessionID))
  696. throw StreamException("Servent loopback");
  698. }else
  699. {
  700. LOG_DEBUG("PCP handshake skip: %s",id.getString().str());
  701. atom.skip(c,dlen);
  702. }
  704.     }
  707. // update server ip/firewall status
  708. if (isTrusted)
  709. {
  710. if (thisHost.isValid())
  711. {
  712. if ((servMgr->serverHost.ip != thisHost.ip) && (servMgr->forceIP.isEmpty()))
  713. {
  714. char ipstr[64];
  715. thisHost.toStr(ipstr);
  716. LOG_DEBUG("Got new ip: %s",ipstr);
  717. servMgr->serverHost.ip = thisHost.ip;
  718. }
  720. if (servMgr->getFirewall() == ServMgr::FW_UNKNOWN)
  721. {
  722. if (thisHost.port && thisHost.globalIP())
  723. servMgr->setFirewall(ServMgr::FW_OFF);
  724. else
  725. servMgr->setFirewall(ServMgr::FW_ON);
  726. }
  727. }
  729. if (disable == 1)
  730. {
  731. LOG_ERROR("client disabled: %d",disable);
  732. servMgr->isDisabled = true;
  733. }else
  734. {
  735. servMgr->isDisabled = false;
  736. }
  737. }
  741. if (!rid.isSet())
  742. {
  743. atom.writeInt(PCP_QUIT,PCP_ERROR_QUIT+PCP_ERROR_NOTIDENTIFIED);
  744. throw StreamException("Remote host not identified");
  745. }
  747. LOG_DEBUG("PCP Outgoing handshake complete.");
  749. }
  750. // -----------------------------------
  751. void Servent::handshakeIncomingPCP(AtomStream &atom, Host &rhost, GnuID &rid, String &agent)
  752. {
  753. int numc,numd;
  754. ID4 id = atom.read(numc,numd);
  757. if (id != PCP_HELO)
  758. {
  759. LOG_DEBUG("PCP incoming reply: %s",id.getString().str());
  760. atom.writeInt(PCP_QUIT,PCP_ERROR_QUIT+PCP_ERROR_BADRESPONSE);
  761. throw StreamException("Got unexpected PCP response");
  762. }
  764. char arg[64];
  766. ID4 osType;
  768. int version=0;
  770. int pingPort=0;
  772. GnuID bcID;
  773. GnuID clientID;
  775. bcID.clear();
  776. clientID.clear();
  778. rhost.port = 0;
  780. for(int i=0; i<numc; i++)
  781. {
  783. int c,dlen;
  784. ID4 id = atom.read(c,dlen);
  786. if (id == PCP_HELO_AGENT)
  787. {
  788. atom.readString(arg,sizeof(arg),dlen);
  789. agent.set(arg);
  791. }else if (id == PCP_HELO_VERSION)
  792. {
  793. version = atom.readInt();
  795. }else if (id == PCP_HELO_SESSIONID)
  796. {
  797. atom.readBytes(rid.id,16);
  798. if (rid.isSame(servMgr->sessionID))
  799. throw StreamException("Servent loopback");
  801. }else if (id == PCP_HELO_BCID)
  802. {
  803. atom.readBytes(bcID.id,16);
  805. }else if (id == PCP_HELO_OSTYPE)
  806. {
  807. osType = atom.readInt();
  808. }else if (id == PCP_HELO_PORT)
  809. {
  810. rhost.port = atom.readShort();
  811. }else if (id == PCP_HELO_PING)
  812. {
  813. pingPort = atom.readShort();
  814. }else
  815. {
  816. LOG_DEBUG("PCP handshake skip: %s",id.getString().str());
  817. atom.skip(c,dlen);
  818. }
  820.     }
  822. if (version)
  823. LOG_DEBUG("Incoming PCP is %s : v%d", agent.cstr(),version);
  826. if (!rhost.globalIP() && servMgr->serverHost.globalIP())
  827. rhost.ip = servMgr->serverHost.ip;
  829. if (pingPort)
  830. {
  831. char ripStr[64];
  832. rhost.toStr(ripStr);
  833. LOG_DEBUG("Incoming firewalled test request: %s ", ripStr);
  834. rhost.port = pingPort;
  835. if (!rhost.globalIP() || !pingHost(rhost,rid))
  836. rhost.port = 0;
  837. }
  839. if (servMgr->isRoot)
  840. {
  841. if (bcID.isSet())
  842. {
  843. if (bcID.getFlags() & 1) // private
  844. {
  845. BCID *bcid = servMgr->findValidBCID(bcID);
  846. if (!bcid || (bcid && !bcid->valid))
  847. {
  848. atom.writeParent(PCP_OLEH,1);
  849. atom.writeInt(PCP_HELO_DISABLE,1);
  850. throw StreamException("Client is banned");
  851. }
  852. }
  853. }
  854. }
  857. atom.writeParent(PCP_OLEH,5);
  858. atom.writeString(PCP_HELO_AGENT,PCX_AGENT);
  859. atom.writeBytes(PCP_HELO_SESSIONID,servMgr->sessionID.id,16);
  860. atom.writeInt(PCP_HELO_VERSION,PCP_CLIENT_VERSION);
  861. atom.writeInt(PCP_HELO_REMOTEIP,rhost.ip);
  862. atom.writeShort(PCP_HELO_PORT,rhost.port);
  864. if (version)
  865. {
  866. if (version < PCP_CLIENT_MINVERSION)
  867. {
  868. atom.writeInt(PCP_QUIT,PCP_ERROR_QUIT+PCP_ERROR_BADAGENT);
  869. throw StreamException("Agent is not valid");
  870. }
  871. }
  873. if (!rid.isSet())
  874. {
  875. atom.writeInt(PCP_QUIT,PCP_ERROR_QUIT+PCP_ERROR_NOTIDENTIFIED);
  876. throw StreamException("Remote host not identified");
  877. }
  881. if (servMgr->isRoot)
  882. {
  883. servMgr->writeRootAtoms(atom,false);
  884. }
  886. LOG_DEBUG("PCP Incoming handshake complete.");
  888. }
  890. // -----------------------------------
  891. void Servent::processIncomingPCP(bool suggestOthers)
  892. {
  893. PCPStream::readVersion(*sock);
  896. AtomStream atom(*sock);
  897. Host rhost = sock->host;
  899. handshakeIncomingPCP(atom,rhost,remoteID,agent);
  902. bool alreadyConnected = (servMgr->findConnection(Servent::T_COUT,remoteID)!=NULL)
  903. || (servMgr->findConnection(Servent::T_CIN,remoteID)!=NULL);
  904. bool unavailable = servMgr->controlInFull();
  905. bool offair = !servMgr->isRoot && !chanMgr->isBroadcasting();
  907. char rstr[64];
  908. rhost.toStr(rstr);
  910. if (unavailable || alreadyConnected || offair)
  911. {
  912. int error;
  914. if (alreadyConnected)
  915. error = PCP_ERROR_QUIT+PCP_ERROR_ALREADYCONNECTED;
  916. else if (unavailable)
  917. error = PCP_ERROR_QUIT+PCP_ERROR_UNAVAILABLE;
  918. else if (offair)
  919. error = PCP_ERROR_QUIT+PCP_ERROR_OFFAIR;
  920. else 
  921. error = PCP_ERROR_QUIT;
  924. if (suggestOthers)
  925. {
  927. ChanHit best;
  928. ChanHitSearch chs;
  930. int cnt=0;
  931. for(int i=0; i<8; i++)
  932. {
  933. best.init();
  935. // find best hit on this network
  936. if (!rhost.globalIP())
  937. {
  938. chs.init();
  939. chs.matchHost = servMgr->serverHost;
  940. chs.waitDelay = 2;
  941. chs.excludeID = remoteID;
  942. chs.trackersOnly = true;
  943. chs.useBusyControls = false;
  944. if (chanMgr->pickHits(chs))
  945. best = chs.best[0];
  946. }
  948. // find best hit on same network
  949. if (!best.host.ip)
  950. {
  951. chs.init();
  952. chs.matchHost = rhost;
  953. chs.waitDelay = 2;
  954. chs.excludeID = remoteID;
  955. chs.trackersOnly = true;
  956. chs.useBusyControls = false;
  957. if (chanMgr->pickHits(chs))
  958. best = chs.best[0];
  959. }
  961. // else find best hit on other networks
  962. if (!best.host.ip)
  963. {
  964. chs.init();
  965. chs.waitDelay = 2;
  966. chs.excludeID = remoteID;
  967. chs.trackersOnly = true;
  968. chs.useBusyControls = false;
  969. if (chanMgr->pickHits(chs))
  970. best = chs.best[0];
  971. }
  973. if (!best.host.ip)
  974. break;
  976. GnuID noID;
  977. noID.clear();
  978. best.writeAtoms(atom,noID);
  979. cnt++;
  980. }
  981. if (cnt)
  982. {
  983. LOG_DEBUG("Sent %d tracker(s) to %s",cnt,rstr);
  984. }
  985. else if (rhost.port)
  986. {
  987. // send push request to best firewalled tracker on other network
  988. chs.init();
  989. chs.waitDelay = 30;
  990. chs.excludeID = remoteID;
  991. chs.trackersOnly = true;
  992. chs.useFirewalled = true;
  993. chs.useBusyControls = false;
  994. if (chanMgr->pickHits(chs))
  995. {
  996. best = chs.best[0];
  997. GnuID noID;
  998. noID.clear();
  999. int cnt = servMgr->broadcastPushRequest(best,rhost,noID,Servent::T_CIN);
  1000. LOG_DEBUG("Broadcasted tracker push request to %d clients for %s",cnt,rstr);
  1001. }
  1002. }
  1003. }
  1006. LOG_ERROR("Sending QUIT to incoming: %d",error);
  1008. atom.writeInt(PCP_QUIT,error);
  1009. return;
  1010. }
  1013. type = T_CIN;
  1014. setStatus(S_CONNECTED);
  1016. atom.writeInt(PCP_OK,0);
  1018. // ask for update
  1019. atom.writeParent(PCP_ROOT,1);
  1020. atom.writeParent(PCP_ROOT_UPDATE,0);
  1022. pcpStream = new PCPStream(remoteID);
  1024. int error = 0;
  1025. BroadcastState bcs;
  1026. while (!error && thread.active && !sock->eof())
  1027. {
  1028. error = pcpStream->readPacket(*sock,bcs);
  1029. sys->sleepIdle();
  1031. if (!servMgr->isRoot && !chanMgr->isBroadcasting())
  1032. error = PCP_ERROR_OFFAIR;
  1033. if (peercastInst->isQuitting)
  1034. error = PCP_ERROR_SHUTDOWN;
  1035. }
  1037. pcpStream->flush(*sock);
  1039. error += PCP_ERROR_QUIT;
  1040. atom.writeInt(PCP_QUIT,error);
  1042. LOG_DEBUG("PCP Incoming to %s closed: %d",rstr,error);
  1044. }
  1046. // ----------------------------------- int Servent::outgoingProc(ThreadInfo *thread) { // thread->lock(); LOG_DEBUG("COUT started");
  1047. Servent *sv = (Servent*)thread->data;
  1048. GnuID noID;
  1049. noID.clear();
  1050. sv->pcpStream = new PCPStream(noID);
  1052. while (sv->thread.active)
  1053. {
  1054. sv->setStatus(S_WAIT);
  1056. if (chanMgr->isBroadcasting() && servMgr->autoServe)
  1057. {
  1058. ChanHit bestHit;
  1059. ChanHitSearch chs;
  1060. char ipStr[64];
  1062. do
  1063. {
  1064. bestHit.init();
  1066. if (servMgr->rootHost.isEmpty())
  1067. break;
  1069. if (sv->pushSock)
  1070. {
  1071. sv->sock = sv->pushSock;
  1072. sv->pushSock = NULL;
  1073. bestHit.host = sv->sock->host;
  1074. break;
  1075. }
  1077. GnuID noID;
  1078. noID.clear();
  1079. ChanHitList *chl = chanMgr->findHitListByID(noID);
  1080. if (chl)
  1081. {
  1082. // find local tracker
  1083. chs.init();
  1084. chs.matchHost = servMgr->serverHost;
  1085. chs.waitDelay = MIN_TRACKER_RETRY;
  1086. chs.excludeID = servMgr->sessionID;
  1087. chs.trackersOnly = true;
  1088. if (!chl->pickHits(chs))
  1089. {
  1090. // else find global tracker
  1091. chs.init();
  1092. chs.waitDelay = MIN_TRACKER_RETRY;
  1093. chs.excludeID = servMgr->sessionID;
  1094. chs.trackersOnly = true;
  1095. chl->pickHits(chs);
  1096. }
  1098. if (chs.numResults)
  1099. {
  1100. bestHit = chs.best[0];
  1101. }
  1102. }
  1105. unsigned int ctime = sys->getTime();
  1107. if ((!bestHit.host.ip) && ((ctime-chanMgr->lastYPConnect) > MIN_YP_RETRY))
  1108. {
  1109. bestHit.host.fromStrName(servMgr->rootHost.cstr(),DEFAULT_PORT);
  1110. bestHit.yp = true;
  1111. chanMgr->lastYPConnect = ctime;
  1112. }
  1113. sys->sleepIdle();
  1115. }while (!bestHit.host.ip && (sv->thread.active));
  1118. if (!bestHit.host.ip) // give up
  1119. {
  1120. LOG_ERROR("COUT giving up");
  1121. break;
  1122. }
  1125. bestHit.host.toStr(ipStr);
  1127. int error=0;
  1128. try  {
  1130. LOG_DEBUG("COUT to %s: Connecting..",ipStr);
  1132. if (!sv->sock)
  1133. {
  1134. sv->setStatus(S_CONNECTING);
  1135. sv->sock = sys->createSocket();
  1136. if (!sv->sock)
  1137. throw StreamException("Unable to create socket");
  1138. sv->sock->open(bestHit.host);
  1139. sv->sock->connect();
  1141. }
  1143. sv->sock->setReadTimeout(30000);
  1144. AtomStream atom(*sv->sock);
  1146. sv->setStatus(S_HANDSHAKE);
  1147. Host rhost = sv->sock->host;
  1148. atom.writeInt(PCP_CONNECT,1);
  1149. handshakeOutgoingPCP(atom,rhost,sv->remoteID,sv->agent,bestHit.yp);
  1151. sv->setStatus(S_CONNECTED);
  1153. LOG_DEBUG("COUT to %s: OK",ipStr);
  1155. sv->pcpStream->init(sv->remoteID);
  1157. BroadcastState bcs;
  1158. error = 0;
  1159. while (!error && sv->thread.active && !sv->sock->eof() && servMgr->autoServe)
  1160. {
  1161. error = sv->pcpStream->readPacket(*sv->sock,bcs);
  1163. sys->sleepIdle();
  1165. if (!chanMgr->isBroadcasting())
  1166. error = PCP_ERROR_OFFAIR;
  1167. if (peercastInst->isQuitting)
  1168. error = PCP_ERROR_SHUTDOWN;
  1170. if (sv->pcpStream->nextRootPacket)
  1171. if (sys->getTime() > (sv->pcpStream->nextRootPacket+30))
  1172. error = PCP_ERROR_NOROOT;
  1173. }
  1174. sv->setStatus(S_CLOSING);
  1176. sv->pcpStream->flush(*sv->sock);
  1178. error += PCP_ERROR_QUIT;
  1179. atom.writeInt(PCP_QUIT,error);
  1181. LOG_ERROR("COUT to %s closed: %d",ipStr,error);
  1182. }catch(TimeoutException &e) { LOG_ERROR("COUT to %s: timeout (%s)",ipStr,e.msg); sv->setStatus(S_TIMEOUT);
  1183. }catch(StreamException &e) { LOG_ERROR("COUT to %s: %s",ipStr,e.msg); sv->setStatus(S_ERROR); }
  1185. try
  1186. {
  1187. if (sv->sock)
  1188. {
  1189. sv->sock->close();
  1190. delete sv->sock;
  1191. sv->sock = NULL;
  1192. }
  1194. }catch(StreamException &) {}
  1196. // don`t discard this hit if we caused the disconnect (stopped broadcasting)
  1197. if (error != (PCP_ERROR_QUIT+PCP_ERROR_OFFAIR))
  1198. chanMgr->deadHit(bestHit);
  1200. }
  1202. sys->sleepIdle();
  1203. } sv->kill(); sys->endThread(thread);
  1204. LOG_DEBUG("COUT ended");
  1205. return 0; } // ----------------------------------- int Servent::incomingProc(ThreadInfo *thread) { // thread->lock(); Servent *sv = (Servent*)thread->data;
  1206. char ipStr[64];
  1207. sv->sock->host.toStr(ipStr);
  1208. try  { sv->handshakeIncoming(); }catch(HTTPException &e) { try { sv->sock->writeLine(e.msg); if (e.code == 401) sv->sock->writeLine("WWW-Authenticate: Basic realm="PeerCast""); sv->sock->writeLine(""); }catch(StreamException &){} LOG_ERROR("Incoming from %s: %s",ipStr,e.msg); }catch(StreamException &e) { LOG_ERROR("Incoming from %s: %s",ipStr,e.msg); }
  1210. sv->kill();
  1211. sys->endThread(thread);
  1212. return 0; } // ----------------------------------- void Servent::processServent() { setStatus(S_HANDSHAKE); handshakeIn(); if (!sock) throw StreamException("Servent has no socket");
  1213. processGnutella(); } // ----------------------------------- void Servent::processStream(bool doneHandshake,ChanInfo &chanInfo) {
  1214. if (!doneHandshake)
  1215. {
  1216. setStatus(S_HANDSHAKE);
  1218. if (!handshakeStream(chanInfo))
  1219. return;
  1220. }
  1222. if (chanInfo.id.isSet())
  1223. {
  1225. chanID = chanInfo.id;
  1227. LOG_CHANNEL("Sending channel: %s ",ChanInfo::getProtocolStr(outputProtocol));
  1229. if (!waitForChannelHeader(chanInfo))
  1230. throw StreamException("Channel not ready");
  1231. servMgr->totalStreams++; Host host = sock->host; host.port = 0; // force to 0 so we ignore the incoming port
  1233. Channel *ch = chanMgr->findChannelByID(chanID);
  1234. if (!ch)
  1235. throw StreamException("Channel not found");
  1237. if (outputProtocol == ChanInfo::SP_HTTP) { if ((addMetadata) && (chanMgr->icyMetaInterval)) sendRawMetaChannel(chanMgr->icyMetaInterval); else 
  1238. sendRawChannel(true,true);
  1240. }else if (outputProtocol == ChanInfo::SP_MMS)
  1241. {
  1242. if (nsSwitchNum)
  1243. {
  1244. sendRawChannel(true,true);
  1245. }else
  1246. {
  1247. sendRawChannel(true,false);
  1248. }
  1250. }else if (outputProtocol  == ChanInfo::SP_PCP) { sendPCPChannel();
  1252. } else if (outputProtocol  == ChanInfo::SP_PEERCAST)
  1253. {
  1254. sendPeercastChannel();
  1255. }
  1256. }
  1257. setStatus(S_CLOSING); } // ----------------------------------------- #if 0 // debug FileStream file; file.openReadOnly("c://test.mp3"); LOG_DEBUG("raw file read"); char buf[4000]; int cnt=0; while (!file.eof()) { LOG_DEBUG("send %d",cnt++); file.read(buf,sizeof(buf)); sock->write(buf,sizeof(buf)); } file.close(); LOG_DEBUG("raw file sent"); return; // debug #endif // ----------------------------------- bool Servent::waitForChannelHeader(ChanInfo &info) { for(int i=0; i<30*10; i++) {
  1258. Channel *ch = chanMgr->findChannelByID(info.id);
  1259. if (!ch)
  1260. return false;
  1261. if (ch->isPlaying() && (ch->rawData.writePos>0)) return true; if (!thread.active || !sock->active()) break; sys->sleep(100); } return false; } // ----------------------------------- void Servent::sendRawChannel(bool sendHead, bool sendData) {
  1262. try {
  1263. sock->setWriteTimeout(DIRECT_WRITE_TIMEOUT*1000);
  1265. Channel *ch = chanMgr->findChannelByID(chanID);
  1266. if (!ch)
  1267. throw StreamException("Channel not found");
  1268. setStatus(S_CONNECTED); LOG_DEBUG("Starting Raw stream of %s at %d",ch->info.name.cstr(),streamPos);
  1270. if (sendHead)
  1271. { ch->headPack.writeRaw(*sock);
  1272. streamPos = ch->headPack.pos + ch->headPack.len;
  1273. LOG_DEBUG("Sent %d bytes header ",ch->headPack.len);
  1274. }
  1276. if (sendData)
  1277. {
  1278. unsigned int streamIndex = ch->streamIndex; unsigned int connectTime = sys->getTime();
  1279. unsigned int lastWriteTime = connectTime;
  1281. while ((thread.active) && sock->active()) {
  1282. ch = chanMgr->findChannelByID(chanID);
  1284. if (ch)
  1285. {
  1287. if (streamIndex != ch->streamIndex)
  1288. {
  1289. streamIndex = ch->streamIndex;
  1290. streamPos = ch->headPack.pos;
  1291. LOG_DEBUG("sendRaw got new stream index");
  1292. }
  1294. ChanPacket rawPack;
  1295. if (ch->rawData.findPacket(streamPos,rawPack))
  1296. {
  1297. if (syncPos != rawPack.sync)
  1298. LOG_ERROR("Send skip: %d",rawPack.sync-syncPos);
  1299. syncPos = rawPack.sync+1;
  1301. if ((rawPack.type == ChanPacket::T_DATA) || (rawPack.type == ChanPacket::T_HEAD))
  1302. {
  1303. rawPack.writeRaw(*sock);
  1304. lastWriteTime = sys->getTime();
  1305. }
  1307. if (rawPack.pos < streamPos)
  1308. LOG_DEBUG("raw: skip back %d",rawPack.pos-streamPos);
  1309. streamPos = rawPack.pos+rawPack.len;
  1310. }
  1311. }
  1313. if ((sys->getTime()-lastWriteTime) > DIRECT_WRITE_TIMEOUT)
  1314. throw TimeoutException();
  1316. sys->sleepIdle(); }
  1317. } }catch(StreamException &e) { LOG_ERROR("Stream channel: %s",e.msg); } }
  1318. #if 0
  1319. // -----------------------------------
  1320. void Servent::sendRawMultiChannel(bool sendHead, bool sendData)
  1321. {
  1322. try
  1323. {
  1324. unsigned int chanStreamIndex[ChanMgr::MAX_CHANNELS];
  1325. unsigned int chanStreamPos[ChanMgr::MAX_CHANNELS];
  1326. GnuID chanIDs[ChanMgr::MAX_CHANNELS];
  1327. int numChanIDs=0;
  1328. for(int i=0; i<ChanMgr::MAX_CHANNELS; i++)
  1329. {
  1330. Channel *ch = &chanMgr->channels[i];
  1331. if (ch->isPlaying())
  1332. chanIDs[numChanIDs++]=ch->info.id;
  1333. }
  1337. setStatus(S_CONNECTED);
  1340. if (sendHead)
  1341. {
  1342. for(int i=0; i<numChanIDs; i++)
  1343. {
  1344. Channel *ch = chanMgr->findChannelByID(chanIDs[i]);
  1345. if (ch)
  1346. {
  1347. LOG_DEBUG("Starting RawMulti stream: %s",ch->info.name.cstr());
  1348. ch->headPack.writeRaw(*sock);
  1349. chanStreamPos[i] = ch->headPack.pos + ch->headPack.len;
  1350. chanStreamIndex[i] = ch->streamIndex;
  1351. LOG_DEBUG("Sent %d bytes header",ch->headPack.len);
  1353. }
  1354. }
  1355. }
  1357. if (sendData)
  1358. {
  1360. unsigned int connectTime=sys->getTime();
  1362. while ((thread.active) && sock->active())
  1363. {
  1365. for(int i=1; i<numChanIDs; i++)
  1366. {
  1367. Channel *ch = chanMgr->findChannelByID(chanIDs[i]);
  1368. if (ch)
  1369. {
  1370. if (chanStreamIndex[i] != ch->streamIndex)
  1371. {
  1372. chanStreamIndex[i] = ch->streamIndex;
  1373. chanStreamPos[i] = ch->headPack.pos;
  1374. LOG_DEBUG("sendRawMulti got new stream index for chan %d",i);
  1375. }
  1377. ChanPacket rawPack;
  1378. if (ch->rawData.findPacket(chanStreamPos[i],rawPack))
  1379. {
  1380. if ((rawPack.type == ChanPacket::T_DATA) || (rawPack.type == ChanPacket::T_HEAD))
  1381. rawPack.writeRaw(*sock);
  1384. if (rawPack.pos < chanStreamPos[i])
  1385. LOG_DEBUG("raw: skip back %d",rawPack.pos-chanStreamPos[i]);
  1386. chanStreamPos[i] = rawPack.pos+rawPack.len;
  1389. //LOG("raw at %d: %d %d",streamPos,ch->rawData.getStreamPos(ch->rawData.firstPos),ch->rawData.getStreamPos(ch->rawData.lastPos));
  1390. }
  1391. }
  1392. break;
  1393. }
  1396. sys->sleepIdle();
  1397. }
  1398. }
  1399. }catch(StreamException &e)
  1400. {
  1401. LOG_ERROR("Stream channel: %s",e.msg);
  1402. }
  1403. }
  1404. #endif
  1405. // ----------------------------------- void Servent::sendRawMetaChannel(int interval) {
  1406. try {
  1407. Channel *ch = chanMgr->findChannelByID(chanID);
  1408. if (!ch)
  1409. throw StreamException("Channel not found");
  1411. sock->setWriteTimeout(DIRECT_WRITE_TIMEOUT*1000);
  1412. setStatus(S_CONNECTED); LOG_DEBUG("Starting Raw Meta stream of %s (metaint: %d) at %d",ch->info.name.cstr(),interval,streamPos); String lastTitle,lastURL; int lastMsgTime=sys->getTime(); bool showMsg=true; char buf[16384]; int bufPos=0; if ((interval > sizeof(buf)) || (interval < 1)) throw StreamException("Bad ICY Meta Interval value");
  1413. unsigned int connectTime = sys->getTime(); unsigned int lastWriteTime = connectTime;
  1415. streamPos = 0; // raw meta channel has no header (its MP3)
  1416. while ((thread.active) && sock->active()) {
  1417. ch = chanMgr->findChannelByID(chanID);
  1419. if (ch)
  1420. {
  1422. ChanPacket rawPack;
  1423. if (ch->rawData.findPacket(streamPos,rawPack))
  1424. {
  1426. if (syncPos != rawPack.sync)
  1427. LOG_ERROR("Send skip: %d",rawPack.sync-syncPos);
  1428. syncPos = rawPack.sync+1;
  1430. MemoryStream mem(rawPack.data,rawPack.len);
  1431. if (rawPack.type == ChanPacket::T_DATA) {
  1432. int len = rawPack.len; char *p = rawPack.data; while (len) { int rl = len; if ((bufPos+rl) > interval) rl = interval-bufPos; memcpy(&buf[bufPos],p,rl); bufPos+=rl; p+=rl; len-=rl; if (bufPos >= interval) { bufPos = 0; sock->write(buf,interval); lastWriteTime = sys->getTime();
  1433. if (chanMgr->broadcastMsgInterval) if ((sys->getTime()-lastMsgTime) >= chanMgr->broadcastMsgInterval) { showMsg ^= true; lastMsgTime = sys->getTime(); } String *metaTitle = &ch->info.track.title; if (!ch->info.comment.isEmpty() && (showMsg)) metaTitle = &ch->info.comment; if (!metaTitle->isSame(lastTitle) || !ch->info.url.isSame(lastURL)) { char tmp[1024]; String title,url; title = *metaTitle; url = ch->info.url; title.convertTo(String::T_META); url.convertTo(String::T_META); sprintf(tmp,"StreamTitle='%s';StreamUrl='%s';",title.cstr(),url.cstr()); int len = ((strlen(tmp) + 15+1) / 16); sock->writeChar(len); sock->write(tmp,len*16); lastTitle = *metaTitle; lastURL = ch->info.url; LOG_DEBUG("StreamTitle: %s, StreamURL: %s",lastTitle.cstr(),lastURL.cstr()); }else { sock->writeChar(0); } }
  1434. } }
  1435. streamPos = rawPack.pos + rawPack.len; }
  1436. }
  1437. if ((sys->getTime()-lastWriteTime) > DIRECT_WRITE_TIMEOUT)
  1438. throw TimeoutException();
  1440. sys->sleepIdle();
  1441. } }catch(StreamException &e) { LOG_ERROR("Stream channel: %s",e.msg); } } // ----------------------------------- void Servent::sendPeercastChannel() {
  1442. try { setStatus(S_CONNECTED);
  1443. Channel *ch = chanMgr->findChannelByID(chanID);
  1444. if (!ch)
  1445. throw StreamException("Channel not found");
  1446. LOG_DEBUG("Starting PeerCast stream: %s",ch->info.name.cstr()); sock->writeTag("PCST"); ChanPacket pack;
  1448. ch->headPack.writePeercast(*sock);
  1449. pack.init(ChanPacket::T_META,ch->insertMeta.data,ch->insertMeta.len,ch->streamPos); pack.writePeercast(*sock); streamPos = 0;
  1450. unsigned int syncPos=0; while ((thread.active) && sock->active()) { ch = chanMgr->findChannelByID(chanID);
  1451. if (ch)
  1452. {
  1454. ChanPacket rawPack;
  1455. if (ch->rawData.findPacket(streamPos,rawPack))
  1456. {
  1457. if ((rawPack.type == ChanPacket::T_DATA) || (rawPack.type == ChanPacket::T_HEAD))
  1458. {
  1459. sock->writeTag("SYNC");
  1460. sock->writeShort(4);
  1461. sock->writeShort(0);
  1462. sock->write(&syncPos,4);
  1463. syncPos++;
  1465. rawPack.writePeercast(*sock);
  1466. }
  1467. streamPos = rawPack.pos + rawPack.len;
  1468. }
  1469. }
  1470. sys->sleepIdle(); } }catch(StreamException &e) { LOG_ERROR("Stream channel: %s",e.msg); } }
  1471. // -----------------------------------
  1472. void Servent::sendPCPChannel()
  1473. {
  1474. Channel *ch = chanMgr->findChannelByID(chanID);
  1475. if (!ch)
  1476. throw StreamException("Channel not found");
  1478. AtomStream atom(*sock);
  1480. pcpStream = new PCPStream(remoteID);
  1481. int error=0;
  1483. try
  1484. {
  1486. LOG_DEBUG("Starting PCP stream of channel at %d",streamPos);
  1489. setStatus(S_CONNECTED);
  1491. atom.writeParent(PCP_CHAN,3 + ((sendHeader)?1:0));
  1492. atom.writeBytes(PCP_CHAN_ID,chanID.id,16);
  1493. ch->info.writeInfoAtoms(atom);
  1494. ch->info.writeTrackAtoms(atom);
  1495. if (sendHeader)
  1496. {
  1497. atom.writeParent(PCP_CHAN_PKT,3);
  1498. atom.writeID4(PCP_CHAN_PKT_TYPE,PCP_CHAN_PKT_HEAD);
  1499. atom.writeInt(PCP_CHAN_PKT_POS,ch->headPack.pos);
  1500. atom.writeBytes(PCP_CHAN_PKT_DATA,ch->headPack.data,ch->headPack.len);
  1502. streamPos = ch->headPack.pos+ch->headPack.len;
  1503. LOG_DEBUG("Sent %d bytes header",ch->headPack.len);
  1504. }
  1506. unsigned int streamIndex = ch->streamIndex;
  1508. while (thread.active)
  1509. {
  1511. Channel *ch = chanMgr->findChannelByID(chanID);
  1513. if (ch)
  1514. {
  1516. if (streamIndex != ch->streamIndex)
  1517. {
  1518. streamIndex = ch->streamIndex;
  1519. streamPos = ch->headPack.pos;
  1520. LOG_DEBUG("sendPCPStream got new stream index");
  1521. }
  1523. ChanPacket rawPack;
  1525. if (ch->rawData.findPacket(streamPos,rawPack))
  1526. {
  1528. if (rawPack.type == ChanPacket::T_HEAD)
  1529. {
  1530. atom.writeParent(PCP_CHAN,2);
  1531. atom.writeBytes(PCP_CHAN_ID,chanID.id,16);
  1532. atom.writeParent(PCP_CHAN_PKT,3);
  1533. atom.writeID4(PCP_CHAN_PKT_TYPE,PCP_CHAN_PKT_HEAD);
  1534. atom.writeInt(PCP_CHAN_PKT_POS,rawPack.pos);
  1535. atom.writeBytes(PCP_CHAN_PKT_DATA,rawPack.data,rawPack.len);
  1538. }else if (rawPack.type == ChanPacket::T_DATA)
  1539. {
  1540. atom.writeParent(PCP_CHAN,2);
  1541. atom.writeBytes(PCP_CHAN_ID,chanID.id,16);
  1542. atom.writeParent(PCP_CHAN_PKT,3);
  1543. atom.writeID4(PCP_CHAN_PKT_TYPE,PCP_CHAN_PKT_DATA);
  1544. atom.writeInt(PCP_CHAN_PKT_POS,rawPack.pos);
  1545. atom.writeBytes(PCP_CHAN_PKT_DATA,rawPack.data,rawPack.len);
  1547. }
  1549. if (rawPack.pos < streamPos)
  1550. LOG_DEBUG("pcp: skip back %d",rawPack.pos-streamPos);
  1552. //LOG_DEBUG("Sending %d-%d (%d,%d,%d)",rawPack.pos,rawPack.pos+rawPack.len,ch->streamPos,ch->rawData.getLatestPos(),ch->rawData.getOldestPos());
  1554. streamPos = rawPack.pos+rawPack.len;
  1555. }
  1557. }
  1558. BroadcastState bcs;
  1559. error = pcpStream->readPacket(*sock,bcs);
  1560. if (error)
  1561. throw StreamException("PCP exception");
  1563. sys->sleepIdle();
  1565. }
  1567. LOG_DEBUG("PCP channel stream closed normally.");
  1569. }catch(StreamException &e)
  1570. {
  1571. LOG_ERROR("Stream channel: %s",e.msg);
  1572. }
  1574. try
  1575. {
  1576. atom.writeInt(PCP_QUIT,error);
  1577. }catch(StreamException &) {}
  1579. }
  1580. // ----------------------------------- int Servent::serverProc(ThreadInfo *thread) { // thread->lock();
  1581. Servent *sv = (Servent*)thread->data; try  { if (!sv->sock) throw StreamException("Server has no socket"); sv->setStatus(S_LISTENING); char servIP[64]; sv->sock->host.toStr(servIP); if (servMgr->isRoot) LOG_DEBUG("Root Server started: %s",servIP); else LOG_DEBUG("Server started: %s",servIP); while ((thread->active) && (sv->sock->active())) {
  1582. if (servMgr->numActiveOnPort(sv->sock->host.port) < servMgr->maxServIn)
  1583. { ClientSocket *cs = sv->sock->accept(); if (cs) {
  1584. LOG_DEBUG("accepted incoming"); Servent *ns = servMgr->allocServent(); if (ns) {
  1585. servMgr->lastIncoming = sys->getTime();
  1586. ns->servPort = sv->sock->host.port; ns->networkID = servMgr->networkID; ns->initIncoming(cs,sv->allow); }else LOG_ERROR("Out of servents"); }
  1587. }
  1588. sys->sleep(100);
  1589. } }catch(StreamException &e) { LOG_ERROR("Server Error: %s:%d",e.msg,e.err); } LOG_DEBUG("Server stopped");
  1590. sv->kill();
  1591. sys->endThread(thread); return 0; }   // -----------------------------------
  1592. bool Servent::writeVariable(Stream &s, const String &var)
  1593. {
  1594. char buf[1024];
  1596. if (var == "type")
  1597. strcpy(buf,getTypeStr());
  1598. else if (var == "status")
  1599. strcpy(buf,getStatusStr());
  1600. else if (var == "address")
  1601. getHost().toStr(buf);
  1602. else if (var == "agent")
  1603. strcpy(buf,agent.cstr());
  1604. else if (var == "bitrate")
  1605. {
  1606. if (sock)
  1607. {
  1608. unsigned int tot = sock->bytesInPerSec+sock->bytesOutPerSec;
  1609. sprintf(buf,"%.1f",BYTES_TO_KBPS(tot));
  1610. }else
  1611. strcpy(buf,"0");
  1612. }else if (var == "uptime")
  1613. {
  1614. String uptime;
  1615. if (lastConnect)
  1616. uptime.setFromStopwatch(sys->getTime()-lastConnect);
  1617. else
  1618. uptime.set("-");
  1619. strcpy(buf,uptime.cstr());
  1620. }else if (var.startsWith("gnet."))
  1621. {
  1623. float ctime = (float)(sys->getTime()-lastConnect);
  1624. if (var == "gnet.packetsIn")
  1625. sprintf(buf,"%d",gnuStream.packetsIn);
  1626. else if (var == "gnet.packetsInPerSec")
  1627. sprintf(buf,"%.1f",ctime>0?((float)gnuStream.packetsIn)/ctime:0);
  1628. else if (var == "gnet.packetsOut")
  1629. sprintf(buf,"%d",gnuStream.packetsOut);
  1630. else if (var == "gnet.packetsOutPerSec")
  1631. sprintf(buf,"%.1f",ctime>0?((float)gnuStream.packetsOut)/ctime:0);
  1632. else if (var == "gnet.normQueue")
  1633. sprintf(buf,"%d",outPacketsNorm.numPending());
  1634. else if (var == "gnet.priQueue")
  1635. sprintf(buf,"%d",outPacketsPri.numPending());
  1636. else if (var == "gnet.flowControl")
  1637. sprintf(buf,"%d",flowControl?1:0);
  1638. else if (var == "gnet.routeTime")
  1639. {
  1640. int nr = seenIDs.numUsed();
  1641. unsigned int tim = sys->getTime()-seenIDs.getOldest();
  1643. String tstr;
  1644. tstr.setFromStopwatch(tim);
  1646. if (nr)
  1647. strcpy(buf,tstr.cstr());
  1648. else
  1649. strcpy(buf,"-");
  1650. }
  1651. else
  1652. return false;
  1654. }else
  1655. return false;
  1657. s.writeString(buf);
  1659. return true;
  1660. }