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