TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp

   1 #include "tao/Messaging/Asynch_Reply_Dispatcher.h"
   2 #include "tao/Pluggable_Messaging_Utils.h"
   3 #include "tao/ORB_Core.h"
   4 #include "tao/debug.h"
   5 #include "tao/ORB_Core.h"
   6 #include "tao/Transport.h"
   7 #include "tao/Transport_Mux_Strategy.h"
   8
   9 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  10
  11 // Constructor.
  12 TAO_Asynch_Reply_Dispatcher::TAO_Asynch_Reply_Dispatcher (
  13     const TAO_Reply_Handler_Stub &reply_handler_stub,
  14     Messaging::ReplyHandler_ptr reply_handler,
  15     TAO_ORB_Core *orb_core,
  16     ACE_Allocator *allocator
  17   )
  18   :TAO_Asynch_Reply_Dispatcher_Base (orb_core, allocator)
  19   , reply_handler_stub_ (reply_handler_stub)
  20   , reply_handler_ (Messaging::ReplyHandler::_duplicate (reply_handler))
  21   , timeout_handler_ (0)
  22 {
  23 }
  24
  25 // Dispatch the reply.
  26 int
  27 TAO_Asynch_Reply_Dispatcher::dispatch_reply (TAO_Pluggable_Reply_Params &params)
  28 {
  29   if (this->timeout_handler_)
  30     {
  31       // If we had registered timeout handlers just cancel them and
  32       // loose ownership of the handlers
  33       this->timeout_handler_->cancel ();
  34       this->timeout_handler_->remove_reference ();
  35       this->timeout_handler_ = 0;
  36       // AMI Timeout Handling End
  37     }
  38
  39   // With Asynch requests the invocation handler can't call idle_after_reply ()
  40   // since it does not handle the reply.
  41   // So we have to do that here in case f.i. the Exclusive TMS left the transport
  42   // busy after the send
  43   if (this->transport_ != 0)
  44     this->transport_->tms ()->idle_after_reply ();
  45
  46   if (!params.input_cdr_)
  47     return -1;
  48
  49   if (!this->try_dispatch_reply ())
  50     return 0;
  51
  52   this->reply_status_ = params.reply_status ();
  53   this->locate_reply_status_ = params.locate_reply_status ();
  54
  55   // Transfer the <params.input_cdr_>'s content to this->reply_cdr_
  56   ACE_Data_Block *db = this->reply_cdr_.clone_from (*params.input_cdr_);
  57
  58   if (db == 0)
  59     {
  60       if (TAO_debug_level > 2)
  61         {
  62           TAOLIB_ERROR ((
  63             LM_ERROR,
  64             ACE_TEXT ("TAO_Messaging (%P|%t) - Asynch_Reply_Dispatcher::dispatch_reply ")
  65             ACE_TEXT ("clone_from failed\n")));
  66         }
  67       return -1;
  68     }
  69
  70   // See whether we need to delete the data block by checking the
  71   // flags. We cannot be happy that we initially allocated the
  72   // datablocks of the stack. If this method is called twice, as is in
  73   // some cases where the same invocation object is used to make two
  74   // invocations like forwarding, the release becomes essential.
  75   if (ACE_BIT_DISABLED (db->flags (), ACE_Message_Block::DONT_DELETE))
  76     {
  77       db->release ();
  78     }
  79
  80   if (!CORBA::is_nil (this->reply_handler_.in ()))
  81     {
  82       // Steal the buffer, that way we don't do any unnecessary copies of
  83       // this data.
  84       CORBA::ULong const max = params.svc_ctx_.maximum ();
  85       CORBA::ULong const len = params.svc_ctx_.length ();
  86       IOP::ServiceContext *context_list = params.svc_ctx_.get_buffer (1);
  87       this->reply_service_info_.replace (max, len, context_list, 1);
  88
  89       if (TAO_debug_level >= 4)
  90         {
  91           TAOLIB_DEBUG ((LM_DEBUG,
  92                       ACE_TEXT ("TAO_Messaging (%P|%t) - Asynch_Reply_Dispatcher")
  93                       ACE_TEXT ("::dispatch_reply status = %d\n"),
  94                                 this->reply_status_));
  95         }
  96
  97       CORBA::ULong reply_error = TAO_AMI_REPLY_NOT_OK;
  98       switch (this->reply_status_)
  99         {
 100         case GIOP::NO_EXCEPTION:
 101           reply_error = TAO_AMI_REPLY_OK;
 102           break;
 103         case GIOP::USER_EXCEPTION:
 104           reply_error = TAO_AMI_REPLY_USER_EXCEPTION;
 105           break;
 106         case GIOP::SYSTEM_EXCEPTION:
 107           reply_error = TAO_AMI_REPLY_SYSTEM_EXCEPTION;
 108           break;
 109         case GIOP::LOCATION_FORWARD:
 110           reply_error = TAO_AMI_REPLY_LOCATION_FORWARD;
 111           break;
 112         case GIOP::LOCATION_FORWARD_PERM:
 113           reply_error = TAO_AMI_REPLY_LOCATION_FORWARD_PERM;
 114           break;
 115
 116         default:
 117           // @@ Michael: Not even the spec mentions this case.
 118           //             We have to think about this case.
 119           // Handle the forwarding and return so the stub restarts the
 120           // request!
 121           reply_error = TAO_AMI_REPLY_NOT_OK;
 122           break;
 123         }
 124
 125       try
 126         {
 127           // Call the Reply Handler's stub.
 128           this->reply_handler_stub_ (this->reply_cdr_,
 129                                      this->reply_handler_.in (),
 130                                      reply_error);
 131         }
 132       catch (const ::CORBA::Exception& ex)
 133         {
 134           if (TAO_debug_level >= 4)
 135             ex._tao_print_exception ("Asynch_Reply_Dispatcher::dispatch_reply, exception during reply handler");
 136         }
 137     }
 138
 139   this->intrusive_remove_ref (this);
 140
 141   return 1;
 142 }
 143
 144 void
 145 TAO_Asynch_Reply_Dispatcher::connection_closed ()
 146 {
 147   try
 148     {
 149       if (this->timeout_handler_)
 150         {
 151           // If we had registered timeout handlers just cancel them and
 152           // loose ownership of the handlers
 153           this->timeout_handler_->cancel ();
 154           this->timeout_handler_->remove_reference ();
 155           this->timeout_handler_ = 0;
 156         }
 157
 158       if (!this->try_dispatch_reply ())
 159         return;
 160
 161       if (!CORBA::is_nil (this->reply_handler_.in ()))
 162         {
 163           // Generate a fake exception....
 164           CORBA::COMM_FAILURE comm_failure (0, CORBA::COMPLETED_MAYBE);
 165
 166           TAO_OutputCDR out_cdr;
 167
 168           comm_failure._tao_encode (out_cdr);
 169
 170           // Turn into an output CDR
 171           TAO_InputCDR cdr (out_cdr);
 172
 173           this->reply_handler_stub_ (cdr,
 174                                      this->reply_handler_.in (),
 175                                      TAO_AMI_REPLY_SYSTEM_EXCEPTION);
 176         }
 177     }
 178   catch (const ::CORBA::Exception& ex)
 179     {
 180       if (TAO_debug_level >= 4)
 181         {
 182           ex._tao_print_exception (
 183             "Asynch_Reply_Dispatcher::connection_closed");
 184         }
 185
 186     }
 187
 188   this->intrusive_remove_ref (this);
 189 }
 190
 191 // AMI Timeout Handling Begin
 192
 193 void
 194 TAO_Asynch_Reply_Dispatcher::reply_timed_out ()
 195 {
 196   try
 197     {
 198       // @@ This check probably is unnecessary..
 199       if (this->timeout_handler_)
 200         {
 201           // If we had registered timeout handlers just cancel them and
 202           // loose ownership of the handlers
 203           this->timeout_handler_->remove_reference ();
 204           this->timeout_handler_ = 0;
 205         }
 206
 207       // With Asynch requests the invocation handler can't call idle_after_reply ()
 208       // since it does not handle the reply.
 209       // So we have to do that here in case f.i. the Exclusive TMS left the transport
 210       // busy after the send
 211       if (this->transport_ != 0)
 212         this->transport_->tms ()->idle_after_reply ();
 213
 214       // This is okay here... Everything relies on our refcount being
 215       // held by the timeout handler, whose refcount in turn is held
 216       // by the reactor.
 217       if (!this->try_dispatch_reply ())
 218         return;
 219
 220       if (!CORBA::is_nil (this->reply_handler_.in ()))
 221         {
 222           // Generate a fake exception....
 223           CORBA::TIMEOUT timeout_failure (
 224             CORBA::SystemException::_tao_minor_code (
 225                 TAO_TIMEOUT_RECV_MINOR_CODE,
 226                 ETIME),
 227              CORBA::COMPLETED_MAYBE);
 228
 229           TAO_OutputCDR out_cdr;
 230
 231           timeout_failure._tao_encode (out_cdr);
 232
 233           // Turn into an output CDR
 234           TAO_InputCDR cdr (out_cdr);
 235
 236           this->reply_handler_stub_ (cdr,
 237                                      this->reply_handler_.in (),
 238                                      TAO_AMI_REPLY_SYSTEM_EXCEPTION);
 239         }
 240     }
 241   catch (const ::CORBA::Exception& ex)
 242     {
 243       if (TAO_debug_level >= 4)
 244         {
 245           ex._tao_print_exception ("Asynch_Reply_Dispatcher::reply_timed_out");
 246         }
 247     }
 248
 249   this->intrusive_remove_ref (this);
 250 }
 251
 252 long
 253 TAO_Asynch_Reply_Dispatcher::schedule_timer (CORBA::ULong request_id,
 254                                              const ACE_Time_Value &max_wait_time)
 255 {
 256   if (this->timeout_handler_ == 0)
 257     {
 258       // @@ Need to use the pool for this..
 259       ACE_NEW_THROW_EX (this->timeout_handler_,
 260                         TAO_Asynch_Timeout_Handler (
 261                           this->transport_->orb_core ()->reactor ()),
 262                         CORBA::NO_MEMORY ());
 263     }
 264
 265   return this->timeout_handler_->schedule_timer (
 266       this->transport_->tms (),
 267       request_id,
 268       max_wait_time);
 269 }
 270
 271 TAO_END_VERSIONED_NAMESPACE_DECL