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

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

P2P编程

开发平台：

Windows_Unix

pcp.cpp：源码内容
  1. // ------------------------------------------------
  2. // File : pcp.cpp
  3. // Date: 1-mar-2004
  4. // Author: giles
  5. //
  6. // (c) 2002-4 peercast.org
  7. // ------------------------------------------------
  8. // This program is free software; you can redistribute it and/or modify
  9. // it under the terms of the GNU General Public License as published by
  10. // the Free Software Foundation; either version 2 of the License, or
  11. // (at your option) any later version.
  13. // This program is distributed in the hope that it will be useful,
  14. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16. // GNU General Public License for more details.
  17. // ------------------------------------------------
  19. #include "atom.h"
  20. #include "pcp.h"
  21. #include "peercast.h"
  22. #include "version2.h"
  24. // ------------------------------------------
  25. void PCPStream::init(GnuID &rid)
  26. {
  27. remoteID = rid;
  28. routeList.clear();
  30. lastPacketTime = 0;
  31. nextRootPacket = 0;  // 0 seconds (never)
  33. inData.init();
  34. inData.accept = ChanPacket::T_PCP;
  36. outData.init();
  37. outData.accept = ChanPacket::T_PCP;
  38. }
  39. // ------------------------------------------
  40. void PCPStream::readVersion(Stream &in)
  41. {
  42. int len = in.readInt();
  44. if (len != 4)
  45. throw StreamException("Invalid PCP");
  47. int ver = in.readInt();
  49. LOG_DEBUG("PCP ver: %d",ver);
  50. }
  51. // ------------------------------------------
  52. void PCPStream::readHeader(Stream &in,Channel *)
  53. {
  54. // AtomStream atom(in);
  56. // if (in.readInt() != PCP_CONNECT)
  57. // throw StreamException("Not PCP");
  59. // readVersion(in);
  60. }
  61. // ------------------------------------------
  62. bool PCPStream::sendPacket(ChanPacket &pack,GnuID &destID)
  63. {
  64. if (destID.isSet())
  65. if (!destID.isSame(remoteID))
  66. if (!routeList.contains(destID))
  67. return false;
  69. return outData.writePacket(pack);
  70. }
  71. // ------------------------------------------
  72. void PCPStream::flush(Stream &in)
  73. {
  74. ChanPacket pack;
  75. // send outward packets
  76. while (outData.numPending())
  77. {
  78. outData.readPacket(pack);
  79. pack.writeRaw(in);
  80. }
  81. }
  82. // ------------------------------------------
  83. int PCPStream::readPacket(Stream &in,Channel *)
  84. {
  85. BroadcastState bcs;
  86. return readPacket(in,bcs);
  87. }
  88. // ------------------------------------------
  89. int PCPStream::readPacket(Stream &in,BroadcastState &bcs)
  90. {
  91. int error = PCP_ERROR_GENERAL;
  92. try
  93. {
  94. AtomStream atom(in);
  96. ChanPacket pack;
  97. MemoryStream mem(pack.data,sizeof(pack.data));
  98. AtomStream patom(mem);
  101. // send outward packets
  102. error = PCP_ERROR_WRITE;
  103. if (outData.numPending())
  104. {
  105. outData.readPacket(pack);
  106. pack.writeRaw(in);
  107. }
  108. error = PCP_ERROR_GENERAL;
  110. if (outData.willSkip())
  111. {
  112. error = PCP_ERROR_WRITE+PCP_ERROR_SKIP;
  113. throw StreamException("Send too slow");
  114. }
  117. error = PCP_ERROR_READ;
  118. // poll for new downward packet
  119. if (in.readReady())
  120. {
  121. int numc,numd;
  122. ID4 id;
  124. id = atom.read(numc,numd);
  126. mem.rewind();
  127. pack.len = patom.writeAtoms(id, in, numc, numd);
  128. pack.type = ChanPacket::T_PCP;
  130. inData.writePacket(pack);
  131. }
  132. error = PCP_ERROR_GENERAL;
  134. // process downward packets
  135. if (inData.numPending())
  136. {
  137. inData.readPacket(pack);
  139. mem.rewind();
  141. int numc,numd;
  142. ID4 id = patom.read(numc,numd);
  144. error = PCPStream::procAtom(patom,id,numc,numd,bcs);
  146. if (error)
  147. throw StreamException("PCP exception");
  148. }
  150. error = 0;
  152. }catch(StreamException &e)
  153. {
  154. LOG_ERROR("PCP readPacket: %s (%d)",e.msg,error);
  155. }
  157. return error;
  158. }
  160. // ------------------------------------------
  161. void PCPStream::readEnd(Stream &,Channel *)
  162. {
  163. }
  166. // ------------------------------------------
  167. void PCPStream::readPushAtoms(AtomStream &atom, int numc,BroadcastState &bcs)
  168. {
  169. Host host;
  170. GnuID chanID;
  172. chanID.clear();
  174. for(int i=0; i<numc; i++)
  175. {
  176. int c,d;
  177. ID4 id = atom.read(c,d);
  179. if (id == PCP_PUSH_IP)
  180. host.ip = atom.readInt();
  181. else if (id == PCP_PUSH_PORT)
  182. host.port = atom.readShort();
  183. else if (id == PCP_PUSH_CHANID)
  184. atom.readBytes(chanID.id,16);
  185. else
  186. {
  187. LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
  188. atom.skip(c,d);
  189. }
  190. }
  193. if (bcs.forMe)
  194. {
  195. char ipstr[64];
  196. host.toStr(ipstr);
  198. Servent *s = NULL;
  200. if (chanID.isSet())
  201. {
  202. Channel *ch = chanMgr->findChannelByID(chanID);
  203. if (ch)
  204. if (ch->isBroadcasting() || !ch->isFull() && !servMgr->relaysFull() && ch->info.id.isSame(chanID))
  205. s = servMgr->allocServent();
  206. }else{
  207. s = servMgr->allocServent();
  208. }
  210. if (s)
  211. {
  212. LOG_DEBUG("GIVing to %s",ipstr);
  213. s->initGIV(host,chanID);
  214. }
  215. }
  217. }
  218. // ------------------------------------------
  219. void PCPStream::readRootAtoms(AtomStream &atom, int numc,BroadcastState &bcs)
  220. {
  221. String url;
  223. for(int i=0; i<numc; i++)
  224. {
  225. int c,d;
  226. ID4 id = atom.read(c,d);
  228. if (id == PCP_ROOT_UPDINT)
  229. {
  230. int si = atom.readInt();
  232. chanMgr->setUpdateInterval(si);
  233. LOG_DEBUG("PCP got new host update interval: %ds",si);
  234. }else if (id == PCP_ROOT_URL)
  235. {
  236. url = "http://www.peercast.org/";
  237. String loc;
  238. atom.readString(loc.data,sizeof(loc.data),d);
  239. url.append(loc);
  241. }else if (id == PCP_ROOT_CHECKVER)
  242. {
  243. unsigned int newVer = atom.readInt();
  244. if (newVer > PCP_CLIENT_VERSION)
  245. {
  246. strcpy(servMgr->downloadURL,url.cstr());
  247. peercastApp->notifyMessage(ServMgr::NT_UPGRADE,"There is a new version available, please click here to upgrade your client.");
  248. }
  249. LOG_DEBUG("PCP got version check: %d / %d",newVer,PCP_CLIENT_VERSION);
  251. }else if (id == PCP_ROOT_NEXT)
  252. {
  253. unsigned int time = atom.readInt();
  255. if (time)
  256. {
  257. unsigned int ctime = sys->getTime();
  258. nextRootPacket = ctime+time;
  259. LOG_DEBUG("PCP expecting next root packet in %ds",time);
  260. }else
  261. {
  262. nextRootPacket = 0;
  263. }
  265. }else if (id == PCP_ROOT_UPDATE)
  266. {
  267. atom.skip(c,d);
  269. chanMgr->broadcastTrackerUpdate(remoteID,true);
  271. }else if ((id == PCP_MESG_ASCII) || (id == PCP_MESG)) // PCP_MESG_ASCII to be depreciated 
  272. {
  273. String newMsg;
  275. atom.readString(newMsg.data,sizeof(newMsg.data),d);
  276. if (!newMsg.isSame(servMgr->rootMsg.cstr()))
  277. {
  278. servMgr->rootMsg = newMsg;
  279. LOG_DEBUG("PCP got new root mesg: %s",servMgr->rootMsg.cstr());
  280. peercastApp->notifyMessage(ServMgr::NT_PEERCAST,servMgr->rootMsg.cstr());
  281. }
  282. }else
  283. {
  284. LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
  285. atom.skip(c,d);
  286. }
  287. }
  288. }
  290. // ------------------------------------------
  291. void PCPStream::readPktAtoms(Channel *ch,AtomStream &atom,int numc,BroadcastState &bcs)
  292. {
  293. ChanPacket pack;
  294. ID4 type;
  297. for(int i=0; i<numc; i++)
  298. {
  299. int c,d;
  300. ID4 id = atom.read(c,d);
  302. if (id == PCP_CHAN_PKT_TYPE)
  303. {
  304. type = atom.readID4();
  306. if (type == PCP_CHAN_PKT_HEAD)
  307. pack.type = ChanPacket::T_HEAD;
  308. else if (type == PCP_CHAN_PKT_DATA)
  309. pack.type = ChanPacket::T_DATA;
  310. else
  311. pack.type = ChanPacket::T_UNKNOWN;
  313. }else if (id == PCP_CHAN_PKT_POS)
  314. {
  315. pack.pos = atom.readInt();
  318. }else if (id == PCP_CHAN_PKT_DATA)
  319. {
  320. pack.len = d;
  321. atom.readBytes(pack.data,pack.len);
  322. }
  323. else
  324. {
  325. LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
  326. atom.skip(c,d);
  327. }
  328. }
  330. if (ch)
  331. {
  333. int diff = pack.pos - ch->streamPos;
  334. if (diff)
  335. LOG_DEBUG("PCP skipping %s%d (%d -> %d)",(diff>0)?"+":"",diff,ch->streamPos,pack.pos);
  338. if (pack.type == ChanPacket::T_HEAD)
  339. {
  340. LOG_DEBUG("New head packet at %d",pack.pos);
  342. // check for stream restart
  343. if (pack.pos == 0)
  344. {
  345. LOG_CHANNEL("PCP resetting stream");
  346. ch->streamIndex++;
  347. ch->rawData.init();
  348. }
  350. ch->headPack = pack;
  352. ch->rawData.writePacket(pack,true);
  353. ch->streamPos = pack.pos+pack.len;
  355. }else if (pack.type == ChanPacket::T_DATA)
  356. {
  357. ch->rawData.writePacket(pack,true);
  358. ch->streamPos = pack.pos+pack.len;
  359. }
  361. }
  363. // update this parent packet stream position
  364. if ((pack.pos) && (!bcs.streamPos || (pack.pos < bcs.streamPos)))
  365. bcs.streamPos = pack.pos;
  367. }
  368. // -----------------------------------
  369. void PCPStream::readHostAtoms(AtomStream &atom, int numc, BroadcastState &bcs)
  370. {
  371. ChanHit hit;
  372. hit.init();
  373. GnuID chanID = bcs.chanID; //use default
  375. bool busy=false;
  377. unsigned int ipNum=0;
  379. for(int i=0; i<numc; i++)
  380. {
  381. int c,d;
  382. ID4 id = atom.read(c,d);
  384. if (id == PCP_HOST_IP)
  385. {
  386. unsigned int ip = atom.readInt();
  387. hit.rhost[ipNum].ip = ip;
  388. }else if (id == PCP_HOST_PORT)
  389. {
  390. int port = atom.readShort();
  391. hit.rhost[ipNum++].port = port;
  393. if (ipNum > 1)
  394. ipNum = 1;
  395. }
  396. else if (id == PCP_HOST_NUML)
  397. hit.numListeners = atom.readInt();
  398. else if (id == PCP_HOST_NUMR)
  399. hit.numRelays = atom.readInt();
  400. else if (id == PCP_HOST_UPTIME)
  401. hit.upTime = atom.readInt();
  402. else if (id == PCP_HOST_OLDPOS)
  403. hit.oldestPos = atom.readInt();
  404. else if (id == PCP_HOST_NEWPOS)
  405. hit.newestPos = atom.readInt();
  406. else if (id == PCP_HOST_VERSION)
  407. hit.version = atom.readInt();
  408. else if (id == PCP_HOST_FLAGS1)
  409. {
  410. int fl1 = atom.readChar();
  412. hit.recv = (fl1 & PCP_HOST_FLAGS1_RECV) !=0;
  413. hit.relay = (fl1 & PCP_HOST_FLAGS1_RELAY) !=0;
  414. hit.direct = (fl1 & PCP_HOST_FLAGS1_DIRECT) !=0;
  415. hit.cin = (fl1 & PCP_HOST_FLAGS1_CIN) !=0;
  416. hit.tracker = (fl1 & PCP_HOST_FLAGS1_TRACKER) !=0;
  417. hit.firewalled = (fl1 & PCP_HOST_FLAGS1_PUSH) !=0;
  420. }else if (id == PCP_HOST_ID)
  421. atom.readBytes(hit.sessionID.id,16);
  422. else if (id == PCP_HOST_CHANID)
  423. atom.readBytes(chanID.id,16);
  424. else
  425. {
  426. LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
  427. atom.skip(c,d);
  428. }
  429. }
  432. hit.host = hit.rhost[0];
  433. hit.chanID = chanID;
  435. hit.numHops = bcs.numHops;
  438. if (hit.recv)
  439. chanMgr->addHit(hit);
  440. else
  441. chanMgr->delHit(hit);
  442. }
  444. // ------------------------------------------
  445. void PCPStream::readChanAtoms(AtomStream &atom,int numc,BroadcastState &bcs)
  446. {
  447. Channel *ch=NULL;
  448. ChanHitList *chl=NULL;
  449. ChanInfo newInfo;
  451. ch = chanMgr->findChannelByID(bcs.chanID);
  452. chl = chanMgr->findHitListByID(bcs.chanID);
  454. if (ch)
  455. newInfo = ch->info;
  456. else if (chl)
  457. newInfo = chl->info;
  460. for(int i=0; i<numc; i++)
  461. {
  463. int c,d;
  464. ID4 id = atom.read(c,d);
  466. if ((id == PCP_CHAN_PKT) && (ch))
  467. {
  468. readPktAtoms(ch,atom,c,bcs);
  469. }else if (id == PCP_CHAN_INFO)
  470. {
  471. newInfo.readInfoAtoms(atom,c);
  473. }else if (id == PCP_CHAN_TRACK)
  474. {
  475. newInfo.readTrackAtoms(atom,c);
  477. }else if (id == PCP_CHAN_BCID)
  478. {
  479. atom.readBytes(newInfo.bcID.id,16);
  481. }else if (id == PCP_CHAN_KEY) // depreciated
  482. {
  483. atom.readBytes(newInfo.bcID.id,16);
  484. newInfo.bcID.id[0] = 0; // clear flags
  486. }else if (id == PCP_CHAN_ID)
  487. {
  488. atom.readBytes(newInfo.id.id,16);
  490. ch = chanMgr->findChannelByID(newInfo.id);
  491. chl = chanMgr->findHitListByID(newInfo.id);
  493. }else
  494. {
  495. LOG_DEBUG("PCP skip: %s,%d,%d",id.getString().str(),c,d);
  496. atom.skip(c,d);
  497. }
  498. }
  500. if (!chl)
  501. chl = chanMgr->addHitList(newInfo);
  503. if (chl)
  504. {
  505. chl->info.update(newInfo);
  507. if (!servMgr->chanLog.isEmpty())
  508. {
  509. //if (chl->numListeners())
  510. {
  511. try
  512. {
  513. FileStream file;
  514. file.openWriteAppend(servMgr->chanLog.cstr());
  516.          XML::Node *rn = new XML::Node("update time="%d"",sys->getTime());
  517.         XML::Node *n = chl->info.createChannelXML();
  518.          n->add(chl->createXML(false));
  519.          n->add(chl->info.createTrackXML());
  520. rn->add(n);
  522. rn->write(file,0);
  523. delete rn;
  524. file.close();
  525. }catch(StreamException &e)
  526. {
  527. LOG_ERROR("Unable to update channel log: %s",e.msg);
  528. }
  529. }
  530. }
  532. }
  534. if (ch && !ch->isBroadcasting())
  535. ch->updateInfo(newInfo);
  538. }
  539. // ------------------------------------------
  540. int PCPStream::readBroadcastAtoms(AtomStream &atom,int numc,BroadcastState &bcs)
  541. {
  542. ChanPacket pack;
  543. int ttl=1;
  544. int ver=0;
  545. GnuID fromID,destID;
  547. fromID.clear();
  548. destID.clear();
  550. bcs.initPacketSettings();
  552. MemoryStream pmem(pack.data,sizeof(pack.data));
  553. AtomStream patom(pmem);
  555. patom.writeParent(PCP_BCST,numc);
  557. for(int i=0; i<numc; i++)
  558. {
  559. int c,d;
  560. ID4 id = atom.read(c,d);
  562. if (id == PCP_BCST_TTL)
  563. {
  564. ttl = atom.readChar()-1;
  565. patom.writeChar(id,ttl);
  567. }else if (id == PCP_BCST_HOPS)
  568. {
  569. bcs.numHops = atom.readChar()+1;
  570. patom.writeChar(id,bcs.numHops);
  572. }else if (id == PCP_BCST_FROM)
  573. {
  574. atom.readBytes(fromID.id,16);
  575. patom.writeBytes(id,fromID.id,16);
  577. routeList.add(fromID);
  578. }else if (id == PCP_BCST_GROUP)
  579. {
  580. bcs.group = atom.readChar();
  581. patom.writeChar(id,bcs.group);
  582. }else if (id == PCP_BCST_DEST)
  583. {
  584. atom.readBytes(destID.id,16);
  585. patom.writeBytes(id,destID.id,16);
  586. bcs.forMe = destID.isSame(servMgr->sessionID);
  588. char idstr1[64];
  589. char idstr2[64];
  591. destID.toStr(idstr1);
  592. servMgr->sessionID.toStr(idstr2);
  594. }else if (id == PCP_BCST_CHANID)
  595. {
  596. atom.readBytes(bcs.chanID.id,16);
  597. patom.writeBytes(id,bcs.chanID.id,16);
  598. }else if (id == PCP_BCST_VERSION)
  599. {
  600. ver = atom.readInt();
  601. patom.writeInt(id,ver);
  602. }else
  603. {
  604. // copy and process atoms
  605. int oldPos = pmem.pos;
  606. patom.writeAtoms(id,atom.io,c,d);
  607. pmem.pos = oldPos;
  608. readAtom(patom,bcs);
  609. }
  610. }
  612. char fromStr[64];
  613. fromStr[0] = 0;
  614. if (fromID.isSet())
  615. fromID.toStr(fromStr);
  616. char destStr[64];
  617. destStr[0] = 0;
  618. if (destID.isSet())
  619. destID.toStr(destStr);
  622. LOG_DEBUG("PCP bcst: group=%d, hops=%d, ver=%d, from=%s, dest=%s",bcs.group,bcs.numHops,ver,fromStr,destStr);
  624. if (fromID.isSet())
  625. if (fromID.isSame(servMgr->sessionID))
  626. {
  627. LOG_ERROR("BCST loopback"); 
  628. return PCP_ERROR_BCST+PCP_ERROR_LOOPBACK;
  629. }
  631. // broadcast back out if ttl > 0 
  632. if ((ttl>0) && (!bcs.forMe))
  633. {
  634. pack.len = pmem.pos;
  635. pack.type = ChanPacket::T_PCP;
  637. if (bcs.group & (PCP_BCST_GROUP_ROOT|PCP_BCST_GROUP_TRACKERS|PCP_BCST_GROUP_RELAYS))
  638. {
  639. chanMgr->broadcastPacketUp(pack,bcs.chanID,remoteID,destID);
  640. }
  642. if (bcs.group & (PCP_BCST_GROUP_ROOT|PCP_BCST_GROUP_TRACKERS|PCP_BCST_GROUP_RELAYS))
  643. {
  644. servMgr->broadcastPacket(pack,bcs.chanID,remoteID,destID,Servent::T_COUT);
  645. }
  647. if (bcs.group & (PCP_BCST_GROUP_RELAYS|PCP_BCST_GROUP_TRACKERS))
  648. {
  649. servMgr->broadcastPacket(pack,bcs.chanID,remoteID,destID,Servent::T_CIN);
  650. }
  652. if (bcs.group & (PCP_BCST_GROUP_RELAYS))
  653. {
  654. servMgr->broadcastPacket(pack,bcs.chanID,remoteID,destID,Servent::T_RELAY);
  655. }
  659. }
  660. return 0;
  661. }
  664. // ------------------------------------------
  665. int PCPStream::procAtom(AtomStream &atom,ID4 id,int numc, int dlen,BroadcastState &bcs)
  666. {
  667. int r=0;
  669. if (id == PCP_CHAN)
  670. {
  671. readChanAtoms(atom,numc,bcs);
  672. }else if (id == PCP_ROOT)
  673. {
  674. if (servMgr->isRoot)
  675. throw StreamException("Unauthorized root message");
  676. else
  677. readRootAtoms(atom,numc,bcs);
  679. }else if (id == PCP_HOST)
  680. {
  681. readHostAtoms(atom,numc,bcs);
  683. }else if ((id == PCP_MESG_ASCII) || (id == PCP_MESG)) // PCP_MESG_ASCII to be depreciated
  684. {
  685. String msg;
  686. atom.readString(msg.data,sizeof(msg.data),dlen);
  687. LOG_DEBUG("PCP got text: %s",msg.cstr());
  688. }else if (id == PCP_BCST)
  689. {
  690. r = readBroadcastAtoms(atom,numc,bcs);
  691. }else if (id == PCP_HELO)
  692. {
  693. atom.skip(numc,dlen);
  694. atom.writeParent(PCP_OLEH,1);
  695. atom.writeBytes(PCP_HELO_SESSIONID,servMgr->sessionID.id,16);
  696. }else if (id == PCP_PUSH)
  697. {
  699. readPushAtoms(atom,numc,bcs);
  700. }else if (id == PCP_OK)
  701. {
  702. atom.readInt();
  704. }else if (id == PCP_QUIT)
  705. {
  706. r = atom.readInt();
  707. if (!r)
  708. r = PCP_ERROR_QUIT;
  710. }else if (id == PCP_ATOM)
  711. {
  712. for(int i=0; i<numc; i++)
  713. {
  714. int nc,nd;
  715. ID4 aid = atom.read(nc,nd);
  716. int ar = procAtom(atom,aid,nc,nd,bcs);
  717. if (ar)
  718. r = ar;
  719. }
  721. }else
  722. {
  723. LOG_CHANNEL("PCP skip: %s",id.getString().str());
  724. atom.skip(numc,dlen);
  725. }
  727. return r;
  729. }
  731. // ------------------------------------------
  732. int PCPStream::readAtom(AtomStream &atom,BroadcastState &bcs)
  733. {
  734. int numc,dlen;
  735. ID4 id = atom.read(numc,dlen);
  737. return procAtom(atom,id,numc,dlen,bcs);
  738. }