clientsubscription.cxx

这是国外的resip协议栈

#include <queue>

#include "resip/stack/Helper.hxx"
#include "rutil/Logger.hxx"
#include "resip/stack/SipFrag.hxx"
#include "resip/stack/SipMessage.hxx"
#include "resip/dum/ClientSubscription.hxx"
#include "resip/dum/Dialog.hxx"
#include "resip/dum/DialogUsageManager.hxx"
#include "resip/dum/SubscriptionHandler.hxx"
#include "resip/dum/SubscriptionCreator.hxx"
#include "resip/dum/UsageUseException.hxx"

#include "resip/dum/AppDialogSet.hxx"

using namespace resip;

#define RESIPROCATE_SUBSYSTEM Subsystem::DUM


ClientSubscription::ClientSubscription(DialogUsageManager& dum, Dialog& dialog,
                                       const SipMessage& request, UInt32 defaultSubExpiration)
   : BaseSubscription(dum, dialog, request),
     mOnNewSubscriptionCalled(mEventType == "refer"),  // don't call onNewSubscription for Refer subscriptions
     mEnded(false),
     mExpires(0),
     mDefaultExpires(defaultSubExpiration),
     mRefreshing(false),
     mHaveQueuedRefresh(false),
     mQueuedRefreshInterval(-1),
     mLargestNotifyCSeq(0)
{
   DebugLog (<< "ClientSubscription::ClientSubscription from " << request.brief());   
   mDialog.makeRequest(*mLastRequest, SUBSCRIBE);
}

ClientSubscription::~ClientSubscription()
{
   mDialog.mClientSubscriptions.remove(this);

   while (!mQueuedNotifies.empty())
   {
      delete mQueuedNotifies.front();
      mQueuedNotifies.pop_front();
   }

   clearDustbin();
}

ClientSubscriptionHandle 
ClientSubscription::getHandle()
{
   return ClientSubscriptionHandle(mDum, getBaseHandle().getId());
}

void
ClientSubscription::dispatch(const SipMessage& msg)
{
   DebugLog (<< "ClientSubscription::dispatch " << msg.brief());
   
   ClientSubscriptionHandler* handler = mDum.getClientSubscriptionHandler(mEventType);
   assert(handler);

   clearDustbin();

   // asserts are checks the correctness of Dialog::dispatch
   if (msg.isRequest() )
   {
      assert( msg.header(h_RequestLine).getMethod() == NOTIFY );
      mRefreshing = false;


      // !dlb! 481 NOTIFY iff state is dead?

      //!dcm! -- heavy, should just store enough information to make response
      //mLastNotify = msg;

      if (!mOnNewSubscriptionCalled && !getAppDialogSet()->isReUsed())
      {
         InfoLog (<< "[ClientSubscription] " << mLastRequest->header(h_To));
         if (msg.exists(h_Contacts))
         {
            mDialog.mRemoteTarget = msg.header(h_Contacts).front();
         }
         
         handler->onNewSubscription(getHandle(), msg);
         mOnNewSubscriptionCalled = true;
      }         

      bool outOfOrder = mLargestNotifyCSeq > msg.header(h_CSeq).sequence();
      if (!outOfOrder)
      {
         mLargestNotifyCSeq = msg.header(h_CSeq).sequence();
      }
      else
      {
         DebugLog(<< "received out of order notify");
      }

      mQueuedNotifies.push_back(new QueuedNotify(msg, outOfOrder));
      if (mQueuedNotifies.size() == 1)
      {
         DebugLog(<< "no queued notify");
         processNextNotify();
         return;
      }
      else
      {
         DebugLog(<< "Notify gets queued");
      }
   }
   else
   {
      DebugLog(<< "processing client subscription response");
      processResponse(msg);
   }
}

