Add support for ICE/STUN/TURN in res_rtp_asterisk and chan_sip.
[asterisk/asterisk.git] / res / pjproject / pjlib / src / pj / ioqueue_symbian.cpp
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 <pj/ioqueue.h>
21 #include <pj/assert.h>
22 #include <pj/errno.h>
23 #include <pj/list.h>
24 #include <pj/lock.h>
25 #include <pj/pool.h>
26 #include <pj/string.h>
27
28 #include "os_symbian.h"
29
30 class CIoqueueCallback;
31
32 /*
33  * IO Queue structure.
34  */
35 struct pj_ioqueue_t
36 {
37     int              eventCount;
38 };
39
40
41 /////////////////////////////////////////////////////////////////////////////
42 // Class to encapsulate asynchronous socket operation.
43 //
44 class CIoqueueCallback : public CActive
45 {
46 public:
47     static CIoqueueCallback* NewL(pj_ioqueue_t *ioqueue,
48                                   pj_ioqueue_key_t *key, 
49                                   pj_sock_t sock, 
50                                   const pj_ioqueue_callback *cb, 
51                                   void *user_data);
52
53     //
54     // Start asynchronous recv() operation
55     //
56     pj_status_t StartRead(pj_ioqueue_op_key_t *op_key, 
57                           void *buf, pj_ssize_t *size, unsigned flags,
58                           pj_sockaddr_t *addr, int *addrlen);
59
60     //
61     // Start asynchronous accept() operation.
62     //
63     pj_status_t StartAccept(pj_ioqueue_op_key_t *op_key,
64                             pj_sock_t *new_sock,
65                             pj_sockaddr_t *local,
66                             pj_sockaddr_t *remote,
67                             int *addrlen );
68
69     //
70     // Completion callback.
71     //
72     void RunL();
73
74     //
75     // CActive's DoCancel()
76     //
77     void DoCancel();
78
79     //
80     // Cancel operation and call callback.
81     //
82     void CancelOperation(pj_ioqueue_op_key_t *op_key, 
83                          pj_ssize_t bytes_status);
84
85     //
86     // Accessors
87     //
88     void* get_user_data() const
89     {
90         return user_data_;
91     }
92     void set_user_data(void *user_data)
93     {
94         user_data_ = user_data;
95     }
96     pj_ioqueue_op_key_t *get_op_key() const
97     {
98         return pending_data_.common_.op_key_;
99     }
100     CPjSocket* get_pj_socket()
101     {
102         return sock_;
103     }
104
105 private:
106     // Type of pending operation.
107     enum Type {
108         TYPE_NONE,
109         TYPE_READ,
110         TYPE_ACCEPT,
111     };
112
113     // Static data.
114     pj_ioqueue_t                *ioqueue_;
115     pj_ioqueue_key_t            *key_;
116     CPjSocket                   *sock_;
117     pj_ioqueue_callback          cb_;
118     void                        *user_data_;
119
120     // Symbian data.
121     TPtr8                        aBufferPtr_;
122     TInetAddr                    aAddress_;
123
124     // Application data.
125     Type                         type_;
126
127     union Pending_Data
128     {
129         struct Common
130         {
131             pj_ioqueue_op_key_t *op_key_;
132         } common_;
133
134         struct Pending_Read
135         {
136             pj_ioqueue_op_key_t *op_key_;
137             pj_sockaddr_t       *addr_;
138             int                 *addrlen_;
139         } read_;
140
141         struct Pending_Accept
142         {
143             pj_ioqueue_op_key_t *op_key_;
144             pj_sock_t           *new_sock_;
145             pj_sockaddr_t       *local_;
146             pj_sockaddr_t       *remote_;
147             int                 *addrlen_;
148         } accept_;
149     };
150
151     union Pending_Data           pending_data_;
152     RSocket                     blank_sock_;
153
154     CIoqueueCallback(pj_ioqueue_t *ioqueue,
155                      pj_ioqueue_key_t *key, pj_sock_t sock, 
156                      const pj_ioqueue_callback *cb, void *user_data)
157     : CActive(CActive::EPriorityStandard),
158           ioqueue_(ioqueue), key_(key), sock_((CPjSocket*)sock), 
159           user_data_(user_data), aBufferPtr_(NULL, 0), type_(TYPE_NONE)
160     {
161         pj_memcpy(&cb_, cb, sizeof(*cb));
162     }
163
164
165     void ConstructL()
166     {
167         CActiveScheduler::Add(this);
168     }
169     
170     void HandleReadCompletion();
171     CPjSocket *HandleAcceptCompletion();
172 };
173
174
175 CIoqueueCallback* CIoqueueCallback::NewL(pj_ioqueue_t *ioqueue,
176                                          pj_ioqueue_key_t *key, 
177                                          pj_sock_t sock, 
178                                          const pj_ioqueue_callback *cb, 
179                                          void *user_data)
180 {
181     CIoqueueCallback *self = new CIoqueueCallback(ioqueue, key, sock, 
182                                                   cb, user_data);
183     CleanupStack::PushL(self);
184     self->ConstructL();
185     CleanupStack::Pop(self);
186
187     return self;
188 }
189
190
191 //
192 // Start asynchronous recv() operation
193 //
194 pj_status_t CIoqueueCallback::StartRead(pj_ioqueue_op_key_t *op_key, 
195                                         void *buf, pj_ssize_t *size, 
196                                         unsigned flags,
197                                         pj_sockaddr_t *addr, int *addrlen)
198 {
199     PJ_ASSERT_RETURN(IsActive()==false, PJ_EBUSY);
200     PJ_ASSERT_RETURN(pending_data_.common_.op_key_==NULL, PJ_EBUSY);
201
202     flags &= ~PJ_IOQUEUE_ALWAYS_ASYNC;
203
204     pending_data_.read_.op_key_ = op_key;
205     pending_data_.read_.addr_ = addr;
206     pending_data_.read_.addrlen_ = addrlen;
207
208     aBufferPtr_.Set((TUint8*)buf, 0, (TInt)*size);
209
210     type_ = TYPE_READ;
211     if (addr && addrlen) {
212         sock_->Socket().RecvFrom(aBufferPtr_, aAddress_, flags, iStatus);
213     } else {
214         aAddress_.SetAddress(0);
215         aAddress_.SetPort(0);
216
217         if (sock_->IsDatagram()) {
218             sock_->Socket().Recv(aBufferPtr_, flags, iStatus);
219         } else {
220             // Using static like this is not pretty, but we don't need to use
221             // the value anyway, hence doing it like this is probably most
222             // optimal.
223             static TSockXfrLength len;
224             sock_->Socket().RecvOneOrMore(aBufferPtr_, flags, iStatus, len);
225         }
226     }
227
228     SetActive();
229     return PJ_EPENDING;
230 }
231
232
233 //
234 // Start asynchronous accept() operation.
235 //
236 pj_status_t CIoqueueCallback::StartAccept(pj_ioqueue_op_key_t *op_key,
237                                           pj_sock_t *new_sock,
238                                           pj_sockaddr_t *local,
239                                           pj_sockaddr_t *remote,
240                                           int *addrlen )
241 {
242     PJ_ASSERT_RETURN(IsActive()==false, PJ_EBUSY);
243     PJ_ASSERT_RETURN(pending_data_.common_.op_key_==NULL, PJ_EBUSY);
244
245     // addrlen must be specified if local or remote is specified
246     PJ_ASSERT_RETURN((!local && !remote) ||
247                      (addrlen && *addrlen), PJ_EINVAL);
248     
249     pending_data_.accept_.op_key_ = op_key;
250     pending_data_.accept_.new_sock_ = new_sock;
251     pending_data_.accept_.local_ = local;
252     pending_data_.accept_.remote_ = remote;
253     pending_data_.accept_.addrlen_ = addrlen;
254
255     // Create blank socket
256     blank_sock_.Open(PjSymbianOS::Instance()->SocketServ());
257
258     type_ = TYPE_ACCEPT;
259     sock_->Socket().Accept(blank_sock_, iStatus);
260
261     SetActive();
262     return PJ_EPENDING;
263 }
264
265
266 //
267 // Handle asynchronous RecvFrom() completion
268 //
269 void CIoqueueCallback::HandleReadCompletion() 
270 {
271     if (pending_data_.read_.addr_ && pending_data_.read_.addrlen_) {
272         PjSymbianOS::Addr2pj(aAddress_, 
273                              *(pj_sockaddr*)pending_data_.read_.addr_,
274                              pending_data_.read_.addrlen_);
275         pending_data_.read_.addr_ = NULL;
276         pending_data_.read_.addrlen_ = NULL;
277     }
278         
279     pending_data_.read_.op_key_ = NULL;
280 }
281
282
283 //
284 // Handle asynchronous Accept() completion.
285 //
286 CPjSocket *CIoqueueCallback::HandleAcceptCompletion() 
287 {
288         CPjSocket *pjNewSock = new CPjSocket(get_pj_socket()->GetAf(), 
289                                              get_pj_socket()->GetSockType(),
290                                              blank_sock_);
291         int addrlen = 0;
292         
293         if (pending_data_.accept_.new_sock_) {
294             *pending_data_.accept_.new_sock_ = (pj_sock_t)pjNewSock;
295             pending_data_.accept_.new_sock_ = NULL;
296         }
297
298         if (pending_data_.accept_.local_) {
299             TInetAddr aAddr;
300             pj_sockaddr *ptr_sockaddr;
301             
302             blank_sock_.LocalName(aAddr);
303             ptr_sockaddr = (pj_sockaddr*)pending_data_.accept_.local_;
304             addrlen = *pending_data_.accept_.addrlen_;
305             PjSymbianOS::Addr2pj(aAddr, *ptr_sockaddr, &addrlen);
306             pending_data_.accept_.local_ = NULL;
307         }
308
309         if (pending_data_.accept_.remote_) {
310             TInetAddr aAddr;
311             pj_sockaddr *ptr_sockaddr;
312
313             blank_sock_.RemoteName(aAddr);
314             ptr_sockaddr = (pj_sockaddr*)pending_data_.accept_.remote_;
315             addrlen = *pending_data_.accept_.addrlen_;
316             PjSymbianOS::Addr2pj(aAddr, *ptr_sockaddr, &addrlen);
317             pending_data_.accept_.remote_ = NULL;
318         }
319
320         if (pending_data_.accept_.addrlen_) {
321             if (addrlen == 0) {
322                 if (pjNewSock->GetAf() == PJ_AF_INET)
323                     addrlen = sizeof(pj_sockaddr_in);
324                 else if (pjNewSock->GetAf() == PJ_AF_INET6)
325                     addrlen = sizeof(pj_sockaddr_in6);
326                 else {
327                     pj_assert(!"Unsupported address family");
328                 }
329             }
330             *pending_data_.accept_.addrlen_ = addrlen;
331             pending_data_.accept_.addrlen_ = NULL;
332         }
333         
334         return pjNewSock;
335 }
336
337
338 //
339 // Completion callback.
340 //
341 void CIoqueueCallback::RunL()
342 {
343     pj_ioqueue_t *ioq = ioqueue_;
344     Type cur_type = type_;
345
346     type_ = TYPE_NONE;
347
348     if (cur_type == TYPE_READ) {
349         //
350         // Completion of asynchronous RecvFrom()
351         //
352
353         /* Clear op_key (save it to temp variable first!) */
354         pj_ioqueue_op_key_t     *op_key = pending_data_.read_.op_key_;
355         pending_data_.read_.op_key_ = NULL;
356
357         // Handle failure condition
358         if (iStatus != KErrNone) {
359             if (cb_.on_read_complete) {
360                 cb_.on_read_complete( key_, op_key, 
361                                       -PJ_RETURN_OS_ERROR(iStatus.Int()));
362             }
363             return;
364         }
365
366         HandleReadCompletion();
367
368         /* Call callback */
369         if (cb_.on_read_complete) {
370             cb_.on_read_complete(key_, op_key, aBufferPtr_.Length());
371         }
372
373     } else if (cur_type == TYPE_ACCEPT) {
374         //
375         // Completion of asynchronous Accept()
376         //
377         
378         /* Clear op_key (save it to temp variable first!) */
379         pj_ioqueue_op_key_t     *op_key = pending_data_.read_.op_key_;
380         pending_data_.read_.op_key_ = NULL;
381
382         // Handle failure condition
383         if (iStatus != KErrNone) {
384             if (pending_data_.accept_.new_sock_)
385                 *pending_data_.accept_.new_sock_ = PJ_INVALID_SOCKET;
386             
387             if (cb_.on_accept_complete) {
388                 cb_.on_accept_complete( key_, op_key, PJ_INVALID_SOCKET,
389                                         -PJ_RETURN_OS_ERROR(iStatus.Int()));
390             }
391             return;
392         }
393
394         CPjSocket *pjNewSock = HandleAcceptCompletion();
395         
396         // Call callback.
397         if (cb_.on_accept_complete) {
398             cb_.on_accept_complete( key_, op_key, (pj_sock_t)pjNewSock, 
399                                     PJ_SUCCESS);
400         }
401     }
402
403     ioq->eventCount++;
404 }
405
406 //
407 // CActive's DoCancel()
408 //
409 void CIoqueueCallback::DoCancel()
410 {
411     if (type_ == TYPE_READ)
412         sock_->Socket().CancelRecv();
413     else if (type_ == TYPE_ACCEPT)
414         sock_->Socket().CancelAccept();
415
416     type_ = TYPE_NONE;
417     pending_data_.common_.op_key_ = NULL;
418 }
419
420 //
421 // Cancel operation and call callback.
422 //
423 void CIoqueueCallback::CancelOperation(pj_ioqueue_op_key_t *op_key, 
424                                        pj_ssize_t bytes_status)
425 {
426     Type cur_type = type_;
427
428     pj_assert(op_key == pending_data_.common_.op_key_);
429
430     Cancel();
431
432     if (cur_type == TYPE_READ) {
433         if (cb_.on_read_complete)
434             cb_.on_read_complete(key_, op_key, bytes_status);
435     } else if (cur_type == TYPE_ACCEPT)
436         ;    
437 }
438
439
440 /////////////////////////////////////////////////////////////////////////////
441 /*
442  * IO Queue key structure.
443  */
444 struct pj_ioqueue_key_t
445 {
446     CIoqueueCallback    *cbObj;
447 };
448
449
450 /*
451  * Return the name of the ioqueue implementation.
452  */
453 PJ_DEF(const char*) pj_ioqueue_name(void)
454 {
455     return "ioqueue-symbian";
456 }
457
458
459 /*
460  * Create a new I/O Queue framework.
461  */
462 PJ_DEF(pj_status_t) pj_ioqueue_create(  pj_pool_t *pool, 
463                                         pj_size_t max_fd,
464                                         pj_ioqueue_t **p_ioqueue)
465 {
466     pj_ioqueue_t *ioq;
467
468     PJ_UNUSED_ARG(max_fd);
469
470     ioq = PJ_POOL_ZALLOC_T(pool, pj_ioqueue_t);
471     *p_ioqueue = ioq;
472     return PJ_SUCCESS;
473 }
474
475
476 /*
477  * Destroy the I/O queue.
478  */
479 PJ_DEF(pj_status_t) pj_ioqueue_destroy( pj_ioqueue_t *ioq )
480 {
481     PJ_UNUSED_ARG(ioq);
482     return PJ_SUCCESS;
483 }
484
485
486 /*
487  * Set the lock object to be used by the I/O Queue. 
488  */
489 PJ_DEF(pj_status_t) pj_ioqueue_set_lock( pj_ioqueue_t *ioq, 
490                                          pj_lock_t *lock,
491                                          pj_bool_t auto_delete )
492 {
493     /* Don't really need lock for now */
494     PJ_UNUSED_ARG(ioq);
495     
496     if (auto_delete) {
497         pj_lock_destroy(lock);
498     }
499
500     return PJ_SUCCESS;
501 }
502
503 PJ_DEF(pj_status_t) pj_ioqueue_set_default_concurrency(pj_ioqueue_t *ioqueue,
504                                                                                                            pj_bool_t allow)
505 {
506         /* Not supported, just return PJ_SUCCESS silently */
507         PJ_UNUSED_ARG(ioqueue);
508         PJ_UNUSED_ARG(allow);
509         return PJ_SUCCESS;
510 }
511
512 /*
513  * Register a socket to the I/O queue framework. 
514  */
515 PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
516                                               pj_ioqueue_t *ioq,
517                                               pj_sock_t sock,
518                                               void *user_data,
519                                               const pj_ioqueue_callback *cb,
520                                               pj_ioqueue_key_t **p_key )
521 {
522     pj_ioqueue_key_t *key;
523
524     key = PJ_POOL_ZALLOC_T(pool, pj_ioqueue_key_t);
525     key->cbObj = CIoqueueCallback::NewL(ioq, key, sock, cb, user_data);
526
527     *p_key = key;
528     return PJ_SUCCESS;
529 }
530
531 /*
532  * Unregister from the I/O Queue framework. 
533  */
534 PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_key_t *key )
535 {
536     if (key == NULL || key->cbObj == NULL)
537         return PJ_SUCCESS;
538
539     // Cancel pending async object
540     if (key->cbObj) {
541         key->cbObj->Cancel();
542     }
543
544     // Close socket.
545     key->cbObj->get_pj_socket()->Socket().Close();
546     delete key->cbObj->get_pj_socket();
547
548     // Delete async object.
549     if (key->cbObj) {
550         delete key->cbObj;
551         key->cbObj = NULL;
552     }
553
554     return PJ_SUCCESS;
555 }
556
557
558 /*
559  * Get user data associated with an ioqueue key.
560  */
561 PJ_DEF(void*) pj_ioqueue_get_user_data( pj_ioqueue_key_t *key )
562 {
563     return key->cbObj->get_user_data();
564 }
565
566
567 /*
568  * Set or change the user data to be associated with the file descriptor or
569  * handle or socket descriptor.
570  */
571 PJ_DEF(pj_status_t) pj_ioqueue_set_user_data( pj_ioqueue_key_t *key,
572                                               void *user_data,
573                                               void **old_data)
574 {
575     if (old_data)
576         *old_data = key->cbObj->get_user_data();
577     key->cbObj->set_user_data(user_data);
578
579     return PJ_SUCCESS;
580 }
581
582
583 /*
584  * Initialize operation key.
585  */
586 PJ_DEF(void) pj_ioqueue_op_key_init( pj_ioqueue_op_key_t *op_key,
587                                      pj_size_t size )
588 {
589     pj_bzero(op_key, size);
590 }
591
592
593 /*
594  * Check if operation is pending on the specified operation key.
595  */
596 PJ_DEF(pj_bool_t) pj_ioqueue_is_pending( pj_ioqueue_key_t *key,
597                                          pj_ioqueue_op_key_t *op_key )
598 {
599     return key->cbObj->get_op_key()==op_key &&
600            key->cbObj->IsActive();
601 }
602
603
604 /*
605  * Post completion status to the specified operation key and call the
606  * appropriate callback. 
607  */
608 PJ_DEF(pj_status_t) pj_ioqueue_post_completion( pj_ioqueue_key_t *key,
609                                                 pj_ioqueue_op_key_t *op_key,
610                                                 pj_ssize_t bytes_status )
611 {
612     if (pj_ioqueue_is_pending(key, op_key)) {
613         key->cbObj->CancelOperation(op_key, bytes_status);
614     }
615     return PJ_SUCCESS;
616 }
617
618
619 #if defined(PJ_HAS_TCP) && PJ_HAS_TCP != 0
620 /**
621  * Instruct I/O Queue to accept incoming connection on the specified 
622  * listening socket.
623  */
624 PJ_DEF(pj_status_t) pj_ioqueue_accept( pj_ioqueue_key_t *key,
625                                        pj_ioqueue_op_key_t *op_key,
626                                        pj_sock_t *new_sock,
627                                        pj_sockaddr_t *local,
628                                        pj_sockaddr_t *remote,
629                                        int *addrlen )
630 {
631     
632     return key->cbObj->StartAccept(op_key, new_sock, local, remote, addrlen);
633 }
634
635
636 /*
637  * Initiate non-blocking socket connect.
638  */
639 PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_key_t *key,
640                                         const pj_sockaddr_t *addr,
641                                         int addrlen )
642 {
643     pj_status_t status;
644     
645     RSocket &rSock = key->cbObj->get_pj_socket()->Socket();
646     TInetAddr inetAddr;
647     TRequestStatus reqStatus;
648
649     // Return failure if access point is marked as down by app.
650     PJ_SYMBIAN_CHECK_CONNECTION();
651     
652     // Convert address
653     status = PjSymbianOS::pj2Addr(*(const pj_sockaddr*)addr, addrlen, 
654                                   inetAddr);
655     if (status != PJ_SUCCESS)
656         return status;
657     
658     // We don't support async connect for now.
659     PJ_TODO(IOQUEUE_SUPPORT_ASYNC_CONNECT);
660
661     rSock.Connect(inetAddr, reqStatus);
662     User::WaitForRequest(reqStatus);
663
664     if (reqStatus == KErrNone)
665         return PJ_SUCCESS;
666
667     return PJ_RETURN_OS_ERROR(reqStatus.Int());
668 }
669
670
671 #endif  /* PJ_HAS_TCP */
672
673 /*
674  * Poll the I/O Queue for completed events.
675  */
676 PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioq,
677                              const pj_time_val *timeout)
678 {
679     /* Polling is not necessary on Symbian, since all async activities
680      * are registered to active scheduler.
681      */
682     PJ_UNUSED_ARG(ioq);
683     PJ_UNUSED_ARG(timeout);
684     return 0;
685 }
686
687
688 /*
689  * Instruct the I/O Queue to read from the specified handle.
690  */
691 PJ_DEF(pj_status_t) pj_ioqueue_recv( pj_ioqueue_key_t *key,
692                                      pj_ioqueue_op_key_t *op_key,
693                                      void *buffer,
694                                      pj_ssize_t *length,
695                                      pj_uint32_t flags )
696 {
697     // If socket has reader, delete it.
698     if (key->cbObj->get_pj_socket()->Reader())
699         key->cbObj->get_pj_socket()->DestroyReader();
700     
701     // Clear flag
702     flags &= ~PJ_IOQUEUE_ALWAYS_ASYNC;
703     return key->cbObj->StartRead(op_key, buffer, length, flags, NULL, NULL);
704 }
705
706
707 /*
708  * This function behaves similarly as #pj_ioqueue_recv(), except that it is
709  * normally called for socket, and the remote address will also be returned
710  * along with the data.
711  */
712 PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_key_t *key,
713                                          pj_ioqueue_op_key_t *op_key,
714                                          void *buffer,
715                                          pj_ssize_t *length,
716                                          pj_uint32_t flags,
717                                          pj_sockaddr_t *addr,
718                                          int *addrlen)
719 {
720     CPjSocket *sock = key->cbObj->get_pj_socket();
721     
722     // If address is specified, check that the length match the
723     // address family
724     if (addr || addrlen) {
725         PJ_ASSERT_RETURN(addr && addrlen && *addrlen, PJ_EINVAL);
726         if (sock->GetAf() == PJ_AF_INET) {
727             PJ_ASSERT_RETURN(*addrlen>=(int)sizeof(pj_sockaddr_in), PJ_EINVAL);
728         } else if (sock->GetAf() == PJ_AF_INET6) {
729             PJ_ASSERT_RETURN(*addrlen>=(int)sizeof(pj_sockaddr_in6), PJ_EINVAL);
730         }
731     }
732     
733     // If socket has reader, delete it.
734     if (sock->Reader())
735         sock->DestroyReader();
736     
737     if (key->cbObj->IsActive())
738         return PJ_EBUSY;
739
740     // Clear flag
741     flags &= ~PJ_IOQUEUE_ALWAYS_ASYNC;
742     return key->cbObj->StartRead(op_key, buffer, length, flags, addr, addrlen);
743 }
744
745
746 /*
747  * Instruct the I/O Queue to write to the handle.
748  */
749 PJ_DEF(pj_status_t) pj_ioqueue_send( pj_ioqueue_key_t *key,
750                                      pj_ioqueue_op_key_t *op_key,
751                                      const void *data,
752                                      pj_ssize_t *length,
753                                      pj_uint32_t flags )
754 {
755     TRequestStatus reqStatus;
756     TPtrC8 aBuffer((const TUint8*)data, (TInt)*length);
757     TSockXfrLength aLen;
758     
759     PJ_UNUSED_ARG(op_key);
760
761     // Forcing pending operation is not supported.
762     PJ_ASSERT_RETURN((flags & PJ_IOQUEUE_ALWAYS_ASYNC)==0, PJ_EINVAL);
763
764     // Return failure if access point is marked as down by app.
765     PJ_SYMBIAN_CHECK_CONNECTION();
766
767     // Clear flag
768     flags &= ~PJ_IOQUEUE_ALWAYS_ASYNC;
769
770     key->cbObj->get_pj_socket()->Socket().Send(aBuffer, flags, reqStatus, aLen);
771     User::WaitForRequest(reqStatus);
772
773     if (reqStatus.Int() != KErrNone)
774         return PJ_RETURN_OS_ERROR(reqStatus.Int());
775
776     //At least in UIQ Emulator, aLen.Length() reports incorrect length
777     //for UDP (some newlc.com users seem to have reported this too).
778     //*length = aLen.Length();
779     return PJ_SUCCESS;
780 }
781
782
783 /*
784  * Instruct the I/O Queue to write to the handle.
785  */
786 PJ_DEF(pj_status_t) pj_ioqueue_sendto( pj_ioqueue_key_t *key,
787                                        pj_ioqueue_op_key_t *op_key,
788                                        const void *data,
789                                        pj_ssize_t *length,
790                                        pj_uint32_t flags,
791                                        const pj_sockaddr_t *addr,
792                                        int addrlen)
793 {
794     TRequestStatus reqStatus;
795     TPtrC8 aBuffer;
796     TInetAddr inetAddr;
797     TSockXfrLength aLen;
798     pj_status_t status;
799     
800     PJ_UNUSED_ARG(op_key);
801
802     // Forcing pending operation is not supported.
803     PJ_ASSERT_RETURN((flags & PJ_IOQUEUE_ALWAYS_ASYNC)==0, PJ_EINVAL);
804
805     // Return failure if access point is marked as down by app.
806     PJ_SYMBIAN_CHECK_CONNECTION();
807
808     // Convert address
809     status = PjSymbianOS::pj2Addr(*(const pj_sockaddr*)addr, addrlen, 
810                                   inetAddr);
811     if (status != PJ_SUCCESS)
812         return status;
813     
814     // Clear flag
815     flags &= ~PJ_IOQUEUE_ALWAYS_ASYNC;
816
817     aBuffer.Set((const TUint8*)data, (TInt)*length);
818     CPjSocket *pjSock = key->cbObj->get_pj_socket();
819
820     pjSock->Socket().SendTo(aBuffer, inetAddr, flags, reqStatus, aLen);
821     User::WaitForRequest(reqStatus);
822
823     if (reqStatus.Int() != KErrNone)
824         return PJ_RETURN_OS_ERROR(reqStatus.Int());
825
826     //At least in UIQ Emulator, aLen.Length() reports incorrect length
827     //for UDP (some newlc.com users seem to have reported this too).
828     //*length = aLen.Length();
829     return PJ_SUCCESS;
830 }
831
832 PJ_DEF(pj_status_t) pj_ioqueue_set_concurrency(pj_ioqueue_key_t *key,
833                                                                                            pj_bool_t allow)
834 {
835         /* Not supported, just return PJ_SUCCESS silently */
836         PJ_UNUSED_ARG(key);
837         PJ_UNUSED_ARG(allow);
838         return PJ_SUCCESS;
839 }
840
841 PJ_DEF(pj_status_t) pj_ioqueue_lock_key(pj_ioqueue_key_t *key)
842 {
843         /* Not supported, just return PJ_SUCCESS silently */
844         PJ_UNUSED_ARG(key);
845         return PJ_SUCCESS;
846 }
847
848 PJ_DEF(pj_status_t) pj_ioqueue_unlock_key(pj_ioqueue_key_t *key)
849 {
850         /* Not supported, just return PJ_SUCCESS silently */
851         PJ_UNUSED_ARG(key);
852         return PJ_SUCCESS;
853 }