ACE/protocols/ace/RMCast/Reassemble.cpp

   1 // author    : Boris Kolpackov <boris@kolpackov.net>
   2 #include "Reassemble.h"
   3 #include "ace/OS_NS_stdlib.h"
   4
   5 namespace ACE_RMCast
   6 {
   7   Reassemble::
   8   Reassemble (Parameters const& )
   9   // : params_ (params)
  10   {
  11   }
  12
  13   void Reassemble::recv (Message_ptr m)
  14   {
  15     Map::ENTRY* e = 0;
  16     Address from (
  17       static_cast<From const*> (m->find (From::id))->address ());
  18
  19     if (Data const* data = static_cast<Data const*> (m->find (Data::id)))
  20     {
  21       if (Part const* part = static_cast<Part const*> (m->find (Part::id)))
  22       {
  23         if (map_.find (from, e) == -1)
  24         {
  25           // First part of the message.
  26           //
  27
  28           if (part->num () != 1)
  29           {
  30             // We assume that we received NoData for one of the preceding
  31             // fragments. Ignore this one.
  32             return;
  33           }
  34
  35           Data_ptr new_data (new Data (data->buf (),
  36                                        static_cast<size_t> (data->size ()),
  37                                        static_cast<size_t> (part->total_size ())));
  38
  39           //std::cerr << "part->total_size (): " << part->total_size () << endl;
  40
  41           map_.bind (from, new_data);
  42         }
  43         else
  44         {
  45           // Next part of the message.
  46           //
  47
  48           if (part->num () == 1)
  49             ACE_OS::abort ();
  50
  51
  52           Data const* data = static_cast<Data const*> (m->find (Data::id));
  53
  54           Data_ptr& new_data = e->int_id_;
  55
  56           ACE_OS::memcpy (new_data->buf () + new_data->size (),
  57                           data->buf (),
  58                           data->size ());
  59
  60           //std::cerr << "data->size (): " << data->size () << endl
  61           //          << "new_data->size (): " << new_data->size () << endl
  62           //          << "new_data->capa (): " << new_data->capacity () << endl;
  63
  64           new_data->size (new_data->size () + data->size ());
  65
  66
  67           if (part->num () == part->of ())
  68           {
  69             // Reassembly is complete.
  70             //
  71             if (part->total_size () != new_data->size ())
  72               ACE_OS::abort ();
  73
  74             Message_ptr new_msg (new Message ());
  75
  76             Address to (
  77               static_cast<To const*> (m->find (To::id))->address ());
  78
  79             new_msg->add (Profile_ptr (new To (to)));
  80             new_msg->add (Profile_ptr (new From (from)));
  81             /*
  82              * Heads up... we need to add the new_data to new_msg then
  83              * unbind the entry that maps to new_data, which will decrement
  84              * its reference count. If the bound/refcounted pointer acted
  85              * polymorphically like a regular pointer does, we'd be able to
  86              * just pass new_data to add(Profile_Ptr) and it would work.
  87              * However, Profile_Ptr and Data_Ptr are not compatible, but
  88              * we can use the secret knowledge that both are instances of the
  89              * same template and that the pointers they contain really are
  90              * hierarchically compatible, and do this funky cast to get
  91              * the result we want.
  92              */
  93             //new_msg->add (*(reinterpret_cast<Profile_ptr*> (&new_data)));
  94
  95             new_msg->add (Profile_ptr (new_data));
  96
  97             map_.unbind (from);
  98
  99             in_->recv (new_msg);
 100           }
 101         }
 102       }
 103       else
 104       {
 105         // Non-fragmented message. Make sure we are in the consistent state
 106         // and forward it up.
 107         //
 108         if (map_.find (from, e) != -1)
 109           ACE_OS::abort ();
 110
 111         in_->recv (m);
 112       }
 113     }
 114     else if (m->find (NoData::id) != 0)
 115     {
 116       if (map_.find (from, e) != -1)
 117       {
 118         // We already received some fragments. Clean everyhting up.
 119         //
 120         map_.unbind (from);
 121       }
 122
 123       in_->recv (m);
 124     }
 125   }
 126 }