void 
ClientSubscription::processResponse(const SipMessage& msg)
{
   ClientSubscriptionHandler* handler = mDum.getClientSubscriptionHandler(mEventType);
   assert(handler);

   mRefreshing = false;

   if (msg.header(h_StatusLine).statusCode() >= 200 && msg.header(h_StatusLine).statusCode() <300)
   {
      if (msg.exists(h_Expires))
      {
         // grab the expires from the 2xx in case there is not one on the NOTIFY .mjf.
         UInt32 expires = msg.header(h_Expires).value();
         UInt32 lastExpires = mLastRequest->header(h_Expires).value();
         if (expires < lastExpires)
         {
            mLastRequest->header(h_Expires).value() = expires;
         }
      }
      sendQueuedRefreshRequest();
   }
   else if (msg.header(h_StatusLine).statusCode() == 481 &&
            msg.exists(h_Expires) && msg.header(h_Expires).value() > 0)
   {
      InfoLog (<< "Received 481 to SUBSCRIBE, reSUBSCRIBEing (presence server probably restarted) "
               << mDialog.mRemoteTarget);

      SharedPtr<SipMessage> sub = mDum.makeSubscription(mDialog.mRemoteTarget, getEventType(), getAppDialogSet()->reuse());
      mDum.send(sub);

      delete this;
      return;
   }
   else if (msg.header(h_StatusLine).statusCode() == 408 ||
            ((msg.header(h_StatusLine).statusCode() == 413 ||
              msg.header(h_StatusLine).statusCode() == 480 ||
              msg.header(h_StatusLine).statusCode() == 486 ||
              msg.header(h_StatusLine).statusCode() == 500 ||
              msg.header(h_StatusLine).statusCode() == 503 ||
              msg.header(h_StatusLine).statusCode() == 600 ||
              msg.header(h_StatusLine).statusCode() == 603) &&
             msg.exists(h_RetryAfter)))
   {
      int retry;

      if (msg.header(h_StatusLine).statusCode() == 408)
      {
         InfoLog (<< "Received 408 to SUBSCRIBE "
                  << mLastRequest->header(h_To));
         retry = handler->onRequestRetry(getHandle(), 0, msg);
      }
      else
      {
         InfoLog (<< "Received non-408 retriable to SUBSCRIBE "
                  << mLastRequest->header(h_To));
         retry = handler->onRequestRetry(getHandle(), msg.header(h_RetryAfter).value(), msg);
      }

      if (retry < 0)
      {
         DebugLog(<< "Application requested failure on Retry-After");
         mEnded = true;
         handler->onTerminated(getHandle(), msg);
         delete this;
         return;
      }
      else if (retry == 0)
      {
         DebugLog(<< "Application requested immediate retry on Retry-After");
         //!dcm! -- why isn't this just a refresh--retry after might be
         //middle element and not indicate dialog destruction       
         if (mDialog.mRemoteTarget.uri().host().empty())
         {
            SharedPtr<SipMessage> sub = mDum.makeSubscription(mLastRequest->header(h_To), getEventType());
            mDum.send(sub);
         }
         else
         {
            SharedPtr<SipMessage> sub = mDum.makeSubscription(mDialog.mRemoteTarget, getEventType(), getAppDialogSet()->reuse());
            mDum.send(sub);
         }
         //!dcm! -- new sub created above, when does this usage get destroyed?
         //return;
      }
      else 
      {
         // leave the usage around until the timeout
         // !dlb! would be nice to set the state to something dead, but not used
         mDum.addTimer(DumTimeout::SubscriptionRetry, 
                       retry, 
                       getBaseHandle(),
                       ++mTimerSeq);
         // leave the usage around until the timeout
         return;
      }
            
      delete this;
      return;
   }
   else if (msg.header(h_StatusLine).statusCode() >= 300)
   {
      if (msg.header(h_StatusLine).statusCode() == 423 
          && msg.exists(h_MinExpires))
      {
         requestRefresh(msg.header(h_MinExpires).value());            
      }
      else
      {
         mEnded = true;
         handler->onTerminated(getHandle(), msg);
         delete this;
         return;
      }
   }
}

