ACE/apps/JAWS3/jaws3/Reactive_IO.cpp

   1 #include "ace/ACE.h"
   2 #include "ace/OS_NS_unistd.h"
   3 #include "ace/Reactor.h"
   4
   5 #ifndef JAWS_BUILD_DLL
   6 #define JAWS_BUILD_DLL
   7 #endif
   8
   9 #include "jaws3/Jaws_IO.h"
  10 #include "jaws3/Reactive_IO.h"
  11 #include "jaws3/Reactive_IO_Helpers.h"
  12 #include "jaws3/Event_Completer.h"
  13
  14 void
  15 JAWS_Reactive_IO::send ( ACE_HANDLE handle
  16                        , ACE_Message_Block *mb
  17                        , JAWS_Event_Completer *completer
  18                        , const ACE_Time_Value &tv
  19                        , void *act
  20                        )
  21 {
  22   if (mb->length () == 0)
  23     {
  24       JAWS_Event_Result io_result ( 0
  25                                   , JAWS_Event_Result::JE_OK
  26                                   , JAWS_Event_Result::JE_SEND_OK
  27                                   );
  28       if (completer)
  29         completer->output_complete (io_result, act);
  30
  31       return;
  32     }
  33
  34   JAWS_IO_Reactive_Send *rs;
  35   rs = JAWS_IO_Reactive_Send::make (handle, mb, completer, tv, act);
  36
  37   if (rs == 0)
  38     {
  39       JAWS_Event_Result io_result ( 0
  40                                   , JAWS_Event_Result::JE_ERROR
  41                                   , JAWS_Event_Result::JE_SEND_FAIL
  42                                   );
  43       if (completer)
  44         completer->output_complete (io_result, act);
  45
  46       return;
  47     }
  48
  49   rs->open ();
  50 }
  51
  52 void
  53 JAWS_Reactive_IO::send ( ACE_HANDLE handle
  54                        , ACE_Message_Block *mb
  55                        , JAWS_Event_Completer *completer
  56                        , void *act
  57                        )
  58 {
  59   this->send (handle, mb, completer, ACE_Time_Value::zero, act);
  60 }
  61
  62
  63 void
  64 JAWS_Reactive_IO::recv ( ACE_HANDLE handle
  65                        , ACE_Message_Block *mb
  66                        , JAWS_Event_Completer *completer
  67                        , const ACE_Time_Value &tv
  68                        , void *act
  69                        )
  70 {
  71   JAWS_IO_Reactive_Recv *rr;
  72   rr = JAWS_IO_Reactive_Recv::make (handle, mb, completer, tv, act);
  73
  74   if (rr == 0)
  75     {
  76       JAWS_Event_Result io_result ( 0
  77                                   , JAWS_Event_Result::JE_ERROR
  78                                   , JAWS_Event_Result::JE_RECV_FAIL
  79                                   );
  80       if (completer)
  81         completer->output_complete (io_result, act);
  82
  83       return;
  84     }
  85
  86   rr->open ();
  87 }
  88
  89
  90 void
  91 JAWS_Reactive_IO::recv ( ACE_HANDLE handle
  92                        , ACE_Message_Block *mb
  93                        , JAWS_Event_Completer *completer
  94                        , void *act
  95                        )
  96 {
  97   this->recv (handle, mb, completer, ACE_Time_Value::zero, act);
  98 }
  99
 100
 101 void
 102 JAWS_Reactive_IO::transmit ( ACE_HANDLE handle
 103                            , ACE_HANDLE source
 104                            , JAWS_Event_Completer *completer
 105                            , const ACE_Time_Value &tv
 106                            , void *act
 107                            , ACE_Message_Block *header
 108                            , ACE_Message_Block *trailer
 109                            )
 110 {
 111   JAWS_IO_Reactive_Transmit *rt;
 112   rt = JAWS_IO_Reactive_Transmit::make ( handle
 113                                        , source
 114                                        , completer
 115                                        , tv
 116                                        , header
 117                                        , trailer
 118                                        , act
 119                                        );
 120
 121   if (rt == 0)
 122     {
 123       JAWS_Event_Result io_result ( 0
 124                                   , JAWS_Event_Result::JE_ERROR
 125                                   , JAWS_Event_Result::JE_TRANSMIT_FAIL
 126                                   );
 127       if (completer)
 128         completer->output_complete (io_result, act);
 129
 130       return;
 131     }
 132
 133   rt->open ();
 134 }
 135
 136
 137 void
 138 JAWS_Reactive_IO::transmit ( ACE_HANDLE handle
 139                            , ACE_HANDLE source
 140                            , JAWS_Event_Completer *completer
 141                            , void *act
 142                            , ACE_Message_Block *header
 143                            , ACE_Message_Block *trailer
 144                            )
 145 {
 146   this->transmit ( handle
 147                  , source
 148                  , completer
 149                  , ACE_Time_Value::zero
 150                  , act
 151                  , header
 152                  , trailer
 153                  );
 154 }
 155
 156
 157 void
 158 JAWS_IO_Reactive_Handler::open ()
 159 {
 160   int result = ACE_Reactor::instance ()->notify (this);
 161
 162   if (result < 0)
 163     this->close (result);
 164 }
 165
 166 void
 167 JAWS_IO_Reactive_Handler::close (int result)
 168 {
 169   if (result < 0)
 170     {
 171       if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::WRITE_MASK))
 172         {
 173           JAWS_Event_Result io_result ( 0
 174                                       , JAWS_Event_Result::JE_ERROR
 175                                       , JAWS_Event_Result::JE_SEND_FAIL
 176                                       );
 177           this->io_result_ = io_result;
 178         }
 179       else if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::READ_MASK))
 180         {
 181           JAWS_Event_Result io_result ( 0
 182                                       , JAWS_Event_Result::JE_ERROR
 183                                       , JAWS_Event_Result::JE_RECV_FAIL
 184                                       );
 185           this->io_result_ = io_result;
 186         }
 187
 188       this->handle_close (this->handle_, this->mask_);
 189     }
 190 }
 191
 192 int
 193 JAWS_IO_Reactive_Handler::handle_timeout (const ACE_Time_Value &, const void *)
 194 {
 195   if (this->was_active_)
 196     {
 197       this->was_active_ = 0;
 198
 199       this->timer_id_ =
 200         ACE_Reactor::instance ()->schedule_timer (this, 0, this->tv_);
 201
 202       return 0;
 203     }
 204
 205   ACE_Reactor::instance ()
 206   ->remove_handler ( this
 207                    , ACE_Event_Handler::RWE_MASK|ACE_Event_Handler::DONT_CALL
 208                    );
 209
 210   this->timer_id_ = -1;
 211
 212   if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::WRITE_MASK))
 213     {
 214       JAWS_Event_Result io_result ( 0
 215                                   , JAWS_Event_Result::JE_ERROR
 216                                   , JAWS_Event_Result::JE_SEND_TIMEOUT
 217                                   , ETIME
 218                                   );
 219       this->io_result_ = io_result;
 220     }
 221   else if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::READ_MASK))
 222     {
 223       JAWS_Event_Result io_result ( 0
 224                                   , JAWS_Event_Result::JE_ERROR
 225                                   , JAWS_Event_Result::JE_RECV_TIMEOUT
 226                                   , ETIME
 227                                   );
 228       this->io_result_ = io_result;
 229     }
 230
 231   return -1;
 232 }
 233
 234 int
 235 JAWS_IO_Reactive_Handler::handle_close (ACE_HANDLE, ACE_Reactor_Mask)
 236 {
 237   if (this->completer_)
 238     {
 239       if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::WRITE_MASK))
 240         this->completer_->output_complete (this->io_result_, this->act_);
 241       else if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::READ_MASK))
 242         this->completer_->input_complete (this->io_result_, this->act_);
 243     }
 244
 245   ACE_Reactor::instance ()
 246   ->remove_handler ( this
 247                    , ACE_Event_Handler::RWE_MASK|ACE_Event_Handler::DONT_CALL
 248                    );
 249
 250   delete this;
 251   return 0;
 252 }
 253
 254 int
 255 JAWS_IO_Reactive_Handler::handle_exception (ACE_HANDLE handle)
 256 {
 257   if (handle == ACE_INVALID_HANDLE)
 258     {
 259       // We are being called back from a notify call.
 260       // This is our cue to register ourselves with the Reactor.
 261
 262       int result;
 263       result =
 264         ACE_Reactor::instance ()
 265         ->register_handler (this, this->mask_|ACE_Event_Handler::EXCEPT_MASK);
 266
 267       if (result < 0)
 268         this->close (result);
 269
 270       return 0;
 271     }
 272
 273   // back to our regularly scheduled except mask handling.
 274
 275   if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::WRITE_MASK))
 276     {
 277       JAWS_Event_Result io_result ( this->bytes_
 278                                   , JAWS_Event_Result::JE_ERROR
 279                                   , JAWS_Event_Result::JE_SEND_SHORT
 280                                   );
 281       this->io_result_ = io_result;
 282     }
 283   else if (ACE_BIT_ENABLED (this->mask_, ACE_Event_Handler::READ_MASK))
 284     {
 285       JAWS_Event_Result io_result ( this->bytes_
 286                                   , JAWS_Event_Result::JE_ERROR
 287                                   , JAWS_Event_Result::JE_RECV_SHORT
 288                                   );
 289       this->io_result_ = io_result;
 290     }
 291
 292   return -1;
 293 }
 294
 295
 296 int
 297 JAWS_IO_Reactive_Send::handle_output (ACE_HANDLE handle)
 298 {
 299   this->was_active_ = 1;
 300
 301   ssize_t count = ACE::send ( handle
 302                             , this->mb_->rd_ptr ()
 303                             , this->mb_->length ()
 304                             );
 305
 306   if (count <= 0 && this->bytes_ == 0)
 307     {
 308       JAWS_Event_Result io_result ( 0
 309                                   , JAWS_Event_Result::JE_ERROR
 310                                   , JAWS_Event_Result::JE_SEND_FAIL
 311                                   );
 312       this->io_result_ = io_result;
 313     }
 314   else if (count <= 0 && this->bytes_ > 0)
 315     {
 316       JAWS_Event_Result io_result ( this->bytes_
 317                                   , JAWS_Event_Result::JE_ERROR
 318                                   , JAWS_Event_Result::JE_SEND_SHORT
 319                                   );
 320       this->io_result_ = io_result;
 321     }
 322   else
 323     {
 324       if (count > 0)
 325         this->mb_->rd_ptr (count);
 326
 327       this->bytes_ += count;
 328
 329       JAWS_Event_Result io_result ( this->bytes_
 330                                   , JAWS_Event_Result::JE_OK
 331                                   , JAWS_Event_Result::JE_SEND_OK
 332                                   );
 333       this->io_result_ = io_result;
 334     }
 335
 336   if (count <= 0 || this->mb_->length () == 0)
 337     return -1;
 338
 339   // Not done yet, so stay registered.
 340   return 0;
 341 }
 342
 343
 344 int
 345 JAWS_IO_Reactive_Recv::handle_input (ACE_HANDLE handle)
 346 {
 347   ssize_t count = ACE::recv ( handle
 348                             , this->mb_->wr_ptr ()
 349                             , this->mb_->space ()
 350                             );
 351
 352   if (count < 0)
 353     {
 354       JAWS_Event_Result io_result ( 0
 355                                   , JAWS_Event_Result::JE_ERROR
 356                                   , JAWS_Event_Result::JE_RECV_FAIL
 357                                   );
 358       this->io_result_ = io_result;
 359     }
 360   else
 361     {
 362       if (count > 0)
 363         this->mb_->wr_ptr (count);
 364
 365       JAWS_Event_Result io_result ( count
 366                                   , JAWS_Event_Result::JE_OK
 367                                   , JAWS_Event_Result::JE_RECV_OK
 368                                   );
 369       this->io_result_ = io_result;
 370     }
 371
 372   return -1;
 373 }
 374
 375
 376 int
 377 JAWS_IO_Reactive_Transmit::handle_output (ACE_HANDLE handle)
 378 {
 379   this->was_active_ = 1;
 380
 381   if (this->header_ && this->header_->length () > 0)
 382     return this->handle_output_header (handle);
 383   else
 384     this->header_ = 0;
 385
 386   if (this->source_ != ACE_INVALID_HANDLE)
 387     return this->handle_output_source (handle);
 388
 389   if (this->trailer_ && this->trailer_->length () > 0)
 390     return this->handle_output_trailer (handle);
 391   else
 392     this->trailer_ = 0;
 393
 394   JAWS_Event_Result io_result ( this->bytes_
 395                               , JAWS_Event_Result::JE_OK
 396                               , JAWS_Event_Result::JE_TRANSMIT_OK
 397                               );
 398   this->io_result_ = io_result;
 399
 400   return -1;
 401 }
 402
 403 int
 404 JAWS_IO_Reactive_Transmit::handle_output_header (ACE_HANDLE handle)
 405 {
 406   return this->handle_output_mb (handle, this->header_);
 407 }
 408
 409 int
 410 JAWS_IO_Reactive_Transmit::handle_output_source (ACE_HANDLE handle)
 411 {
 412   ACE_Message_Block *mb = this->source_buf_;
 413
 414   if (mb == 0)
 415     return -1;
 416
 417   // Try to read data into the mb if data is still available.
 418   if (mb->space () && this->source_ != ACE_INVALID_HANDLE)
 419     {
 420       ssize_t count;
 421       count = ACE_OS::read (this->source_, mb->wr_ptr (), mb->space ());
 422
 423       if (count < 0)
 424         {
 425           this->source_ = ACE_INVALID_HANDLE;
 426           this->source_buf_ = 0;
 427
 428           if (this->bytes_ == 0)
 429             {
 430               JAWS_Event_Result io_result ( 0
 431                                           , JAWS_Event_Result::JE_ERROR
 432                                           , JAWS_Event_Result::JE_TRANSMIT_FAIL
 433                                           );
 434               this->io_result_ = io_result;
 435             }
 436           else if (this->bytes_ > 0)
 437             {
 438               JAWS_Event_Result io_result ( this->bytes_
 439                                           , JAWS_Event_Result::JE_ERROR
 440                                           , JAWS_Event_Result::JE_TRANSMIT_SHORT
 441                                           );
 442               this->io_result_ = io_result;
 443             }
 444
 445           return -1;
 446         }
 447       else if (count == 0)
 448         this->source_ = ACE_INVALID_HANDLE;
 449       else
 450         mb->wr_ptr (count);
 451     }
 452
 453   int result = 0;
 454
 455   if (mb->length () > 0)
 456     result = this->handle_output_mb (handle, mb);
 457
 458   if (result < 0)
 459     {
 460       this->source_ = ACE_INVALID_HANDLE;
 461       this->source_buf_ = 0;
 462     }
 463   else if (mb == 0 && this->source_ == ACE_INVALID_HANDLE)
 464     this->source_buf_ = 0;
 465   else
 466     this->source_buf_->crunch ();
 467
 468   return result;
 469 }
 470
 471 int
 472 JAWS_IO_Reactive_Transmit::handle_output_trailer (ACE_HANDLE handle)
 473 {
 474   int result = this->handle_output_mb (handle, this->trailer_);
 475
 476   if (result == 0 && this->trailer_ == 0)
 477     {
 478       JAWS_Event_Result io_result ( this->bytes_
 479                                   , JAWS_Event_Result::JE_ERROR
 480                                   , JAWS_Event_Result::JE_TRANSMIT_SHORT
 481                                   );
 482       this->io_result_ = io_result;
 483       return -1;
 484     }
 485
 486   return result;
 487 }
 488
 489 int
 490 JAWS_IO_Reactive_Transmit::handle_output_mb ( ACE_HANDLE handle
 491                                             , ACE_Message_Block *&mb
 492                                             )
 493 {
 494   ssize_t count = ACE::send (handle, mb->rd_ptr (), mb->length ());
 495
 496   if (count <= 0 && this->bytes_ == 0)
 497     {
 498       JAWS_Event_Result io_result ( 0
 499                                   , JAWS_Event_Result::JE_ERROR
 500                                   , JAWS_Event_Result::JE_TRANSMIT_FAIL
 501                                   );
 502       this->io_result_ = io_result;
 503     }
 504   else if (count <= 0 && this->bytes_ > 0)
 505     {
 506       JAWS_Event_Result io_result ( this->bytes_
 507                                   , JAWS_Event_Result::JE_ERROR
 508                                   , JAWS_Event_Result::JE_TRANSMIT_SHORT
 509                                   );
 510       this->io_result_ = io_result;
 511     }
 512   else
 513     {
 514       mb->rd_ptr (count);
 515       this->bytes_ += count;
 516     }
 517
 518   if (count <= 0)
 519     return -1;
 520
 521   if (mb->length () == 0)
 522     mb = 0;
 523
 524   return 0;
 525 }
 526
 527 void
 528 JAWS_IO_Reactive_Transmit::close (int result)
 529 {
 530   if (result < 0)
 531     {
 532       JAWS_Event_Result io_result ( 0
 533                                   , JAWS_Event_Result::JE_ERROR
 534                                   , JAWS_Event_Result::JE_TRANSMIT_FAIL
 535                                   );
 536       this->io_result_ = io_result;
 537
 538       this->handle_close (this->handle_, this->mask_);
 539     }
 540 }
 541
 542 int
 543 JAWS_IO_Reactive_Transmit::handle_timeout (const ACE_Time_Value &, const void *)
 544 {
 545   if (this->was_active_)
 546     {
 547       this->was_active_ = 0;
 548
 549       this->timer_id_ =
 550         ACE_Reactor::instance ()->schedule_timer (this, 0, this->tv_);
 551
 552       return 0;
 553     }
 554
 555   ACE_Reactor::instance ()
 556   ->remove_handler ( this
 557                    , ACE_Event_Handler::RWE_MASK|ACE_Event_Handler::DONT_CALL
 558                    );
 559
 560   this->timer_id_ = -1;
 561
 562   JAWS_Event_Result io_result ( 0
 563                               , JAWS_Event_Result::JE_ERROR
 564                               , JAWS_Event_Result::JE_TRANSMIT_TIMEOUT
 565                               , ETIME
 566                               );
 567
 568   this->io_result_ = io_result;
 569
 570   return -1;
 571 }
 572
 573 int
 574 JAWS_IO_Reactive_Transmit::handle_exception (ACE_HANDLE handle)
 575 {
 576   if (handle == ACE_INVALID_HANDLE)
 577     {
 578       // We are being called back from a notify call.
 579       // This is our cue to register ourselves with the Reactor.
 580
 581       int result;
 582       result =
 583         ACE_Reactor::instance ()
 584         ->register_handler (this, this->mask_|ACE_Event_Handler::EXCEPT_MASK);
 585
 586       if (result < 0)
 587         this->close (result);
 588
 589       return 0;
 590     }
 591
 592   // back to our regularly scheduled except mask handling.
 593
 594   JAWS_Event_Result io_result ( this->bytes_
 595                               , JAWS_Event_Result::JE_ERROR
 596                               , JAWS_Event_Result::JE_TRANSMIT_SHORT
 597                               );
 598   this->io_result_ = io_result;
 599
 600   return -1;
 601 }