9
3

EQStream.cpp 49 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759
  1. /*
  2. EQ2Emulator: Everquest II Server Emulator
  3. Copyright (C) 2007 EQ2EMulator Development Team (http://www.eq2emulator.net)
  4. This file is part of EQ2Emulator.
  5. EQ2Emulator is free software: you can redistribute it and/or modify
  6. it under the terms of the GNU General Public License as published by
  7. the Free Software Foundation, either version 3 of the License, or
  8. (at your option) any later version.
  9. EQ2Emulator is distributed in the hope that it will be useful,
  10. but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. GNU General Public License for more details.
  13. You should have received a copy of the GNU General Public License
  14. along with EQ2Emulator. If not, see <http://www.gnu.org/licenses/>.
  15. */
  16. #ifdef WIN32
  17. #include <WinSock2.h>
  18. #include <windows.h>
  19. #endif
  20. #include "debug.h"
  21. #include <string>
  22. #include <iomanip>
  23. #include <iostream>
  24. #include <vector>
  25. #include <time.h>
  26. #include <sys/types.h>
  27. #ifdef WIN32
  28. #include <time.h>
  29. #else
  30. #include <sys/socket.h>
  31. #include <netinet/in.h>
  32. #include <sys/time.h>
  33. #include <sys/socket.h>
  34. #include <netdb.h>
  35. #include <fcntl.h>
  36. #include <arpa/inet.h>
  37. #endif
  38. #include "EQPacket.h"
  39. #include "EQStream.h"
  40. #include "EQStreamFactory.h"
  41. #include "misc.h"
  42. #include "Mutex.h"
  43. #include "op_codes.h"
  44. #include "CRC16.h"
  45. #include "packet_dump.h"
  46. #ifdef LOGIN
  47. #include "../LoginServer/login_structs.h"
  48. #endif
  49. #include "EQ2_Common_Structs.h"
  50. #include "Log.h"
  51. uint16 EQStream::MaxWindowSize=2048;
  52. void EQStream::init(bool resetSession) {
  53. if (resetSession)
  54. {
  55. streamactive = false;
  56. sessionAttempts = 0;
  57. }
  58. timeout_delays = 0;
  59. MInUse.lock();
  60. active_users = 0;
  61. MInUse.unlock();
  62. Session=0;
  63. Key=0;
  64. MaxLen=0;
  65. NextInSeq=0;
  66. NextOutSeq=0;
  67. CombinedAppPacket=NULL;
  68. MAcks.lock();
  69. MaxAckReceived = -1;
  70. NextAckToSend = -1;
  71. LastAckSent = -1;
  72. MAcks.unlock();
  73. LastSeqSent=-1;
  74. MaxSends=5;
  75. LastPacket=Timer::GetCurrentTime2();
  76. oversize_buffer=NULL;
  77. oversize_length=0;
  78. oversize_offset=0;
  79. Factory = NULL;
  80. MRate.lock();
  81. RateThreshold=RATEBASE/250;
  82. DecayRate=DECAYBASE/250;
  83. MRate.unlock();
  84. BytesWritten=0;
  85. SequencedBase = 0;
  86. AverageDelta = 500;
  87. crypto->setRC4Key(0);
  88. retransmittimer = Timer::GetCurrentTime2();
  89. retransmittimeout = 500 * RETRANSMIT_TIMEOUT_MULT;
  90. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  91. LogWrite(PACKET__DEBUG, 9, "Packet", "init Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  92. }
  93. }
  94. EQStream::EQStream(sockaddr_in addr){
  95. crypto = new Crypto();
  96. resend_que_timer = new Timer(1000);
  97. combine_timer = new Timer(250); //250 milliseconds
  98. combine_timer->Start();
  99. resend_que_timer->Start();
  100. init();
  101. remote_ip=addr.sin_addr.s_addr;
  102. remote_port=addr.sin_port;
  103. State=CLOSED;
  104. StreamType=UnknownStream;
  105. compressed=true;
  106. encoded=false;
  107. app_opcode_size=2;
  108. #ifdef WIN32
  109. ZeroMemory(&stream, sizeof(z_stream));
  110. #else
  111. bzero(&stream, sizeof(z_stream));
  112. #endif
  113. stream.zalloc = (alloc_func)0;
  114. stream.zfree = (free_func)0;
  115. stream.opaque = (voidpf)0;
  116. deflateInit2(&stream, 9, Z_DEFLATED, 13, 9, Z_DEFAULT_STRATEGY);
  117. //deflateInit(&stream, 5);
  118. compressed_offset = 0;
  119. client_version = 0;
  120. received_packets = 0;
  121. sent_packets = 0;
  122. #ifdef WRITE_PACKETS
  123. write_packets = 0;
  124. char write_packets_filename[64];
  125. snprintf(write_packets_filename, sizeof(write_packets_filename), "PacketLog%i.log", Timer::GetCurrentTime2());
  126. write_packets = fopen(write_packets_filename, "w+");
  127. #endif
  128. }
  129. EQProtocolPacket* EQStream::ProcessEncryptedData(uchar* data, int32 size, int16 opcode){
  130. //cout << "B4:\n";
  131. //DumpPacket(data, size);
  132. /*if(size >= 2 && data[0] == 0 && data[1] == 0){
  133. cout << "Attempting to fix packet!\n";
  134. //Have to fix bad packet from client or it will screw up encryption :P
  135. size--;
  136. data++;
  137. }*/
  138. crypto->RC4Decrypt(data,size);
  139. int8 offset = 0;
  140. if(data[0] == 0xFF && size > 2){
  141. offset = 3;
  142. memcpy(&opcode, data+sizeof(int8), sizeof(int16));
  143. }
  144. else{
  145. offset = 1;
  146. memcpy(&opcode, data, sizeof(int8));
  147. }
  148. //cout << "After:\n";
  149. //DumpPacket(data, size);
  150. return new EQProtocolPacket(opcode, data+offset, size - offset);
  151. }
  152. EQProtocolPacket* EQStream::ProcessEncryptedPacket(EQProtocolPacket *p){
  153. EQProtocolPacket* ret = NULL;
  154. if(p->opcode == OP_Packet && p->size > 2)
  155. ret = ProcessEncryptedData(p->pBuffer+2, p->size-2, p->opcode);
  156. else
  157. ret = ProcessEncryptedData(p->pBuffer, p->size, p->opcode);
  158. return ret;
  159. }
  160. bool EQStream::HandleEmbeddedPacket(EQProtocolPacket *p, int16 offset, int16 length){
  161. if(p && p->size >= ((uint32)(offset+2))){
  162. if(p->pBuffer[offset] == 0 && p->pBuffer[offset+1] == 0x19){
  163. if(length == 0)
  164. length = p->size-2-offset;
  165. else
  166. length-=2;
  167. #ifdef LE_DEBUG
  168. printf( "Creating OP_AppCombined Packet!\n");
  169. #endif
  170. EQProtocolPacket *subp=new EQProtocolPacket(OP_AppCombined, p->pBuffer+2+offset, length);
  171. subp->copyInfo(p);
  172. ProcessPacket(subp, p);
  173. safe_delete(subp);
  174. return true;
  175. }
  176. else if (p->pBuffer[offset] == 0 && p->pBuffer[offset + 1] == 0) {
  177. if (length == 0)
  178. length = p->size - 1 - offset;
  179. else
  180. length--;
  181. #ifdef LE_DEBUG
  182. LogWrite(PACKET__DEBUG, 0, "Packet", "Creating Opcode 0 Packet!");
  183. DumpPacket(p->pBuffer + 1 + offset, length);
  184. #endif
  185. EQProtocolPacket* newpacket = ProcessEncryptedData(p->pBuffer + 1 + offset, length, OP_Packet);
  186. if (newpacket) {
  187. #ifdef LE_DEBUG
  188. LogWrite(PACKET__DEBUG, 0, "Packet", "Result: ");
  189. DumpPacket(newpacket);
  190. #endif
  191. EQApplicationPacket* ap = newpacket->MakeApplicationPacket(2);
  192. if (ap->version == 0)
  193. ap->version = client_version;
  194. InboundQueuePush(ap);
  195. #ifdef WRITE_PACKETS
  196. WritePackets(ap->GetOpcodeName(), p->pBuffer + 1 + offset, length, false);
  197. #endif
  198. safe_delete(newpacket);
  199. }
  200. else
  201. LogWrite(PACKET__ERROR, 0, "Packet", "No Packet!");
  202. return true;
  203. }
  204. }
  205. return false;
  206. }
  207. void EQStream::ProcessPacket(EQProtocolPacket *p, EQProtocolPacket* lastp)
  208. {
  209. uint32 processed=0,subpacket_length=0;
  210. if (p) {
  211. if (p->opcode!=OP_SessionRequest && p->opcode!=OP_SessionResponse && !Session) {
  212. #ifdef EQN_DEBUG
  213. LogWrite(PACKET__ERROR, 0, "Packet", "*** Session not initialized, packet ignored ");
  214. //p->DumpRaw();
  215. #endif
  216. return;
  217. }
  218. //cout << "Received " << (int)p->opcode << ":\n";
  219. //DumpPacket(p->pBuffer, p->size);
  220. switch (p->opcode) {
  221. case OP_Combined: {
  222. processed=0;
  223. int8 offset = 0;
  224. int count = 0;
  225. #ifdef LE_DEBUG
  226. printf( "OP_Combined:\n");
  227. DumpPacket(p);
  228. #endif
  229. while(processed<p->size) {
  230. if ((subpacket_length=(unsigned char)*(p->pBuffer+processed))==0xff) {
  231. subpacket_length = ntohs(*(uint16*)(p->pBuffer + processed + 1));
  232. offset = 3;
  233. }
  234. else
  235. offset = 1;
  236. count++;
  237. #ifdef LE_DEBUG
  238. printf( "OP_Combined Packet %i (%u) (%u):\n", count, subpacket_length, processed);
  239. #endif
  240. bool isSubPacket = EQProtocolPacket::IsProtocolPacket(p->pBuffer + processed + offset, subpacket_length, false);
  241. if (isSubPacket) {
  242. EQProtocolPacket* subp = new EQProtocolPacket(p->pBuffer + processed + offset, subpacket_length);
  243. subp->copyInfo(p);
  244. //I've seen some garbage packets get sent with wrong protocol opcodes but the rest of the combine is still correct
  245. //So don't break if GetProtocolPacket fails
  246. #ifdef LE_DEBUG
  247. printf( "Opcode %i:\n", subp->opcode);
  248. DumpPacket(subp);
  249. #endif
  250. ProcessPacket(subp, p);
  251. #ifdef LE_DEBUG
  252. DumpPacket(subp);
  253. #endif
  254. delete subp;
  255. }
  256. else if (ntohs(*reinterpret_cast<uint16_t*>(p->pBuffer + processed + offset)) > 0x1e) {
  257. //Garbage packet?
  258. crypto->RC4Decrypt(p->pBuffer + processed + offset, subpacket_length);
  259. LogWrite(PACKET__ERROR, 0, "Packet", "Garbage packet?!:");
  260. printf("!!!!!!!!!Garbage Packet!!!!!!!!!!!!!:\n");
  261. DumpPacket(p->pBuffer + processed + offset, subpacket_length);
  262. }
  263. processed+=subpacket_length+offset;
  264. }
  265. break;
  266. }
  267. case OP_AppCombined: {
  268. processed=0;
  269. EQProtocolPacket* newpacket = 0;
  270. int8 offset = 0;
  271. #ifdef LE_DEBUG
  272. printf( "OP_AppCombined: \n");
  273. DumpPacket(p);
  274. #endif
  275. int count = 0;
  276. while(processed<p->size) {
  277. count++;
  278. if ((subpacket_length=(unsigned char)*(p->pBuffer+processed))==0xff) {
  279. subpacket_length=ntohs(*(uint16 *)(p->pBuffer+processed+1));
  280. offset = 3;
  281. } else
  282. offset = 1;
  283. if(crypto->getRC4Key()==0 && p->size >= 70){
  284. processRSAKey(p);
  285. }
  286. else if(crypto->isEncrypted()){
  287. #ifdef LE_DEBUG
  288. printf( "OP_AppCombined Packet %i (%u) (%u): \n", count, subpacket_length, processed);
  289. DumpPacket(p->pBuffer+processed+offset, subpacket_length);
  290. #endif
  291. if(!HandleEmbeddedPacket(p, processed + offset, subpacket_length)){
  292. #ifdef LE_DEBUG
  293. printf( "OP_AppCombined Here:\n");
  294. #endif
  295. MCombineQueueLock.lock();
  296. newpacket = ProcessEncryptedData(p->pBuffer+processed + offset, subpacket_length, OP_AppCombined);
  297. MCombineQueueLock.unlock();
  298. if(newpacket){
  299. #ifdef LE_DEBUG
  300. printf( "Opcode %i:\n", newpacket->opcode);
  301. DumpPacket(newpacket);
  302. #endif
  303. EQApplicationPacket* ap = newpacket->MakeApplicationPacket(2);
  304. #ifdef LE_DEBUG
  305. printf( "OP_AppCombined Here2:\n");
  306. DumpPacket(ap);
  307. #endif
  308. if (ap->version == 0)
  309. ap->version = client_version;
  310. #ifdef WRITE_PACKETS
  311. WritePackets(ap->GetOpcodeName(), p->pBuffer + processed + offset, subpacket_length, false);
  312. #endif
  313. InboundQueuePush(ap);
  314. safe_delete(newpacket);
  315. }
  316. }
  317. }
  318. processed+=subpacket_length+offset;
  319. }
  320. }
  321. break;
  322. case OP_Packet: {
  323. if (!p->pBuffer || (p->Size() < 4))
  324. {
  325. break;
  326. }
  327. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  328. sint8 check=CompareSequence(NextInSeq,seq);
  329. if (check == SeqFuture) {
  330. #ifdef EQN_DEBUG
  331. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Future packet: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  332. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  333. p->DumpRawHeader(seq);
  334. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  335. #endif
  336. OutOfOrderpackets[seq] = p->Copy();
  337. // Image (2020): Removed as this is bad contributes to infinite loop
  338. //SendOutOfOrderAck(seq);
  339. } else if (check == SeqPast) {
  340. #ifdef EQN_DEBUG
  341. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Duplicate packet: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  342. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  343. p->DumpRawHeader(seq);
  344. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  345. #endif
  346. // Image (2020): Removed as this is bad contributes to infinite loop
  347. //OutOfOrderpackets[seq] = p->Copy();
  348. SendOutOfOrderAck(seq);
  349. } else {
  350. EQProtocolPacket* qp = RemoveQueue(seq);
  351. if (qp) {
  352. LogWrite(PACKET__DEBUG, 1, "Packet", "OP_Fragment: Removing older queued packet with sequence %i", seq);
  353. delete qp;
  354. }
  355. SetNextAckToSend(seq);
  356. NextInSeq++;
  357. if(HandleEmbeddedPacket(p))
  358. break;
  359. if(crypto->getRC4Key()==0 && p && p->size >= 70){
  360. processRSAKey(p);
  361. }
  362. else if(crypto->isEncrypted() && p){
  363. MCombineQueueLock.lock();
  364. EQProtocolPacket* newpacket = ProcessEncryptedPacket(p);
  365. MCombineQueueLock.unlock();
  366. if(newpacket){
  367. EQApplicationPacket* ap = newpacket->MakeApplicationPacket(2);
  368. if (ap->version == 0)
  369. ap->version = client_version;
  370. #ifdef WRITE_PACKETS
  371. WritePackets(ap->GetOpcodeName(), p->pBuffer, p->size, false);
  372. #endif
  373. InboundQueuePush(ap);
  374. safe_delete(newpacket);
  375. }
  376. }
  377. }
  378. }
  379. break;
  380. case OP_Fragment: {
  381. if (!p->pBuffer || (p->Size() < 4))
  382. {
  383. break;
  384. }
  385. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  386. sint8 check=CompareSequence(NextInSeq,seq);
  387. if (check == SeqFuture) {
  388. #ifdef EQN_DEBUG
  389. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Future packet2: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  390. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  391. //p->DumpRawHeader(seq);
  392. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  393. #endif
  394. OutOfOrderpackets[seq] = p->Copy();
  395. //SendOutOfOrderAck(seq);
  396. } else if (check == SeqPast) {
  397. #ifdef EQN_DEBUG
  398. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Duplicate packet2: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  399. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  400. //p->DumpRawHeader(seq);
  401. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  402. #endif
  403. //OutOfOrderpackets[seq] = p->Copy();
  404. SendOutOfOrderAck(seq);
  405. } else {
  406. // In case we did queue one before as well.
  407. EQProtocolPacket* qp = RemoveQueue(seq);
  408. if (qp) {
  409. LogWrite(PACKET__DEBUG, 1, "Packet", "OP_Fragment: Removing older queued packet with sequence %i", seq);
  410. delete qp;
  411. }
  412. SetNextAckToSend(seq);
  413. NextInSeq++;
  414. if (oversize_buffer) {
  415. memcpy(oversize_buffer+oversize_offset,p->pBuffer+2,p->size-2);
  416. oversize_offset+=p->size-2;
  417. //cout << "Oversized is " << oversize_offset << "/" << oversize_length << " (" << (p->size-2) << ") Seq=" << seq << endl;
  418. if (oversize_offset==oversize_length) {
  419. if (*(p->pBuffer+2)==0x00 && *(p->pBuffer+3)==0x19) {
  420. EQProtocolPacket *subp=new EQProtocolPacket(oversize_buffer,oversize_offset);
  421. subp->copyInfo(p);
  422. ProcessPacket(subp, p);
  423. delete subp;
  424. } else {
  425. if(crypto->isEncrypted() && p && p->size > 2){
  426. MCombineQueueLock.lock();
  427. EQProtocolPacket* p2 = ProcessEncryptedData(oversize_buffer, oversize_offset, p->opcode);
  428. MCombineQueueLock.unlock();
  429. EQApplicationPacket* ap = p2->MakeApplicationPacket(2);
  430. ap->copyInfo(p);
  431. if (ap->version == 0)
  432. ap->version = client_version;
  433. #ifdef WRITE_PACKETS
  434. WritePackets(ap->GetOpcodeName(), oversize_buffer, oversize_offset, false);
  435. #endif
  436. ap->copyInfo(p);
  437. InboundQueuePush(ap);
  438. safe_delete(p2);
  439. }
  440. }
  441. delete[] oversize_buffer;
  442. oversize_buffer=NULL;
  443. oversize_offset=0;
  444. }
  445. } else if (!oversize_buffer) {
  446. oversize_length=ntohl(*(uint32 *)(p->pBuffer+2));
  447. oversize_buffer=new unsigned char[oversize_length];
  448. memcpy(oversize_buffer,p->pBuffer+6,p->size-6);
  449. oversize_offset=p->size-6;
  450. //cout << "Oversized is " << oversize_offset << "/" << oversize_length << " (" << (p->size-6) << ") Seq=" << seq << endl;
  451. }
  452. }
  453. }
  454. break;
  455. case OP_KeepAlive: {
  456. #ifndef COLLECTOR
  457. NonSequencedPush(new EQProtocolPacket(p->opcode,p->pBuffer,p->size));
  458. #endif
  459. }
  460. break;
  461. case OP_Ack: {
  462. if (!p->pBuffer || (p->Size() < 4))
  463. {
  464. LogWrite(PACKET__DEBUG, 9, "Packet", "Received OP_Ack that was of malformed size");
  465. break;
  466. }
  467. uint16 seq = ntohs(*(uint16*)(p->pBuffer));
  468. AckPackets(seq);
  469. retransmittimer = Timer::GetCurrentTime2();
  470. }
  471. break;
  472. case OP_SessionRequest: {
  473. if (p->Size() < sizeof(SessionRequest))
  474. {
  475. break;
  476. }
  477. if (GetState() == ESTABLISHED) {
  478. //_log(NET__ERROR, _L "Received OP_SessionRequest in ESTABLISHED state (%d) streamactive (%i) attempt (%i)" __L, GetState(), streamactive, sessionAttempts);
  479. // client seems to try a max of 4 times (initial +3 retries) then gives up, giving it a few more attempts just in case
  480. // streamactive means we identified the opcode, we cannot re-establish this connection
  481. if (streamactive || (sessionAttempts > 30))
  482. {
  483. SendDisconnect(false);
  484. SetState(CLOSED);
  485. break;
  486. }
  487. }
  488. sessionAttempts++;
  489. init(GetState() != ESTABLISHED);
  490. OutboundQueueClear();
  491. SessionRequest *Request=(SessionRequest *)p->pBuffer;
  492. Session=ntohl(Request->Session);
  493. SetMaxLen(ntohl(Request->MaxLength));
  494. #ifndef COLLECTOR
  495. NextInSeq=0;
  496. Key=0x33624702;
  497. SendSessionResponse();
  498. #endif
  499. SetState(ESTABLISHED);
  500. }
  501. break;
  502. case OP_SessionResponse: {
  503. if (p->Size() < sizeof(SessionResponse))
  504. {
  505. break;
  506. }
  507. init();
  508. OutboundQueueClear();
  509. SetActive(true);
  510. SessionResponse *Response=(SessionResponse *)p->pBuffer;
  511. SetMaxLen(ntohl(Response->MaxLength));
  512. Key=ntohl(Response->Key);
  513. NextInSeq=0;
  514. SetState(ESTABLISHED);
  515. if (!Session)
  516. Session=ntohl(Response->Session);
  517. compressed=(Response->Format&FLAG_COMPRESSED);
  518. encoded=(Response->Format&FLAG_ENCODED);
  519. // Kinda kludgy, but trie for now
  520. if (compressed) {
  521. if (remote_port==9000 || (remote_port==0 && p->src_port==9000))
  522. SetStreamType(WorldStream);
  523. else
  524. SetStreamType(ZoneStream);
  525. } else if (encoded)
  526. SetStreamType(ChatOrMailStream);
  527. else
  528. SetStreamType(LoginStream);
  529. }
  530. break;
  531. case OP_SessionDisconnect: {
  532. //NextInSeq=0;
  533. SendDisconnect();
  534. //SetState(CLOSED);
  535. }
  536. break;
  537. case OP_OutOfOrderAck: {
  538. if (!p->pBuffer || (p->Size() < 4))
  539. {
  540. LogWrite(PACKET__DEBUG, 9, "Packet", "Received OP_OutOfOrderAck that was of malformed size");
  541. break;
  542. }
  543. uint16 seq = ntohs(*(uint16*)(p->pBuffer));
  544. MOutboundQueue.lock();
  545. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  546. LogWrite(PACKET__DEBUG, 9, "Packet", "Pre-OOA Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  547. }
  548. //if the packet they got out of order is between our last acked packet and the last sent packet, then its valid.
  549. if (CompareSequence(SequencedBase, seq) != SeqPast && CompareSequence(NextOutSeq, seq) == SeqPast) {
  550. uint16 sqsize = SequencedQueue.size();
  551. uint16 index = seq - SequencedBase;
  552. LogWrite(PACKET__DEBUG, 9, "Packet", "OP_OutOfOrderAck marking packet acked in queue (queue index = %u, queue size = %u)", index, sqsize);
  553. if (index < sqsize) {
  554. SequencedQueue[index]->acked = true;
  555. // flag packets for a resend
  556. uint16 count = 0;
  557. uint32 timeout = AverageDelta * 2 + 100;
  558. for (auto sitr = SequencedQueue.begin(); sitr != SequencedQueue.end() && count < index; ++sitr, ++count) {
  559. if (!(*sitr)->acked && (*sitr)->sent_time > 0 && (((*sitr)->sent_time + timeout) < Timer::GetCurrentTime2())) {
  560. (*sitr)->sent_time = 0;
  561. LogWrite(PACKET__DEBUG, 9, "Packet", "OP_OutOfOrderAck Flagging packet %u for retransmission", SequencedBase + count);
  562. }
  563. }
  564. }
  565. if (RETRANSMIT_TIMEOUT_MULT) {
  566. retransmittimer = Timer::GetCurrentTime2();
  567. }
  568. }
  569. else {
  570. LogWrite(PACKET__DEBUG, 9, "Packet", "Received OP_OutOfOrderAck for out-of-window %u. Window (%u->%u)", seq, SequencedBase, NextOutSeq);
  571. }
  572. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  573. LogWrite(PACKET__DEBUG, 9, "Packet", "Post-OOA Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  574. }
  575. MOutboundQueue.unlock();
  576. }
  577. break;
  578. case OP_ServerKeyRequest:{
  579. if (p->Size() < sizeof(ClientSessionStats))
  580. {
  581. //_log(NET__ERROR, _L "Received OP_SessionStatRequest that was of malformed size" __L);
  582. break;
  583. }
  584. ClientSessionStats* Stats = (ClientSessionStats*)p->pBuffer;
  585. int16 request_id = Stats->RequestID;
  586. AdjustRates(ntohl(Stats->average_delta));
  587. ServerSessionStats* stats=(ServerSessionStats*)p->pBuffer;
  588. memset(stats, 0, sizeof(ServerSessionStats));
  589. stats->RequestID = request_id;
  590. stats->current_time = ntohl(Timer::GetCurrentTime2());
  591. stats->sent_packets = ntohl(sent_packets);
  592. stats->sent_packets2 = ntohl(sent_packets);
  593. stats->received_packets = ntohl(received_packets);
  594. stats->received_packets2 = ntohl(received_packets);
  595. NonSequencedPush(new EQProtocolPacket(OP_SessionStatResponse,p->pBuffer,p->size));
  596. if(!crypto->isEncrypted())
  597. SendKeyRequest();
  598. }
  599. break;
  600. case OP_SessionStatResponse: {
  601. LogWrite(PACKET__INFO, 0, "Packet", "OP_SessionStatResponse");
  602. }
  603. break;
  604. case OP_OutOfSession: {
  605. LogWrite(PACKET__INFO, 0, "Packet", "OP_OutOfSession");
  606. SendDisconnect();
  607. SetState(CLOSED);
  608. }
  609. break;
  610. default:
  611. //EQApplicationPacket *ap = p->MakeApplicationPacket(app_opcode_size);
  612. //InboundQueuePush(ap);
  613. EQApplicationPacket* ap = p->MakeApplicationPacket(app_opcode_size);
  614. if (ap->version == 0)
  615. ap->version = client_version;
  616. #ifdef WRITE_PACKETS
  617. WritePackets(ap->GetOpcodeName(), p->pBuffer, p->size, false);
  618. #endif
  619. //InboundQueuePush(ap);
  620. LogWrite(PACKET__INFO, 0, "Packet", "Received unknown packet type, not adding to inbound queue");
  621. safe_delete(ap);
  622. //SendDisconnect();
  623. break;
  624. }
  625. }
  626. }
  627. int8 EQStream::EQ2_Compress(EQ2Packet* app, int8 offset){
  628. #ifdef LE_DEBUG
  629. printf( "Before Compress in %s, line %i:\n", __FUNCTION__, __LINE__);
  630. DumpPacket(app);
  631. #endif
  632. uchar* pDataPtr = app->pBuffer + offset;
  633. int xpandSize = app->size * 2;
  634. uchar* deflate_buff = new uchar[xpandSize];
  635. MCompressData.lock();
  636. stream.next_in = pDataPtr;
  637. stream.avail_in = app->size - offset;
  638. stream.next_out = deflate_buff;
  639. stream.avail_out = xpandSize;
  640. int ret = deflate(&stream, Z_SYNC_FLUSH);
  641. if (ret != Z_OK)
  642. {
  643. printf("ZLIB COMPRESSION RETFAIL: %i, %i (Ret: %i)\n", app->size, stream.avail_out, ret);
  644. MCompressData.unlock();
  645. safe_delete_array(deflate_buff);
  646. return 0;
  647. }
  648. int32 newsize = xpandSize - stream.avail_out;
  649. safe_delete_array(app->pBuffer);
  650. app->size = newsize + offset;
  651. app->pBuffer = new uchar[app->size];
  652. app->pBuffer[(offset - 1)] = 1;
  653. memcpy(app->pBuffer + offset, deflate_buff, newsize);
  654. MCompressData.unlock();
  655. safe_delete_array(deflate_buff);
  656. #ifdef LE_DEBUG
  657. printf( "After Compress in %s, line %i:\n", __FUNCTION__, __LINE__);
  658. DumpPacket(app);
  659. #endif
  660. return offset - 1;
  661. }
  662. int16 EQStream::processRSAKey(EQProtocolPacket *p){
  663. /*int16 limit = 0;
  664. int8 offset = 13;
  665. int8 offset2 = 0;
  666. if(p->pBuffer[2] == 0)
  667. limit = p->pBuffer[9];
  668. else{
  669. limit = p->pBuffer[5];
  670. offset2 = 5;
  671. offset-=1;
  672. }
  673. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + offset + (limit-8), 8));
  674. return (limit + offset +1) - offset2;*/
  675. if(p->pBuffer[0] == 0)
  676. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + 62, 8));
  677. else
  678. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + 61, 8));
  679. return 0;
  680. }
  681. void EQStream::SendKeyRequest(){
  682. int32 crypto_key_size = 60;
  683. int16 size = sizeof(KeyGen_Struct) + sizeof(KeyGen_End_Struct) + crypto_key_size;
  684. EQ2Packet *outapp=new EQ2Packet(OP_WSLoginRequestMsg,NULL,size);
  685. memcpy(&outapp->pBuffer[0], &crypto_key_size, sizeof(int32));
  686. memset(&outapp->pBuffer[4], 0xFF, crypto_key_size);
  687. memset(&outapp->pBuffer[size-5], 1, 1);
  688. memset(&outapp->pBuffer[size-1], 1, 1);
  689. EQ2QueuePacket(outapp);
  690. }
  691. void EQStream::EncryptPacket(EQ2Packet* app, int8 compress_offset, int8 offset){
  692. if(app->size>2 && crypto->isEncrypted()){
  693. app->packet_encrypted = true;
  694. uchar* crypt_buff = app->pBuffer;
  695. if(app->eq2_compressed)
  696. crypto->RC4Encrypt(crypt_buff + compress_offset, app->size - compress_offset);
  697. else
  698. crypto->RC4Encrypt(crypt_buff + 2 + offset, app->size - 2 - offset);
  699. }
  700. }
  701. void EQStream::EQ2QueuePacket(EQ2Packet* app, bool attempted_combine){
  702. if(CheckActive()){
  703. if(!attempted_combine){
  704. MCombineQueueLock.lock();
  705. combine_queue.push_back(app);
  706. MCombineQueueLock.unlock();
  707. }
  708. else{
  709. MCombineQueueLock.lock();
  710. PreparePacket(app);
  711. MCombineQueueLock.unlock();
  712. #ifdef LE_DEBUG
  713. printf( "After B in %s, line %i:\n", __FUNCTION__, __LINE__);
  714. DumpPacket(app);
  715. #endif
  716. SendPacket(app);
  717. }
  718. }
  719. }
  720. void EQStream::UnPreparePacket(EQ2Packet* app){
  721. if(app->pBuffer[2] == 0 && app->pBuffer[3] == 19){
  722. uchar* new_buffer = new uchar[app->size-3];
  723. memcpy(new_buffer+2, app->pBuffer+5, app->size-3);
  724. delete[] app->pBuffer;
  725. app->size-=3;
  726. app->pBuffer = new_buffer;
  727. }
  728. }
  729. #ifdef WRITE_PACKETS
  730. char EQStream::GetChar(uchar in)
  731. {
  732. if (in < ' ' || in > '~')
  733. return '.';
  734. return (char)in;
  735. }
  736. void EQStream::WriteToFile(char* pFormat, ...) {
  737. va_list args;
  738. va_start(args, pFormat);
  739. vfprintf(write_packets, pFormat, args);
  740. va_end(args);
  741. }
  742. void EQStream::WritePackets(const char* opcodeName, uchar* data, int32 size, bool outgoing) {
  743. MWritePackets.lock();
  744. struct in_addr ip_addr;
  745. ip_addr.s_addr = remote_ip;
  746. char timebuffer[80];
  747. time_t rawtime;
  748. struct tm* timeinfo;
  749. time(&rawtime);
  750. timeinfo = localtime(&rawtime);
  751. strftime(timebuffer, 80, "%m/%d/%Y %H:%M:%S", timeinfo);
  752. if (outgoing)
  753. WriteToFile("-- %s --\n%s\nSERVER -> %s\n", opcodeName, timebuffer, inet_ntoa(ip_addr));
  754. else
  755. WriteToFile("-- %s --\n%s\n%s -> SERVER\n", opcodeName, timebuffer, inet_ntoa(ip_addr));
  756. int i;
  757. int nLines = size / 16;
  758. int nExtra = size % 16;
  759. uchar* pPtr = data;
  760. for (i = 0; i < nLines; i++)
  761. {
  762. WriteToFile("%4.4X:\t%2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %c%c%c%c%c%c%c%c%c%c%c%c%c%c%c%c\n", i * 16, pPtr[0], pPtr[1], pPtr[2], pPtr[3], pPtr[4], pPtr[5], pPtr[6], pPtr[7], pPtr[8], pPtr[9], pPtr[10], pPtr[11], pPtr[12], pPtr[13], pPtr[14], pPtr[15], GetChar(pPtr[0]), GetChar(pPtr[1]), GetChar(pPtr[2]), GetChar(pPtr[3]), GetChar(pPtr[4]), GetChar(pPtr[5]), GetChar(pPtr[6]), GetChar(pPtr[7]), GetChar(pPtr[8]), GetChar(pPtr[9]), GetChar(pPtr[10]), GetChar(pPtr[11]), GetChar(pPtr[12]), GetChar(pPtr[13]), GetChar(pPtr[14]), GetChar(pPtr[15]));
  763. pPtr += 16;
  764. }
  765. if (nExtra)
  766. {
  767. WriteToFile("%4.4X\t", nLines * 16);
  768. for (i = 0; i < nExtra; i++)
  769. {
  770. WriteToFile("%2.2X ", pPtr[i]);
  771. }
  772. for (i; i < 16; i++)
  773. WriteToFile(" ");
  774. for (i = 0; i < nExtra; i++)
  775. {
  776. WriteToFile("%c", GetChar(pPtr[i]));
  777. }
  778. WriteToFile("\n");
  779. }
  780. WriteToFile("\n\n");
  781. fflush(write_packets);
  782. MWritePackets.unlock();
  783. }
  784. void EQStream::WritePackets(EQ2Packet* app, bool outgoing) {
  785. if (app->version == 0)
  786. app->version = client_version;
  787. WritePackets(app->GetOpcodeName(), app->pBuffer, app->size, outgoing);
  788. }
  789. #endif
  790. void EQStream::PreparePacket(EQ2Packet* app, int8 offset){
  791. app->setVersion(client_version);
  792. compressed_offset = 0;
  793. #ifdef LE_DEBUG
  794. printf( "Before A in %s, line %i:\n", __FUNCTION__, __LINE__);
  795. DumpPacket(app);
  796. #endif
  797. if(!app->packet_prepared){
  798. if(app->PreparePacket(MaxLen) == 255) //invalid version
  799. return;
  800. }
  801. #ifdef LE_DEBUG
  802. printf( "After Prepare in %s, line %i:\n", __FUNCTION__, __LINE__);
  803. DumpPacket(app);
  804. #endif
  805. #ifdef WRITE_PACKETS
  806. if (!app->eq2_compressed && !app->packet_encrypted)
  807. WritePackets(app, true);
  808. #endif
  809. if(!app->eq2_compressed && app->size>128){
  810. compressed_offset = EQ2_Compress(app);
  811. if (compressed_offset)
  812. app->eq2_compressed = true;
  813. }
  814. if(!app->packet_encrypted){
  815. EncryptPacket(app, compressed_offset, offset);
  816. if(app->size > 2 && app->pBuffer[2] == 0){
  817. uchar* new_buffer = new uchar[app->size+1];
  818. new_buffer[2] = 0;
  819. memcpy(new_buffer+3, app->pBuffer+2, app->size-2);
  820. delete[] app->pBuffer;
  821. app->pBuffer = new_buffer;
  822. app->size++;
  823. }
  824. }
  825. #ifdef LE_DEBUG
  826. printf( "After A in %s, line %i:\n", __FUNCTION__, __LINE__);
  827. DumpPacket(app);
  828. #endif
  829. }
  830. void EQStream::SendPacket(EQProtocolPacket *p)
  831. {
  832. uint32 chunksize,used;
  833. uint32 length;
  834. // Convert the EQApplicationPacket to 1 or more EQProtocolPackets
  835. if (p->size>( MaxLen-8)) { // proto-op(2), seq(2), app-op(2) ... data ... crc(2)
  836. uchar* tmpbuff=p->pBuffer;
  837. length=p->size - 2;
  838. EQProtocolPacket *out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  839. *(uint32 *)(out->pBuffer+2)=htonl(length);
  840. used=MaxLen-10;
  841. memcpy(out->pBuffer+6,tmpbuff+2,used);
  842. #ifdef LE_DEBUG
  843. printf("(%s, %i) New Fragment:\n ", __FUNCTION__, __LINE__);
  844. DumpPacket(out);
  845. #endif
  846. SequencedPush(out);
  847. while (used<length) {
  848. chunksize=min(length-used,MaxLen-6);
  849. out=new EQProtocolPacket(OP_Fragment,NULL,chunksize+2);
  850. //memcpy(out->pBuffer+2,tmpbuff,1);
  851. memcpy(out->pBuffer+2,tmpbuff+used+2,chunksize);
  852. #ifdef LE_DEBUG
  853. printf("Chunk: \n");
  854. DumpPacket(out);
  855. #endif
  856. SequencedPush(out);
  857. used+=chunksize;
  858. }
  859. #ifdef LE_DEBUG
  860. printf( "ChunkDelete: \n");
  861. DumpPacket(out);
  862. //cerr << "1: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  863. #endif
  864. delete p;
  865. } else {
  866. SequencedPush(p);
  867. }
  868. }
  869. void EQStream::SendPacket(EQApplicationPacket *p)
  870. {
  871. uint32 chunksize,used;
  872. uint32 length;
  873. // Convert the EQApplicationPacket to 1 or more EQProtocolPackets
  874. if (p->size>(MaxLen-8)) { // proto-op(2), seq(2), app-op(2) ... data ... crc(2)
  875. //cout << "Making oversized packet for: " << endl;
  876. //cout << p->size << endl;
  877. //p->DumpRawHeader();
  878. //dump_message(p->pBuffer,p->size,timestamp());
  879. //cout << p->size << endl;
  880. unsigned char *tmpbuff=new unsigned char[p->size+2];
  881. //cout << hex << (int)tmpbuff << dec << endl;
  882. length=p->serialize(tmpbuff);
  883. EQProtocolPacket *out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  884. *(uint32 *)(out->pBuffer+2)=htonl(p->Size());
  885. memcpy(out->pBuffer+6,tmpbuff,MaxLen-10);
  886. used=MaxLen-10;
  887. SequencedPush(out);
  888. //cout << "Chunk #" << ++i << " size=" << used << ", length-used=" << (length-used) << endl;
  889. while (used<length) {
  890. out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  891. chunksize=min(length-used,MaxLen-6);
  892. memcpy(out->pBuffer+2,tmpbuff+used,chunksize);
  893. out->size=chunksize+2;
  894. SequencedPush(out);
  895. used+=chunksize;
  896. //cout << "Chunk #"<< ++i << " size=" << chunksize << ", length-used=" << (length-used) << endl;
  897. }
  898. //cerr << "1: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  899. delete p;
  900. delete[] tmpbuff;
  901. } else {
  902. EQProtocolPacket *out=new EQProtocolPacket(OP_Packet,NULL,p->Size()+2);
  903. p->serialize(out->pBuffer+2);
  904. SequencedPush(out);
  905. //cerr << "2: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  906. delete p;
  907. }
  908. }
  909. void EQStream::SequencedPush(EQProtocolPacket *p)
  910. {
  911. p->setVersion(client_version);
  912. MOutboundQueue.lock();
  913. *(uint16 *)(p->pBuffer)=htons(NextOutSeq);
  914. SequencedQueue.push_back(p);
  915. p->sequence = NextOutSeq;
  916. NextOutSeq++;
  917. MOutboundQueue.unlock();
  918. }
  919. void EQStream::NonSequencedPush(EQProtocolPacket *p)
  920. {
  921. p->setVersion(client_version);
  922. MOutboundQueue.lock();
  923. NonSequencedQueue.push(p);
  924. MOutboundQueue.unlock();
  925. }
  926. void EQStream::SendAck(uint16 seq)
  927. {
  928. uint16 Seq=htons(seq);
  929. SetLastAckSent(seq);
  930. NonSequencedPush(new EQProtocolPacket(OP_Ack,(unsigned char *)&Seq,sizeof(uint16)));
  931. }
  932. void EQStream::SendOutOfOrderAck(uint16 seq)
  933. {
  934. uint16 Seq=htons(seq);
  935. NonSequencedPush(new EQProtocolPacket(OP_OutOfOrderAck,(unsigned char *)&Seq,sizeof(uint16)));
  936. }
  937. bool EQStream::CheckCombineQueue(){
  938. bool ret = true; //processed all packets
  939. MCombineQueueLock.lock();
  940. if(combine_queue.size() > 0){
  941. EQ2Packet* first = combine_queue.front();
  942. combine_queue.pop_front();
  943. if(combine_queue.size() == 0){ //nothing to combine this with
  944. EQ2QueuePacket(first, true);
  945. }
  946. else{
  947. PreparePacket(first);
  948. EQ2Packet* second = 0;
  949. bool combine_worked = false;
  950. int16 count = 0;
  951. while(combine_queue.size()){
  952. count++;
  953. second = combine_queue.front();
  954. combine_queue.pop_front();
  955. PreparePacket(second);
  956. /*if(first->GetRawOpcode() != OP_AppCombined && first->pBuffer[2] == 0){
  957. EQ2Packet* tmp = second;
  958. second = first;
  959. first = tmp;
  960. }*/
  961. if(!first->AppCombine(second)){
  962. first->SetProtocolOpcode(OP_Packet);
  963. if(combine_worked){
  964. SequencedPush(first);
  965. }
  966. else{
  967. EQ2QueuePacket(first, true);
  968. }
  969. first = second;
  970. combine_worked = false;
  971. }
  972. else{
  973. combine_worked = true;
  974. //DumpPacket(first);
  975. }
  976. if(count >= 60 || first->size > 4000){ //other clients need packets too
  977. ret = false;
  978. break;
  979. }
  980. }
  981. if(first){
  982. first->SetProtocolOpcode(OP_Packet);
  983. if(combine_worked){
  984. SequencedPush(first);
  985. }
  986. else{
  987. EQ2QueuePacket(first, true);
  988. }
  989. }
  990. }
  991. }
  992. MCombineQueueLock.unlock();
  993. return ret;
  994. }
  995. void EQStream::CheckResend(int eq_fd){
  996. int32 curr = Timer::GetCurrentTime2();
  997. EQProtocolPacket* packet = 0;
  998. deque<EQProtocolPacket*>::iterator itr;
  999. MResendQue.lock();
  1000. for(itr=resend_que.begin();itr!=resend_que.end();itr++){
  1001. packet = *itr;
  1002. if(packet->attempt_count >= 5){//tried to resend this packet 5 times, client must already have it but didnt ack it
  1003. safe_delete(packet);
  1004. itr = resend_que.erase(itr);
  1005. if(itr == resend_que.end())
  1006. break;
  1007. }
  1008. else{
  1009. if((curr - packet->sent_time) < 1000)
  1010. continue;
  1011. packet->sent_time -=1000;
  1012. packet->attempt_count++;
  1013. WritePacket(eq_fd, packet);
  1014. }
  1015. }
  1016. MResendQue.unlock();
  1017. }
  1018. //returns SeqFuture if `seq` is later than `expected_seq`
  1019. EQStream::SeqOrder EQStream::CompareSequence(uint16 expected_seq, uint16 seq)
  1020. {
  1021. if (expected_seq == seq) {
  1022. // Curent
  1023. return SeqInOrder;
  1024. }
  1025. else if ((seq > expected_seq && (uint32)seq < ((uint32)expected_seq + EQStream::MaxWindowSize)) || seq < (expected_seq - EQStream::MaxWindowSize)) {
  1026. // Future
  1027. return SeqFuture;
  1028. }
  1029. else {
  1030. // Past
  1031. return SeqPast;
  1032. }
  1033. }
  1034. void EQStream::AckPackets(uint16 seq)
  1035. {
  1036. std::deque<EQProtocolPacket*>::iterator itr, tmp;
  1037. MOutboundQueue.lock();
  1038. SeqOrder ord = CompareSequence(SequencedBase, seq);
  1039. if (ord == SeqInOrder) {
  1040. //they are not acking anything new...
  1041. LogWrite(PACKET__DEBUG, 9, "Packet", "Received an ack with no window advancement (seq %u)", seq);
  1042. }
  1043. else if (ord == SeqPast) {
  1044. //they are nacking blocks going back before our buffer, wtf?
  1045. LogWrite(PACKET__DEBUG, 9, "Packet", "Received an ack with backward window advancement (they gave %u, our window starts at %u). This is bad" , seq, SequencedBase);
  1046. }
  1047. else {
  1048. LogWrite(PACKET__DEBUG, 9, "Packet", "Received an ack up through sequence %u. Our base is %u", seq, SequencedBase);
  1049. //this is a good ack, we get to ack some blocks.
  1050. seq++; //we stop at the block right after their ack, counting on the wrap of both numbers.
  1051. while (SequencedBase != seq) {
  1052. if (SequencedQueue.empty()) {
  1053. LogWrite(PACKET__DEBUG, 9, "Packet", "OUT OF PACKETS acked packet with sequence %u. Next send is %u before this", (unsigned long)SequencedBase, SequencedQueue.size());
  1054. SequencedBase = NextOutSeq;
  1055. break;
  1056. }
  1057. LogWrite(PACKET__DEBUG, 9, "Packet", "Removing acked packet with sequence %u", (unsigned long)SequencedBase);
  1058. //clean out the acked packet
  1059. delete SequencedQueue.front();
  1060. SequencedQueue.pop_front();
  1061. //advance the base sequence number to the seq of the block after the one we just got rid of.
  1062. SequencedBase++;
  1063. }
  1064. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  1065. LogWrite(PACKET__DEBUG, 9, "Packet", "Post-Ack on %u Invalid Sequenced queue: BS %u + SQ %u != NOS %u", seq, SequencedBase, SequencedQueue.size(), NextOutSeq);
  1066. }
  1067. }
  1068. MOutboundQueue.unlock();
  1069. }
  1070. void EQStream::Write(int eq_fd)
  1071. {
  1072. queue<EQProtocolPacket *> ReadyToSend;
  1073. long maxack;
  1074. // Check our rate to make sure we can send more
  1075. MRate.lock();
  1076. sint32 threshold=RateThreshold;
  1077. MRate.unlock();
  1078. if (BytesWritten > threshold) {
  1079. //cout << "Over threshold: " << BytesWritten << " > " << threshold << endl;
  1080. return;
  1081. }
  1082. MCombinedAppPacket.lock();
  1083. EQApplicationPacket *CombPack=CombinedAppPacket;
  1084. CombinedAppPacket=NULL;
  1085. MCombinedAppPacket.unlock();
  1086. if (CombPack) {
  1087. SendPacket(CombPack);
  1088. }
  1089. // If we got more packets to we need to ack, send an ack on the highest one
  1090. MAcks.lock();
  1091. maxack=MaxAckReceived;
  1092. // Added from peaks findings
  1093. if (NextAckToSend>LastAckSent || LastAckSent == 0x0000ffff)
  1094. SendAck(NextAckToSend);
  1095. MAcks.unlock();
  1096. // Lock the outbound queues while we process
  1097. MOutboundQueue.lock();
  1098. // Adjust where we start sending in case we get a late ack
  1099. //if (maxack>LastSeqSent)
  1100. // LastSeqSent=maxack;
  1101. // Place to hold the base packet t combine into
  1102. EQProtocolPacket *p=NULL;
  1103. std::deque<EQProtocolPacket*>::iterator sitr;
  1104. // Find the next sequenced packet to send from the "queue"
  1105. sitr = SequencedQueue.begin();
  1106. uint16 count = 0;
  1107. // get to start of packets
  1108. while (sitr != SequencedQueue.end() && (*sitr)->sent_time > 0) {
  1109. ++sitr;
  1110. ++count;
  1111. }
  1112. bool SeqEmpty = false, NonSeqEmpty = false;
  1113. // Loop until both are empty or MaxSends is reached
  1114. while (!SeqEmpty || !NonSeqEmpty) {
  1115. // See if there are more non-sequenced packets left
  1116. if (!NonSequencedQueue.empty()) {
  1117. if (!p) {
  1118. // If we don't have a packet to try to combine into, use this one as the base
  1119. // And remove it form the queue
  1120. p = NonSequencedQueue.front();
  1121. LogWrite(PACKET__DEBUG, 9, "Packet", "Starting combined packet with non-seq packet of len %u",p->size);
  1122. NonSequencedQueue.pop();
  1123. }
  1124. else if (!p->combine(NonSequencedQueue.front())) {
  1125. // Trying to combine this packet with the base didn't work (too big maybe)
  1126. // So just send the base packet (we'll try this packet again later)
  1127. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined packet full at len %u, next non-seq packet is len %u", p->size, (NonSequencedQueue.front())->size);
  1128. ReadyToSend.push(p);
  1129. BytesWritten += p->size;
  1130. p = nullptr;
  1131. if (BytesWritten > threshold) {
  1132. // Sent enough this round, lets stop to be fair
  1133. LogWrite(PACKET__DEBUG, 9, "Packet", "Exceeded write threshold in nonseq (%u > %u)", BytesWritten, threshold);
  1134. break;
  1135. }
  1136. }
  1137. else {
  1138. // Combine worked, so just remove this packet and it's spot in the queue
  1139. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined non-seq packet of len %u, yeilding %u combined", (NonSequencedQueue.front())->size, p->size);
  1140. delete NonSequencedQueue.front();
  1141. NonSequencedQueue.pop();
  1142. }
  1143. }
  1144. else {
  1145. // No more non-sequenced packets
  1146. NonSeqEmpty = true;
  1147. }
  1148. if (sitr != SequencedQueue.end()) {
  1149. uint16 seq_send = SequencedBase + count; //just for logging...
  1150. if (SequencedQueue.empty()) {
  1151. LogWrite(PACKET__DEBUG, 9, "Packet", "Tried to write a packet with an empty queue (%u is past next out %u)", seq_send, NextOutSeq);
  1152. SeqEmpty = true;
  1153. continue;
  1154. }
  1155. if ((*sitr)->acked || (*sitr)->sent_time != 0) {
  1156. ++sitr;
  1157. ++count;
  1158. if (p) {
  1159. LogWrite(PACKET__DEBUG, 9, "Packet", "Final combined packet not full, len %u", p->size);
  1160. ReadyToSend.push(p);
  1161. BytesWritten += p->size;
  1162. p = nullptr;
  1163. }
  1164. LogWrite(PACKET__DEBUG, 9, "Packet", "Not retransmitting seq packet %u because already marked as acked", seq_send);
  1165. }
  1166. else if (!p) {
  1167. // If we don't have a packet to try to combine into, use this one as the base
  1168. // Copy it first as it will still live until it is acked
  1169. p = (*sitr)->Copy();
  1170. LogWrite(PACKET__DEBUG, 9, "Packet", "Starting combined packet with seq packet %u of len %u", seq_send, p->size);
  1171. (*sitr)->sent_time = Timer::GetCurrentTime2();
  1172. ++sitr;
  1173. ++count;
  1174. }
  1175. else if (!p->combine(*sitr)) {
  1176. // Trying to combine this packet with the base didn't work (too big maybe)
  1177. // So just send the base packet (we'll try this packet again later)
  1178. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined packet full at len %u, next seq packet %u is len %u", p->size, seq_send + 1, (*sitr)->size);
  1179. ReadyToSend.push(p);
  1180. BytesWritten += p->size;
  1181. p = nullptr;
  1182. if ((*sitr)->opcode != OP_Fragment && BytesWritten > threshold) {
  1183. // Sent enough this round, lets stop to be fair
  1184. LogWrite(PACKET__DEBUG, 9, "Packet", "Exceeded write threshold in seq (%u > %u)", BytesWritten, threshold);
  1185. break;
  1186. }
  1187. }
  1188. else {
  1189. // Combine worked
  1190. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined seq packet %u of len %u, yeilding %u combined", seq_send, (*sitr)->size, p->size);
  1191. (*sitr)->sent_time = Timer::GetCurrentTime2();
  1192. ++sitr;
  1193. ++count;
  1194. }
  1195. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  1196. LogWrite(PACKET__DEBUG, 9, "Packet", "Post send Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  1197. }
  1198. }
  1199. else {
  1200. // No more sequenced packets
  1201. SeqEmpty = true;
  1202. }
  1203. }
  1204. MOutboundQueue.unlock(); // Unlock the queue
  1205. // We have a packet still, must have run out of both seq and non-seq, so send it
  1206. if (p) {
  1207. LogWrite(PACKET__DEBUG, 9, "Packet", "Final combined packet not full, len %u", p->size);
  1208. ReadyToSend.push(p);
  1209. BytesWritten += p->size;
  1210. }
  1211. // Send all the packets we "made"
  1212. while (!ReadyToSend.empty()) {
  1213. p = ReadyToSend.front();
  1214. WritePacket(eq_fd, p);
  1215. delete p;
  1216. ReadyToSend.pop();
  1217. }
  1218. //see if we need to send our disconnect and finish our close
  1219. if (SeqEmpty && NonSeqEmpty) {
  1220. //no more data to send
  1221. if (GetState() == CLOSING) {
  1222. MOutboundQueue.lock();
  1223. if (SequencedQueue.size() > 0 )
  1224. LogWrite(PACKET__DEBUG, 9, "Packet", "All outgoing data flushed, client should be disconnecting, awaiting acknowledgement of SequencedQueue.");
  1225. else
  1226. {
  1227. LogWrite(PACKET__DEBUG, 9, "Packet", "All outgoing data flushed, disconnecting client.");
  1228. //we are waiting for the queues to empty, now we can do our disconnect.
  1229. //this packet will not actually go out until the next call to Write().
  1230. SendDisconnect();
  1231. //SetState(CLOSED);
  1232. }
  1233. MOutboundQueue.unlock();
  1234. }
  1235. }
  1236. }
  1237. void EQStream::WritePacket(int eq_fd, EQProtocolPacket *p)
  1238. {
  1239. uint32 length = 0;
  1240. sockaddr_in address;
  1241. unsigned char tmpbuffer[2048];
  1242. address.sin_family = AF_INET;
  1243. address.sin_addr.s_addr=remote_ip;
  1244. address.sin_port=remote_port;
  1245. #ifdef NOWAY
  1246. uint32 ip=address.sin_addr.s_addr;
  1247. cout << "Sending to: "
  1248. << (int)*(unsigned char *)&ip
  1249. << "." << (int)*((unsigned char *)&ip+1)
  1250. << "." << (int)*((unsigned char *)&ip+2)
  1251. << "." << (int)*((unsigned char *)&ip+3)
  1252. << "," << (int)ntohs(address.sin_port) << "(" << p->size << ")" << endl;
  1253. p->DumpRaw();
  1254. cout << "-------------" << endl;
  1255. #endif
  1256. length=p->serialize(buffer);
  1257. if (p->opcode!=OP_SessionRequest && p->opcode!=OP_SessionResponse) {
  1258. if (compressed) {
  1259. BytesWritten -= p->size;
  1260. uint32 newlen=EQProtocolPacket::Compress(buffer,length,tmpbuffer,2048);
  1261. memcpy(buffer,tmpbuffer,newlen);
  1262. length=newlen;
  1263. BytesWritten += newlen;
  1264. }
  1265. if (encoded) {
  1266. EQProtocolPacket::ChatEncode(buffer,length,Key);
  1267. }
  1268. *(uint16 *)(buffer+length)=htons(CRC16(buffer,length,Key));
  1269. length+=2;
  1270. }
  1271. sent_packets++;
  1272. //dump_message_column(buffer,length,"Writer: ");
  1273. //cout << "Raw Data:\n";
  1274. //DumpPacket(buffer, length);
  1275. sendto(eq_fd,(char *)buffer,length,0,(sockaddr *)&address,sizeof(address));
  1276. }
  1277. EQProtocolPacket *EQStream::Read(int eq_fd, sockaddr_in *from)
  1278. {
  1279. int socklen;
  1280. int length=0;
  1281. unsigned char buffer[2048];
  1282. EQProtocolPacket *p=NULL;
  1283. char temp[15];
  1284. socklen=sizeof(sockaddr);
  1285. #ifdef WIN32
  1286. length=recvfrom(eq_fd, (char *)buffer, 2048, 0, (struct sockaddr*)from, (int *)&socklen);
  1287. #else
  1288. length=recvfrom(eq_fd, buffer, 2048, 0, (struct sockaddr*)from, (socklen_t *)&socklen);
  1289. #endif
  1290. if (length>=2) {
  1291. DumpPacket(buffer, length);
  1292. p=new EQProtocolPacket(buffer[1],&buffer[2],length-2);
  1293. uint32 ip=from->sin_addr.s_addr;
  1294. sprintf(temp,"%d.%d.%d.%d:%d",
  1295. *(unsigned char *)&ip,
  1296. *((unsigned char *)&ip+1),
  1297. *((unsigned char *)&ip+2),
  1298. *((unsigned char *)&ip+3),
  1299. ntohs(from->sin_port));
  1300. //cout << timestamp() << "Data from: " << temp << " OpCode 0x" << hex << setw(2) << setfill('0') << (int)p->opcode << dec << endl;
  1301. //dump_message(p->pBuffer,p->size,timestamp());
  1302. }
  1303. return p;
  1304. }
  1305. void EQStream::SendSessionResponse()
  1306. {
  1307. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionResponse,NULL,sizeof(SessionResponse));
  1308. SessionResponse *Response=(SessionResponse *)out->pBuffer;
  1309. Response->Session=htonl(Session);
  1310. Response->MaxLength=htonl(MaxLen);
  1311. Response->UnknownA=2;
  1312. Response->Format=0;
  1313. if (compressed)
  1314. Response->Format|=FLAG_COMPRESSED;
  1315. if (encoded)
  1316. Response->Format|=FLAG_ENCODED;
  1317. Response->Key=htonl(Key);
  1318. out->size=sizeof(SessionResponse);
  1319. NonSequencedPush(out);
  1320. }
  1321. void EQStream::SendSessionRequest()
  1322. {
  1323. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionRequest,NULL,sizeof(SessionRequest));
  1324. SessionRequest *Request=(SessionRequest *)out->pBuffer;
  1325. memset(Request,0,sizeof(SessionRequest));
  1326. Request->Session=htonl(time(NULL));
  1327. Request->MaxLength=htonl(512);
  1328. NonSequencedPush(out);
  1329. }
  1330. void EQStream::SendDisconnect(bool setstate)
  1331. {
  1332. try{
  1333. if(GetState() != ESTABLISHED)
  1334. return;
  1335. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionDisconnect,NULL,sizeof(uint32)+sizeof(int16));
  1336. *(uint32 *)out->pBuffer=htonl(Session);
  1337. out->pBuffer[4] = 0;
  1338. out->pBuffer[5] = 6;
  1339. NonSequencedPush(out);
  1340. if(setstate)
  1341. SetState(CLOSING);
  1342. }
  1343. catch(...){}
  1344. }
  1345. void EQStream::InboundQueuePush(EQApplicationPacket *p)
  1346. {
  1347. MInboundQueue.lock();
  1348. InboundQueue.push_back(p);
  1349. MInboundQueue.unlock();
  1350. }
  1351. EQApplicationPacket *EQStream::PopPacket()
  1352. {
  1353. EQApplicationPacket *p=NULL;
  1354. MInboundQueue.lock();
  1355. if (InboundQueue.size()) {
  1356. p=InboundQueue.front();
  1357. InboundQueue.pop_front();
  1358. }
  1359. MInboundQueue.unlock();
  1360. if(p)
  1361. p->setVersion(client_version);
  1362. return p;
  1363. }
  1364. void EQStream::InboundQueueClear()
  1365. {
  1366. MInboundQueue.lock();
  1367. while(InboundQueue.size()){
  1368. delete InboundQueue.front();
  1369. InboundQueue.pop_front();
  1370. }
  1371. MInboundQueue.unlock();
  1372. }
  1373. void EQStream::EncryptPacket(uchar* data, int16 size){
  1374. if(size>6){
  1375. }
  1376. }
  1377. bool EQStream::HasOutgoingData()
  1378. {
  1379. bool flag;
  1380. //once closed, we have nothing more to say
  1381. if(CheckClosed())
  1382. return(false);
  1383. MOutboundQueue.lock();
  1384. flag=(!NonSequencedQueue.empty());
  1385. if (!flag) {
  1386. flag = (!SequencedQueue.empty());
  1387. }
  1388. MOutboundQueue.unlock();
  1389. if (!flag) {
  1390. MAcks.lock();
  1391. flag= (NextAckToSend>LastAckSent);
  1392. MAcks.unlock();
  1393. }
  1394. if (!flag) {
  1395. MCombinedAppPacket.lock();
  1396. flag=(CombinedAppPacket!=NULL);
  1397. MCombinedAppPacket.unlock();
  1398. }
  1399. return flag;
  1400. }
  1401. void EQStream::OutboundQueueClear()
  1402. {
  1403. MOutboundQueue.lock();
  1404. while(NonSequencedQueue.size()) {
  1405. delete NonSequencedQueue.front();
  1406. NonSequencedQueue.pop();
  1407. }
  1408. while(SequencedQueue.size()) {
  1409. delete SequencedQueue.front();
  1410. SequencedQueue.pop_front();
  1411. }
  1412. MOutboundQueue.unlock();
  1413. }
  1414. void EQStream::Process(const unsigned char *buffer, const uint32 length)
  1415. {
  1416. received_packets++;
  1417. static unsigned char newbuffer[2048];
  1418. uint32 newlength=0;
  1419. #ifdef LE_DEBUG
  1420. printf("ProcessBuffer:\n");
  1421. DumpPacket(buffer, length);
  1422. #endif
  1423. if (EQProtocolPacket::ValidateCRC(buffer,length,Key)) {
  1424. if (compressed) {
  1425. newlength=EQProtocolPacket::Decompress(buffer,length,newbuffer,2048);
  1426. #ifdef LE_DEBUG
  1427. printf("ProcessBufferDecompress:\n");
  1428. DumpPacket(buffer, newlength);
  1429. #endif
  1430. } else {
  1431. memcpy(newbuffer,buffer,length);
  1432. newlength=length;
  1433. if (encoded)
  1434. EQProtocolPacket::ChatDecode(newbuffer,newlength-2,Key);
  1435. }
  1436. if (buffer[1]!=0x01 && buffer[1]!=0x02 && buffer[1]!=0x1d)
  1437. newlength-=2;
  1438. #ifdef LE_DEBUG
  1439. printf("ResultProcessBuffer:\n");
  1440. DumpPacket(buffer, newlength);
  1441. #endif
  1442. EQProtocolPacket p(newbuffer,newlength);
  1443. ProcessPacket(&p);
  1444. ProcessQueue();
  1445. } else {
  1446. #ifdef EQN_DEBUG
  1447. cout << "Incoming packet failed checksum:" <<endl;
  1448. dump_message_column(const_cast<unsigned char *>(buffer),length,"CRC failed: ");
  1449. #endif
  1450. }
  1451. }
  1452. long EQStream::GetMaxAckReceived()
  1453. {
  1454. MAcks.lock();
  1455. long l=MaxAckReceived;
  1456. MAcks.unlock();
  1457. return l;
  1458. }
  1459. long EQStream::GetNextAckToSend()
  1460. {
  1461. MAcks.lock();
  1462. long l=NextAckToSend;
  1463. MAcks.unlock();
  1464. return l;
  1465. }
  1466. long EQStream::GetLastAckSent()
  1467. {
  1468. MAcks.lock();
  1469. long l=LastAckSent;
  1470. MAcks.unlock();
  1471. return l;
  1472. }
  1473. void EQStream::SetMaxAckReceived(uint32 seq)
  1474. {
  1475. deque<EQProtocolPacket *>::iterator itr;
  1476. MAcks.lock();
  1477. MaxAckReceived=seq;
  1478. MAcks.unlock();
  1479. MOutboundQueue.lock();
  1480. if (long(seq) > LastSeqSent)
  1481. LastSeqSent=seq;
  1482. MResendQue.lock();
  1483. EQProtocolPacket* packet = 0;
  1484. for(itr=resend_que.begin();itr!=resend_que.end();itr++){
  1485. packet = *itr;
  1486. if(packet && packet->sequence <= seq){
  1487. safe_delete(packet);
  1488. itr = resend_que.erase(itr);
  1489. if(itr == resend_que.end())
  1490. break;
  1491. }
  1492. }
  1493. MResendQue.unlock();
  1494. MOutboundQueue.unlock();
  1495. }
  1496. void EQStream::SetNextAckToSend(uint32 seq)
  1497. {
  1498. MAcks.lock();
  1499. NextAckToSend=seq;
  1500. MAcks.unlock();
  1501. }
  1502. void EQStream::SetLastAckSent(uint32 seq)
  1503. {
  1504. MAcks.lock();
  1505. LastAckSent=seq;
  1506. MAcks.unlock();
  1507. }
  1508. void EQStream::SetLastSeqSent(uint32 seq)
  1509. {
  1510. MOutboundQueue.lock();
  1511. LastSeqSent=seq;
  1512. MOutboundQueue.unlock();
  1513. }
  1514. void EQStream::SetStreamType(EQStreamType type)
  1515. {
  1516. StreamType=type;
  1517. switch (StreamType) {
  1518. case LoginStream:
  1519. app_opcode_size=1;
  1520. compressed=false;
  1521. encoded=false;
  1522. break;
  1523. case EQ2Stream:
  1524. app_opcode_size=2;
  1525. compressed=false;
  1526. encoded=false;
  1527. break;
  1528. case ChatOrMailStream:
  1529. case ChatStream:
  1530. case MailStream:
  1531. app_opcode_size=1;
  1532. compressed=false;
  1533. encoded=true;
  1534. break;
  1535. case ZoneStream:
  1536. case WorldStream:
  1537. default:
  1538. app_opcode_size=2;
  1539. compressed=true;
  1540. encoded=false;
  1541. break;
  1542. }
  1543. }
  1544. void EQStream::ProcessQueue()
  1545. {
  1546. if (OutOfOrderpackets.empty()) {
  1547. return;
  1548. }
  1549. EQProtocolPacket* qp = NULL;
  1550. while ((qp = RemoveQueue(NextInSeq)) != NULL) {
  1551. //_log(NET__DEBUG, _L "Processing Queued Packet: Seq=%d" __L, NextInSeq);
  1552. ProcessPacket(qp);
  1553. delete qp;
  1554. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1555. }
  1556. }
  1557. EQProtocolPacket* EQStream::RemoveQueue(uint16 seq)
  1558. {
  1559. map<unsigned short, EQProtocolPacket*>::iterator itr;
  1560. EQProtocolPacket* qp = NULL;
  1561. if ((itr = OutOfOrderpackets.find(seq)) != OutOfOrderpackets.end()) {
  1562. qp = itr->second;
  1563. OutOfOrderpackets.erase(itr);
  1564. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1565. }
  1566. return qp;
  1567. }
  1568. void EQStream::Decay()
  1569. {
  1570. MRate.lock();
  1571. uint32 rate=DecayRate;
  1572. MRate.unlock();
  1573. if (BytesWritten>0) {
  1574. BytesWritten-=rate;
  1575. if (BytesWritten<0)
  1576. BytesWritten=0;
  1577. }
  1578. int count = 0;
  1579. MOutboundQueue.lock();
  1580. for (auto sitr = SequencedQueue.begin(); sitr != SequencedQueue.end(); ++sitr, count++) {
  1581. if (!(*sitr)->acked && (*sitr)->sent_time > 0 && ((*sitr)->sent_time + retransmittimeout) < Timer::GetCurrentTime2()) {
  1582. (*sitr)->sent_time = 0;
  1583. LogWrite(PACKET__DEBUG, 9, "Packet", "Timeout exceeded for seq %u. Flagging packet for retransmission", SequencedBase + count);
  1584. }
  1585. }
  1586. MOutboundQueue.unlock();
  1587. }
  1588. void EQStream::AdjustRates(uint32 average_delta)
  1589. {
  1590. if (average_delta && (average_delta <= AVERAGE_DELTA_MAX)) {
  1591. MRate.lock();
  1592. AverageDelta = average_delta;
  1593. RateThreshold = RATEBASE / average_delta;
  1594. DecayRate = DECAYBASE / average_delta;
  1595. if (BytesWritten > RateThreshold)
  1596. BytesWritten = RateThreshold + DecayRate;
  1597. MRate.unlock();
  1598. }
  1599. else {
  1600. AverageDelta = AVERAGE_DELTA_MAX;
  1601. }
  1602. }