MpFlowGraphBase.h

Go to the documentation of this file.

//  
// Copyright (C) 2006-2013 SIPez LLC.  All rights reserved.
//
// Copyright (C) 2004-2008 SIPfoundry Inc.
// Licensed by SIPfoundry under the LGPL license.
//
// Copyright (C) 2004-2006 Pingtel Corp.  All rights reserved.
// Licensed to SIPfoundry under a Contributor Agreement.
//
// $$


#ifndef _MpFlowGraphBase_h_
#define _MpFlowGraphBase_h_

// SYSTEM INCLUDES

#include "rtcp/RtcpConfig.h"

// APPLICATION INCLUDES
#include <utl/UtlHashMap.h>

#include "utl/UtlString.h"
#include "os/OsDefs.h"
#include "os/OsMsgQ.h"
#include "os/OsStatus.h"
#include "os/OsRWMutex.h"
#include "os/OsTime.h"
#include "mp/MpResource.h"
#include "os/OsMsgDispatcher.h"
#include "mp/MpResNotificationMsg.h"
#ifdef INCLUDE_RTCP /* [ */
#include "rtcp/RTCManager.h"
#endif /* INCLUDE_RTCP ] */

// DEFINES
// MACROS
// EXTERNAL FUNCTIONS
// EXTERNAL VARIABLES
// CONSTANTS
// STRUCTS
// TYPEDEFS

// FORWARD DECLARATIONS
class MpFlowGraphMsg;
class OsMsg;

class MpFlowGraphBase
#ifdef INCLUDE_RTCP /* [ */
                      : public CBaseClass,
                        public IRTCPNotify