void 
ClientSubscription::processNextNotify()
{
   //!dcm! There is a timing issue in this code which can cause this to be
   //!called when there are no queued NOTIFY messages. Probably a subscription
   //!teardown/timer crossover.
   //assert(!mQueuedNotifies.empty());
   if (mQueuedNotifies.empty())
   {
      return;
   }

   QueuedNotify* qn = mQueuedNotifies.front();
   ClientSubscriptionHandler* handler = mDum.getClientSubscriptionHandler(mEventType);
   assert(handler);

   unsigned long refreshInterval = 0;
   if (!qn->outOfOrder())
   {
      UInt32 expires = 0;
      //default to 3600 seconds so non-compliant endpoints don't result in leaked usages
      if (qn->notify().exists(h_SubscriptionState) && qn->notify().header(h_SubscriptionState).exists(p_expires))
      {
         expires = qn->notify().header(h_SubscriptionState).param(p_expires);
      }
      else if (mLastRequest->exists(h_Expires))
      {
         expires = mLastRequest->header(h_Expires).value();
      }
      else if (mDefaultExpires)
      {
         /* if we haven't gotten an expires value from:
            1. the subscription state from this notify
            2. the last request
            then use the default expires (meaning it came from the 2xx in response
            to the initial SUBSCRIBE). .mjf.
          */
         expires = mDefaultExpires;
      }
      else
      {
         expires = 3600;
      }
      
      if (!mLastRequest->exists(h_Expires))
      {
         DebugLog(<< "No expires header in last request, set to " << expires);
         mLastRequest->header(h_Expires).value() = expires;
      }
      UInt64 now = Timer::getTimeSecs();
      
      if (mExpires == 0 || now + expires < mExpires)
      {
         refreshInterval = Helper::aBitSmallerThan((unsigned long)expires);
         mExpires = now + refreshInterval;
      }
   }
   //if no subscription state header, treat as an extension. Only allow for
   //refer to handle non-compliant implementations
   if (!qn->notify().exists(h_SubscriptionState))
   {
      if (qn->notify().exists(h_Event) && qn->notify().header(h_Event).value() == "refer")
      {
         SipFrag* frag  = dynamic_cast<SipFrag*>(qn->notify().getContents());
         if (frag)
         {
            if (frag->message().isResponse())
            {
               int code = frag->message().header(h_StatusLine).statusCode();
               if (code < 200)
               {
                  handler->onUpdateExtension(getHandle(), qn->notify(), qn->outOfOrder());
               }
               else
               {
                  acceptUpdate();
                  mEnded = true;                     
                  handler->onTerminated(getHandle(), qn->notify());
                  delete this;
               }
            }
         }
         else
         {
            acceptUpdate();
            mEnded = true;
            handler->onTerminated(getHandle(), qn->notify());
            delete this;
         }
      }
      else
      {            
         mDialog.makeResponse(*mLastResponse, qn->notify(), 400);
         mLastResponse->header(h_StatusLine).reason() = "Missing Subscription-State header";
         send(mLastResponse);
         mEnded = true;
         handler->onTerminated(getHandle(), qn->notify());
         delete this;
      }
      return;
   }

   if (!mEnded && isEqualNoCase(qn->notify().header(h_SubscriptionState).value(), Symbols::Active))
   {
      if (refreshInterval)
      {
         mDum.addTimer(DumTimeout::Subscription, refreshInterval, getBaseHandle(), ++mTimerSeq);
         InfoLog (<< "[ClientSubscription] reSUBSCRIBE in " << refreshInterval);
      }
         
      handler->onUpdateActive(getHandle(), qn->notify(), qn->outOfOrder());
   }
   else if (!mEnded && isEqualNoCase(qn->notify().header(h_SubscriptionState).value(), Symbols::Pending))
   {
      if (refreshInterval)
      {
         mDum.addTimer(DumTimeout::Subscription, refreshInterval, getBaseHandle(), ++mTimerSeq);
         InfoLog (<< "[ClientSubscription] reSUBSCRIBE in " << refreshInterval);
      }

      handler->onUpdatePending(getHandle(), qn->notify(), qn->outOfOrder());
   }
   else if (isEqualNoCase(qn->notify().header(h_SubscriptionState).value(), Symbols::Terminated))
   {
      acceptUpdate();
      mEnded = true;
      handler->onTerminated(getHandle(), qn->notify());
      DebugLog (<< "[ClientSubscription] " << mLastRequest->header(h_To) << "[ClientSubscription] Terminated");                   
      delete this;
      return;
   }
   else if (!mEnded)
   {
      handler->onUpdateExtension(getHandle(), qn->notify(), qn->outOfOrder());
   }
}

void
ClientSubscription::dispatch(const DumTimeout& timer)
{
   if (timer.seq() == mTimerSeq)