Add support for ICE/STUN/TURN in res_rtp_asterisk and chan_sip.
[asterisk/asterisk.git] / res / pjproject / pjlib / src / pjlib-test / ioq_tcp.c
1 /* $Id$ */
2 /* 
3  * Copyright (C) 2008-2011 Teluu Inc. (http://www.teluu.com)
4  * Copyright (C) 2003-2008 Benny Prijono <benny@prijono.org>
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License as published by
8  * the Free Software Foundation; either version 2 of the License, or
9  * (at your option) any later version.
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  * GNU General Public License for more details.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
19  */
20 #include "test.h"
21
22 /**
23  * \page page_pjlib_ioqueue_tcp_test Test: I/O Queue (TCP)
24  *
25  * This file provides implementation to test the
26  * functionality of the I/O queue when TCP socket is used.
27  *
28  *
29  * This file is <b>pjlib-test/ioq_tcp.c</b>
30  *
31  * \include pjlib-test/ioq_tcp.c
32  */
33
34
35 #if INCLUDE_TCP_IOQUEUE_TEST
36
37 #include <pjlib.h>
38
39 #if PJ_HAS_TCP
40
41 #define THIS_FILE           "test_tcp"
42 #define NON_EXISTANT_PORT   50123
43 #define LOOP                100
44 #define BUF_MIN_SIZE        32
45 #define BUF_MAX_SIZE        2048
46 #define SOCK_INACTIVE_MIN   (4-2)
47 #define SOCK_INACTIVE_MAX   (PJ_IOQUEUE_MAX_HANDLES - 2)
48 #define POOL_SIZE           (2*BUF_MAX_SIZE + SOCK_INACTIVE_MAX*128 + 2048)
49
50 static pj_ssize_t            callback_read_size,
51                              callback_write_size,
52                              callback_accept_status,
53                              callback_connect_status;
54 static unsigned              callback_call_count;
55 static pj_ioqueue_key_t     *callback_read_key,
56                             *callback_write_key,
57                             *callback_accept_key,
58                             *callback_connect_key;
59 static pj_ioqueue_op_key_t  *callback_read_op,
60                             *callback_write_op,
61                             *callback_accept_op;
62
63 static void on_ioqueue_read(pj_ioqueue_key_t *key, 
64                             pj_ioqueue_op_key_t *op_key,
65                             pj_ssize_t bytes_read)
66 {
67     callback_read_key = key;
68     callback_read_op = op_key;
69     callback_read_size = bytes_read;
70     callback_call_count++;
71 }
72
73 static void on_ioqueue_write(pj_ioqueue_key_t *key, 
74                              pj_ioqueue_op_key_t *op_key,
75                              pj_ssize_t bytes_written)
76 {
77     callback_write_key = key;
78     callback_write_op = op_key;
79     callback_write_size = bytes_written;
80     callback_call_count++;
81 }
82
83 static void on_ioqueue_accept(pj_ioqueue_key_t *key, 
84                               pj_ioqueue_op_key_t *op_key,
85                               pj_sock_t sock, 
86                               int status)
87 {
88     if (sock == PJ_INVALID_SOCKET) {
89
90         if (status != PJ_SUCCESS) {
91             /* Ignore. Could be blocking error */
92             app_perror(".....warning: received error in on_ioqueue_accept() callback",
93                        status);
94         } else {
95             callback_accept_status = -61;
96             PJ_LOG(3,("", "..... on_ioqueue_accept() callback was given "
97                           "invalid socket and status is %d", status));
98         }
99     } else {
100         pj_sockaddr addr;
101         int client_addr_len;
102
103         client_addr_len = sizeof(addr);
104         status = pj_sock_getsockname(sock, &addr, &client_addr_len);
105         if (status != PJ_SUCCESS) {
106             app_perror("...ERROR in pj_sock_getsockname()", status);
107         }
108
109         callback_accept_key = key;
110         callback_accept_op = op_key;
111         callback_accept_status = status;
112         callback_call_count++;
113     }
114 }
115
116 static void on_ioqueue_connect(pj_ioqueue_key_t *key, int status)
117 {
118     callback_connect_key = key;
119     callback_connect_status = status;
120     callback_call_count++;
121 }
122
123 static pj_ioqueue_callback test_cb = 
124 {
125     &on_ioqueue_read,
126     &on_ioqueue_write,
127     &on_ioqueue_accept,
128     &on_ioqueue_connect,
129 };
130
131 static int send_recv_test(pj_ioqueue_t *ioque,
132                           pj_ioqueue_key_t *skey,
133                           pj_ioqueue_key_t *ckey,
134                           void *send_buf,
135                           void *recv_buf,
136                           pj_ssize_t bufsize,
137                           pj_timestamp *t_elapsed)
138 {
139     pj_status_t status;
140     pj_ssize_t bytes;
141     pj_time_val timeout;
142     pj_timestamp t1, t2;
143     int pending_op = 0;
144     pj_ioqueue_op_key_t read_op, write_op;
145
146     // Start reading on the server side.
147     bytes = bufsize;
148     status = pj_ioqueue_recv(skey, &read_op, recv_buf, &bytes, 0);
149     if (status != PJ_SUCCESS && status != PJ_EPENDING) {
150         app_perror("...pj_ioqueue_recv error", status);
151         return -100;
152     }
153     
154     if (status == PJ_EPENDING)
155         ++pending_op;
156     else {
157         /* Does not expect to return error or immediate data. */
158         return -115;
159     }
160
161     // Randomize send buffer.
162     pj_create_random_string((char*)send_buf, bufsize);
163
164     // Starts send on the client side.
165     bytes = bufsize;
166     status = pj_ioqueue_send(ckey, &write_op, send_buf, &bytes, 0);
167     if (status != PJ_SUCCESS && bytes != PJ_EPENDING) {
168         return -120;
169     }
170     if (status == PJ_EPENDING) {
171         ++pending_op;
172     }
173
174     // Begin time.
175     pj_get_timestamp(&t1);
176
177     // Reset indicators
178     callback_read_size = callback_write_size = 0;
179     callback_read_key = callback_write_key = NULL;
180     callback_read_op = callback_write_op = NULL;
181
182     // Poll the queue until we've got completion event in the server side.
183     status = 0;
184     while (pending_op > 0) {
185         timeout.sec = 1; timeout.msec = 0;
186 #ifdef PJ_SYMBIAN
187         PJ_UNUSED_ARG(ioque);
188         status = pj_symbianos_poll(-1, 1000);
189 #else
190         status = pj_ioqueue_poll(ioque, &timeout);
191 #endif
192         if (status > 0) {
193             if (callback_read_size) {
194                 if (callback_read_size != bufsize)
195                     return -160;
196                 if (callback_read_key != skey)
197                     return -161;
198                 if (callback_read_op != &read_op)
199                     return -162;
200             }
201             if (callback_write_size) {
202                 if (callback_write_key != ckey)
203                     return -163;
204                 if (callback_write_op != &write_op)
205                     return -164;
206             }
207             pending_op -= status;
208         }
209         if (status == 0) {
210             PJ_LOG(3,("", "...error: timed out"));
211         }
212         if (status < 0) {
213             return -170;
214         }
215     }
216
217     // Pending op is zero.
218     // Subsequent poll should yield zero too.
219     timeout.sec = timeout.msec = 0;
220 #ifdef PJ_SYMBIAN
221     status = pj_symbianos_poll(-1, 1);
222 #else
223     status = pj_ioqueue_poll(ioque, &timeout);
224 #endif
225     if (status != 0)
226         return -173;
227
228     // End time.
229     pj_get_timestamp(&t2);
230     t_elapsed->u32.lo += (t2.u32.lo - t1.u32.lo);
231
232     // Compare recv buffer with send buffer.
233     if (pj_memcmp(send_buf, recv_buf, bufsize) != 0) {
234         return -180;
235     }
236
237     // Success
238     return 0;
239 }
240
241
242 /*
243  * Compliance test for success scenario.
244  */
245 static int compliance_test_0(pj_bool_t allow_concur)
246 {
247     pj_sock_t ssock=-1, csock0=-1, csock1=-1;
248     pj_sockaddr_in addr, client_addr, rmt_addr;
249     int client_addr_len;
250     pj_pool_t *pool = NULL;
251     char *send_buf, *recv_buf;
252     pj_ioqueue_t *ioque = NULL;
253     pj_ioqueue_key_t *skey=NULL, *ckey0=NULL, *ckey1=NULL;
254     pj_ioqueue_op_key_t accept_op;
255     int bufsize = BUF_MIN_SIZE;
256     pj_ssize_t status = -1;
257     int pending_op = 0;
258     pj_timestamp t_elapsed;
259     pj_str_t s;
260     pj_status_t rc;
261
262     // Create pool.
263     pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);
264
265     // Allocate buffers for send and receive.
266     send_buf = (char*)pj_pool_alloc(pool, bufsize);
267     recv_buf = (char*)pj_pool_alloc(pool, bufsize);
268
269     // Create server socket and client socket for connecting
270     rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &ssock);
271     if (rc != PJ_SUCCESS) {
272         app_perror("...error creating socket", rc);
273         status=-1; goto on_error;
274     }
275
276     rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &csock1);
277     if (rc != PJ_SUCCESS) {
278         app_perror("...error creating socket", rc);
279         status=-1; goto on_error;
280     }
281
282     // Bind server socket.
283     pj_sockaddr_in_init(&addr, 0, 0);
284     if ((rc=pj_sock_bind(ssock, &addr, sizeof(addr))) != 0 ) {
285         app_perror("...bind error", rc);
286         status=-10; goto on_error;
287     }
288
289     // Get server address.
290     client_addr_len = sizeof(addr);
291     rc = pj_sock_getsockname(ssock, &addr, &client_addr_len);
292     if (rc != PJ_SUCCESS) {
293         app_perror("...ERROR in pj_sock_getsockname()", rc);
294         status=-15; goto on_error;
295     }
296     addr.sin_addr = pj_inet_addr(pj_cstr(&s, "127.0.0.1"));
297
298     // Create I/O Queue.
299     rc = pj_ioqueue_create(pool, PJ_IOQUEUE_MAX_HANDLES, &ioque);
300     if (rc != PJ_SUCCESS) {
301         app_perror("...ERROR in pj_ioqueue_create()", rc);
302         status=-20; goto on_error;
303     }
304
305     // Concurrency
306     rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
307     if (rc != PJ_SUCCESS) {
308         app_perror("...ERROR in pj_ioqueue_set_default_concurrency()", rc);
309         status=-21; goto on_error;
310     }
311
312     // Register server socket and client socket.
313     rc = pj_ioqueue_register_sock(pool, ioque, ssock, NULL, &test_cb, &skey);
314     if (rc == PJ_SUCCESS)
315         rc = pj_ioqueue_register_sock(pool, ioque, csock1, NULL, &test_cb, 
316                                       &ckey1);
317     else
318         ckey1 = NULL;
319     if (rc != PJ_SUCCESS) {
320         app_perror("...ERROR in pj_ioqueue_register_sock()", rc);
321         status=-23; goto on_error;
322     }
323
324     // Server socket listen().
325     if (pj_sock_listen(ssock, 5)) {
326         app_perror("...ERROR in pj_sock_listen()", rc);
327         status=-25; goto on_error;
328     }
329
330     // Server socket accept()
331     client_addr_len = sizeof(pj_sockaddr_in);
332     status = pj_ioqueue_accept(skey, &accept_op, &csock0, 
333                                &client_addr, &rmt_addr, &client_addr_len);
334     if (status != PJ_EPENDING) {
335         app_perror("...ERROR in pj_ioqueue_accept()", rc);
336         status=-30; goto on_error;
337     }
338     if (status==PJ_EPENDING) {
339         ++pending_op;
340     }
341
342     // Client socket connect()
343     status = pj_ioqueue_connect(ckey1, &addr, sizeof(addr));
344     if (status!=PJ_SUCCESS && status != PJ_EPENDING) {
345         app_perror("...ERROR in pj_ioqueue_connect()", rc);
346         status=-40; goto on_error;
347     }
348     if (status==PJ_EPENDING) {
349         ++pending_op;
350     }
351
352     // Poll until connected
353     callback_read_size = callback_write_size = 0;
354     callback_accept_status = callback_connect_status = -2;
355     callback_call_count = 0;
356
357     callback_read_key = callback_write_key = 
358         callback_accept_key = callback_connect_key = NULL;
359     callback_accept_op = callback_read_op = callback_write_op = NULL;
360
361     while (pending_op) {
362         pj_time_val timeout = {1, 0};
363
364 #ifdef PJ_SYMBIAN
365         callback_call_count = 0;
366         pj_symbianos_poll(-1, 1000);
367         status = callback_call_count;
368 #else
369         status = pj_ioqueue_poll(ioque, &timeout);
370 #endif
371         if (status > 0) {
372             if (callback_accept_status != -2) {
373                 if (callback_accept_status != 0) {
374                     status=-41; goto on_error;
375                 }
376                 if (callback_accept_key != skey) {
377                     status=-42; goto on_error;
378                 }
379                 if (callback_accept_op != &accept_op) {
380                     status=-43; goto on_error;
381                 }
382                 callback_accept_status = -2;
383             }
384
385             if (callback_connect_status != -2) {
386                 if (callback_connect_status != 0) {
387                     status=-50; goto on_error;
388                 }
389                 if (callback_connect_key != ckey1) {
390                     status=-51; goto on_error;
391                 }
392                 callback_connect_status = -2;
393             }
394
395             if (status > pending_op) {
396                 PJ_LOG(3,(THIS_FILE,
397                           "...error: pj_ioqueue_poll() returned %d "
398                           "(only expecting %d)",
399                           status, pending_op));
400                 return -52;
401             }
402             pending_op -= status;
403
404             if (pending_op == 0) {
405                 status = 0;
406             }
407         }
408     }
409
410     // There's no pending operation.
411     // When we poll the ioqueue, there must not be events.
412     if (pending_op == 0) {
413         pj_time_val timeout = {1, 0};
414 #ifdef PJ_SYMBIAN
415         status = pj_symbianos_poll(-1, 1000);
416 #else
417         status = pj_ioqueue_poll(ioque, &timeout);
418 #endif
419         if (status != 0) {
420             status=-60; goto on_error;
421         }
422     }
423
424     // Check accepted socket.
425     if (csock0 == PJ_INVALID_SOCKET) {
426         status = -69;
427         app_perror("...accept() error", pj_get_os_error());
428         goto on_error;
429     }
430
431     // Register newly accepted socket.
432     rc = pj_ioqueue_register_sock(pool, ioque, csock0, NULL, 
433                                   &test_cb, &ckey0);
434     if (rc != PJ_SUCCESS) {
435         app_perror("...ERROR in pj_ioqueue_register_sock", rc);
436         status = -70;
437         goto on_error;
438     }
439
440     // Test send and receive.
441     t_elapsed.u32.lo = 0;
442     status = send_recv_test(ioque, ckey0, ckey1, send_buf, 
443                             recv_buf, bufsize, &t_elapsed);
444     if (status != 0) {
445         goto on_error;
446     }
447
448     // Success
449     status = 0;
450
451 on_error:
452     if (skey != NULL)
453         pj_ioqueue_unregister(skey);
454     else if (ssock != PJ_INVALID_SOCKET)
455         pj_sock_close(ssock);
456     
457     if (ckey1 != NULL)
458         pj_ioqueue_unregister(ckey1);
459     else if (csock1 != PJ_INVALID_SOCKET)
460         pj_sock_close(csock1);
461     
462     if (ckey0 != NULL)
463         pj_ioqueue_unregister(ckey0);
464     else if (csock0 != PJ_INVALID_SOCKET)
465         pj_sock_close(csock0);
466     
467     if (ioque != NULL)
468         pj_ioqueue_destroy(ioque);
469     pj_pool_release(pool);
470     return status;
471
472 }
473
474 /*
475  * Compliance test for failed scenario.
476  * In this case, the client connects to a non-existant service.
477  */
478 static int compliance_test_1(pj_bool_t allow_concur)
479 {
480     pj_sock_t csock1=PJ_INVALID_SOCKET;
481     pj_sockaddr_in addr;
482     pj_pool_t *pool = NULL;
483     pj_ioqueue_t *ioque = NULL;
484     pj_ioqueue_key_t *ckey1 = NULL;
485     pj_ssize_t status = -1;
486     int pending_op = 0;
487     pj_str_t s;
488     pj_status_t rc;
489
490     // Create pool.
491     pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);
492
493     // Create I/O Queue.
494     rc = pj_ioqueue_create(pool, PJ_IOQUEUE_MAX_HANDLES, &ioque);
495     if (!ioque) {
496         status=-20; goto on_error;
497     }
498
499     // Concurrency
500     rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
501     if (rc != PJ_SUCCESS) {
502         status=-21; goto on_error;
503     }
504
505     // Create client socket
506     rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &csock1);
507     if (rc != PJ_SUCCESS) {
508         app_perror("...ERROR in pj_sock_socket()", rc);
509         status=-1; goto on_error;
510     }
511
512     // Register client socket.
513     rc = pj_ioqueue_register_sock(pool, ioque, csock1, NULL, 
514                                   &test_cb, &ckey1);
515     if (rc != PJ_SUCCESS) {
516         app_perror("...ERROR in pj_ioqueue_register_sock()", rc);
517         status=-23; goto on_error;
518     }
519
520     // Initialize remote address.
521     pj_sockaddr_in_init(&addr, pj_cstr(&s, "127.0.0.1"), NON_EXISTANT_PORT);
522
523     // Client socket connect()
524     status = pj_ioqueue_connect(ckey1, &addr, sizeof(addr));
525     if (status==PJ_SUCCESS) {
526         // unexpectedly success!
527         status = -30;
528         goto on_error;
529     }
530     if (status != PJ_EPENDING) {
531         // success
532     } else {
533         ++pending_op;
534     }
535
536     callback_connect_status = -2;
537     callback_connect_key = NULL;
538
539     // Poll until we've got result
540     while (pending_op) {
541         pj_time_val timeout = {1, 0};
542
543 #ifdef PJ_SYMBIAN
544         callback_call_count = 0;
545         pj_symbianos_poll(-1, 1000);
546         status = callback_call_count;
547 #else
548         status = pj_ioqueue_poll(ioque, &timeout);
549 #endif
550         if (status > 0) {
551             if (callback_connect_key==ckey1) {
552                 if (callback_connect_status == 0) {
553                     // unexpectedly connected!
554                     status = -50;
555                     goto on_error;
556                 }
557             }
558
559             if (status > pending_op) {
560                 PJ_LOG(3,(THIS_FILE,
561                           "...error: pj_ioqueue_poll() returned %d "
562                           "(only expecting %d)",
563                           status, pending_op));
564                 return -552;
565             }
566
567             pending_op -= status;
568             if (pending_op == 0) {
569                 status = 0;
570             }
571         }
572     }
573
574     // There's no pending operation.
575     // When we poll the ioqueue, there must not be events.
576     if (pending_op == 0) {
577         pj_time_val timeout = {1, 0};
578 #ifdef PJ_SYMBIAN
579         status = pj_symbianos_poll(-1, 1000);
580 #else
581         status = pj_ioqueue_poll(ioque, &timeout);
582 #endif
583         if (status != 0) {
584             status=-60; goto on_error;
585         }
586     }
587
588     // Success
589     status = 0;
590
591 on_error:
592     if (ckey1 != NULL)
593         pj_ioqueue_unregister(ckey1);
594     else if (csock1 != PJ_INVALID_SOCKET)
595         pj_sock_close(csock1);
596     
597     if (ioque != NULL)
598         pj_ioqueue_destroy(ioque);
599     pj_pool_release(pool);
600     return status;
601 }
602
603
604 /*
605  * Repeated connect/accept on the same listener socket.
606  */
607 static int compliance_test_2(pj_bool_t allow_concur)
608 {
609 #if defined(PJ_SYMBIAN) && PJ_SYMBIAN!=0
610     enum { MAX_PAIR = 1, TEST_LOOP = 2 };
611 #else
612     enum { MAX_PAIR = 4, TEST_LOOP = 2 };
613 #endif
614
615     struct listener
616     {
617         pj_sock_t            sock;
618         pj_ioqueue_key_t    *key;
619         pj_sockaddr_in       addr;
620         int                  addr_len;
621     } listener;
622
623     struct server
624     {
625         pj_sock_t            sock;
626         pj_ioqueue_key_t    *key;
627         pj_sockaddr_in       local_addr;
628         pj_sockaddr_in       rem_addr;
629         int                  rem_addr_len;
630         pj_ioqueue_op_key_t  accept_op;
631     } server[MAX_PAIR];
632
633     struct client
634     {
635         pj_sock_t            sock;
636         pj_ioqueue_key_t    *key;
637     } client[MAX_PAIR];
638
639     pj_pool_t *pool = NULL;
640     char *send_buf, *recv_buf;
641     pj_ioqueue_t *ioque = NULL;
642     int i, bufsize = BUF_MIN_SIZE;
643     pj_ssize_t status;
644     int test_loop, pending_op = 0;
645     pj_timestamp t_elapsed;
646     pj_str_t s;
647     pj_status_t rc;
648
649     listener.sock = PJ_INVALID_SOCKET;
650     listener.key = NULL;
651     
652     for (i=0; i<MAX_PAIR; ++i) {
653         server[i].sock = PJ_INVALID_SOCKET;
654         server[i].key = NULL;
655     }
656     
657     for (i=0; i<MAX_PAIR; ++i) {
658         client[i].sock = PJ_INVALID_SOCKET;
659         client[i].key = NULL;   
660     }
661     
662     // Create pool.
663     pool = pj_pool_create(mem, NULL, POOL_SIZE, 4000, NULL);
664
665
666     // Create I/O Queue.
667     rc = pj_ioqueue_create(pool, PJ_IOQUEUE_MAX_HANDLES, &ioque);
668     if (rc != PJ_SUCCESS) {
669         app_perror("...ERROR in pj_ioqueue_create()", rc);
670         return -10;
671     }
672
673
674     // Concurrency
675     rc = pj_ioqueue_set_default_concurrency(ioque, allow_concur);
676     if (rc != PJ_SUCCESS) {
677         app_perror("...ERROR in pj_ioqueue_set_default_concurrency()", rc);
678         return -11;
679     }
680
681     // Allocate buffers for send and receive.
682     send_buf = (char*)pj_pool_alloc(pool, bufsize);
683     recv_buf = (char*)pj_pool_alloc(pool, bufsize);
684
685     // Create listener socket
686     rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &listener.sock);
687     if (rc != PJ_SUCCESS) {
688         app_perror("...error creating socket", rc);
689         status=-20; goto on_error;
690     }
691
692     // Bind listener socket.
693     pj_sockaddr_in_init(&listener.addr, 0, 0);
694     if ((rc=pj_sock_bind(listener.sock, &listener.addr, sizeof(listener.addr))) != 0 ) {
695         app_perror("...bind error", rc);
696         status=-30; goto on_error;
697     }
698
699     // Get listener address.
700     listener.addr_len = sizeof(listener.addr);
701     rc = pj_sock_getsockname(listener.sock, &listener.addr, &listener.addr_len);
702     if (rc != PJ_SUCCESS) {
703         app_perror("...ERROR in pj_sock_getsockname()", rc);
704         status=-40; goto on_error;
705     }
706     listener.addr.sin_addr = pj_inet_addr(pj_cstr(&s, "127.0.0.1"));
707
708
709     // Register listener socket.
710     rc = pj_ioqueue_register_sock(pool, ioque, listener.sock, NULL, &test_cb, 
711                                   &listener.key);
712     if (rc != PJ_SUCCESS) {
713         app_perror("...ERROR", rc);
714         status=-50; goto on_error;
715     }
716
717
718     // Listener socket listen().
719     if (pj_sock_listen(listener.sock, 5)) {
720         app_perror("...ERROR in pj_sock_listen()", rc);
721         status=-60; goto on_error;
722     }
723
724
725     for (test_loop=0; test_loop < TEST_LOOP; ++test_loop) {
726         // Client connect and server accept.
727         for (i=0; i<MAX_PAIR; ++i) {
728             rc = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &client[i].sock);
729             if (rc != PJ_SUCCESS) {
730                 app_perror("...error creating socket", rc);
731                 status=-70; goto on_error;
732             }
733
734             rc = pj_ioqueue_register_sock(pool, ioque, client[i].sock, NULL, 
735                                           &test_cb, &client[i].key);
736             if (rc != PJ_SUCCESS) {
737                 app_perror("...error ", rc);
738                 status=-80; goto on_error;
739             }
740
741             // Server socket accept()
742             pj_ioqueue_op_key_init(&server[i].accept_op, 
743                                    sizeof(server[i].accept_op));
744             server[i].rem_addr_len = sizeof(pj_sockaddr_in);
745             status = pj_ioqueue_accept(listener.key, &server[i].accept_op, 
746                                        &server[i].sock, &server[i].local_addr, 
747                                        &server[i].rem_addr, 
748                                        &server[i].rem_addr_len);
749             if (status!=PJ_SUCCESS && status != PJ_EPENDING) {
750                 app_perror("...ERROR in pj_ioqueue_accept()", rc);
751                 status=-90; goto on_error;
752             }
753             if (status==PJ_EPENDING) {
754                 ++pending_op;
755             }
756
757
758             // Client socket connect()
759             status = pj_ioqueue_connect(client[i].key, &listener.addr, 
760                                         sizeof(listener.addr));
761             if (status!=PJ_SUCCESS && status != PJ_EPENDING) {
762                 app_perror("...ERROR in pj_ioqueue_connect()", rc);
763                 status=-100; goto on_error;
764             }
765             if (status==PJ_EPENDING) {
766                 ++pending_op;
767             }
768
769             // Poll until connection of this pair established
770             while (pending_op) {
771                 pj_time_val timeout = {1, 0};
772
773 #ifdef PJ_SYMBIAN
774                 status = pj_symbianos_poll(-1, 1000);
775 #else
776                 status = pj_ioqueue_poll(ioque, &timeout);
777 #endif
778                 if (status > 0) {
779                     if (status > pending_op) {
780                         PJ_LOG(3,(THIS_FILE,
781                                   "...error: pj_ioqueue_poll() returned %d "
782                                   "(only expecting %d)",
783                                   status, pending_op));
784                         return -110;
785                     }
786                     pending_op -= status;
787
788                     if (pending_op == 0) {
789                         status = 0;
790                     }
791                 }
792             }
793         }
794
795         // There's no pending operation.
796         // When we poll the ioqueue, there must not be events.
797         if (pending_op == 0) {
798             pj_time_val timeout = {1, 0};
799 #ifdef PJ_SYMBIAN
800             status = pj_symbianos_poll(-1, 1000);
801 #else
802             status = pj_ioqueue_poll(ioque, &timeout);
803 #endif
804             if (status != 0) {
805                 status=-120; goto on_error;
806             }
807         }
808
809         for (i=0; i<MAX_PAIR; ++i) {
810             // Check server socket.
811             if (server[i].sock == PJ_INVALID_SOCKET) {
812                 status = -130;
813                 app_perror("...accept() error", pj_get_os_error());
814                 goto on_error;
815             }
816
817             // Check addresses
818             if (server[i].local_addr.sin_family != pj_AF_INET() ||
819                 server[i].local_addr.sin_addr.s_addr == 0 ||
820                 server[i].local_addr.sin_port == 0)
821             {
822                 app_perror("...ERROR address not set", rc);
823                 status = -140;
824                 goto on_error;
825             }
826
827             if (server[i].rem_addr.sin_family != pj_AF_INET() ||
828                 server[i].rem_addr.sin_addr.s_addr == 0 ||
829                 server[i].rem_addr.sin_port == 0)
830             {
831                 app_perror("...ERROR address not set", rc);
832                 status = -150;
833                 goto on_error;
834             }
835
836
837             // Register newly accepted socket.
838             rc = pj_ioqueue_register_sock(pool, ioque, server[i].sock, NULL,
839                                           &test_cb, &server[i].key);
840             if (rc != PJ_SUCCESS) {
841                 app_perror("...ERROR in pj_ioqueue_register_sock", rc);
842                 status = -160;
843                 goto on_error;
844             }
845
846             // Test send and receive.
847             t_elapsed.u32.lo = 0;
848             status = send_recv_test(ioque, server[i].key, client[i].key, 
849                                     send_buf, recv_buf, bufsize, &t_elapsed);
850             if (status != 0) {
851                 goto on_error;
852             }
853         }
854
855         // Success
856         status = 0;
857
858         for (i=0; i<MAX_PAIR; ++i) {
859             if (server[i].key != NULL) {
860                 pj_ioqueue_unregister(server[i].key);
861                 server[i].key = NULL;
862                 server[i].sock = PJ_INVALID_SOCKET;
863             } else if (server[i].sock != PJ_INVALID_SOCKET) {
864                 pj_sock_close(server[i].sock);
865                 server[i].sock = PJ_INVALID_SOCKET;
866             }
867
868             if (client[i].key != NULL) {
869                 pj_ioqueue_unregister(client[i].key);
870                 client[i].key = NULL;
871                 client[i].sock = PJ_INVALID_SOCKET;
872             } else if (client[i].sock != PJ_INVALID_SOCKET) {
873                 pj_sock_close(client[i].sock);
874                 client[i].sock = PJ_INVALID_SOCKET;
875             }
876         }
877     }
878
879     status = 0;
880
881 on_error:
882     for (i=0; i<MAX_PAIR; ++i) {
883         if (server[i].key != NULL) {
884             pj_ioqueue_unregister(server[i].key);
885             server[i].key = NULL;
886             server[i].sock = PJ_INVALID_SOCKET;
887         } else if (server[i].sock != PJ_INVALID_SOCKET) {
888             pj_sock_close(server[i].sock);
889             server[i].sock = PJ_INVALID_SOCKET;
890         }
891
892         if (client[i].key != NULL) {
893             pj_ioqueue_unregister(client[i].key);
894             client[i].key = NULL;
895             server[i].sock = PJ_INVALID_SOCKET;
896         } else if (client[i].sock != PJ_INVALID_SOCKET) {
897             pj_sock_close(client[i].sock);
898             client[i].sock = PJ_INVALID_SOCKET;
899         }
900     }
901
902     if (listener.key) {
903         pj_ioqueue_unregister(listener.key);
904         listener.key = NULL;
905     } else if (listener.sock != PJ_INVALID_SOCKET) {
906         pj_sock_close(listener.sock);
907         listener.sock = PJ_INVALID_SOCKET;
908     }
909
910     if (ioque != NULL)
911         pj_ioqueue_destroy(ioque);
912     pj_pool_release(pool);
913     return status;
914
915 }
916
917
918 static int tcp_ioqueue_test_impl(pj_bool_t allow_concur)
919 {
920     int status;
921
922     PJ_LOG(3,(THIS_FILE, "..testing with concurency=%d", allow_concur));
923
924     PJ_LOG(3, (THIS_FILE, "..%s compliance test 0 (success scenario)",
925                pj_ioqueue_name()));
926     if ((status=compliance_test_0(allow_concur)) != 0) {
927         PJ_LOG(1, (THIS_FILE, "....FAILED (status=%d)\n", status));
928         return status;
929     }
930     PJ_LOG(3, (THIS_FILE, "..%s compliance test 1 (failed scenario)",
931                pj_ioqueue_name()));
932     if ((status=compliance_test_1(allow_concur)) != 0) {
933         PJ_LOG(1, (THIS_FILE, "....FAILED (status=%d)\n", status));
934         return status;
935     }
936
937     PJ_LOG(3, (THIS_FILE, "..%s compliance test 2 (repeated accept)",
938                pj_ioqueue_name()));
939     if ((status=compliance_test_2(allow_concur)) != 0) {
940         PJ_LOG(1, (THIS_FILE, "....FAILED (status=%d)\n", status));
941         return status;
942     }
943
944     return 0;
945 }
946
947 int tcp_ioqueue_test()
948 {
949     int rc;
950
951     rc = tcp_ioqueue_test_impl(PJ_TRUE);
952     if (rc != 0)
953         return rc;
954
955     rc = tcp_ioqueue_test_impl(PJ_FALSE);
956     if (rc != 0)
957         return rc;
958
959     return 0;
960 }
961
962 #endif  /* PJ_HAS_TCP */
963
964
965 #else
966 /* To prevent warning about "translation unit is empty"
967  * when this test is disabled. 
968  */
969 int dummy_uiq_tcp;
970 #endif  /* INCLUDE_TCP_IOQUEUE_TEST */
971
972