#endif /* INCLUDE_RTCP ] */
{
/* //////////////////////////// PUBLIC //////////////////////////////////// */
public:
   friend OsStatus MpResource::postMessage(MpFlowGraphMsg& rMsg);

   typedef enum
   {
      STARTED,   
      STOPPED    
   } FlowGraphState;

   typedef enum 
   {
      BASE_FLOWGRAPH       = 0,
      CALL_FLOWGRAPH       = 1,
      TOPOLOGY_FLOWGRAPH   = 2
   } FlowGraphType;
   //: FlowGraph Types

/* ============================ CREATORS ================================== */


   MpFlowGraphBase(int samplesPerFrame = 0, 
                   int samplesPerSec = 0, 
                   OsMsgDispatcher *pNotifDispatcher = NULL);

   virtual
   ~MpFlowGraphBase();


/* ============================ MANIPULATORS ============================== */

   OsStatus addLink(MpResource& rFrom, int outPortIdx,
                    MpResource& rTo,   int inPortIdx);
   OsMsgDispatcher*  
   setNotificationDispatcher(OsMsgDispatcher* notifyDispatcher);
   OsStatus addResource(MpResource& rResource, UtlBoolean makeNameUnique=TRUE);
   OsStatus destroyResources(void);
   OsStatus disable(void);
   OsStatus enable(void);
   virtual OsStatus gainFocus(void);
   OsStatus insertResourceAfter(MpResource& rResource,
                                MpResource& rUpstreamResource,
                                int outPortIdx);
   OsStatus insertResourceBefore(MpResource& rResource,
                                 MpResource& rDownstreamResource,
                                 int inPortIdx);
   virtual OsStatus loseFocus(void);
   virtual OsStatus postNotification(const MpResNotificationMsg& msg);
   virtual OsStatus processNextFrame(void);
   OsStatus removeLink(MpResource& rFrom, int outPortIdx);
   OsStatus removeResource(MpResource& rResource);
   OsStatus destroyResource(const UtlString& resourceName);
   OsStatus setNotificationsEnabled(bool enabled, 
                                    const UtlString& resourceName = NULL);

   OsStatus start(void);
   OsStatus stop(void);

   void synchronize(const char* tag=NULL, int val=0);

/* ============================ ACCESSORS ================================= */

   static void flowGraphInfo(MpFlowGraphBase* pFlowGraph);


   virtual FlowGraphType getType();

   inline int getSamplesPerFrame(void) const;

   inline int getSamplesPerSec(void) const;

   inline int getState(void) const;

   inline OsMsgDispatcher* getNotificationDispatcher() const;

   OsStatus lookupResource(const UtlString& name,
                           MpResource*& rpResource);
   inline int numLinks(void) const;

   inline int numFramesProcessed(void) const;

   inline int numResources(void) const;

   inline OsMsgQ* getMsgQ(void) ;

   inline int getFlowgraphNum() const;

   inline UtlString getFlowgraphName() const;

   OsStatus getLatencyForPath(MpResource *pStartResource,
                              int startResourceInput,
                              const UtlString &endResourceName,
                              int endResourceOutput,
                              UtlBoolean includeEndResourceLatency,
                              int &latency);
   OsStatus getLatencyForPathReverse(MpResource *pStartResource,
                                     int startResourceOutput,
                                     const UtlString &endResourceName,
                                     int endResourceInput,
                                     UtlBoolean includeEndResourceLatency,
                                     int &latency);
#ifdef INCLUDE_RTCP /* [ */
   IRTCPSession* getRTCPSessionPtr(void);

   IRTCPConnection* getRTCPConnectionPtr(MpConnectionID connId, int mediaType, int streamId);

#endif /* INCLUDE_RTCP ] */

   


/* ============================ INQUIRY =================================== */


   UtlBoolean isStarted(void) const;

/* ============================ CALLBACKS ================================= */
#ifdef INCLUDE_RTCP /* [ */

    unsigned long GetEventInterest(void);

    void LocalSSRCCollision(IRTCPConnection    *piRTCPConnection,
                            IRTCPSession       *piRTCPSession);


    void RemoteSSRCCollision(IRTCPConnection    *piRTCPConnection,
                             IRTCPSession       *piRTCPSession);


DECLARE_IBASE_M
#endif /* INCLUDE_RTCP ] */
/* //////////////////////////// PROTECTED ///////////////////////////////// */
protected:
   OsRWMutex        mRWMutex;      // reader/writer lock for synchronization

   virtual OsStatus postMessage(const MpFlowGraphMsg& rMsg,
                                const OsTime& rTimeout=OsTime::OS_INFINITY);
   virtual UtlBoolean handleMessage(OsMsg& rMsg);
   UtlBoolean handleRemoveLink(MpResource* pFrom, int outPortIdx);
   UtlBoolean handleRemoveResource(MpResource* pResource);
   UtlBoolean handleDestroyResource(const UtlString &name);
   UtlBoolean handleSynchronize(MpFlowGraphMsg& rMsg);
/* //////////////////////////// PRIVATE /////////////////////////////////// */
private:

   enum { MAX_FLOWGRAPH_MESSAGES  = 150};
   enum { MAX_FLOWGRAPH_RESOURCES = 50};

   static int       gFgMaxNumber; 
   int              mFgNumber; 

   UtlHashMap mResourceDict;   
   MpResource* mExecOrder[MAX_FLOWGRAPH_RESOURCES]; 
   MpResource* mUnsorted[MAX_FLOWGRAPH_RESOURCES];  
   int       mCurState;        
   OsMsgQ    mMessages;        
   OsMsgDispatcher* mNotifyDispatcher; 
   int       mPeriodCnt;       
   int       mLinkCnt;         
   int       mResourceCnt;     
   UtlBoolean mRecomputeOrder; 
   int       mSamplesPerFrame; 
   int       mSamplesPerSec;   
   MpResource* mpResourceInProcess; 
   static const OsTime smProcessMessagesTimeout; 

#ifdef INCLUDE_RTCP /* [ */
   int hashRtcpConnection(MpConnectionID connId, int mediaType, int streamId);
   UtlBoolean createRtcpConnection(MpConnectionID connId, int mediaType, int streamId);
   UtlBoolean deleteRtcpConnection(MpConnectionID connId, int mediaType, int streamId);

   IRTCPSession* mpiRTCPSession;
   unsigned long mulEventInterest;
   OsMutex mRtcpConnMutex;
   UtlHashMap mRtcpConnMap;    
#endif /* INCLUDE_RTCP ] */


   OsStatus computeOrder(void);
   int getExecOrderIndex(const MpResource* resource) const;
   UtlBoolean disconnectAllInputs(MpResource* pResource);
   UtlBoolean disconnectAllOutputs(MpResource* pResource);
   UtlBoolean handleAddLink(MpResource* pFrom, int outPortIdx,
                            MpResource* pTo,   int inPortIdx);
   UtlBoolean handleAddResource(MpResource* pResource,
                               UtlBoolean makeNameUnique);
   UtlBoolean handleDestroyResources(void);
   UtlBoolean handleDisable(void);
   UtlBoolean handleEnable(void);
   UtlBoolean handleStart(void);
   UtlBoolean handleStop(void);
   OsStatus handleGetLatencyForPath(MpResource *pStartResource,
                                    int startResourceInput,
                                    const UtlString &endResourceName,
                                    int endResourceOutput,
                                    UtlBoolean includeEndResourceLatency,
                                    int &latency);

   OsStatus handleGetLatencyForPathReverse(MpResource *pStartResource,
                                           int startResourceOutput,
                                           const UtlString &endResourceName,
                                           int endResourceInput,
                                           UtlBoolean includeEndResourceLatency,
                                           int &latency);

   OsStatus lookupResourcePrivate(const UtlString& name,
                                  MpResource*& rpResource);
   OsStatus processMessages(void);
   MpFlowGraphBase(const MpFlowGraphBase& rMpFlowGraph);

   MpFlowGraphBase& operator=(const MpFlowGraphBase& rhs);

};

/* ============================ INLINE METHODS ============================ */

int MpFlowGraphBase::getSamplesPerFrame(void) const
{
   return mSamplesPerFrame;
}

int MpFlowGraphBase::getSamplesPerSec(void) const
{
   return mSamplesPerSec;
}

int MpFlowGraphBase::getState(void) const
{
   return mCurState;
}

OsMsgDispatcher* MpFlowGraphBase::getNotificationDispatcher(void) const
{
   return mNotifyDispatcher;
}

int MpFlowGraphBase::numLinks(void) const
{
   return mLinkCnt;
}

int MpFlowGraphBase::numFramesProcessed(void) const
{
   return mPeriodCnt;
}

int MpFlowGraphBase::numResources(void) const
{
   return mResourceCnt;
}

OsMsgQ* MpFlowGraphBase::getMsgQ(void) 
{
   return &mMessages ;
}

int MpFlowGraphBase::getFlowgraphNum() const
{
   return mFgNumber;
}

UtlString MpFlowGraphBase::getFlowgraphName() const
{
   char tmpStr[10];
   snprintf(tmpStr, sizeof(tmpStr), "FG%d", getFlowgraphNum());
   return tmpStr;
}

#ifdef INCLUDE_RTCP /* [ */
inline IRTCPSession *MpFlowGraphBase::getRTCPSessionPtr(void)
{
    return(mpiRTCPSession);
}



inline unsigned long MpFlowGraphBase::GetEventInterest(void)
{

    return(mulEventInterest);
}
#endif /* INCLUDE_RTCP ] */

#endif  // _MpFlowGraphBase_h_