TAO/orbsvcs/tests/AVStreams/Pluggable_Flow_Protocol/TimeStamp.cpp

   1 #include "TimeStamp.h"
   2 #include "orbsvcs/AV/AVStreams_i.h"
   3 #include "ace/High_Res_Timer.h"
   4 #include "ace/OS_NS_strings.h"
   5
   6 //TimeStamp Protocol Object
   7
   8 TimeStamp_Protocol_Object::TimeStamp_Protocol_Object (TAO_AV_Callback *callback,
   9                                                       TAO_AV_Transport *transport)
  10   :TAO_AV_Protocol_Object (callback,transport)
  11 {
  12   ACE_DEBUG ((LM_DEBUG,
  13               "TimeStamp_Protocol_Object::TimeStamp_Protocol_Object\n"));
  14   ACE_NEW (this->frame_,
  15            ACE_Message_Block);
  16
  17   this->frame_->size (4 * this->transport_->mtu ());
  18 }
  19
  20
  21 int
  22 TimeStamp_Protocol_Object::handle_input ()
  23 {
  24   ssize_t n = this->transport_->recv (this->frame_->rd_ptr (),
  25                                       this->frame_->size ());
  26   if (n == -1)
  27     ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_UDP_Flow_Handler::handle_input recv failed\n"),-1);
  28   if (n == -1)
  29     ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_UDP_Flow_Handler::handle_input connection closed\n"),-1);
  30   this->frame_->wr_ptr (this->frame_->rd_ptr () + n);
  31
  32   return this->callback_->receive_frame (this->frame_);
  33 }
  34
  35 //  int
  36 //  TimeStamp_Protocol_Object::handle_input ()
  37 //  {
  38 //    iovec iov;
  39 //    int iovcnt;
  40 //    int n = this->transport_->recv (&iov, iovcnt);
  41
  42 //    int frame_size = BUFSIZ;
  43
  44 //    int begin = 1;
  45 //    ACE_Message_Block* prev;
  46 //    int iov_base = 0;
  47 //    while (n >= frame_size)
  48 //      {
  49 //        ACE_DEBUG ((LM_DEBUG,
  50 //                "(%N|%l) Frame Size %d %d\n",
  51 //                n,
  52 //                frame_size));
  53
  54 //        ACE_Message_Block* mb = 0;
  55 //        ACE_NEW_RETURN (mb,
  56 //                    ACE_Message_Block(frame_size),
  57 //                    -1);
  58
  59 //        ACE_OS::memmove (mb->rd_ptr (), iov.iov_base, frame_size);
  60 //        mb->wr_ptr (mb->rd_ptr () + frame_size);
  61
  62 //        //      iov_base += frame_size;
  63
  64 //        n -= frame_size;
  65
  66 //        if (begin)
  67 //      {
  68 //        prev = mb;
  69 //        this->frame_ = mb;
  70 //        begin = 0;
  71 //      }
  72 //        else
  73 //      {
  74 //        prev->cont (mb);
  75 //        prev = mb;
  76
  77 //      }
  78 //      }
  79
  80 //    if (n > 0)
  81 //      if (begin)
  82 //        {
  83 //      ACE_DEBUG ((LM_DEBUG,
  84 //                  "(%N|%l) Frame Size %d\n",
  85 //                    n));
  86 //      ACE_OS::memmove (this->frame_->rd_ptr (), iov.iov_base, n);
  87 //      this->frame_->wr_ptr (this->frame_->rd_ptr () + n);
  88 //        }
  89 //      else
  90 //        {
  91 //      ACE_DEBUG ((LM_DEBUG,
  92 //                  "(%N|%l) Frame Size %d\n",
  93 //                  n));
  94
  95 //      ACE_Message_Block* mb = 0;
  96 //      ACE_NEW_RETURN (mb,
  97 //                      ACE_Message_Block (frame_size),
  98 //                      -1);
  99
 100 //      ACE_OS::memmove (mb->rd_ptr (), iov.iov_base, n);
 101 //      mb->wr_ptr (mb->rd_ptr () + n);
 102 //      prev->cont (mb);
 103 //        }
 104
 105 //    ACE_DEBUG ((LM_DEBUG,
 106 //            "IOVEC SIZE %d %d\n",
 107 //            n,
 108 //            iov.iov_len));
 109
 110 //      if (n == -1)
 111 //      ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Flow_Handler::handle_input recv failed\n"),-1);
 112 //    if (n == 0)
 113 //      ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Flow_Handler::handle_input connection closed\n"),-1);
 114
 115 //    return this->callback_->receive_frame (this->frame_);
 116 //  }
 117
 118 /// send a data frame.
 119 int
 120 TimeStamp_Protocol_Object::send_frame (ACE_Message_Block *frame,
 121                                        TAO_AV_frame_info *)
 122 {
 123   ACE_DEBUG ((LM_DEBUG,
 124               "TimeStamp_Protocol_Object::send_frame\n"));
 125
 126   ACE_Message_Block* timestamp;
 127   ACE_NEW_RETURN (timestamp,
 128                   ACE_Message_Block (BUFSIZ),
 129                   -1);
 130
 131   ACE_hrtime_t now = ACE_OS::gethrtime ();
 132
 133   ACE_UINT64 usec = now;
 134   ACE_UINT32 val_1 = ACE_CU64_TO_CU32 (usec);
 135   ACE_DEBUG ((LM_DEBUG,
 136               "Time Stamp %u usecs\n",
 137               val_1));
 138
 139   ACE_OS::memcpy (timestamp->wr_ptr (), &now, sizeof (now));
 140   timestamp->wr_ptr (sizeof (now));
 141
 142   frame->cont (timestamp);
 143
 144   ssize_t result = this->transport_->send (frame);
 145   if (result < 0)
 146     return result;
 147   return 0;
 148 }
 149
 150 int
 151 TimeStamp_Protocol_Object::send_frame (iovec const* iov,
 152                                        int iovcnt,
 153                                        TAO_AV_frame_info *)
 154 {
 155   return this->transport_->send (iov,iovcnt);
 156 }
 157
 158 int
 159 TimeStamp_Protocol_Object::send_frame (const char* buf,
 160                                        size_t len)
 161 {
 162   int result = this->transport_->send (buf, len, 0);
 163   if (result < 0)
 164     return result;
 165   return 0;
 166 }
 167
 168 /// end the stream.
 169 int
 170 TimeStamp_Protocol_Object::destroy ()
 171 {
 172   this->callback_->handle_destroy ();
 173   return 0;
 174 }
 175
 176
 177 TimeStamp_Protocol_Factory::TimeStamp_Protocol_Factory ()
 178 {
 179 }
 180
 181 TimeStamp_Protocol_Factory::~TimeStamp_Protocol_Factory ()
 182 {
 183 }
 184
 185 int
 186 TimeStamp_Protocol_Factory::init (int, ACE_TCHAR **)
 187 {
 188   ACE_DEBUG ((LM_DEBUG,
 189               "TimeStamp_Protocol_Factory::init\n"));
 190   return 0;
 191 }
 192
 193 int
 194 TimeStamp_Protocol_Factory::match_protocol (const char *flow_string)
 195 {
 196   ACE_DEBUG ((LM_DEBUG,
 197               "TimeStamp_Protocol_Factory::match_protocol\n"));
 198   if (ACE_OS::strcasecmp (flow_string,"TS") == 0)
 199     return 1;
 200   return 0;
 201 }
 202
 203 TAO_AV_Protocol_Object*
 204 TimeStamp_Protocol_Factory::make_protocol_object (TAO_FlowSpec_Entry *entry,
 205                                                   TAO_Base_StreamEndPoint *endpoint,
 206                                                   TAO_AV_Flow_Handler *handler,
 207                                                   TAO_AV_Transport *transport)
 208 {
 209   TAO_AV_Callback *callback = 0;
 210   endpoint->get_callback (entry->flowname (),
 211                           callback);
 212
 213
 214   TimeStamp_Protocol_Object *object = 0;
 215   ACE_NEW_RETURN (object,
 216                   TimeStamp_Protocol_Object (callback,
 217                                              transport),
 218                   0);
 219   callback->open (object,
 220                   handler);
 221   endpoint->set_protocol_object (entry->flowname (),
 222                                  object);
 223   return object;
 224 }
 225
 226 ACE_FACTORY_DEFINE (TAO_TS, TimeStamp_Protocol_Factory)
 227 ACE_STATIC_SVC_DEFINE (TimeStamp_Protocol_Factory,
 228                        ACE_TEXT ("TimeStamp_Protocol_Factory"),
 229                        ACE_SVC_OBJ_T,
 230                        &ACE_SVC_NAME (TimeStamp_Protocol_Factory),
 231                        ACE_Service_Type::DELETE_THIS |
 232                        ACE_Service_Type::DELETE_OBJ,
 233                        0)
 234
 235
 236
 237
 238