?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Upstream tarball 10068
[amule.git] / src / libs / ec / cpp / ECSocket.cpp
blob11d8f00366909d61a0f073c28b02b48b3a2a6c21
1 //
2 // This file is part of the aMule Project.
3 //
4 // Copyright (c) 2004-2008 aMule Team ( admin@amule.org / http://www.amule.org )
5 // Copyright (c) 2004-2008 Angel Vidal ( kry@amule.org )
6 //
7 // Any parts of this program derived from the xMule, lMule or eMule project,
8 // or contributed by third-party developers are copyrighted by their
9 // respective authors.
10 //
11 // This program is free software; you can redistribute it and/or modify
12 // it under the terms of the GNU General Public License as published by
13 // the Free Software Foundation; either version 2 of the License, or
14 // (at your option) any later version.
15 //
16 // This program is distributed in the hope that it will be useful,
17 // but WITHOUT ANY WARRANTY; without even the implied warranty of
18 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 // GNU General Public License for more details.
20 //
21 // You should have received a copy of the GNU General Public License
22 // along with this program; if not, write to the Free Software
23 // Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
24 //
25
26 #include "ECSocket.h"
27
28 #include <sstream>
29 #include <iostream>
30 #include <algorithm>
31
32 using namespace std;
33
34 #include "ECPacket.h" // Needed for CECPacket
35
36 #define EC_COMPRESSION_LEVEL Z_BEST_COMPRESSION
37 #define EC_MAX_UNCOMPRESSED 1024
38
39 #ifndef __GNUC__
40 #define __attribute__(x)
41 #endif
42
43 // If your compiler gives errors on these lines, just remove them.
44 int utf8_mbtowc(wchar_t *p, const unsigned char *s, int n) __attribute__((__visibility__("internal")));
45 int utf8_wctomb(unsigned char *s, wchar_t wc, int maxlen) __attribute__((__visibility__("internal")));
46 int utf8_mb_remain(char c) __attribute__((__pure__));
47
48 /*----------=> Import from the Linux kernel <=----------*/
49 /*
50 * linux/fs/nls_base.c
51 */
52
53 /*
54 * Sample implementation from Unicode home page.
55 * http://www.stonehand.com/unicode/standard/fss-utf.html
56 */
57 struct utf8_table {
58 int cmask;
59 int cval;
60 int shift;
61 uint32_t lmask;
62 uint32_t lval;
63 };
64
65 static const struct utf8_table utf8_table[] =
66 {
67 {0x80, 0x00, 0*6, 0x7F, 0, /* 1 byte sequence */},
68 {0xE0, 0xC0, 1*6, 0x7FF, 0x80, /* 2 byte sequence */},
69 {0xF0, 0xE0, 2*6, 0xFFFF, 0x800, /* 3 byte sequence */},
70 {0xF8, 0xF0, 3*6, 0x1FFFFF, 0x10000, /* 4 byte sequence */},
71 {0xFC, 0xF8, 4*6, 0x3FFFFFF, 0x200000, /* 5 byte sequence */},
72 {0xFE, 0xFC, 5*6, 0x7FFFFFFF, 0x4000000, /* 6 byte sequence */},
73 {0, 0, 0, 0, 0, /* end of table */}
74 };
75
76 int utf8_mbtowc(uint32_t *p, const unsigned char *s, int n)
77 {
78 uint32_t l;
79 int c0, c, nc;
80 const struct utf8_table *t;
81
82 nc = 0;
83 c0 = *s;
84 l = c0;
85 for (t = utf8_table; t->cmask; t++) {
86 nc++;
87 if ((c0 & t->cmask) == t->cval) {
88 l &= t->lmask;
89 if (l < t->lval)
90 return -1;
91 *p = l;
92 return nc;
93 }
94 if (n <= nc)
95 return -1;
96 s++;
97 c = (*s ^ 0x80) & 0xFF;
98 if (c & 0xC0)
99 return -1;
100 l = (l << 6) | c;
101 }
102 return -1;
103 }
104
105 int utf8_wctomb(unsigned char *s, uint32_t wc, int maxlen)
106 {
107 uint32_t l;
108 int c, nc;
109 const struct utf8_table *t;
110
111 l = wc;
112 nc = 0;
113 for (t = utf8_table; t->cmask && maxlen; t++, maxlen--) {
114 nc++;
115 if (l <= t->lmask) {
116 c = t->shift;
117 *s = t->cval | (l >> c);
118 while (c > 0) {
119 c -= 6;
120 s++;
121 *s = 0x80 | ((l >> c) & 0x3F);
122 }
123 return nc;
124 }
125 }
126 return -1;
127 }
128 /*----------=> End of Import <=----------*/
129
130 int utf8_mb_remain(char c)
131 {
132 int i;
133 for (i = 0; i < 5; ++i) {
134 if ((c & utf8_table[i].cmask) == utf8_table[i].cval) break;
135 }
136 return i;
137 }
138
139
140 void CQueuedData::Write(const void *data, size_t len)
141 {
142 const size_t canWrite = std::min(GetRemLength(), len);
143 wxASSERT(len == canWrite);
144
145 memcpy(m_wr_ptr, data, canWrite);
146 m_wr_ptr += canWrite;
147 }
148
149
150 void CQueuedData::WriteAt(const void *data, size_t len, size_t offset)
151 {
152 wxASSERT(len + offset <= m_data.size());
153 if (offset > m_data.size()) {
154 return;
155 } else if (offset + len > m_data.size()) {
156 len = m_data.size() - offset;
157 }
158
159 memcpy(&m_data[0] + offset, data, len);
160 }
161
162
163 void CQueuedData::Read(void *data, size_t len)
164 {
165 const size_t canRead = std::min(GetUnreadDataLength(), len);
166 wxASSERT(len == canRead);
167
168 memcpy(data, m_rd_ptr, canRead);
169 m_rd_ptr += canRead;
170 }
171
172
173 void CQueuedData::WriteToSocket(CECSocket *sock)
174 {
175 wxCHECK_RET(m_rd_ptr < m_wr_ptr,
176 wxT("Reading past written data in WriteToSocket"));
177
178 sock->SocketWrite(m_rd_ptr, GetUnreadDataLength());
179 m_rd_ptr += sock->GetLastCount();
180 }
181
182
183 void CQueuedData::ReadFromSocket(CECSocket *sock, size_t len)
184 {
185 const size_t canWrite = std::min(GetRemLength(), len);
186 wxASSERT(len == canWrite);
187
188 sock->SocketRead(m_wr_ptr, canWrite);
189 m_wr_ptr += sock->GetLastCount();
190 }
191
192
193 size_t CQueuedData::ReadFromSocketAll(CECSocket *sock, size_t len)
194 {
195 size_t read_rem = std::min(GetRemLength(), len);
196 wxASSERT(read_rem == len);
197
198 // We get here when socket is truly blocking
199 do {
200 // Give socket a 10 sec chance to recv more data.
201 if ( !sock->WaitSocketRead(10, 0) ) {
202 break;
203 }
204
205 wxASSERT(m_wr_ptr + read_rem <= &m_data[0] + m_data.size());
206 sock->SocketRead(m_wr_ptr, read_rem);
207 m_wr_ptr += sock->GetLastCount();
208 read_rem -= sock->GetLastCount();
209
210 if (sock->SocketError() && !sock->WouldBlock()) {
211 break;
212 }
213 } while (read_rem);
214
215 return len - read_rem;
216 }
217
218
219 size_t CQueuedData::GetLength() const
220 {
221 return m_data.size();
222 }
223
224
225 size_t CQueuedData::GetDataLength() const
226 {
227 const size_t len = m_wr_ptr - &m_data[0];
228 wxCHECK_MSG(len <= m_data.size(), m_data.size(),
229 wxT("Write-pointer past end of buffer"));
230
231 return len;
232 }
233
234
235 size_t CQueuedData::GetRemLength() const
236 {
237 return m_data.size() - GetDataLength();
238 }
239
240
241 size_t CQueuedData::GetUnreadDataLength() const
242 {
243 wxCHECK_MSG(m_wr_ptr >= m_rd_ptr, 0,
244 wxT("Read position past write position."));
245
246 return m_wr_ptr - m_rd_ptr;
247 }
248
249
250
251 //
252 // CECSocket API - User interface functions
253 //
254
255 CECSocket::CECSocket(bool use_events)
256 :
257 m_use_events(use_events),
258 m_in_ptr(EC_SOCKET_BUFFER_SIZE),
259 m_out_ptr(EC_SOCKET_BUFFER_SIZE),
260 m_curr_rx_data(new CQueuedData(EC_SOCKET_BUFFER_SIZE)),
261 m_curr_tx_data(new CQueuedData(EC_SOCKET_BUFFER_SIZE)),
262 m_rx_flags(0),
263 m_tx_flags(0),
264 m_my_flags(0x20 | EC_FLAG_ZLIB | EC_FLAG_UTF8_NUMBERS | EC_FLAG_ACCEPTS),
265 // setup initial state: 4 flags + 4 length
266 m_bytes_needed(8),
267 m_in_header(true)
268 {
269
270 }
271
272 bool CECSocket::HaveNotificationSupport()
273 {
274 return (m_rx_flags & EC_FLAG_NOTIFY) != 0;
275 }
276
277 CECSocket::~CECSocket()
278 {
279 while (!m_output_queue.empty()) {
280 CQueuedData *data = m_output_queue.front();
281 m_output_queue.pop_front();
282 delete data;
283 }
284 }
285
286 bool CECSocket::ConnectSocket(uint32_t ip, uint16_t port)
287 {
288 bool res;
289 #if wxCHECK_VERSION(2, 8, 8)
290 res = InternalConnect(ip, port, !m_use_events);
291 #else
292 res = InternalConnect(ip, port, false);
293 if ( !m_use_events ) {
294 res = WaitSocketConnect(10, 0) && InternalIsConnected();
295 if ( res ) {
296 OnConnect();
297 } else {
298 OnLost();
299 }
300 }
301 #endif
302 return !SocketError() && res;
303 }
304
305 void CECSocket::SendPacket(const CECPacket *packet)
306 {
307 uint32 len = WritePacket(packet);
308 packet->DebugPrint(false, len);
309 OnOutput();
310 }
311
312 const CECPacket *CECSocket::SendRecvPacket(const CECPacket *packet)
313 {
314 SendPacket(packet);
315 m_curr_rx_data->ReadFromSocketAll(this, 2 * sizeof(uint32_t));
316 if (SocketError() && !WouldBlock()) {
317 OnError();
318 return 0;
319 }
320
321 m_curr_rx_data->Read(&m_rx_flags, sizeof(m_rx_flags));
322 m_rx_flags = ENDIAN_NTOHL(m_rx_flags);
323 m_curr_rx_data->Read(&m_curr_packet_len, sizeof(m_curr_packet_len));
324 m_curr_packet_len = ENDIAN_NTOHL(m_curr_packet_len);
325
326 if ( m_curr_rx_data->GetLength() < (m_curr_packet_len+2*sizeof(uint32_t)) ) {
327 m_curr_rx_data.reset(new CQueuedData(m_curr_packet_len));
328 }
329 m_curr_rx_data->ReadFromSocketAll(this, m_curr_packet_len);
330 if (SocketError() && !WouldBlock()) {
331 OnError();
332 return 0;
333 }
334 const CECPacket *reply = ReadPacket();
335 m_curr_rx_data->Rewind();
336 return reply;
337 }
338
339 std::string CECSocket::GetLastErrorMsg()
340 {
341 int code = InternalGetLastError();
342 switch(code) {
343 case EC_ERROR_NOERROR:
344 return "No error happened";
345 case EC_ERROR_INVOP:
346 return "Invalid operation";
347 case EC_ERROR_IOERR:
348 return "Input/Output error";
349 case EC_ERROR_INVADDR:
350 return "Invalid address passed to wxSocket";
351 case EC_ERROR_INVSOCK:
352 return "Invalid socket (uninitialized)";
353 case EC_ERROR_NOHOST:
354 return "No corresponding host";
355 case EC_ERROR_INVPORT:
356 return "Invalid port";
357 case EC_ERROR_WOULDBLOCK:
358 return "The socket is non-blocking and the operation would block";
359 case EC_ERROR_TIMEDOUT:
360 return "The timeout for this operation expired";
361 case EC_ERROR_MEMERR:
362 return "Memory exhausted";
363 }
364 ostringstream error_string;
365 error_string << "Error code " << code << " unknown.";
366 return error_string.str();
367 }
368
369 void CECSocket::OnError()
370 {
371 #ifdef __DEBUG__
372 cout << GetLastErrorMsg() << endl;
373 #endif
374 }
375
376 void CECSocket::OnLost()
377 {
378 }
379
380 //
381 // Event handlers
382 //
383 void CECSocket::OnConnect()
384 {
385 }
386
387 void CECSocket::OnInput()
388 {
389 size_t bytes_rx = 0;
390 do {
391 if (m_curr_rx_data.get()) {
392 m_curr_rx_data->ReadFromSocket(this, m_bytes_needed);
393 } else {
394 return;
395 }
396 if (SocketError() && !WouldBlock()) {
397 OnError();
398 // socket already disconnected in this point
399 m_curr_rx_data.reset(0);
400 return;
401 }
402 bytes_rx = GetLastCount();
403 m_bytes_needed -= bytes_rx;
404 } while (m_bytes_needed && bytes_rx);
405
406 if (!m_bytes_needed) {
407 if (m_in_header) {
408 m_in_header = false;
409 m_curr_rx_data->Read(&m_rx_flags, sizeof(m_rx_flags));
410 m_rx_flags = ENDIAN_NTOHL(m_rx_flags);
411 if (m_rx_flags & EC_FLAG_ACCEPTS) {
412 // Client sends its capabilities, update the internal mask.
413 m_curr_rx_data->Read(&m_my_flags, sizeof(m_my_flags));
414 m_my_flags = ENDIAN_NTOHL(m_my_flags);
415 //printf("Reading accepts mask: %x\n", m_my_flags);
416 wxASSERT(m_my_flags & 0x20);
417 // There has to be 4 more bytes. THERE HAS TO BE, DAMN IT.
418 m_curr_rx_data->ReadFromSocketAll(this, sizeof(m_curr_packet_len));
419 }
420 m_curr_rx_data->Read(&m_curr_packet_len, sizeof(m_curr_packet_len));
421 m_curr_packet_len = ENDIAN_NTOHL(m_curr_packet_len);
422 m_bytes_needed = m_curr_packet_len;
423 // packet bigger that 16Mb looks more like broken request
424 if (m_bytes_needed > 16*1024*1024) {
425 CloseSocket();
426 return;
427 }
428 size_t needed_size = m_bytes_needed + ((m_rx_flags & EC_FLAG_ACCEPTS) ? 12 : 8);
429 if (!m_curr_rx_data.get() ||
430 m_curr_rx_data->GetLength() < needed_size) {
431 m_curr_rx_data.reset(new CQueuedData(needed_size));
432 }
433 //#warning Kry TODO: Read packet?
434 } else {
435 //m_curr_rx_data->DumpMem();
436 std::auto_ptr<const CECPacket> packet(ReadPacket());
437 m_curr_rx_data->Rewind();
438 if (packet.get()) {
439 std::auto_ptr<const CECPacket> reply(OnPacketReceived(packet.get(), m_curr_packet_len));
440 if (reply.get()) {
441 SendPacket(reply.get());
442 }
443 }
444 m_bytes_needed = 8;
445 m_in_header = true;
446 }
447 }
448 }
449
450 void CECSocket::OnOutput()
451 {
452 while (!m_output_queue.empty()) {
453 CQueuedData* data = m_output_queue.front();
454 data->WriteToSocket(this);
455 if (!data->GetUnreadDataLength()) {
456 m_output_queue.pop_front();
457 delete data;
458 }
459 if (SocketError()) {
460 if (!WouldBlock()) {
461 // real error, abort
462 OnError();
463 return;
464 }
465 // Now it's just a blocked socket.
466 if ( m_use_events ) {
467 // Event driven logic: return, OnOutput() will be called again later
468 return;
469 }
470 // Syncronous call: wait (for max 10 secs)
471 if ( !WaitSocketWrite(10, 0) ) {
472 // Still not through ?
473 if (WouldBlock()) {
474 // WouldBlock() is only EAGAIN or EWOULD_BLOCK,
475 // and those shouldn't create an infinite wait.
476 // So give it another chance.
477 continue;
478 } else {
479 OnError();
480 break;
481 }
482 }
483 }
484 }
485 //
486 // All outstanding data sent to socket
487 // (used for push clients)
488 //
489 WriteDoneAndQueueEmpty();
490 }
491
492 bool CECSocket::DataPending()
493 {
494 return !m_output_queue.empty();
495 }
496
497 //
498 // Socket I/O
499 //
500
501 size_t CECSocket::ReadBufferFromSocket(void *buffer, size_t required_len)
502 {
503 wxASSERT(required_len);
504
505 if (m_curr_rx_data->GetUnreadDataLength() < required_len) {
506 // need more data that we have. Looks like nothing will help here
507 return 0;
508 }
509 m_curr_rx_data->Read(buffer, required_len);
510 return required_len;
511 }
512
513 void CECSocket::WriteBufferToSocket(const void *buffer, size_t len)
514 {
515 unsigned char *wr_ptr = (unsigned char *)buffer;
516 while ( len ) {
517 size_t curr_free = m_curr_tx_data->GetRemLength();
518 if ( len > curr_free ) {
519
520 m_curr_tx_data->Write(wr_ptr, curr_free);
521 len -= curr_free;
522 wr_ptr += curr_free;
523 m_output_queue.push_back(m_curr_tx_data.release());
524 m_curr_tx_data.reset(new CQueuedData(EC_SOCKET_BUFFER_SIZE));
525 } else {
526 m_curr_tx_data->Write(wr_ptr, len);
527 break;
528 }
529 }
530 }
531
532
533 //
534 // ZLib "error handler"
535 //
536
537 void ShowZError(int zerror, z_streamp strm)
538 {
539 const char *p = NULL;
540
541 switch (zerror) {
542 case Z_STREAM_END: p = "Z_STREAM_END"; break;
543 case Z_NEED_DICT: p = "Z_NEED_DICT"; break;
544 case Z_ERRNO: p = "Z_ERRNO"; break;
545 case Z_STREAM_ERROR: p = "Z_STREAM_ERROR"; break;
546 case Z_DATA_ERROR: p = "Z_DATA_ERROR"; break;
547 case Z_MEM_ERROR: p = "Z_MEM_ERROR"; break;
548 case Z_BUF_ERROR: p = "Z_BUF_ERROR"; break;
549 case Z_VERSION_ERROR: p = "Z_VERSION_ERROR"; break;
550 }
551 printf("ZLib operation returned %s\n", p);
552 printf("ZLib error message: %s\n", strm->msg);
553 printf("zstream state:\n\tnext_in=%p\n\tavail_in=%u\n\ttotal_in=%lu\n\tnext_out=%p\n\tavail_out=%u\n\ttotal_out=%lu\n",
554 strm->next_in, strm->avail_in, strm->total_in, strm->next_out, strm->avail_out, strm->total_out);
555 }
556
557
558 bool CECSocket::ReadNumber(void *buffer, size_t len)
559 {
560 if (m_rx_flags & EC_FLAG_UTF8_NUMBERS) {
561 unsigned char mb[6];
562 uint32_t wc;
563 if (!ReadBuffer(mb, 1)) return false;
564 int remains = utf8_mb_remain(mb[0]);
565 if (remains) if (!ReadBuffer(&(mb[1]), remains)) return false;
566 if (utf8_mbtowc(&wc, mb, 6) == -1) return false; // Invalid UTF-8 code sequence
567 switch (len) {
568 case 1: PokeUInt8( buffer, wc ); break;
569 case 2: RawPokeUInt16( buffer, wc ); break;
570 case 4: RawPokeUInt32( buffer, wc ); break;
571 }
572 } else {
573 if ( !ReadBuffer(buffer, len) ) {
574 return false;
575 }
576 switch (len) {
577 case 2:
578 RawPokeUInt16( buffer, ENDIAN_NTOHS( RawPeekUInt16( buffer ) ) );
579 break;
580 case 4:
581 RawPokeUInt32( buffer, ENDIAN_NTOHL( RawPeekUInt32( buffer ) ) );
582 break;
583 }
584 }
585 return true;
586 }
587
588 bool CECSocket::WriteNumber(const void *buffer, size_t len)
589 {
590 if (m_tx_flags & EC_FLAG_UTF8_NUMBERS) {
591 unsigned char mb[6];
592 uint32_t wc = 0;
593 int mb_len;
594 switch (len) {
595 case 1: wc = PeekUInt8( buffer ); break;
596 case 2: wc = RawPeekUInt16( buffer ); break;
597 case 4: wc = RawPeekUInt32( buffer ); break;
598 default: return false;
599 }
600 if ((mb_len = utf8_wctomb(mb, wc, 6)) == -1) return false; // Something is terribly wrong...
601 return WriteBuffer(mb, mb_len);
602 } else {
603 char tmp[8];
604
605 switch (len) {
606 case 1: PokeUInt8( tmp, PeekUInt8( buffer ) ); break;
607 case 2: RawPokeUInt16( tmp, ENDIAN_NTOHS( RawPeekUInt16( buffer ) ) ); break;
608 case 4: RawPokeUInt32( tmp, ENDIAN_NTOHL( RawPeekUInt32( buffer ) ) ); break;
609 }
610 return WriteBuffer(tmp, len);
611 }
612 }
613
614 bool CECSocket::ReadBuffer(void *buffer, size_t len)
615 {
616 if (m_rx_flags & EC_FLAG_ZLIB) {
617 if ( !m_z.avail_in ) {
618 // no reason for this situation: all packet should be
619 // buffered by now
620 return false;
621 }
622 m_z.avail_out = (uInt)len;
623 m_z.next_out = (Bytef*)buffer;
624 int zerror = inflate(&m_z, Z_SYNC_FLUSH);
625 if ((zerror != Z_OK) && (zerror != Z_STREAM_END)) {
626 ShowZError(zerror, &m_z);
627 return false;
628 }
629 return true;
630 } else {
631 // using uncompressed buffered i/o
632 return ReadBufferFromSocket(buffer, len) == len;
633 }
634 }
635
636 bool CECSocket::WriteBuffer(const void *buffer, size_t len)
637 {
638 if (m_tx_flags & EC_FLAG_ZLIB) {
639
640 unsigned char *rd_ptr = (unsigned char *)buffer;
641 do {
642 unsigned int remain_in = EC_SOCKET_BUFFER_SIZE - m_z.avail_in;
643 if ( remain_in >= len ) {
644 memcpy(m_z.next_in+m_z.avail_in, rd_ptr, len);
645 m_z.avail_in += (uInt)len;
646 len = 0;
647 } else {
648 memcpy(m_z.next_in+m_z.avail_in, rd_ptr, remain_in);
649 m_z.avail_in += remain_in;
650 len -= remain_in;
651 rd_ptr += remain_in;
652 // buffer is full, calling zlib
653 do {
654 m_z.next_out = &m_out_ptr[0];
655 m_z.avail_out = EC_SOCKET_BUFFER_SIZE;
656 int zerror = deflate(&m_z, Z_NO_FLUSH);
657 if ( zerror != Z_OK ) {
658 ShowZError(zerror, &m_z);
659 return false;
660 }
661 WriteBufferToSocket(&m_out_ptr[0],
662 EC_SOCKET_BUFFER_SIZE - m_z.avail_out);
663 } while ( m_z.avail_out == 0 );
664 // all input should be used by now
665 wxASSERT(m_z.avail_in == 0);
666 m_z.next_in = &m_in_ptr[0];
667 }
668 } while ( len );
669 return true;
670 } else {
671 // using uncompressed buffered i/o
672 WriteBufferToSocket(buffer, len);
673 return true;
674 }
675 }
676
677 bool CECSocket::FlushBuffers()
678 {
679 if (m_tx_flags & EC_FLAG_ZLIB) {
680 do {
681 m_z.next_out = &m_out_ptr[0];
682 m_z.avail_out = EC_SOCKET_BUFFER_SIZE;
683 int zerror = deflate(&m_z, Z_FINISH);
684 if ( zerror == Z_STREAM_ERROR ) {
685 ShowZError(zerror, &m_z);
686 return false;
687 }
688 WriteBufferToSocket(&m_out_ptr[0],
689 EC_SOCKET_BUFFER_SIZE - m_z.avail_out);
690 } while ( m_z.avail_out == 0 );
691 }
692 if ( m_curr_tx_data->GetDataLength() ) {
693 m_output_queue.push_back(m_curr_tx_data.release());
694 m_curr_tx_data.reset(new CQueuedData(EC_SOCKET_BUFFER_SIZE));
695 }
696 return true;
697 }
698
699 //
700 // Packet I/O
701 //
702
703 uint32 CECSocket::WritePacket(const CECPacket *packet)
704 {
705 if (SocketError() && !WouldBlock()) {
706 OnError();
707 return 0;
708 }
709 // Check if output queue is empty. If not, memorize the current end.
710 std::list<CQueuedData*>::iterator outputStart = m_output_queue.begin();
711 uint32 outputQueueSize = m_output_queue.size();
712 for (uint32 i = 1; i < outputQueueSize; i++) {
713 outputStart++;
714 }
715
716 uint32_t flags = 0x20;
717
718 if ( packet->GetPacketLength() > EC_MAX_UNCOMPRESSED ) {
719 flags |= EC_FLAG_ZLIB;
720 } else {
721 flags |= EC_FLAG_UTF8_NUMBERS;
722 }
723
724 flags &= m_my_flags;
725 m_tx_flags = flags;
726
727 if (flags & EC_FLAG_ZLIB) {
728 m_z.zalloc = Z_NULL;
729 m_z.zfree = Z_NULL;
730 m_z.opaque = Z_NULL;
731 m_z.avail_in = 0;
732 m_z.next_in = &m_in_ptr[0];
733 int zerror = deflateInit(&m_z, EC_COMPRESSION_LEVEL);
734 if (zerror != Z_OK) {
735 // don't use zlib if init failed
736 flags &= ~EC_FLAG_ZLIB;
737 ShowZError(zerror, &m_z);
738 }
739 }
740
741 uint32_t tmp_flags = ENDIAN_HTONL(flags/* | EC_FLAG_ACCEPTS*/);
742 WriteBufferToSocket(&tmp_flags, sizeof(uint32));
743
744 // preallocate 4 bytes in buffer for packet length
745 uint32_t packet_len = 0;
746 WriteBufferToSocket(&packet_len, sizeof(uint32));
747
748 packet->WritePacket(*this);
749
750 // Finalize zlib compression and move current data to outout queue
751 FlushBuffers();
752
753 // find the beginning of our data in the output queue
754 if (outputQueueSize) {
755 outputStart++;
756 } else {
757 outputStart = m_output_queue.begin();
758 }
759 // now calculate actual size of data
760 for(std::list<CQueuedData*>::iterator it = outputStart; it != m_output_queue.end(); it++) {
761 packet_len += (uint32_t)(*it)->GetDataLength();
762 }
763 // 4 flags and 4 length are not counted
764 packet_len -= 8;
765 // now write actual length at offset 4
766 uint32 packet_len_E = ENDIAN_HTONL(packet_len);
767 (*outputStart)->WriteAt(&packet_len_E, 4, 4);
768
769 if (flags & EC_FLAG_ZLIB) {
770 int zerror = deflateEnd(&m_z);
771 if ( zerror != Z_OK ) {
772 ShowZError(zerror, &m_z);
773 }
774 }
775 return packet_len;
776 }
777
778
779 const CECPacket *CECSocket::ReadPacket()
780 {
781 CECPacket *packet = 0;
782
783 uint32_t flags = m_rx_flags;
784
785 if ( ((flags & 0x60) != 0x20) || (flags & EC_FLAG_UNKNOWN_MASK) ) {
786 // Protocol error - other end might use an older protocol
787 cout << "ReadPacket: packet have invalid flags " << flags << endl;
788 CloseSocket();
789 return 0;
790 }
791
792 if (flags & EC_FLAG_ZLIB) {
793
794 m_z.zalloc = Z_NULL;
795 m_z.zfree = Z_NULL;
796 m_z.opaque = Z_NULL;
797 m_z.avail_in = 0;
798 m_z.next_in = 0;
799
800 int zerror = inflateInit(&m_z);
801 if (zerror != Z_OK) {
802 ShowZError(zerror, &m_z);
803 cout << "ReadPacket: failed zlib init" << endl;
804 CloseSocket();
805 return 0;
806 }
807 }
808
809 m_curr_rx_data->ToZlib(m_z);
810 packet = new CECPacket();
811
812 if (!packet->ReadFromSocket(*this)) {
813 cout << "ReadPacket: error in packet read" << endl;
814 delete packet;
815 packet = NULL;
816 CloseSocket();
817 }
818
819 if (flags & EC_FLAG_ZLIB) {
820 int zerror = inflateEnd(&m_z);
821 if ( zerror != Z_OK ) {
822 ShowZError(zerror, &m_z);
823 cout << "ReadPacket: failed zlib free" << endl;
824 CloseSocket();
825 }
826 }
827
828 return packet;
829 }
830
831 const CECPacket *CECSocket::OnPacketReceived(const CECPacket *, uint32)
832 {
833 return 0;
834 }
835 // File_checked_for_headers
aMule P2P client