Google Git
Sign in
android / platform / external / opencore / 94d15606702ba7469bbb71739df707bfce1c4c3a / . / nodes / pvsocketnode / src / pvmf_socket_node.cpp
blob: f6da7ef790912883d58e843b5c59259bc63e35b4 [file] [log] [blame]
/* ------------------------------------------------------------------
* Copyright (C) 1998-2009 PacketVideo
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
* express or implied.
* See the License for the specific language governing permissions
* and limitations under the License.
* -------------------------------------------------------------------
*/
#include "pvmf_socket_node.h"
#include "oscl_string_utils.h"
#include "pvmf_basic_errorinfomessage.h"
#include "pvmf_errorinfomessage_extension.h"
#include "pvmf_socket_node_events.h"
#include "oscl_byte_order.h"
#include "pvmf_protocol_engine_command_format_ids.h"
#include "pvmf_media_cmd.h"
#include "pvmf_media_msg_format_ids.h"
#include "oscl_tickcount.h"
#include "oscl_rand.h"
#include "oscl_time.h"
#include "pvmf_socket_node_tunables.h"
#include "oscl_bin_stream.h"
// Use default DLL entry point for Symbian
#include "oscl_dll.h"
//////////////////////////////////////////////////
// PVMFSocketNodeMemPool
//////////////////////////////////////////////////
PVMFSocketNodeMemPool::PVMFSocketNodeMemPool(uint32 aMemPoolNumBufs)
: iMediaDataMemPool(NULL)
{
iMediaDataMemPool = OSCL_NEW(OsclMemPoolFixedChunkAllocator, (aMemPoolNumBufs, MEDIA_DATA_CLASS_SIZE));
iInternalAlloc = NULL;
iSharedBufferAlloc = NULL;
iMediaMsgAllocator = NULL;
iMediaFragGroupAlloc = NULL;
iMediaFragGroupAllocMempool = NULL;
}
void PVMFSocketNodeMemPool::CreateAllocators(const OSCL_HeapString<OsclMemAllocator>& iMime, uint32 aSize, uint32 aExpectedNumberOfBlocksPerBuffer, uint32 aResizeSize, uint32 aMaxNumResizes)
{
if (!iSharedBufferAlloc)
{
iSharedDataBufferInfo.Init(aSize, aExpectedNumberOfBlocksPerBuffer, aResizeSize, aMaxNumResizes);
const uint32 maxAllocatorNameLen = 255;
char allocatorName[maxAllocatorNameLen];
const uint32 strLenForMimeName = maxAllocatorNameLen - oscl_strlen("InternalSocketBuffer");
oscl_strncpy(allocatorName, iMime.get_cstr(), strLenForMimeName);
allocatorName[strLenForMimeName] = '\0';
oscl_strcat(allocatorName, "InternalSocketBuffer");
CreateDefaultDataBufferAllocator(allocatorName);
}
#if SNODE_ENABLE_UDP_MULTI_PACKET
CreateUDPMultipleRecvAllocator();
#endif
}
OsclMemPoolResizableAllocator* PVMFSocketNodeMemPool::CreateResizableDataBufferAllocator(const char* allocatorName)
{
OSCL_UNUSED_ARG(allocatorName);
if (iMediaMsgAllocator)
{
/*
* It may be possible that the allocator was binded to iInternalAlloc/iSharedBufferAlloc
* So destruct it.
* Deallocation of iInternalAlloc/iSharedBufferAlloc will be done as part of Reset/Destruction of node
*/
OSCL_DELETE(iMediaMsgAllocator);
iMediaMsgAllocator = NULL;
}
OSCL_ASSERT(iSharedBufferAlloc == NULL);
if (iSharedBufferAlloc)
return NULL;
iSharedBufferAlloc = OSCL_NEW(OsclMemPoolResizableAllocator, (iSharedDataBufferInfo.iSize, (1 + iSharedDataBufferInfo.iMaxNumResizes),
iSharedDataBufferInfo.iExpectedNumberOfBlocksPerBuffer));
iSharedBufferAlloc->setMaxSzForNewMemPoolBuffer(iSharedDataBufferInfo.iResizeSize);
iMediaMsgAllocator = OSCL_NEW(PVMFResizableSimpleMediaMsgAlloc, (iSharedBufferAlloc));
return iSharedBufferAlloc;
}
void PVMFSocketNodeMemPool::CreateDefaultDataBufferAllocator(const char* allocatorName)
{
OSCL_UNUSED_ARG(allocatorName);
if (iMediaMsgAllocator)
{
/*
* It may be possible that the allocator was binded to iInternalAlloc/iSharedBufferAlloc
* So destruct it.
* Deallocation of iInternalAlloc/iSharedBufferAlloc will be done as part of Reset/Destruction of node
*/
OSCL_DELETE(iMediaMsgAllocator);
iMediaMsgAllocator = NULL;
}
OSCL_ASSERT(iInternalAlloc == NULL);
if (iInternalAlloc)
return;
iInternalAlloc = OSCL_NEW(OsclMemPoolResizableAllocator, (iSharedDataBufferInfo.iSize, (1 + iSharedDataBufferInfo.iMaxNumResizes),
iSharedDataBufferInfo.iExpectedNumberOfBlocksPerBuffer));
iInternalAlloc->setMaxSzForNewMemPoolBuffer(iSharedDataBufferInfo.iResizeSize);
iMediaMsgAllocator = OSCL_NEW(PVMFResizableSimpleMediaMsgAlloc, (iInternalAlloc));
}
void PVMFSocketNodeMemPool::CreateUDPMultipleRecvAllocator()
{
#if SNODE_ENABLE_UDP_MULTI_PACKET
//Create the multiple-receive allocator
iMediaFragGroupAllocMempool
= OSCL_NEW(OsclMemPoolFixedChunkAllocator
, (SNODE_UDP_MULTI_FRAG_ALLOCATOR_MAX_MSGS)
);
iMediaFragGroupAlloc
= OSCL_NEW(PVMFMediaFragGroupCombinedAlloc<PVMFSocketNodeAllocator>
, (SNODE_UDP_MULTI_FRAG_ALLOCATOR_MAX_MSGS
, SNODE_UDP_MULTI_FRAG_ALLOCATOR_MAX_FRAGS_PER_MSG
, iMediaFragGroupAllocMempool)
);
iMediaFragGroupAlloc->create();
#endif
}
uint32 PVMFSocketNodeMemPool::GetMaxSizeMediaMsgLen()
{
OsclMemPoolResizableAllocator* resizableAllocator = NULL;
const uint32 wrappingOverhead = iMediaMsgAllocator->GetMediaMsgAllocationOverheadBytes();
if (iSharedBufferAlloc)
{
resizableAllocator = OSCL_STATIC_CAST(OsclMemPoolResizableAllocator*, iSharedBufferAlloc);
}
else if (iInternalAlloc)
{
resizableAllocator = OSCL_STATIC_CAST(OsclMemPoolResizableAllocator*, iInternalAlloc);
}
if (resizableAllocator)
{
return (resizableAllocator->getLargestContiguousFreeBlockSize() - wrappingOverhead);
}
return 0;
}
void PVMFSocketNodeMemPool::DestroyAllocators()
{
if (iInternalAlloc != NULL)
{
iInternalAlloc->CancelFreeChunkAvailableCallback();
iInternalAlloc->removeRef();
}
if (iSharedBufferAlloc != NULL)
{
iSharedBufferAlloc->CancelFreeChunkAvailableCallback();
iSharedBufferAlloc->removeRef();
}
if (iMediaMsgAllocator)
{
OSCL_DELETE(iMediaMsgAllocator);
iMediaMsgAllocator = NULL;
}
#if SNODE_ENABLE_UDP_MULTI_PACKET
DestroyUDPMultipleRecvAllocator();
#endif
}
void PVMFSocketNodeMemPool::DestroyUDPMultipleRecvAllocator()
{
#if SNODE_ENABLE_UDP_MULTI_PACKET
if (iMediaFragGroupAlloc)
{
iMediaFragGroupAlloc->removeRef();
iMediaFragGroupAlloc = NULL;
}
if (iMediaFragGroupAllocMempool)
{
iMediaFragGroupAllocMempool->removeRef();
iMediaFragGroupAllocMempool = NULL;
}
#endif
}
//////////////////////////////////////////////////
// SocketNodeStats
//////////////////////////////////////////////////
#if(ENABLE_SOCKET_NODE_STATS)
void SocketNodeStats::Log(PVMFPortVector<PVMFSocketPort, PVMFSocketNodeAllocator>& aPortVec)
{
if (iLogged)
return;//prevent multiple logging.
if (!iLogger)
return;//prevent logging after thread logoff.
iLogged = true;
PVLOGGER_LOGMSG(PVLOGMSG_INST_PROF, iLogger, PVLOGMSG_ERR,
(0, "SocketNodeStats:@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@"));
PVLOGGER_LOGMSG(PVLOGMSG_INST_PROF, iLogger, PVLOGMSG_ERR,
(0, "SocketNodeStats: %8d Num Run", iNumRun));
PVLOGGER_LOGMSG(PVLOGMSG_INST_PROF, iLogger, PVLOGMSG_ERR,
(0, "SocketNodeStats: %8d Num QueueNodeCmd", iNumQueueNodeCmd));
PVLOGGER_LOGMSG(PVLOGMSG_INST_PROF, iLogger, PVLOGMSG_ERR,
(0, "SocketNodeStats: %8d Num Bind", iNumBind));
for (uint32 i = 0; i < aPortVec.size(); i++)
{
if (aPortVec[i]->iConfig)
aPortVec[i]->iConfig->iPortStats.Log(iLogger, aPortVec[i]->iConfig->iMime);
}
PVLOGGER_LOGMSG(PVLOGMSG_INST_PROF, iLogger, PVLOGMSG_ERR,
(0, "SocketNodeStats:@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@"));
}
#endif //ENABLE_SOCKET_NODE_STATS
//////////////////////////////////////////////////
// SocketPortConfig
//////////////////////////////////////////////////
void SocketPortConfig::CreateAllocators(uint32 aSize, uint32 aExpectedNumberOfBlocksPerBuffer, uint32 aResizeSize, uint32 aMaxNumResizes)
{
iMemPool->CreateAllocators(iMime, aSize, aExpectedNumberOfBlocksPerBuffer, aResizeSize, aMaxNumResizes);
}
void SocketPortConfig::CleanupMemPools(Oscl_DefAlloc& aAlloc)
{
if (iMemPool)
{
iMemPool->DestroyAllocators();
iMemPool->~PVMFSocketNodeMemPool();
aAlloc.deallocate((OsclAny*)(iMemPool));
iMemPool = NULL;
}
}
//////////////////////////////////////////////////
// End SocketPortConfig
//////////////////////////////////////////////////
//////////////////////////////////////////////////
// Standard Node APIs
//////////////////////////////////////////////////
OSCL_EXPORT_REF PVMFSocketNode::PVMFSocketNode(int32 aPriority)
: OsclActiveObject(aPriority, "PVMFSocketNode")
, TIMEOUT_CONNECT(30000)
, TIMEOUT_SEND(3000)
, TIMEOUT_SENDTO(3000)
, TIMEOUT_RECV(-1)
, TIMEOUT_RECVFROM(-1)
, TIMEOUT_SHUTDOWN(10000)
, UDP_PORT_RANGE(2000)
, MAX_UDP_PACKET_SIZE(MAX_SOCKET_BUFFER_SIZE)
, MIN_UDP_PACKET_SIZE(MIN_SOCKET_BUFFER_SIZE)
{
iLogger = NULL;
iDataPathLogger = NULL;
iDataPathLoggerRTP = NULL;
iDataPathLoggerRTCP = NULL;
iOsclErrorTrapImp = NULL;
iSockServ = NULL;
iMaxTcpRecvBufferSize = SNODE_DEFAULT_SOCKET_TCP_BUFFER_SIZE;
iMaxTcpRecvBufferCount = SNODE_DEFAULT_SOCKET_TCP_BUFFER_COUNT;
iSocketID = 0;
iCommandErrorCode = PVMFSocketNodeErrorEventStart;
iErrorEventErrorCode = PVMFSocketNodeErrorEventStart;
iExtensionInterface = NULL;
iInSocketCallback = false;
iNumStopPortActivityPending = (-1);//inactive.
int32 err;
OSCL_TRY(err,
//Create the input command queue. Use a reserve to avoid lots of
//dynamic memory allocation.
iPendingCmdQueue.Construct(PVMF_SOCKET_NODE_COMMAND_ID_START,
PVMF_SOCKET_NODE_COMMAND_VECTOR_RESERVE);
//Create the "current command" queue. It will only contain one
//command at a time, so use a reserve of 1.
iCurrentCmdQueue.Construct(0, 1);
iCancelCmdQueue.Construct(0, 1);
//Create the port vector.
iPortVector.Construct(PVMF_SOCKET_NODE_PORT_VECTOR_RESERVE);
//Set the node capability data.
//This node can support an unlimited number of ports.
iCapability.iCanSupportMultipleInputPorts = true;
iCapability.iCanSupportMultipleOutputPorts = true;
iCapability.iHasMaxNumberOfPorts = false;
iCapability.iMaxNumberOfPorts = 0;//no maximum
iCapability.iInputFormatCapability.push_back(PVMF_MIME_INET_UDP);
iCapability.iInputFormatCapability.push_back(PVMF_MIME_INET_TCP);
iCapability.iOutputFormatCapability.push_back(PVMF_MIME_INET_TCP);
iCapability.iOutputFormatCapability.push_back(PVMF_MIME_INET_UDP);
iDnsCache.NewL();
);
if (err != OsclErrNone)
{
//if a leave happened, cleanup and re-throw the error
iPendingCmdQueue.clear();
iCurrentCmdQueue.clear();
iCancelCmdQueue.clear();
iPortVector.clear();
iCapability.iInputFormatCapability.clear();
iCapability.iOutputFormatCapability.clear();
OSCL_CLEANUP_BASE_CLASS(PVMFNodeInterface);
OSCL_CLEANUP_BASE_CLASS(OsclActiveObject);
OSCL_LEAVE(err);
}
}
OSCL_EXPORT_REF PVMFSocketNode::~PVMFSocketNode()
{
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.Log(iPortVector);
#endif
Cancel();
//thread logoff
if (IsAdded())
RemoveFromScheduler();
if (iExtensionInterface)
{
iExtensionInterface->removeRef();
}
/* Cleanup allocated ports */
CleanupPorts();
iPortVector.clear();
CleanupClosedTCPSockets();
CleanupClosedUDPSockets();
CleanupClosedDNS();
//Cleanup commands
//The command queues are self-deleting, but we want to
//notify the observer of unprocessed commands.
while (!iCurrentCmdQueue.empty())
{
CommandComplete(iCurrentCmdQueue, iCurrentCmdQueue.front(), PVMFFailure);
}
while (!iPendingCmdQueue.empty())
{
CommandComplete(iPendingCmdQueue, iPendingCmdQueue.front(), PVMFFailure);
}
while (!iCancelCmdQueue.empty())
{
CommandComplete(iCancelCmdQueue, iCancelCmdQueue.front(), PVMFFailure);
}
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "Goin to delete Sock Serv"));
if (iSockServ)
{
iSockServ->Close();
iSockServ->~OsclSocketServ();
iAlloc.deallocate(iSockServ);
iSockServ = NULL;
}
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:~PVMFSocketNode out"));
}
//Do thread-specific node creation and go to "Idle" state.
OSCL_EXPORT_REF PVMFStatus PVMFSocketNode::ThreadLogon()
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:ThreadLogon"));
switch (iInterfaceState)
{
case EPVMFNodeCreated:
if (!IsAdded())
AddToScheduler();
iLogger = PVLogger::GetLoggerObject("PVMFSocketNode");
iDataPathLogger = PVLogger::GetLoggerObject("datapath.socketnode");
iDataPathLoggerRTP = PVLogger::GetLoggerObject("datapath.socketnode.rtp");
iDataPathLoggerRTCP = PVLogger::GetLoggerObject("datapath.socketnode.rtcp");
iOsclErrorTrapImp = OsclErrorTrap::GetErrorTrapImp();
SetState(EPVMFNodeIdle);
return PVMFSuccess;
default:
return PVMFErrInvalidState;
}
}
//Do thread-specific node cleanup and go to "Created" state.
OSCL_EXPORT_REF PVMFStatus PVMFSocketNode::ThreadLogoff()
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:ThreadLogoff"));
switch (iInterfaceState)
{
case EPVMFNodeIdle:
{
if (IsAdded())
RemoveFromScheduler();
iLogger = NULL;
iDataPathLogger = NULL;
iDataPathLoggerRTP = NULL;
iDataPathLoggerRTCP = NULL;
iOsclErrorTrapImp = NULL;
SetState(EPVMFNodeCreated);
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.Logoff();
#endif
return PVMFSuccess;
}
// break; This break statement was removed to avoid compiler warning for Unreachable Code
default:
return PVMFErrInvalidState;
// break; This break statement was removed to avoid compiler warning for Unreachable Code
}
}
//retrieve node capabilities.
OSCL_EXPORT_REF PVMFStatus PVMFSocketNode::GetCapability(PVMFNodeCapability& aNodeCapability)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:GetCapability"));
aNodeCapability = iCapability;
return PVMFSuccess;
}
//retrieve a port iterator.
OSCL_EXPORT_REF PVMFPortIter* PVMFSocketNode::GetPorts(const PVMFPortFilter* aFilter)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:GetPorts"));
OSCL_UNUSED_ARG(aFilter);//port filter is not implemented.
iPortVector.Reset();
return &iPortVector;
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::QueryUUID(PVMFSessionId s, const PvmfMimeString& aMimeType,
Oscl_Vector<PVUuid, OsclMemAllocator>& aUuids,
bool aExactUuidsOnly,
const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:QueryUUID"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_QUERYUUID, aMimeType, aUuids, aExactUuidsOnly, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::QueryInterface(PVMFSessionId s, const PVUuid& aUuid,
PVInterface*& aInterfacePtr,
const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:QueryInterface"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_QUERYINTERFACE, aUuid, aInterfacePtr, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::RequestPort(PVMFSessionId s, int32 aPortTag, const PvmfMimeString* aPortConfig, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:RequestPort"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_REQUESTPORT, aPortTag, aPortConfig, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::ReleasePort(PVMFSessionId s, PVMFPortInterface& aPort, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:ReleasePort"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_RELEASEPORT, aPort, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::Init(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:Init"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_INIT, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::Prepare(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:Init"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_PREPARE, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::Start(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:Start"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_START, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::Stop(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:Stop"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_STOP, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::Flush(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:Flush"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_FLUSH, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::Pause(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:Pause"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_PAUSE, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::Reset(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:Reset"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_RESET, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::CancelAllCommands(PVMFSessionId s, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:CancelAllCommands"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_CANCELALLCOMMANDS, aContext);
return QueueCommandL(cmd);
}
OSCL_EXPORT_REF PVMFCommandId PVMFSocketNode::CancelCommand(PVMFSessionId s, PVMFCommandId aCmdId, const OsclAny* aContext)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:CancelCommand"));
PVMFSocketNodeCommand cmd;
cmd.PVMFSocketNodeCommandBase::Construct(s, PVMF_GENERIC_NODE_CANCELCOMMAND, aCmdId, aContext);
return QueueCommandL(cmd);
}
//Port activity handler
void PVMFSocketNode::HandlePortActivity(const PVMFPortActivity &aActivity)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE,
(0, "0x%x PVMFSocketNode::HandlePortActivity IN: port=0x%x, type=%d",
this, aActivity.iPort, aActivity.iType));
switch (aActivity.iType)
{
case PVMF_PORT_ACTIVITY_CONNECT:
{
PVMFSocketPort* sockPort = OSCL_STATIC_CAST(PVMFSocketPort*, aActivity.iPort);
OSCL_ASSERT(sockPort && sockPort->iConfig);
SocketPortConfig& aSockConfig = *sockPort->iConfig;
aSockConfig.CreateAllocators(iMaxTcpRecvBufferSize *(iMaxTcpRecvBufferCount - 1), iMaxTcpRecvBufferCount - 1, iMaxTcpRecvBufferSize, 1);
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumPortEventConnect++;
#endif
//Receives may have been blocked waiting on the port to be connected, so check here.
//Note: This is a known use case for HTTP streaming, where the port request and connect
//can happen after the node start
if (CanReceive(aSockConfig))
StartRecvOperation(aSockConfig);
}
break;
case PVMF_PORT_ACTIVITY_INCOMING_MSG:
{
PVMFSocketPort* sockPort = (PVMFSocketPort*)(aActivity.iPort);
OSCL_ASSERT(sockPort && sockPort->iConfig);
SocketPortConfig& aSockConfig = *sockPort->iConfig;
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumPortEventIncomingMsg++;
#endif
//Try to process this message now.
if (CanProcessIncomingMsg(aSockConfig))
ProcessIncomingMsg(aSockConfig);
//Otherwise, ignore this event now. Other code will check
//the input queue as needed.
}
break;
case PVMF_PORT_ACTIVITY_CONNECTED_PORT_READY:
{
PVMFSocketPort* sockPort = (PVMFSocketPort*)(aActivity.iPort);
OSCL_ASSERT(sockPort && sockPort->iConfig);
SocketPortConfig& aSockConfig = *sockPort->iConfig;
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumPortEventConnectedPortReady++;
#endif
//See if the port was waiting on this event, and if so, continue processing.
if (aSockConfig.iState.iConnectOperation == EPVSocketPortConnectOperation_WaitOnConnectedPort)
{
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.EndConnectTime(aSockConfig.iState.iConnectOperation);
#endif
ConnectOperationComplete(aSockConfig, PVMFSuccess, NULL);
}
if (aSockConfig.iState.iRecvOperation == EPVSocketPortRecvOperation_WaitOnConnectedPort)
{
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.EndRecvTime(aSockConfig.iState.iRecvOperation);
#endif
RecvOperationComplete(aSockConfig, PVMFSuccess, NULL);
}
//Otherwise ignore this event now. Other code will check connected
//port status as needed.
}
break;
default:
//all other events can be ignored.
break;
}
}
//////////////////////////////////////////////////
// End Standard Node APIs
//////////////////////////////////////////////////
//////////////////////////////////////////////////
// Additional Public APIs unique to Socket Node
//////////////////////////////////////////////////
OSCL_EXPORT_REF bool PVMFSocketNode::GetPortConfig(PVMFPortInterface &aPort, OsclNetworkAddress &aLocalAdd, OsclNetworkAddress &aRemoteAdd)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::GetPortConfig IN"));
aLocalAdd.ipAddr.Set("0.0.0.0");
aLocalAdd.port = 0;
aRemoteAdd.ipAddr.Set("");
aRemoteAdd.port = 0;
for (uint32 i = 0; i < iPortVector.size(); i++)
{
if (iPortVector[i]->iConfig
&& iPortVector[i]->iConfig->iPVMFPort == &aPort)
{
aLocalAdd = iPortVector[i]->iConfig->iAddr.iLocalAdd;
aRemoteAdd = iPortVector[i]->iConfig->iAddr.iRemoteAdd;
return true;
}
}
return false;
}
OSCL_EXPORT_REF bool PVMFSocketNode::SetPortConfig(PVMFPortInterface &aPort, OsclNetworkAddress aLocalAdd, OsclNetworkAddress aRemoteAdd)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::SetPortConfig IN"));
for (uint32 i = 0; i < iPortVector.size(); i++)
{
if (iPortVector[i]->iConfig
&& iPortVector[i]->iConfig->iPVMFPort == &aPort)
{
iPortVector[i]->iConfig->iAddr.iLocalAdd = aLocalAdd;
iPortVector[i]->iConfig->iAddr.iRemoteAdd = aRemoteAdd;
return true;
}
}
return false;
}
//////////////////////////////////////////////////
// End Additional Public APIs unique to Socket Node
//////////////////////////////////////////////////
//Create UDP sockets on consecutive ports.
//This is needed because 3GPP streaming servers require consecutive ports.
//If successfully created, the sockets are saved in the "allocated port" vector.
//The created sockets will be associated with PVMF ports later during RequestPort
//commands.
PVMFStatus PVMFSocketNode::AllocateConsecutivePorts(PvmfMimeString* aPortConfig,
uint32& aLowerPortNum,
uint32& aHigherPortNum, uint32& aStartPortNum)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::AllocateConsecutivePorts IN"));
//used in case of 3GPP streaming to allocate consecutive udp ports
//for rtp and rtcp.
//lower port number = rtp => PVMF_SOCKET_NODE_PORT_TYPE_SOURCE
//higher port number = rtcp => PVMF_SOCKET_NODE_PORT_TYPE_SINK
aLowerPortNum = 0;
aHigherPortNum = 0;
uint32 startSockID = iSocketID;
PVMFStatus status = PVMFFailure;
SocketPortConfig* lower_sock_config = OSCL_NEW(SocketPortConfig, ());
SocketPortConfig* higher_sock_config = OSCL_NEW(SocketPortConfig, ());
for (int maxNumOfBind = UDP_PORT_RANGE; maxNumOfBind >= 0; maxNumOfBind--)
{
OSCL_HeapString<OsclMemAllocator> rtpportConfigWithMime;
rtpportConfigWithMime += aPortConfig->get_cstr();
rtpportConfigWithMime += _STRLIT_CHAR("/rtp");
ParseTransportConfig(&rtpportConfigWithMime,
lower_sock_config->iAddr,
lower_sock_config->iMime);
//this should only be UDP
if (lower_sock_config->iAddr.iProtocol != INET_UDP)
{
status = PVMFErrArgument;
goto Exit;
}
lower_sock_config->iContainer = this;
lower_sock_config->iAddr.iLocalAdd.port = aStartPortNum;
lower_sock_config->iSockId = startSockID++;
lower_sock_config->iRTP = true;
if (NULL == (lower_sock_config->iUDPSocket = (OsclUDPSocket*)CreateOsclSocketAndBind(lower_sock_config->iAddr, lower_sock_config->iSockId)))
{
status = PVMFFailure;
goto Exit;
}
OSCL_HeapString<OsclMemAllocator> rtcpportConfigWithMime;
rtcpportConfigWithMime += aPortConfig->get_cstr();
rtcpportConfigWithMime += _STRLIT_CHAR("/rtcp");
ParseTransportConfig(&rtcpportConfigWithMime,
higher_sock_config->iAddr,
higher_sock_config->iMime);
//this should only be UDP
if (higher_sock_config->iAddr.iProtocol != INET_UDP)
{
status = PVMFErrArgument;
goto Exit;
}
higher_sock_config->iContainer = this;
higher_sock_config->iAddr.iLocalAdd.port =
lower_sock_config->iAddr.iLocalAdd.port + 1;
higher_sock_config->iSockId = startSockID++;
higher_sock_config->iRTCP = true;
if (NULL == (higher_sock_config->iUDPSocket = (OsclUDPSocket*)CreateOsclSocketAndBind(higher_sock_config->iAddr, higher_sock_config->iSockId)))
{
status = PVMFFailure;
goto Exit;
}
//We were able to create and bind, now see whether we got consecutive ports or not.
if (higher_sock_config->iAddr.iLocalAdd.port ==
(lower_sock_config->iAddr.iLocalAdd.port + 1))
{
aLowerPortNum = lower_sock_config->iAddr.iLocalAdd.port;
lower_sock_config->iPVMFPort = NULL;
lower_sock_config->iTag = PVMF_SOCKET_NODE_PORT_TYPE_SOURCE;
iAllocatedPortVector.push_back(lower_sock_config);
aHigherPortNum = higher_sock_config->iAddr.iLocalAdd.port;
higher_sock_config->iPVMFPort = NULL;
higher_sock_config->iTag = PVMF_SOCKET_NODE_PORT_TYPE_SINK;
iAllocatedPortVector.push_back(higher_sock_config);
iSocketID = startSockID;
aStartPortNum = higher_sock_config->iAddr.iLocalAdd.port + 1;
status = PVMFSuccess;
goto Exit;
}
else
{
//Delete and try again
OsclUDPSocket* udpSocket1 = lower_sock_config->iUDPSocket;
udpSocket1->~OsclUDPSocket();
iAlloc.deallocate(udpSocket1);
lower_sock_config->iUDPSocket = NULL;
OsclUDPSocket* udpSocket2 = higher_sock_config->iUDPSocket;
udpSocket2->~OsclUDPSocket();
iAlloc.deallocate(udpSocket2);
higher_sock_config->iUDPSocket = NULL;
aStartPortNum = higher_sock_config->iAddr.iLocalAdd.port + 1;
startSockID = iSocketID;
}
}//for max port bind attempts
status = PVMFFailure;
Exit:
if (status != PVMFSuccess)
{
OSCL_DELETE(lower_sock_config);
OSCL_DELETE(higher_sock_config);
}
return status;
}
OSCL_EXPORT_REF PVMFStatus PVMFSocketNode::SetMaxTCPRecvBufferSize(uint32 aBufferSize)
{
if ((aBufferSize > 0) && (aBufferSize < SNODE_DEFAULT_MAX_TCP_RECV_BUFFER_SIZE))
{
iMaxTcpRecvBufferSize = aBufferSize;
return PVMFSuccess;
}
return PVMFErrArgument;
}
OSCL_EXPORT_REF PVMFStatus PVMFSocketNode::GetMaxTCPRecvBufferSize(uint32& aSize)
{
aSize = iMaxTcpRecvBufferSize;
return PVMFSuccess;
}
OSCL_EXPORT_REF PVMFStatus PVMFSocketNode::SetMaxTCPRecvBufferCount(uint32 aCount)
{
iMaxTcpRecvBufferCount = aCount;
return PVMFSuccess;
}
OSCL_EXPORT_REF PVMFStatus PVMFSocketNode::GetMaxTCPRecvBufferCount(uint32& aCount)
{
aCount = iMaxTcpRecvBufferCount;
return PVMFSuccess;
}
OsclMemPoolResizableAllocator* PVMFSocketNode::CreateSharedBuffer(const PVMFPortInterface* aPort , uint32 aBufferSize, uint32 aExpectedNumberOfBlocksPerBuffer, uint32 aResizeSize, uint32 aMaxNumResizes)
{
//validate input params if required..
OSCL_ASSERT(aPort);
if (!aPort)
return NULL;
//If data buffer for the port is already created and is existing, then destroy the prev buffer and create new one (with requested attributes).
//CreateResizableDataBufferAllocator func will take care of this.
SocketPortConfig* portConfig = OSCL_STATIC_CAST(PVMFSocketPort*, aPort)->iConfig;
PVMFSocketNodeMemPool * portMemPool = portConfig->iMemPool;
portMemPool->iSharedDataBufferInfo.Init(aBufferSize, aExpectedNumberOfBlocksPerBuffer, aResizeSize, aMaxNumResizes);
return portMemPool->CreateResizableDataBufferAllocator(portConfig->iMime.get_cstr());
}
///////////////////////////
// End Socket Extension Interface
///////////////////////////
//////////////////////////////////////
// Incoming Message Handlers
//////////////////////////////////////
//Tell whether it's possible to process an incoming message on a port.
bool PVMFSocketNode::CanProcessIncomingMsg(SocketPortConfig& aSockConfig)
{
return
//node is started
iInterfaceState == EPVMFNodeStarted
//port has input messages
&& aSockConfig.iPVMFPort && aSockConfig.iPVMFPort->IncomingMsgQueueSize() > 0
//port is not busy with any sequence
&& aSockConfig.iState.iSequence == EPVSocketPortSequence_None;
}
//Process an incoming message on this port.
//Assume caller already verified "can process incoming msg".
void PVMFSocketNode::ProcessIncomingMsg(SocketPortConfig& aSockConfig)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::ProcessIncomingMsg: aPort=0x%x", aSockConfig.iPVMFPort));
//Dequeue the incoming message
PVMFSharedMediaMsgPtr msg;
PVMFStatus status = aSockConfig.iPVMFPort->DequeueIncomingMsg(msg);
if (status != PVMFSuccess)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::ProcessIncomingDataMsg: Error - DequeueIncomingMsg failed"));
ReportErrorEvent(PVMFErrPortProcessing);
return ;
}
//calling logic should prevent processing a message when some other sequence
//is active on the port
OSCL_ASSERT(aSockConfig.iState.iSequence == EPVSocketPortSequence_None);
//Handle 3 types of messages
if (msg->getFormatID() == PVMF_MEDIA_CMD_SOCKET_CONNECT_FORMAT_ID)
{
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumDequeueIncomingConnectMsg++;
#endif
//Connect message
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ProcessIncomingMsg - CONNECT SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_cstr()));
//Parse the address
OsclRefCounterMemFrag newSockAddr;
msg->getFormatSpecificInfo(newSockAddr);
if (newSockAddr.getMemFragPtr())
{
if (!ParseTransportConfig((char*)newSockAddr.getMemFragPtr(),
newSockAddr.getMemFragSize(),
aSockConfig.iAddr,
aSockConfig.iMime))
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::ProcessIncomingMsg: ERROR - Invalid port config"));
ReportSocketNodeError(PVMFErrPortProcessing, PVMFSocketNodeErrorInvalidPortTag);
return ;
}
}
//Start the sequence
StartSequence(aSockConfig, EPVSocketPortSequence_InputConnectMsg);
}
else if (msg->getFormatID() == PVMF_MEDIA_CMD_SOCKET_DISCONNECT_FORMAT_ID)
{
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumDequeueIncomingDisconnectMsg++;
#endif
//Disconnect message
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ProcessIncomingMsg - DISCONNECT msg - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_cstr()));
//Start the sequence
StartSequence(aSockConfig, EPVSocketPortSequence_InputDisconnectMsg);
}
else if (msg->getFormatID() == PVMF_MEDIA_MSG_DATA_FORMAT_ID)
{
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumDequeueIncomingDataMsg++;
#endif
//Media Data message
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ProcessIncomingMsg - MEDIA DATA msg - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_cstr()));
//calling logic should prevent any simultaneous send/connect
OSCL_ASSERT(aSockConfig.iState.iConnectOperation == EPVSocketPortConnectOperation_None);
//Start the sequence
StartSequence(aSockConfig, EPVSocketPortSequence_InputDataMsg, (OsclAny*)&msg);
}
else
{
//unexpected message type
ReportErrorEvent(PVMFErrPortProcessing);
}
}
//////////////////////////////////////
// End Incoming Message Handlers
//////////////////////////////////////
//////////////////////////////////////
// Connect Operation Handlers
//////////////////////////////////////
//Start the given "connect" operation on the port.
//If the connect fails, this will call "connect operation complete" and continue to
//the next operation.
PVMFStatus PVMFSocketNode::StartConnectOperation(SocketPortConfig& aSockConfig, TPVSocketPortConnectOperation aOperation)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::StartConnectOperation() In Operation %d", aOperation));
//connect operation should be idle now.
OSCL_ASSERT(aSockConfig.iState.iConnectOperation == EPVSocketPortConnectOperation_None);
//Update the state
aSockConfig.iState.iConnectOperation = aOperation;
PVMFStatus status = PVMFFailure;
switch (aOperation)
{
case EPVSocketPortConnectOperation_GetHostByName:
{
//This is the beginning of TCP socket create/connect.
//this should be TCP.
OSCL_ASSERT(aSockConfig.iAddr.iProtocol == INET_TCP);
//Any old socket should be cleaned up or else queued for cleanup at this point.
OSCL_ASSERT(!aSockConfig.iTCPSocket);
//Create the socket
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation GetHostByName, Creating socket - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iTCPSocket = (OsclTCPSocket*)CreateOsclSocketAndBind(aSockConfig.iAddr, aSockConfig.iSockId);
if (!aSockConfig.iTCPSocket)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation GetHostByName, Creating socket FAILED - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::StartConnectOperation: ERROR - CreateSocket()"));
if (aSockConfig.iState.iSequence == EPVSocketPortSequence_RequestPort)
{
iCommandErrorCode = PVMFSocketNodeErrorSocketFailure;
}
else
{
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeErrorSocketFailure);
}
status = PVMFErrResource;
break;
}
//see if DNS is needed
if (OsclValidInetAddr(aSockConfig.iAddr.iRemoteHost.get_cstr()))
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation GetHostByName, DNS is not needed - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iAddr.iRemoteAdd.ipAddr.Set(aSockConfig.iAddr.iRemoteHost.get_cstr());
status = PVMFSuccess;//no lookup needed-- continue to next operation.
break;
}
//DNS lookup needed.
//see if we can use the DNS cache instead of an actual lookup
aSockConfig.iAddr.iRemoteAdd.ipAddr.Set("");
if (iDnsCache.QueryGetHostByName(aSockConfig.iAddr.iRemoteHost.get_str(), (char*)aSockConfig.iAddr.iRemoteAdd.ipAddr.Str(), aSockConfig.iAddr.iRemoteAdd.ipAddr.MaxLen()))
{
//DNS cache lookup success
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation GetHostByName, DNS cache lookup success - SockId=%d, Mime=%s, Host=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str(), aSockConfig.iAddr.iRemoteHost.get_str()));
status = PVMFSuccess;//no lookup needed-- continue to next operation.
break;
}
//Create DNS object
if (!aSockConfig.iDNS)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation GetHostByName, Creating DNS object - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iDNS = OsclDNS::NewL(iAlloc, *iSockServ, *this, aSockConfig.iSockId);
if (!aSockConfig.iDNS)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation GetHostByName, Creating DNS FAILED - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::StartConnectOperation: ERROR - Create DNS()"));
if (aSockConfig.iState.iSequence == EPVSocketPortSequence_RequestPort)
{
iCommandErrorCode = PVMFSocketNodeError_DNSLookup;
}
else
{
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeError_DNSLookup);
}
status = PVMFErrResource;
break;
}
}
//start the asynchronous DNS operation.
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumGetHostByName++;
aSockConfig.iPortStats.StartConnectTime(aSockConfig.iState.iConnectOperation);
#endif
TPVDNSEvent aEvent = aSockConfig.iDNS->GetHostByName(aSockConfig.iAddr.iRemoteHost.get_str(),
aSockConfig.iAddr.iRemoteAdd, -1);
if (EPVDNSPending != aEvent)
{
aSockConfig.iState.iConnectOperation = EPVSocketPortConnectOperation_None;
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartConnectOperation: GetHostByName Failed - SockId=%d, Mime=%s, Host=%s",
aSockConfig.iSockId, aSockConfig.iMime.get_str(),
aSockConfig.iAddr.iRemoteHost.get_str()));
if (aSockConfig.iState.iSequence == EPVSocketPortSequence_RequestPort)
{
iCommandErrorCode = PVMFSocketNodeError_DNSLookup;
}
else
{
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeError_DNSLookup);
}
status = PVMFErrResource;
break;
}
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation: Waiting On DNS - SockId=%d, Mime=%s, Host=%s",
aSockConfig.iSockId, aSockConfig.iMime.get_str(),
aSockConfig.iAddr.iRemoteHost.get_str()));
#if (ENABLE_SOCKET_NODE_DEBUG_INFO_EVENT)
ReportInfoEvent(PVMFSocketNodeInfoEventRequestedDNSResolution);
#endif
status = PVMFPending;
//wait on the DNS lookup to complete in HandleDNSEvent callback
}
break;
case EPVSocketPortConnectOperation_Connect:
//Initiate a socket connect, after DNS lookup was successful or address is already IP.
//Will return "pending" or an error.
{
if (!aSockConfig.iTCPSocket)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartConnectOperation: ERROR socket already exists - SockId=%d, Mime=%s",
aSockConfig.iSockId, aSockConfig.iMime.get_str()));
status = PVMFFailure;//unexpected
break;
}
//start the asynchronous connect operation.
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumConnect++;
aSockConfig.iPortStats.StartConnectTime(aSockConfig.iState.iConnectOperation);
#endif
TPVSocketEvent ret = aSockConfig.iTCPSocket->Connect(aSockConfig.iAddr.iRemoteAdd, TIMEOUT_CONNECT);
if (ret != EPVSocketPending)
{
//failure
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartConnectOperation: Starting Connect FAILED - SockId=%d, Mime=%s, RemoteIP=%s",
aSockConfig.iSockId, aSockConfig.iMime.get_str(),
(char*)(aSockConfig.iAddr.iRemoteAdd.ipAddr.Str())));
if (aSockConfig.iState.iSequence == EPVSocketPortSequence_RequestPort)
{
iCommandErrorCode = PVMFSocketNodeError_TCPSocketConnect;
}
else
{
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeError_TCPSocketConnect);
}
status = PVMFErrResource;
break;
}
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartConnectOperation - Socket Connect Pending...SockId=%d, Mime=%s, RemoteIP=%s"
, aSockConfig.iSockId, aSockConfig.iMime.get_str(),
(char*)(aSockConfig.iAddr.iRemoteAdd.ipAddr.Str())));
#if (ENABLE_SOCKET_NODE_DEBUG_INFO_EVENT)
ReportInfoEvent(PVMFSocketNodeInfoEventConnectRequestPending);
#endif
status = PVMFPending;
//wait on the socket connect to complete in HandleSocketEvent callback
}
break;
case EPVSocketPortConnectOperation_Shutdown:
{
if (!aSockConfig.iTCPSocket)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartConnectOperation: ERROR socket already exists - SockId=%d, Mime=%s",
aSockConfig.iSockId, aSockConfig.iMime.get_str()));
status = PVMFFailure;//unexpected
break;
}
//Initiate a socket shutdown.
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumShutdown++;
aSockConfig.iPortStats.StartConnectTime(aSockConfig.iState.iConnectOperation);
#endif
TPVSocketEvent ret = aSockConfig.iTCPSocket->Shutdown(EPVSocketBothShutdown, TIMEOUT_SHUTDOWN);
if (ret == EPVSocketPending)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartConnectOperation - Shutdown Pending...SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
status = PVMFPending;
//wait on the socket shutdown to complete in HandleSocketEvent callback
}
else
{
status = PVMFFailure;
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartConnectOperation - Shutdown FAILED - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
}
}
break;
case EPVSocketPortConnectOperation_WaitOnConnectedPort:
if (aSockConfig.iPVMFPort
&& aSockConfig.iPVMFPort->IsOutgoingQueueBusy())
{
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumWaitOnConnectedPort++;
aSockConfig.iPortStats.StartConnectTime(aSockConfig.iState.iConnectOperation);
#endif
status = PVMFPending;
//wait for CONNECTED_PORT_READY event in HandlePortActivity
}
else
{
status = PVMFSuccess;//nothing needed.
}
break;
case EPVSocketPortConnectOperation_WaitOnSendRecvIdle:
if (aSockConfig.iState.iRecvOperation == EPVSocketPortRecvOperation_None
&& aSockConfig.iState.iSendOperation == EPVSocketPortSendOperation_None)
{
//send/recv are already idle.
status = PVMFSuccess;
}
else
{
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumWaitOnSendRecvIdle++;
aSockConfig.iPortStats.StartConnectTime(aSockConfig.iState.iConnectOperation);
#endif
status = PVMFPending;
//wait for send/recv completion (in RecvOperationComplete or SendOperationComplete)
}
break;
default:
//add code to handle this case
status = PVMFFailure;
OSCL_ASSERT(0);
break;
}
//Handle synchronous completion or failure.
if (status != PVMFPending)
status = ConnectOperationComplete(aSockConfig, status, NULL);
return status;
}
//The current connect operation is complete.
//Continue with the next operation, or finish the sequence.
//The status is in "aStatus". If there was a socket callback, the result is in "aSocketActivity".
PVMFStatus PVMFSocketNode::ConnectOperationComplete(SocketPortConfig& aSockConfig, PVMFStatus aStatus, PVMFSocketActivity* aSocketActivity)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::ConnectOperationComplete() in"));
//status should never be "pending" here
OSCL_ASSERT(aStatus != PVMFPending);
PVMFStatus status = aStatus;
TPVSocketPortConnectOperation curOp = aSockConfig.iState.iConnectOperation;
aSockConfig.iState.iConnectOperation = EPVSocketPortConnectOperation_None;
aSockConfig.iState.iConnectOperationStatus = aStatus;
aSockConfig.iState.iConnectOperationCanceled = false;
switch (curOp)
{
case EPVSocketPortConnectOperation_None:
//unexpected.
break;
case EPVSocketPortConnectOperation_GetHostByName:
//TCP Get Host By Name is complete
//catch completion of connect cancel during the cleanup sequence
if (aSockConfig.iState.iSequence == EPVSocketPortSequence_SocketCleanup)
{
//Continue the shutdown sequence
status = StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_Shutdown);
break;
}
//else normal connect sequence.
if (aStatus == PVMFSuccess)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ConnectOperationComplete GetHostByName Success - SockId=%d, Mime=%s Host= %s", aSockConfig.iSockId, aSockConfig.iMime.get_str()
, aSockConfig.iAddr.iRemoteHost.get_str()));
//Save lookup results in the DNS cache
if (aSocketActivity)
{
iDnsCache.UpdateCacheRecord(aSockConfig.iAddr.iRemoteHost.get_str(), (char*)aSockConfig.iAddr.iRemoteAdd.ipAddr.Str());
}
//DNS Lookup is always followed by a Connect
status = StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_Connect);
}
else
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ConnectOperationComplete GetHostByName FAILED - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
//If DNS lookup failed, then fail the current sequence with
//an error.
switch (aSockConfig.iState.iSequence)
{
case EPVSocketPortSequence_RequestPort:
//report the error in the command complete
iCommandErrorCode = PVMFSocketNodeError_DNSLookup;
break;
default:
//report the error in an error message
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeError_DNSLookup);
break;
}
//Fail the sequence.
SequenceComplete(aSockConfig, aStatus);
}
break;
case EPVSocketPortConnectOperation_Connect:
//TCP Connect is complete
//catch completion of connect cancel during the cleanup sequence
if (aSockConfig.iState.iSequence == EPVSocketPortSequence_SocketCleanup)
{
//Continue the shutdown sequence
status = StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_Shutdown);
break;
}
//else normal connect sequence.
if (aStatus == PVMFSuccess)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ConnectOperationComplete Connect Success - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
}
else
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ConnectOperationComplete Connect FAILED - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
//Report connect errors.
switch (aSockConfig.iState.iSequence)
{
case EPVSocketPortSequence_RequestPort:
//for a node command, we report the error in the node command completion message
iCommandErrorCode = PVMFSocketNodeError_TCPSocketConnect;
break;
case EPVSocketPortSequence_InputConnectMsg:
default:
//for other sequences, we report the error in an error message.
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeError_TCPSocketConnect);
break;
}
}
//Connect complete is always the end of a sequence.
SequenceComplete(aSockConfig, aStatus);
break;
case EPVSocketPortConnectOperation_Shutdown:
//TCP Shutdown is complete
if (aStatus == PVMFSuccess)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ConnectOperationComplete Shutdown Success - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
}
else
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::ConnectOperationComplete Shutdown FAILED - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
if (aSocketActivity)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::ConnectOperationComplete: ERROR aEvent=%d, Ln %d", aSocketActivity->iEvent, __LINE__));
}
}
//Note: ignore shutdown failures and just continue onto the
//next operation.
//Shutdown is followed by waiting for send/recv to go idle.
//On most OS the shutdown command is sufficient to cause send/recv to complete,
//but on some it doesn't, so cancel any outstanding send/recv op here.
CancelSendOperation(aSockConfig);
CancelRecvOperation(aSockConfig);
status = StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_WaitOnSendRecvIdle);
break;
case EPVSocketPortConnectOperation_WaitOnConnectedPort:
//TCP Wait on connected port is complete.
//Proceed to send the EOS and close the socket
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::ConnectOperationComplete WaitOnConnectedPort Calling CleanupTCP - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
CleanupTCP(aSockConfig);
//Check the sequence...
switch (aSockConfig.iState.iSequence)
{
case EPVSocketPortSequence_InputConnectMsg:
//a disconnect is followed by a host lookup + connect
//for an input connect message
status = StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_GetHostByName);
break;
default:
//for all other sequences, this is the end
SequenceComplete(aSockConfig, PVMFSuccess);
status = PVMFSuccess;
break;
}
break;
case EPVSocketPortConnectOperation_WaitOnSendRecvIdle:
//UDP or TCP "Wait on Send/Recv Idle" is complete.
//If TCP, wait on connected port to send EOS
if (aSockConfig.iTCPSocket)
{
status = StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_WaitOnConnectedPort);
}
//If UDP, proceed to close & delete
else if (aSockConfig.iUDPSocket)
{
CleanupUDP(aSockConfig);
//This is the end of the sequence.
SequenceComplete(aSockConfig, PVMFSuccess);
status = PVMFSuccess;
}
else
{
//unexpected, but end the sequence anyway.
SequenceComplete(aSockConfig, PVMFSuccess);
status = PVMFSuccess;
}
break;
default:
OSCL_ASSERT(0);//unexpected, add code for this case.
break;
}
return status;
}
PVMFStatus PVMFSocketNode::CancelConnectOperation(SocketPortConfig& aSockConfig)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::CancelConnectOperation() In"));
PVMFStatus status = PVMFSuccess;
switch (aSockConfig.iState.iConnectOperation)
{
case EPVSocketPortConnectOperation_None:
break;
case EPVSocketPortConnectOperation_GetHostByName:
if (aSockConfig.iDNS)
{
if (!aSockConfig.iState.iConnectOperationCanceled)
{
aSockConfig.iState.iConnectOperationCanceled = true;
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelConnectOperation Calling CancelGetHostByName - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iDNS->CancelGetHostByName();
}
status = PVMFPending;
//wait for host lookup to complete in HandleDNSEvent
}
break;
case EPVSocketPortConnectOperation_Connect:
if (aSockConfig.iTCPSocket)
{
if (!aSockConfig.iState.iConnectOperationCanceled)
{
aSockConfig.iState.iConnectOperationCanceled = true;
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelConnectOperation Calling CancelConnect - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iTCPSocket->CancelConnect();
}
status = PVMFPending;
//wait for connect to complete in HandleSocketEvent
}
break;
case EPVSocketPortConnectOperation_Shutdown:
if (aSockConfig.iTCPSocket)
{
if (!aSockConfig.iState.iConnectOperationCanceled)
{
aSockConfig.iState.iConnectOperationCanceled = true;
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelConnectOperation Calling CancelShutdown - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iTCPSocket->CancelShutdown();
}
status = PVMFPending;
//wait for shutdown to complete in HandleSocketEvent
}
break;
case EPVSocketPortConnectOperation_WaitOnConnectedPort:
//just clear the state
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelConnectOperation Cancelling WaitOnConnectedPort - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iState.iConnectOperation = EPVSocketPortConnectOperation_None;
break;
case EPVSocketPortConnectOperation_WaitOnSendRecvIdle:
//just clear the state
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelConnectOperation Cancelling WaitOnSendRecvIdle - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iState.iConnectOperation = EPVSocketPortConnectOperation_None;
break;
default:
OSCL_ASSERT(0);//add code for this case
status = PVMFFailure;
break;
}
return status;
}
//////////////////////////////////////
// End Connect Operation Handlers
//////////////////////////////////////
//////////////////////////////////
// Receive Operation Handlers
//////////////////////////////////
//Tell if it's possible to initiate a recv or recvfrom operation
//on this port.
bool PVMFSocketNode::CanReceive(SocketPortConfig& aSockConfig)
{
return
//node is started
iInterfaceState == EPVMFNodeStarted
//port is connected
&& aSockConfig.iPVMFPort && aSockConfig.iPVMFPort->IsConnected()
//socket exists (gets created during request port or connect sequence)
&& (aSockConfig.iUDPSocket || aSockConfig.iTCPSocket)
//port is not busy with any sequence other than sending data
&& (aSockConfig.iState.iSequence == EPVSocketPortSequence_None
|| aSockConfig.iState.iSequence == EPVSocketPortSequence_InputDataMsg)
//port is not busy with any receive operation.
&& aSockConfig.iState.iRecvOperation == EPVSocketPortRecvOperation_None
//there's no node stop going on.
&& iNumStopPortActivityPending < 0;
}
//Enter the "wait on memory" state
void PVMFSocketNode::StartRecvWaitOnMemory(SocketPortConfig& aSockConfig, int32 aSize)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::StartRecvWaitOnMemory() In"));
aSockConfig.iState.iRecvOperation = EPVSocketPortRecvOperation_WaitOnMemory;
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumWaitOnMemory++;
aSockConfig.iPortStats.StartRecvTime(aSockConfig.iState.iRecvOperation);
#endif
if (aSize)
{//wait on data buffer
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartRecvWaitOnMemory, wait on data buffer - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
if (aSockConfig.iMemPool->iSharedBufferAlloc)
{
aSockConfig.iMemPool->iSharedBufferAlloc->notifyfreeblockavailable(aSockConfig, aSize, NULL);
}
else
{
if (aSockConfig.iMemPool->iInternalAlloc)
aSockConfig.iMemPool->iInternalAlloc->notifyfreeblockavailable(aSockConfig, aSize, NULL);
}
}
else
{//wait on media data wrapper
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartRecvWaitOnMemory, wait on data wrapper - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iMemPool->iMediaDataMemPool->notifyfreechunkavailable(aSockConfig, NULL);
}
}
//Enter the "wait on connected port" state
void PVMFSocketNode::StartRecvWaitOnConnectedPort(SocketPortConfig& aSockConfig, PVMFSocketActivity& aSocketActivity)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::StartRecvWaitOnConnectedPort() In"));
//outgoing queue is busy-- must queue this event for later processing
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartRecvWaitOnConnectedPort, wait for connected port - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
switch (aSocketActivity.iFxn)
{
case EPVSocketRecv:
case EPVSocketRecvFrom:
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumQueueSocketRecv++;
#endif
aSockConfig.iSocketRecvActivity.Set(aSocketActivity.iStatus
, aSocketActivity.iId
, aSocketActivity.iFxn
, aSocketActivity.iEvent
, aSocketActivity.iError);
break;
default:
OSCL_ASSERT(false);//invalid input arg.
break;
}
//current state shoudl be idle
OSCL_ASSERT(aSockConfig.iState.iRecvOperation == EPVSocketPortRecvOperation_None);
aSockConfig.iState.iRecvOperation = EPVSocketPortRecvOperation_WaitOnConnectedPort;
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumWaitOnConnectedPort++;
aSockConfig.iPortStats.StartRecvTime(aSockConfig.iState.iRecvOperation);
#endif
}
//Start a receive operation on the port's socket.
//Assume the caller already verified that the port can receive now.
//If there is a failure, this routine will call the "recv operation complete" routine to
//do error handling or continue to the next state.
PVMFStatus PVMFSocketNode::StartRecvOperation(SocketPortConfig& aSockConfig)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::StartRecvOperation() In"));
//caller should have verified we can receive data right now.
OSCL_ASSERT(aSockConfig.iState.iRecvOperation == EPVSocketPortRecvOperation_None);
//there should be either a UDP or TCP socket on this port and a memory pool.
if (!aSockConfig.iTCPSocket && !aSockConfig.iUDPSocket)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartRecvOperation ERROR no socket - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
return PVMFFailure;
}
if (!aSockConfig.iMemPool)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartRecvOperation ERROR no mem pool - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
return PVMFFailure;
}
PVMFStatus status = PVMFFailure;
if (aSockConfig.iUDPSocket)
{
aSockConfig.iState.iRecvOperation = EPVSocketPortRecvOperation_RecvFrom;
//Create the memory buffer
int32 memSize;
int32 multiRecvLimitBytes = 0;
switch (aSockConfig.iMemPool->iPortTag)
{
case PVMF_SOCKET_NODE_PORT_TYPE_SOURCE:
//source tag is used for RTP
#if SNODE_ENABLE_UDP_MULTI_PACKET
//must allocate enough space for the max bytes per receive plus
//one more max size packet, since we can't peek ahead to see
//how much data is on the socket.
memSize = SNODE_UDP_MULTI_MAX_BYTES_PER_RECV + MAX_UDP_PACKET_SIZE;
//this is the amt of "free space" required to continue receiving
//additional packets. Use the max packet size to avoid truncated
//packets.
multiRecvLimitBytes = MAX_UDP_PACKET_SIZE;
#else
//for single packet recvs.
memSize = MAX_UDP_PACKET_SIZE;
#endif
break;
case PVMF_SOCKET_NODE_PORT_TYPE_SINK:
//sink tag is used for RTCP
default:
memSize = MAX_FEEDBACK_SOCKET_BUFFER_SIZE;
break;
}
//Socket node provides the buffer to receive the data on the socket
//Socket node gets this from the mem pool.
//It may be possible that socket node may not be able to get the
//buffer of required size from the mempool
//If we are not able to get the packets from the OS, then the packets
//received from server will eventually be lost.
//In general, streaming server will be sending the UDP packets of small size
//(say 2048 bytes.. = MIN_UDP_PACKET_SIZE)
//So its better to attempt to allocate the memory of size
//SNODE_UDP_MULTI_MIN_BYTES_PER_RECV + MIN_UDP_PACKET_SIZE/MIN_UDP_PACKET_SIZE
//in such cases.
const int32 largestMediaMsgSz = aSockConfig.iMemPool->GetMaxSizeMediaMsgLen();
PVMF_SOCKETNODE_LOGINFO((0, "PVMFSocketNode::StartRecvOperation - MaxSizeMediMsg %d memSize %d", largestMediaMsgSz, memSize));
if (memSize > largestMediaMsgSz)
{
#if SNODE_ENABLE_UDP_MULTI_PACKET
multiRecvLimitBytes = MIN_UDP_PACKET_SIZE;
const int32 minBufferToContinueRecv = (SNODE_UDP_MULTI_MIN_BYTES_PER_RECV + MIN_UDP_PACKET_SIZE);
if (largestMediaMsgSz >= minBufferToContinueRecv)
{
memSize = largestMediaMsgSz;
}
else
{
memSize = minBufferToContinueRecv;
}
#else
if (largestMediaMsgSz >= MIN_UDP_PACKET_SIZE)
{
memSize = largestMediaMsgSz;
}
else
{
memSize = MIN_UDP_PACKET_SIZE;
}
#endif
PVMF_SOCKETNODE_LOGINFO((0, "PVMFSocketNode::StartRecvOperation - Resizing Mem Request Attributes## NewSz %d recvLimitBytes %d", memSize, multiRecvLimitBytes));
}
//Create the media data impl.
int32 err;
OsclSharedPtr<PVMFMediaDataImpl> mediaDataImpl;
err = GetMediaDataImpl(aSockConfig, mediaDataImpl, memSize);
if (err != OsclErrNone)
{
StartRecvWaitOnMemory(aSockConfig, memSize);
status = PVMFPending;
//wait on memory pool callback "notifyfreechunkavailable"
goto Exit;
}
//Create the media data buffer.
err = CreateMediaData(aSockConfig, mediaDataImpl);
if (err != OsclErrNone)
{
StartRecvWaitOnMemory(aSockConfig);
status = PVMFPending;
//wait on memory pool callback "notifyfreechunkavailable"
goto Exit;
}
// Retrieve memory fragment to write to
OsclRefCounterMemFrag refCtrMemFragOut;
aSockConfig.iPendingRecvMediaData->getMediaFragment(0, refCtrMemFragOut);
//Issue the Oscl socket request
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumRecvFrom++;
aSockConfig.iPortStats.StartRecvTime(EPVSocketPortRecvOperation_RecvFrom);
#endif
TPVSocketEvent retVal = aSockConfig.iUDPSocket->RecvFrom((uint8*)refCtrMemFragOut.getMemFragPtr()
, refCtrMemFragOut.getCapacity()
, aSockConfig.iSourceAddr
, TIMEOUT_RECVFROM
, multiRecvLimitBytes
, &aSockConfig.iRecvFromPacketLen);
if (EPVSocketPending != retVal)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::RecvFrom() failed retVal=%d Ln %d", retVal, __LINE__));
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartRecvOperation RecvFrom FAILED - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
status = PVMFFailure;
goto Exit;
}
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartRecvOperation RecvFrom Pending - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
status = PVMFPending;
//wait on the socket callback HandleSocketEvent
}
else if (aSockConfig.iTCPSocket)
{
aSockConfig.iState.iRecvOperation = EPVSocketPortRecvOperation_Recv;
//Allocate memory
int32 err;
OsclSharedPtr<PVMFMediaDataImpl> mediaDataImpl;
mediaDataImpl = aSockConfig.iMemPool->getMediaDataImpl(iMaxTcpRecvBufferSize);
if (NULL == mediaDataImpl.GetRep())
{
StartRecvWaitOnMemory(aSockConfig, iMaxTcpRecvBufferSize);
status = PVMFPending;
//wait on memory pool callback "notifyfreechunkavailable"
goto Exit;
}
OSCL_TRY(err,
aSockConfig.iPendingRecvMediaData = PVMFMediaData::createMediaData(mediaDataImpl, aSockConfig.iMemPool->iMediaDataMemPool););
if (err != OsclErrNone)
{
StartRecvWaitOnMemory(aSockConfig);
status = PVMFPending;
//wait on memory pool callback "notifyfreechunkavailable"
goto Exit;
}
// Retrieve memory fragment to write to
OsclRefCounterMemFrag refCtrMemFragOut;
aSockConfig.iPendingRecvMediaData->getMediaFragment(0, refCtrMemFragOut);
//Issue the Oscl socket request.
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumRecv++;
aSockConfig.iPortStats.StartRecvTime(EPVSocketPortRecvOperation_Recv);
#endif
TPVSocketEvent retVal = aSockConfig.iTCPSocket->Recv((uint8*)refCtrMemFragOut.getMemFragPtr()
, refCtrMemFragOut.getCapacity()
, TIMEOUT_RECV);
if (EPVSocketPending != retVal)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::Recv() failed retVal=%d Ln %d", retVal, __LINE__));
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartRecvOperation Recv FAILED - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
status = PVMFFailure;
goto Exit;
}
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartRecvOperation Recv Pending - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
status = PVMFPending;
//wait on HandleSocketEvent callback.
}
else
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::StartRecvOperation() Unexpected error, no socket"));
}
Exit:
//Handle synchronous completion or failure.
if (status != PVMFPending
&& status != PVMFFailure)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::StartRecvOperation: Error. status=%d", status));
}
if (status != PVMFPending)
status = RecvOperationComplete(aSockConfig, status, NULL);
return status;
}
//The current recv operation is complete. Process the result and continue with the next operation.
//The status is in "aStatus".
//If there was a socket callback, the result is in "aSocketActivity".
PVMFStatus PVMFSocketNode::RecvOperationComplete(SocketPortConfig& aSockConfig, PVMFStatus aStatus, PVMFSocketActivity* aSocketActivity)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::RecvOperationComplete() In"));
//status should never be pending here
OSCL_ASSERT(aStatus != PVMFPending);
PVMFStatus status = aStatus;
bool recvOperationCanceled = aSockConfig.iState.iRecvOperationCanceled;
//Update the state
TPVSocketPortRecvOperation curOp = aSockConfig.iState.iRecvOperation;
aSockConfig.iState.iRecvOperation = EPVSocketPortRecvOperation_None;
aSockConfig.iState.iRecvOperationStatus = aStatus;
aSockConfig.iState.iRecvOperationCanceled = false;
switch (curOp)
{
case EPVSocketPortRecvOperation_WaitOnMemory:
//a memory wait is complete.
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::RecvOperationComplete WaitOnMemory - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
break;
case EPVSocketPortRecvOperation_WaitOnConnectedPort:
//a port wait is complete
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::RecvOperationComplete WaitOnConnectedPort - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
//there may be a prior recv or recvfrom result that was waiting on this
if (aSockConfig.iSocketRecvActivity.iValid)
{
aSockConfig.iSocketRecvActivity.iValid = false;
switch (aSockConfig.iSocketRecvActivity.iFxn)
{
case EPVSocketRecv:
HandleRecvComplete(aSockConfig
, aSockConfig.iSocketRecvActivity.iStatus
, &aSockConfig.iSocketRecvActivity, recvOperationCanceled);
break;
case EPVSocketRecvFrom:
HandleRecvFromComplete(aSockConfig
, aSockConfig.iSocketRecvActivity.iStatus
, &aSockConfig.iSocketRecvActivity, recvOperationCanceled);
break;
default:
OSCL_ASSERT(0);//invalid arg
break;
}
}
break;
case EPVSocketPortRecvOperation_Recv:
//TCP receive is complete
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::RecvOperationComplete Recv - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
HandleRecvComplete(aSockConfig, aStatus, aSocketActivity, recvOperationCanceled);
break;
case EPVSocketPortRecvOperation_RecvFrom:
//UDP receive from is complete
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::RecvOperationComplete RecvFrom - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
HandleRecvFromComplete(aSockConfig, aStatus, aSocketActivity, recvOperationCanceled);
break;
default:
OSCL_ASSERT(0);//add code for this case
break;
}
//Receive may have been blocked waiting on this operation to complete,
//so start/resume receives now if possible
if (CanReceive(aSockConfig))
status = StartRecvOperation(aSockConfig);
//Detect connect "Wait on send/recv idle" complete
if (aSockConfig.iState.iConnectOperation == EPVSocketPortConnectOperation_WaitOnSendRecvIdle
&& aSockConfig.iState.iRecvOperation == EPVSocketPortRecvOperation_None
&& aSockConfig.iState.iSendOperation == EPVSocketPortSendOperation_None)
{
ConnectOperationComplete(aSockConfig, PVMFSuccess, NULL);
}
return status;
}
//Cancel any current receive operation.
//Return Success if the operation is canceled or Pending if we need to wait.
PVMFStatus PVMFSocketNode::CancelRecvOperation(SocketPortConfig& aSockConfig)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::CancelRecvOperation() In"));
PVMFStatus status = PVMFSuccess;
switch (aSockConfig.iState.iRecvOperation)
{
case EPVSocketPortRecvOperation_None:
break;
case EPVSocketPortRecvOperation_Recv:
if (aSockConfig.iTCPSocket)
{
if (!aSockConfig.iState.iRecvOperationCanceled)
{
aSockConfig.iState.iRecvOperationCanceled = true;
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelRecvOperation Calling CancelRecv - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iTCPSocket->CancelRecv();
}
status = PVMFPending;
//wait on recv to complete in HandleSocketEvent
}
break;
case EPVSocketPortRecvOperation_RecvFrom:
if (aSockConfig.iUDPSocket)
{
if (!aSockConfig.iState.iRecvOperationCanceled)
{
aSockConfig.iState.iRecvOperationCanceled = true;
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelRecvOperation Calling CancelRecvFrom - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iUDPSocket->CancelRecvFrom();
}
status = PVMFPending;
//wait on recvfrom to complete in HandleSocketEvent
}
break;
case EPVSocketPortRecvOperation_WaitOnConnectedPort:
//just clear the state
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelRecvOperation Cancelling WaitOnConnectedPort - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iState.iRecvOperation = EPVSocketPortRecvOperation_None;
//also discard the received data and the associated recv activity
if (aSockConfig.iSocketRecvActivity.iValid)
aSockConfig.iSocketRecvActivity.iValid = false;
if (aSockConfig.iPendingRecvMediaData.GetRep())
aSockConfig.iPendingRecvMediaData.Unbind();
break;
case EPVSocketPortRecvOperation_WaitOnMemory:
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelRecvOperation Cancelling WaitOnMemory - SockId=%d, Mime=%s ", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
if (aSockConfig.iMemPool != NULL)
{
if (aSockConfig.iMemPool->iInternalAlloc)
aSockConfig.iMemPool->iInternalAlloc->CancelFreeChunkAvailableCallback();
if (aSockConfig.iMemPool->iSharedBufferAlloc)
aSockConfig.iMemPool->iSharedBufferAlloc->CancelFreeChunkAvailableCallback();
}
//clear the state
aSockConfig.iState.iRecvOperation = EPVSocketPortRecvOperation_None;
break;
default:
OSCL_ASSERT(0);//add code for this case
status = PVMFFailure;
break;
}
return status;
}
//a memory pool callback
void SocketPortConfig::freechunkavailable(OsclAny* aContextData)
{
#if(ENABLE_SOCKET_NODE_STATS)
iPortStats.iNumFreeChunkCallback++;
#endif
OSCL_UNUSED_ARG(aContextData);
//complete the "wait on memory" state
if (iState.iRecvOperation == EPVSocketPortRecvOperation_WaitOnMemory)
{
#if(ENABLE_SOCKET_NODE_STATS)
iPortStats.EndRecvTime(iState.iRecvOperation);
#endif
iContainer->RecvOperationComplete(*this, PVMFSuccess, NULL);
}
}
void SocketPortConfig::freeblockavailable(OsclAny* aContextData)
{
freechunkavailable(aContextData);
}
//Handler for a "recv" complete. This handles two input cases:
// - Operation failed to initiate. Status will be in aStatus and aSocketActivity will be NULL.
// - Operation completed asynchronously via HandleSocketEvent callback. Status will be in aStatus
// and socket callback result will be in aSocketActivity.
//
//The end result of processing will be one of the following:
// - Received data sent to connected port
// - Result queued on socket activity queue for later processing.
// - Error or cancellation processed.
void PVMFSocketNode::HandleRecvComplete(SocketPortConfig& aSockConfig, PVMFStatus aStatus, PVMFSocketActivity* aSocketActivity, bool aRecvOperationCanceled)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::HandleRecvComplete() In"));
//operation should be complete when this is called.
OSCL_ASSERT(aStatus != PVMFPending);
int32 sockActivityEvent = aSocketActivity->iEvent;
PVMFStatus status = aStatus;
int32 tmplen;
aSockConfig.iTCPSocket->GetRecvData(&tmplen);
// If data length is <= zero and aSocketActivity->iEvent is EPVSocketSuccess, it should be treated as failure.
// On some platforms, its possible to get data length as zero and aSocketActivity->iEvent as EPVSocketSuccess.
if (EPVSocketSuccess == sockActivityEvent && tmplen <= 0)
{
PVMF_SOCKETNODE_LOGINFO((0, "PVMFSocketNode::HandleRecvFromComplete() Sucessful Recv With Zero Length. Treating as failure."));
sockActivityEvent = EPVSocketFailure;
status = PVMFFailure;
}
//If there's no socket activity input, then this must be a failure in initiating
//a Recv operation.
if (!aSocketActivity)
{
OSCL_ASSERT(status != PVMFSuccess);
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvComplete: Request Failed - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeErrorSocketFailure);
//release media data
if (aSockConfig.iPendingRecvMediaData.GetRep())
aSockConfig.iPendingRecvMediaData.Unbind();
return;
}
//Otherwise this is a result from Oscl Sockets of a Recv call.
//See whether we can send this data to the connected port
if (sockActivityEvent == EPVSocketSuccess)
{
if (!aSockConfig.iPVMFPort
|| !aSockConfig.iPVMFPort->IsConnected())
{
//port already disconnected-- just discard the data
if (aSockConfig.iPendingRecvMediaData.GetRep())
aSockConfig.iPendingRecvMediaData.Unbind();
return;
}
if (aSockConfig.iPVMFPort->IsOutgoingQueueBusy())
{
//wait on port so we can send recv data.
StartRecvWaitOnConnectedPort(aSockConfig, *aSocketActivity);
return;
}
}
//If we get here then it's time to process the recv result.
//Release media data on failure
if (status != PVMFSuccess)
{
if (aSockConfig.iPendingRecvMediaData.GetRep())
aSockConfig.iPendingRecvMediaData.Unbind();
}
switch (sockActivityEvent)
{
case EPVSocketSuccess:
{
//Get data length and set media buffer size
int32 len;
aSockConfig.iTCPSocket->GetRecvData(&len);
aSockConfig.iPendingRecvMediaData->setMediaFragFilledLen(0, len);
// Resize the buffer
if (aSockConfig.iMemPool)
{
OsclSharedPtr<PVMFMediaDataImpl> mediaMsgImpl;
aSockConfig.iPendingRecvMediaData->getMediaDataImpl(mediaMsgImpl);
aSockConfig.iMemPool->resizeSocketDataBuffer(mediaMsgImpl);
}
else
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleRecvComplete() ERROR:mempool not found"));
OSCL_ASSERT(0);
return ;
}
// set system time as timestamp
TimeValue currentTime;
currentTime.set_to_current_time();
int32 currentMilliSec = currentTime.to_msec();
aSockConfig.iPendingRecvMediaData->setTimestamp((PVMFTimestamp)currentMilliSec);
//queue to next port
PVMFSharedMediaMsgPtr aMediaMsgPtr;
convertToPVMFMediaMsg(aMediaMsgPtr, aSockConfig.iPendingRecvMediaData);
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumQueueOutgoingTCPMsg++;
#endif
PVMFStatus status = aSockConfig.iPVMFPort->QueueOutgoingMsg(aMediaMsgPtr);
if (status != PVMFSuccess)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvComplete: QueueOutgoingMsg Failed - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleRecvComplete() ERROR:%d, Outgoing queue size=%d. Data discarded!", status, aSockConfig.iPVMFPort->OutgoingMsgQueueSize()));
ReportErrorEvent(PVMFErrPortProcessing);
return ;
}
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::HandleRecvComplete: Request Succcess - SockId=%d, Mime=%s, Len=%d", aSockConfig.iSockId, aSockConfig.iMime.get_str(), len));
}
break;
case EPVSocketTimeout:
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvComplete: Request TimedOut - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleRecvComplete() ERROR EPVSocketTimeout Ln %d", __LINE__));
ReportSocketNodeError(PVMFErrTimeout, PVMFSocketNodeErrorSocketTimeOut);
}
break;
case EPVSocketCancel:
//On some OS we may get EPVSocketCancel for recv failures, so check whether this
//was response to a cancel command or not.
if (aRecvOperationCanceled)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvComplete: Request Cancelled - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
PVMF_SOCKETNODE_LOGINFO((0, "PVMFSocketNode::HandleRecvComplete() EPVSocketCancel"));
break;
}
//else fallthrough to the failure processing...
case EPVSocketFailure:
{
//After a receive failure, we may need to do a TCP shutdown.
//Check what else is currently happening on the port.
switch (aSockConfig.iState.iSequence)
{
case EPVSocketPortSequence_RequestPort:
case EPVSocketPortSequence_InputConnectMsg:
case EPVSocketPortSequence_InputDisconnectMsg:
case EPVSocketPortSequence_SocketCleanup:
//some of these conditions may be impossible, but in any case,
//it doesn't make sense to start a shutdown sequence during
//any of these. Ignore the error.
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvComplete: Request Failed, ignoring - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
break;
case EPVSocketPortSequence_None:
case EPVSocketPortSequence_InputDataMsg:
//for these cases, start a shutdown sequence
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvComplete: Request Failed, starting shutdown sequence - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
//start the sequence
StartSequence(aSockConfig, EPVSocketPortSequence_SocketCleanup);
break;
default:
//need code to handle this case.
OSCL_ASSERT(0);
break;
}
}
break;
default:
OSCL_ASSERT(0);
break;
}
}
//Handler for a "recv from" complete. This handles two input cases:
// - Operation failed to initiate. Status will be in aStatus and aSocketActivity will be NULL.
// - Operation completed asynchronously via HandleSocketEvent callback. Status will be in aStatus
// and socket callback result will be in aSocketActivity.
//
//The end result of processing will be one of the following:
// - Received data sent to connected port
// - Result queued on socket activity queue for later processing.
// - Error or cancellation processed.
void PVMFSocketNode::HandleRecvFromComplete(SocketPortConfig& aSockConfig, PVMFStatus aStatus, PVMFSocketActivity* aSocketActivity, bool aRecvOperationCanceled)
{
OSCL_UNUSED_ARG(aStatus);
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::HandleRecvFromComplete() In"));
//operation should be complete when this is called.
OSCL_ASSERT(aStatus != PVMFPending);
//If there's no socket activity input, then this must be a failure in initiating
//a RecvFrom operation.
if (!aSocketActivity)
{
OSCL_ASSERT(aStatus != PVMFSuccess);
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvFromComplete: Request Failed - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeErrorSocketFailure);
//release media data
if (aSockConfig.iPendingRecvMediaData.GetRep())
aSockConfig.iPendingRecvMediaData.Unbind();
return;
}
//Otherwise this is a result from Oscl Sockets of a RecvFrom call.
//Handle "truncated packet" error, in which case we still have data even though
//the result is EPVSocketFailure.
int32 dataLen = 0;
uint8* dataPtr = NULL;
if (aSocketActivity->iEvent == EPVSocketFailure
|| aSocketActivity->iEvent == EPVSocketSuccess)
{
if (aSockConfig.iUDPSocket)
dataPtr = aSockConfig.iUDPSocket->GetRecvData(&dataLen);
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumRecvFromPackets += aSockConfig.iRecvFromPacketLen.size();
if (aSockConfig.iRecvFromPacketLen.size() > aSockConfig.iPortStats.iMaxRecvFromPackets)
aSockConfig.iPortStats.iMaxRecvFromPackets = aSockConfig.iRecvFromPacketLen.size();
#endif
}
//Handle cases where we got some data
if (dataLen > 0)
{
//See if we can pass this data along to the connected port. If not, then queue the
//result for later processing.
if (aSockConfig.iPVMFPort->IsOutgoingQueueBusy())
{
//outgoing queue is busy-- must queue this for later.
StartRecvWaitOnConnectedPort(aSockConfig, *aSocketActivity);
}
else
{
//report truncated packet.
if (aSocketActivity->iEvent == EPVSocketFailure)
{
PVUuid eventuuid = PVMFSocketNodeEventTypeUUID;
int32 infocode = PVMFSocketNodeInfoEventPacketTruncated;
ReportInfoEvent(PVMFErrCorrupt, NULL, &eventuuid, &infocode);
}
//set the media buffer size.
aSockConfig.iPendingRecvMediaData->setMediaFragFilledLen(0, dataLen);
//resize the buffer.
if (aSockConfig.iMemPool)
{
OsclSharedPtr<PVMFMediaDataImpl> mediaMsgImpl;
aSockConfig.iPendingRecvMediaData->getMediaDataImpl(mediaMsgImpl);
aSockConfig.iMemPool->resizeSocketDataBuffer(mediaMsgImpl);
}
else
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_ERR, (0, "PVMFSocketNode::HandleRecvFromComplete() ERROR:mempool not found"));
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvFromComplete: ERROR mempool not found - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
return;//unexpected, cleanup
}
#if SNODE_ENABLE_UDP_MULTI_PACKET
//The media buffer may contain multiple UDP packets. We need to
//re-package these into a media frag group.
//Allocate a new media frag group.
int32 err;
OsclSharedPtr< PVMFMediaDataImpl > mediaFragGroup;
err = Allocate(aSockConfig, mediaFragGroup);
if (err != OsclErrNone)
{
//Unexpected error. With current usage by JB node, this pool should
//never run out.
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvFromComplete: Unexpected mempool error - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeErrorSocketFailure);
//release media data
if (aSockConfig.iPendingRecvMediaData.GetRep())
aSockConfig.iPendingRecvMediaData.Unbind();
return;
}
//get the single frag data buffer pointer.
OsclRefCounterMemFrag singleFrag;
aSockConfig.iPendingRecvMediaData->getMediaFragment(0, singleFrag);
for (uint32 i = 0; i < aSockConfig.iRecvFromPacketLen.size(); dataPtr += aSockConfig.iRecvFromPacketLen[i++])
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::HandleRecvFromComplete: Fragment %d Len %d - SockId=%d, Mime=%s"
, i, aSockConfig.iRecvFromPacketLen[i], aSockConfig.iSockId, aSockConfig.iMime.get_str()));
//Create a new fragment with the existing refcount
//and the proper packet boundaries.
OsclRefCounterMemFrag multiFrag(singleFrag);
//now adjust the single frag to the actual packet boundary.
multiFrag.getMemFrag().ptr = dataPtr;
multiFrag.getMemFrag().len = aSockConfig.iRecvFromPacketLen[i];
mediaFragGroup->appendMediaFragment(multiFrag);
#if (PVLOGGER_INST_LEVEL > PVLOGMSG_INST_LLDBG)
if (aSockConfig.iRTP)
{
LogRTPHeaderFields(aSockConfig, multiFrag);
}
else if (aSockConfig.iRTCP)
{
LogRTCPHeaderFields(aSockConfig, multiFrag);
}
#endif
}
//Replace the original media data message with the new one.
//since we've linked the new frags to the original refcounter, we
//can just unbind the current message then re-allocate it with the
//new data.
aSockConfig.iPendingRecvMediaData.Unbind();
err = CreateMediaData(aSockConfig, mediaFragGroup);
if (err != OsclErrNone)
{
//unexpected since we just freed one message
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvFromComplete: Unexpected mempool error - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeErrorSocketFailure);
return;
}
#else //SNODE_ENABLE_UDP_MULTI_PACKET
#if (PVLOGGER_INST_LEVEL > PVLOGMSG_INST_LLDBG)
if (aSockConfig.iRTP)
{
OsclRefCounterMemFrag memfrag;
aSockConfig.iPendingRecvMediaData->getMediaFragment(0, memfrag);
LogRTPHeaderFields(aSockConfig, memfrag);
}
else if (aSockConfig.iRTCP)
{
OsclRefCounterMemFrag memfrag;
aSockConfig.iPendingRecvMediaData->getMediaFragment(0, memfrag);
LogRTCPHeaderFields(aSockConfig, memfrag);
}
#endif
#endif //SNODE_ENABLE_UDP_MULTI_PACKET
//push the received data to the connected port
PVMFSharedMediaMsgPtr mediaMsgPtr;
convertToPVMFMediaMsg(mediaMsgPtr, aSockConfig.iPendingRecvMediaData);
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumQueueOutgoingUDPMsg++;
#endif
PVMFStatus status = aSockConfig.iPVMFPort->QueueOutgoingMsg(mediaMsgPtr);
if (status != PVMFSuccess)
{
//should never get here because we already checked outgoign queue earlier.
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleRecvFromComplete() ERROR:%d, Outgoing queue size=%d. Data discarded!", status, aSockConfig.iPVMFPort->OutgoingMsgQueueSize()));
ReportErrorEvent(PVMFInfoOverflow);
}
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::HandleRecvFromComplete: Request Succcess - SockId=%d, Mime=%s, Len=%d", aSockConfig.iSockId, aSockConfig.iMime.get_str(), dataLen));
}
}
else
{
//handle error, cancel, and timeout cases.
//release media data.
if (aSockConfig.iPendingRecvMediaData.GetRep())
aSockConfig.iPendingRecvMediaData.Unbind();
//report error events.
switch (aSocketActivity->iEvent)
{
case EPVSocketCancel:
//On some OS we may get EPVSocketCancel for recv failures, so check whether this
//was response to a cancel command or not.
if (aRecvOperationCanceled)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvFromComplete: Request Cancelled - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
break;
}
//else fallthrough to failure processing...
case EPVSocketFailure:
//report RTP failures.
//Note: do not report RTCP errors, because some servers send zero-byte packets
//and we want to be tolerant of that case.
if (aSockConfig.iTag == PVMF_SOCKET_NODE_PORT_TYPE_SOURCE)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvFromComplete: Request Failed - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeErrorSocketFailure);
}
break;
case EPVSocketTimeout:
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::HandleRecvFromComplete: Request TimedOut - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleRecvFromComplete() ERROR EPVSocketTimeout Ln %d", __LINE__));
ReportSocketNodeError(PVMFErrTimeout, PVMFSocketNodeErrorSocketTimeOut);
}
break;
default:
break;
}
}
}
//////////////////////////////////
// End Receive Operation Handlers
//////////////////////////////////
//////////////////////////////////
// Send Operation Handlers
//////////////////////////////////
//Start a send operation on the port's socket.
//Assume the caller already verified that the port can send now.
//If there is a failure, this routine will call the "send operation complete" routine to
//do error handling or continue to the next state.
PVMFStatus PVMFSocketNode::StartSendOperation(SocketPortConfig& aSockConfig, PVMFSharedMediaMsgPtr& aMsg)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::StartSendOperation() In"));
//caller should have verified we can send now.
OSCL_ASSERT(aSockConfig.iState.iSendOperation == EPVSocketPortSendOperation_None);
//caller should provide a media data message as input
OSCL_ASSERT(aMsg->getFormatID() == PVMF_MEDIA_MSG_DATA_FORMAT_ID);
//there should be either a UDP or TCP socket on this port
if (!aSockConfig.iTCPSocket && !aSockConfig.iUDPSocket)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartSendOperation ERROR no socket - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
return PVMFFailure;
}
aSockConfig.iPVMFPort->iNumFramesConsumed++;
// Retrieve memory fragment to write to
if (aSockConfig.iPendingSendMediaData.GetRep())
aSockConfig.iPendingSendMediaData.Unbind();
convertToPVMFMediaData(aSockConfig.iPendingSendMediaData, aMsg);
OsclRefCounterMemFrag refCtrMemFragOut;
aSockConfig.iPendingSendMediaData->getMediaFragment(0, refCtrMemFragOut);
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartSendOperation - SockId=%d, Mime=%s, Len=%d", aSockConfig.iSockId, aSockConfig.iMime.get_str(), refCtrMemFragOut.getMemFragSize()));
PVMFStatus status = PVMFFailure;
if (aSockConfig.iUDPSocket)
{
aSockConfig.iState.iSendOperation = EPVSocketPortSendOperation_SendTo;
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumSendTo++;
aSockConfig.iPortStats.StartSendTime(aSockConfig.iState.iSendOperation);
#endif
TPVSocketEvent retVal = aSockConfig.iUDPSocket->SendTo((uint8*)refCtrMemFragOut.getMemFragPtr()
, refCtrMemFragOut.getMemFragSize()
, aSockConfig.iAddr.iRemoteAdd
, TIMEOUT_SENDTO);
if (retVal == EPVSocketPending)
status = PVMFPending;//wait on HandleSocketEvent callback
else
status = PVMFFailure;
if (PVMFPending != status)
{
//error
PVMF_SOCKETNODE_LOGERROR((0, "0x%x PVMFSocketNode::SendTo() failed retVal=%d Ln %d", this, retVal, __LINE__));
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::StartSendOperation: UDP - Error. status=%d", status));
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartSendOperation SendTo FAILED - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
}
else
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartSendOperation SendTo Pending - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
}
}
else if (aSockConfig.iTCPSocket)
{
aSockConfig.iState.iSendOperation = EPVSocketPortSendOperation_Send;
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumSend++;
aSockConfig.iPortStats.StartSendTime(aSockConfig.iState.iSendOperation);
#endif
TPVSocketEvent retVal = aSockConfig.iTCPSocket->Send((uint8*)refCtrMemFragOut.getMemFragPtr(),
refCtrMemFragOut.getMemFragSize(),
TIMEOUT_SEND);
if (retVal == EPVSocketPending)
status = PVMFPending;//wait on HandleSocketEvent callback
else
status = PVMFFailure;
if (PVMFPending != status)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::StartSendOperation: TCP - Error. status=%d", status));
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::StartSendOperation Calling Send FAILED - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
}
else
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartSendOperation Send Pending - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
}
}
//Handle synchronous completion or failures
if (status != PVMFPending)
status = SendOperationComplete(aSockConfig, status, NULL);
return status;
}
//The current send operation is complete and status is in "aStatus".
//If there was a socket callback, the result is in "aSocketActivity".
//Process the result and continue with the next operation if needed.
PVMFStatus PVMFSocketNode::SendOperationComplete(SocketPortConfig& aSockConfig, PVMFStatus aStatus, PVMFSocketActivity* aSocketActivity)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::SendOperationComplete()"));
OSCL_ASSERT(aStatus != PVMFPending);
PVMFStatus status = aStatus;
//Update the send state
TPVSocketPortSendOperation curOp = aSockConfig.iState.iSendOperation;
aSockConfig.iState.iSendOperation = EPVSocketPortSendOperation_None;
aSockConfig.iState.iSendOperationStatus = aStatus;
aSockConfig.iState.iSendOperationCanceled = false;
//Release the media buffer after each send operation, regardless of success/fail.
if (aSockConfig.iPendingSendMediaData.GetRep() != NULL)
aSockConfig.iPendingSendMediaData.Unbind();
if (aSocketActivity)
{
//Datapath logging
switch (aSocketActivity->iEvent)
{
case EPVSocketSuccess:
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::SendOperationComplete - Success - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
break;
case EPVSocketTimeout:
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::SendOperationComplete - TimeOut - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
break;
case EPVSocketFailure:
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::SendOperationComplete - Failed - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
break;
case EPVSocketCancel:
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::SendOperationComplete - Cancelled - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
break;
default:
OSCL_ASSERT(0);
break;
}
}
//report TCP errors.
if (aStatus != PVMFSuccess
&& curOp == EPVSocketPortSendOperation_Send)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::SendOperationComplete - Send Failed - "
"SockId=%d, Mime=%s, Operaton=%d",
aSockConfig.iSockId, aSockConfig.iMime.get_str(), aSockConfig.iState.iSendOperation));
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::SendOperationComplete: Send/SendTo() ERROR status=%d, Ln %d", status, __LINE__));
ReportSocketNodeError(PVMFErrResource, PVMFSocketNodeErrorTCPSocketSendError);
}
//This completes an input data message sequence
if (aSockConfig.iState.iSequence == EPVSocketPortSequence_InputDataMsg)
SequenceComplete(aSockConfig, aStatus);
//Detect connect "Wait on send/recv idle" complete
if (aSockConfig.iState.iConnectOperation == EPVSocketPortConnectOperation_WaitOnSendRecvIdle
&& aSockConfig.iState.iRecvOperation == EPVSocketPortRecvOperation_None
&& aSockConfig.iState.iSendOperation == EPVSocketPortSendOperation_None)
{
ConnectOperationComplete(aSockConfig, PVMFSuccess, NULL);
}
return status;
}
//Cancel the current send operation.
//Return Success if the operation is canceled, or Pending if we need to wait.
PVMFStatus PVMFSocketNode::CancelSendOperation(SocketPortConfig& aSockConfig)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::CancelSendOperation()"));
PVMFStatus status = PVMFSuccess;
switch (aSockConfig.iState.iSendOperation)
{
case EPVSocketPortSendOperation_None:
break;
case EPVSocketPortSendOperation_Send:
if (aSockConfig.iTCPSocket)
{
if (!aSockConfig.iState.iSendOperationCanceled)
{
aSockConfig.iState.iSendOperationCanceled = true;
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelSendOperation Calling CancelSend - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iTCPSocket->CancelSend();
}
status = PVMFPending;
//wait on send to complete in HandleSocketEvent
}
break;
case EPVSocketPortSendOperation_SendTo:
if (aSockConfig.iUDPSocket)
{
if (!aSockConfig.iState.iSendOperationCanceled)
{
aSockConfig.iState.iSendOperationCanceled = true;
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CancelSendOperation Calling CancelSendTo - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
aSockConfig.iUDPSocket->CancelSendTo();
}
status = PVMFPending;
//wait on sendto to complete in HandleSocketEvent
}
break;
default:
OSCL_ASSERT(0);
status = PVMFFailure;
break;
}
return status;
}
//////////////////////////////////
// End Send Operation Handlers
//////////////////////////////////
//////////////////////////////////////
// Sequence Handler
//////////////////////////////////////
static const char* const TPVSocketPortSequenceStr[] =
{
"EPVSocketPortSequence_None"
, "EPVSocketPortSequence_RequestPort"
, "EPVSocketPortSequence_InputConnectMsg"
, "EPVSocketPortSequence_InputDataMsg"
, "EPVSocketPortSequence_InputDisconnectMsg"
, "EPVSocketPortSequence_SocketCleanup"
, "EPVSocketPortSequence_Last"
} ;
//Start a sequence on a port.
//This routine will complete all steps possible unless some operation has
//asynchronous completion.
//This call may actually complete the sequence.
//This routine documents the series of steps for each sequence.
PVMFStatus PVMFSocketNode::StartSequence(SocketPortConfig& aSockConfig, TPVSocketPortSequence aSequence, OsclAny* aParam)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartSequence, '%s' - SockId=%d, Mime=%s"
, TPVSocketPortSequenceStr[aSequence]
, aSockConfig.iSockId, aSockConfig.iMime.get_cstr()));
//Set the sequence.
aSockConfig.iState.iSequence = aSequence;
switch (aSequence)
{
case EPVSocketPortSequence_RequestPort:
//Sequence:
// 1. GetHostByName,
// 2. Connect
// 3. Command complete.
return StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_GetHostByName);
case EPVSocketPortSequence_InputConnectMsg:
//Sequence:
// If already connected,
// 1. Shutdown
// 2. Cancel send/recv operation if needed.
// 3. Wait for send/recv idle if needed.
// 4. Wait for connected port if needed.
// 5. Close socket + send EOS to connected port + delete socket
// or add to cleanup queue & trigger AO.
// 6. Create
// 7. GetHostByName
// 8. Connect
// Else if not already connected,
// 1. GetHostByName
// 2. Connect
if (aSockConfig.iTCPSocket)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::StartSequence - Disconnecting before Connect SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_cstr()));
return StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_Shutdown);
}
else
{
return StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_GetHostByName);
}
break;
case EPVSocketPortSequence_InputDataMsg:
//Sequence:
// 1. Send
OSCL_ASSERT(aParam);
return StartSendOperation(aSockConfig, *((PVMFSharedMediaMsgPtr*)aParam));
case EPVSocketPortSequence_InputDisconnectMsg:
//Sequence:
// 1. Shutdown
// 2. Cancel send/recv operation if needed.
// 3. Wait for send/recv idle if needed.
// 4. Wait for connected port if needed.
// 5. Close socket + send EOS to connected port + delete socket
// or add to cleanup queue & trigger AO.
return StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_Shutdown);
case EPVSocketPortSequence_SocketCleanup:
if (aSockConfig.iTCPSocket)
{
//TCP Shutdown Sequence:
// 1. Cancel current connect operation if needed.
// 2. Cancel send/recv operation if needed.
// 3. Shutdown
// 4. Wait for send/recv idle if needed.
// 5. Wait for connected port if needed.
// 6. Close socket + send EOS to connected port + delete socket
// or add to cleanup queue & trigger AO.
if (aSockConfig.iState.iConnectOperation == EPVSocketPortConnectOperation_Shutdown)
{
return PVMFPending;//wait on completion in HandleSocketEvent
}
if (CancelConnectOperation(aSockConfig) == PVMFPending)
{
return PVMFPending;//wait on connect operation to cancel
//(in HandleSocketEvent or HandleDNSEvent)
}
else
{
//Go ahead and start any send/recv cancel since they can happen simultaneously
//with the Shutdown operation.
CancelRecvOperation(aSockConfig);
CancelSendOperation(aSockConfig);
return StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_Shutdown);
}
}
else if (aSockConfig.iUDPSocket)
{
//UDP Shutdown Sequence:
// 1. Cancel all current operations
// 2. Wait for send/recv idle if needed.
// 3. Close socket + delete socket or add to cleanup queue & trigger AO.
switch (aSockConfig.iState.iConnectOperation)
{
case EPVSocketPortConnectOperation_WaitOnSendRecvIdle:
//keep waiting for completion (in RecvOperationComplete or SendOperationComplete)
return PVMFPending;
case EPVSocketPortConnectOperation_None:
CancelRecvOperation(aSockConfig);
CancelSendOperation(aSockConfig);
return StartConnectOperation(aSockConfig, EPVSocketPortConnectOperation_WaitOnSendRecvIdle);
default:
//unexpected
OSCL_ASSERT(0);
return PVMFFailure;
}
}
//else
{
//nothing needed.
SequenceComplete(aSockConfig, PVMFSuccess);
return PVMFSuccess;
}
break;
default:
//add code to handle this case
OSCL_ASSERT(0);
return PVMFFailure;
}
}
//The current multi-operation sequence is complete.
//Process the result and continue with the next operation if needed.
void PVMFSocketNode::SequenceComplete(SocketPortConfig& aSockConfig, PVMFStatus aStatus)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::SequenceComplete() Sequence %d Status %d ", aSockConfig.iState.iSequence, aStatus));
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::SequenceComplete '%s' - SockId=%d, Mime=%s"
, TPVSocketPortSequenceStr[aSockConfig.iState.iSequence]
, aSockConfig.iSockId, aSockConfig.iMime.get_str()));
//Update the state
TPVSocketPortSequence curSequence = aSockConfig.iState.iSequence;
aSockConfig.iState.iSequence = EPVSocketPortSequence_None;
aSockConfig.iState.iSequenceStatus = aStatus;
switch (curSequence)
{
case EPVSocketPortSequence_RequestPort:
//may need to complete the node command
if (iCurrentCmdQueue.size()
&& iCurrentCmdQueue.front().iCmd == PVMF_GENERIC_NODE_REQUESTPORT)
{
//@TODO would be nice to cleanup the allocated port in case of a failure.
CommandComplete(iCurrentCmdQueue, iCurrentCmdQueue.front(), aStatus, iRequestedPort);
}
break;
case EPVSocketPortSequence_InputConnectMsg:
case EPVSocketPortSequence_InputDisconnectMsg:
case EPVSocketPortSequence_InputDataMsg:
//If we've just completed an input port message, this may complete a node Flush command
if (!iCurrentCmdQueue.empty()
&& iCurrentCmdQueue.front().iCmd == PVMF_GENERIC_NODE_FLUSH)
{
//Flush is done when all input ports are empty.
for (uint32 i = 0; i < iPortVector.size(); i++)
{
if (iPortVector[i]->IncomingMsgQueueSize() > 0)
return;//keep waiting
}
//All ports empty-- complete the Flush.
CommandComplete(iCurrentCmdQueue, iCurrentCmdQueue.front(), PVMFSuccess);
}
break;
case EPVSocketPortSequence_SocketCleanup:
//When socket cleanup is complete, decrement the "stop node activity"
//counter and look for completion of all ports.
//This port is done-- decrement the counter
iNumStopPortActivityPending--;
//When counter reaches zero, all ports are done and the sequence is complete.
if (iNumStopPortActivityPending == 0)
{
//Reset the counter to the "idle" value.
iNumStopPortActivityPending = (-1);
//There may be a Reset or CancelAll command waiting on
//this to complete.
if (!iCurrentCmdQueue.empty()
&& iCurrentCmdQueue.front().iCmd == PVMF_GENERIC_NODE_RESET)
{
CommandComplete(iCurrentCmdQueue, iCurrentCmdQueue.front(), PVMFSuccess);
}
else if (!iCancelCmdQueue.empty()
&& iCancelCmdQueue.front().iCmd == PVMF_GENERIC_NODE_CANCELALLCOMMANDS)
{
CommandComplete(iCancelCmdQueue, iCancelCmdQueue.front(), PVMFSuccess);
}
//ports may be cleaned up now, so exit this routine now to
//avoid a crash from further port processing.
return;
}
break;
default:
break;
}
//All sequences other than input data message will
//block receive operations, so may need to start or resume receives now.
if (curSequence != EPVSocketPortSequence_InputDataMsg
&& (aStatus == PVMFSuccess)
&& CanReceive(aSockConfig))
{
StartRecvOperation(aSockConfig);
}
//Input message processing may have been blocked waiting on
//this sequence to complete, so resume now.
if (CanProcessIncomingMsg(aSockConfig)
&& (aStatus == PVMFSuccess))
{
ProcessIncomingMsg(aSockConfig);
}
}
//////////////////////////////////////
// End Sequence Handler
//////////////////////////////////////
//////////////////////////////////////
// Command Handlers
//////////////////////////////////////
//See if we can process a new input command right now.
bool PVMFSocketNode::CanProcessCommand()
{
//We may need to wait on
//1) a new command to arrive
//2) some prior command to complete
//Note: this allows a Cancel to interrupt the current command, but it does
//not allow a Cancel to interrupt another cancel.
return (!iPendingCmdQueue.empty()
&& (iCurrentCmdQueue.empty()
|| (iPendingCmdQueue.front().hipri() && iCancelCmdQueue.empty())));
}
//Process an input command.
void PVMFSocketNode::ProcessCommand(PVMFSocketNodeCmdQ& aCmdQ, PVMFSocketNodeCommand& aCmd)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNodeCommand::ProcessCommand() in"));
PVMFStatus status = PVMFFailure;
iCommandErrorCode = PVMFSocketNodeErrorEventStart; //no error
if (aCmd.hipri())
{
//calling logic should prevent multiple cancels in progress.
OSCL_ASSERT(iCancelCmdQueue.empty());
switch (aCmd.iCmd)
{
case PVMF_GENERIC_NODE_CANCELALLCOMMANDS:
status = DoCancelAllCommands(aCmd);
break;
case PVMF_GENERIC_NODE_CANCELCOMMAND:
status = DoCancelCommand(aCmd);
break;
default://unknown command type
status = PVMFFailure;
break;
}
//If the command was not finished in this call, then move it to the
//"cancel cmd" queue where it will remain until it is completed.
if (status == PVMFPending)
{
//move the new cmd to the "cancel cmd" queue where it will
//remain until complete.
int32 err;
OSCL_TRY(err, iCancelCmdQueue.StoreL(aCmd););
//we reserved space in the CancelCmdQueue already, so
//it should not be possible to fail here.
OSCL_ASSERT(err == OsclErrNone);
//erase the original command
aCmdQ.Erase(&aCmd);
}
else
{
CommandComplete(aCmdQ, aCmd, status, NULL);
}
}
else
{
//calling logic should prevent multiple commands in progress.
OSCL_ASSERT(iCurrentCmdQueue.empty());
OsclAny* eventData = NULL;
switch (aCmd.iCmd)
{
case PVMF_GENERIC_NODE_QUERYUUID:
status = DoQueryUuid(aCmd);
break;
case PVMF_GENERIC_NODE_QUERYINTERFACE:
status = DoQueryInterface(aCmd);
break;
case PVMF_GENERIC_NODE_REQUESTPORT:
{
iRequestedPort = NULL;
status = DoRequestPort(aCmd, iRequestedPort);
eventData = iRequestedPort;
break;
}
case PVMF_GENERIC_NODE_RELEASEPORT:
status = DoReleasePort(aCmd);
break;
case PVMF_GENERIC_NODE_INIT:
status = DoInit(aCmd);
break;
case PVMF_GENERIC_NODE_PREPARE:
status = DoPrepare(aCmd);
if (status == PVMFSuccess)
{
ChangeExternalState(EPVMFNodePrepared);
}
break;
case PVMF_GENERIC_NODE_START:
status = DoStart(aCmd);
break;
case PVMF_GENERIC_NODE_STOP:
status = DoStop(aCmd);
break;
case PVMF_GENERIC_NODE_FLUSH:
status = DoFlush(aCmd);
break;
case PVMF_GENERIC_NODE_PAUSE:
status = DoPause(aCmd);
break;
case PVMF_GENERIC_NODE_RESET:
status = DoReset(aCmd);
break;
default://unknown command type
status = PVMFFailure;
break;
}
//If the command was not finished in this call, then move it to the
//"current cmd" queue where it will remain until it is completed.
if (status == PVMFPending)
{
//move the new cmd to the "current cmd" queue where it will
//remain until complete.
int32 err;
OSCL_TRY(err, iCurrentCmdQueue.StoreL(aCmd););
//we reserved space in the CurrentCmdQueue already, so
//it should not be possible to fail here.
OSCL_ASSERT(err == OsclErrNone);
//erase the original command
aCmdQ.Erase(&aCmd);
}
else
{
CommandComplete(aCmdQ, aCmd, status, eventData);
}
}
}
//Called to complete a node command.
void PVMFSocketNode::CommandComplete(PVMFSocketNodeCmdQ& aCmdQ,
PVMFSocketNodeCommand& aCmd,
PVMFStatus aStatus,
OsclAny* aEventData,
PVUuid* aEventUUID,
int32* aEventCode)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode:CommandComplete Id %d Cmd %d Status %d Context %d Data %d"
, aCmd.iId, aCmd.iCmd, aStatus, aCmd.iContext, aEventData));
//See whether command processing was blocked on entry to this routine, so we can
//decide later whether it's necessary to trigger the AO to resume processsing.
bool canProcess = CanProcessCommand();
PVMFStatus status = aStatus;
//do standard state changes & other cmd completion.
if (aStatus == PVMFSuccess)
{
switch (aCmd.iCmd)
{
case PVMF_GENERIC_NODE_INIT:
ChangeExternalState(EPVMFNodeInitialized);
break;
case PVMF_GENERIC_NODE_PREPARE:
ChangeExternalState(EPVMFNodePrepared);
break;
case PVMF_GENERIC_NODE_START:
ChangeExternalState(EPVMFNodeStarted);
break;
case PVMF_GENERIC_NODE_STOP:
ChangeExternalState(EPVMFNodePrepared);
#if(ENABLE_SOCKET_NODE_STATS)
//dump the stats.
iSocketNodeStats.Log(iPortVector);
#endif
break;
case PVMF_GENERIC_NODE_PAUSE:
ChangeExternalState(EPVMFNodePaused);
break;
case PVMF_GENERIC_NODE_RESET:
ChangeExternalState(EPVMFNodeCreated);
//Complete the reset command.
{
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.Log(iPortVector);
#endif
//cleanup all ports.
CleanupPorts();
//go back to Idle state
iPortVector.Reconstruct();
SetState(EPVMFNodeIdle);
}
break;
case PVMF_GENERIC_NODE_CANCELALLCOMMANDS:
//Complete the reset command.
{
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.Log(iPortVector);
#endif
//Since "cancel all" is effectively used as a Reset,
//go ahead and cleanup all ports.
CleanupPorts();
}
break;
default:
break;
}
}
//create extended error message if an explicit error was provided in args
PVInterface* extif = NULL;
PVMFBasicErrorInfoMessage* errormsg = NULL;
if (aEventUUID && aEventCode)
{
errormsg = OSCL_NEW(PVMFBasicErrorInfoMessage, (*aEventCode, *aEventUUID, NULL));
extif = OSCL_STATIC_CAST(PVInterface*, errormsg);
}
//else if no error input, see if "command error code" was set.
else if (iCommandErrorCode != PVMFSocketNodeErrorEventStart)
{
PVUuid eventuuid = PVMFSocketNodeEventTypeUUID;
errormsg = OSCL_NEW(PVMFBasicErrorInfoMessage, (iCommandErrorCode, eventuuid, NULL));
extif = OSCL_STATIC_CAST(PVInterface*, errormsg);
}
//create response
PVMFCmdResp resp(aCmd.iId, aCmd.iContext, status, extif, aEventData);
PVMFSessionId session = aCmd.iSession;
//Erase the command from the queue.
aCmdQ.Erase(&aCmd);
//Report completion to the session observer.
ReportCmdCompleteEvent(session, resp);
if (errormsg)
{
errormsg->removeRef();
}
iCommandErrorCode = PVMFSocketNodeErrorEventStart;//reset.
//See if there was a pending cancel waiting on the current command to
//complete.
if (!iCancelCmdQueue.empty()
&& iCancelCmdQueue.front().iCmd == PVMF_GENERIC_NODE_CANCELCOMMAND)
{
CommandComplete(iCancelCmdQueue, iCancelCmdQueue.front(), PVMFSuccess);
}
//May need to resume command handling if the AO was blocked
//waiting on asynchronous command completion, but it's unblocked now.
if (!canProcess
&& CanProcessCommand()
&& IsAdded())
{
RunIfNotReady();
}
}
void PVMFSocketNode::ReportErrorEvent(PVMFEventType aEventType,
OsclAny* aEventData,
PVUuid* aEventUUID,
int32* aEventCode)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_ERR,
(0, "PVMFSocketNode:NodeErrorEvent Type %d Data %d"
, aEventType, aEventData));
if (aEventUUID && aEventCode)
{
PVMFBasicErrorInfoMessage* eventmsg =
OSCL_NEW(PVMFBasicErrorInfoMessage, (*aEventCode, *aEventUUID, NULL));
PVMFAsyncEvent asyncevent(PVMFErrorEvent,
aEventType,
NULL,
OSCL_STATIC_CAST(PVInterface*, eventmsg),
aEventData,
NULL,
0);
PVMFNodeInterface::ReportErrorEvent(asyncevent);
eventmsg->removeRef();
}
else
{
PVMFNodeInterface::ReportErrorEvent(aEventType, aEventData);
}
}
void PVMFSocketNode::ReportInfoEvent(PVMFEventType aEventType,
OsclAny* aEventData,
PVUuid* aEventUUID,
int32* aEventCode)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_MLDBG, iLogger, PVLOGMSG_STACK_TRACE,
(0, "PVMFSocketNode:NodeInfoEvent Type %d Data %d"
, aEventType, aEventData));
if (aEventUUID && aEventCode)
{
PVMFBasicErrorInfoMessage* eventmsg =
OSCL_NEW(PVMFBasicErrorInfoMessage, (*aEventCode, *aEventUUID, NULL));
PVMFAsyncEvent asyncevent(PVMFInfoEvent,
aEventType,
NULL,
OSCL_STATIC_CAST(PVInterface*, eventmsg),
aEventData,
NULL,
0);
PVMFNodeInterface::ReportInfoEvent(asyncevent);
eventmsg->removeRef();
}
else
{
PVMFNodeInterface::ReportInfoEvent(aEventType, aEventData);
}
}
PVMFCommandId PVMFSocketNode::QueueCommandL(PVMFSocketNodeCommand& aCmd)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::QueueCommandL()"));
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.iNumQueueNodeCmd++;
#endif
PVMFCommandId id;
id = iPendingCmdQueue.AddL(aCmd);
//This may be a processing trigger.
//Wakeup the AO if needed.
if (IsAdded()
&& CanProcessCommand())
{
RunIfNotReady();
}
return id;
}
PVMFStatus PVMFSocketNode::DoReset(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoReset()"));
/* This node allows a reset from any state */
return DoStopNodeActivity();
}
PVMFStatus PVMFSocketNode::DoQueryUuid(PVMFSocketNodeCommand& aCmd)
{
//This node supports Query UUID from any state
OSCL_String* mimetype;
Oscl_Vector<PVUuid, OsclMemAllocator> *uuidvec;
bool exactmatch;
aCmd.PVMFSocketNodeCommandBase::Parse(mimetype, uuidvec, exactmatch);
//Try to match the input mimetype against any of
//the custom interfaces for this node
//Match against custom interface1...
if (*mimetype == PVMF_SOCKET_NODE_EXTENSION_INTERFACE_MIMETYPE
//also match against base mimetypes for custom interface1,
//unless exactmatch is set.
|| (!exactmatch && *mimetype == PVMF_SOCKET_NODE_MIMETYPE)
|| (!exactmatch && *mimetype == PVMF_SOCKET_NODE_BASEMIMETYPE))
{
PVUuid uuid(PVMF_SOCKET_NODE_EXTENSION_INTERFACE_UUID);
uuidvec->push_back(uuid);
}
return PVMFSuccess;
}
bool PVMFSocketNode::queryInterface(const PVUuid& uuid, PVInterface*& iface)
{
iface = NULL;
if (uuid == PVUuid(PVMF_SOCKET_NODE_EXTENSION_INTERFACE_UUID))
{
if (!iExtensionInterface)
{
iExtensionInterface = OSCL_NEW(PVMFSocketNodeExtensionInterfaceImpl, (this));
}
if (iExtensionInterface)
{
return (iExtensionInterface->queryInterface(uuid, iface));
}
else
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::queryInterface: Error - Out of memory"));
OSCL_LEAVE(OsclErrNoMemory);
return false;
}
}
else
{
return false;
}
}
PVMFStatus PVMFSocketNode::DoQueryInterface(PVMFSocketNodeCommand& aCmd)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE,
(0, "PVMFSocketNode::DoQueryInterface"));
PVUuid* uuid;
PVInterface** ptr;
aCmd.PVMFSocketNodeCommandBase::Parse(uuid, ptr);
if (*uuid == PVUuid(PVMF_SOCKET_NODE_EXTENSION_INTERFACE_UUID))
{
if (!iExtensionInterface)
{
iExtensionInterface = OSCL_NEW(PVMFSocketNodeExtensionInterfaceImpl, (this));
}
if (iExtensionInterface)
{
if (iExtensionInterface->queryInterface(*uuid, *ptr))
{
return PVMFSuccess;
}
else
{
return PVMFErrNotSupported;
}
}
else
{
return PVMFErrNoMemory;
}
}
else
{//not supported
*ptr = NULL;
return PVMFErrNotSupported;
}
}
PVMFStatus PVMFSocketNode::DoRequestPort(PVMFSocketNodeCommand& aCmd, PVMFSocketPort* &aPort)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::DoRequestPort() In"));
//retrieve port tag.
int32 tag;
OSCL_String* portconfig;
aCmd.PVMFSocketNodeCommandBase::Parse(tag, portconfig);
//validate the tag...
switch (tag)
{
case PVMF_SOCKET_NODE_PORT_TYPE_SOURCE:
case PVMF_SOCKET_NODE_PORT_TYPE_SINK:
case PVMF_SOCKET_NODE_PORT_TYPE_PASSTHRU:
break;
default:
//bad port tag
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: Error - Invalid port tag"));
iCommandErrorCode = PVMFSocketNodeErrorInvalidPortTag;
return PVMFErrArgument;
}
SOCKET_ADDR sock_add;
sock_add.iTag = tag;
OSCL_HeapString<OsclMemAllocator> mime;
if (!ParseTransportConfig(portconfig, sock_add, mime))
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: ERROR - Invalid port config"));
iCommandErrorCode = PVMFSocketNodeErrorInvalidPortTag;
return PVMFErrArgument;
}
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::DoRequestPort tag %d config %s",
tag, portconfig->get_cstr()));
switch (sock_add.iProtocol)
{
case INET_TCP:
{
//Create a new port with no socket.
PVMFStatus status = AddPort(tag, aPort);
if (status != PVMFSuccess)
return status;
//create the socket config
SocketPortConfig* sock_config = OSCL_NEW(SocketPortConfig, ());
aPort->iConfig = sock_config;
sock_config->iPVMFPort = aPort;
sock_config->iMime = mime;
sock_config->iContainer = this;
sock_config->iSockId = iSocketID++;
sock_config->iTag = tag;
sock_config->iAddr = sock_add;
//create the mem pool
PVMFSocketNodeMemPool* memPool;
status = AllocatePortMemPool(tag, memPool);
if (status == PVMFSuccess)
sock_config->iMemPool = memPool;
if (status != PVMFSuccess)
return status;
//optionally create the socket and launch the connect sequence.
if (oscl_strstr(portconfig->get_cstr(), ";no-connect-on-create") != NULL)
{
//if "no create on connect" was selected then port creation is complete now.
status = PVMFSuccess;
}
else
{
//else create the socket and launch the connect sequence.
OSCL_ASSERT(sock_config->iState.iSequence == EPVSocketPortSequence_None);
status = StartSequence(*sock_config, EPVSocketPortSequence_RequestPort);
//may need to wait on connect completion here, or else there could
//be an error.
}
return status;
}
case INET_UDP:
{
//We should have already created this UDP socket in a call to AllocateConsecutivePorts.
SocketPortConfig* sockConfig = FindSocketPortConfig(sock_add);
if (!sockConfig)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: ERROR - Invalid protocol for port request"));
return PVMFFailure;
}
//The udp socket was already created using AllocateConsecutivePorts(),
//the but PVMFPort and memory pool were not yet created.
//Allocate a new PVMF port
PVMFStatus status = AddPort(tag, aPort);
if (status == PVMFSuccess)
{
PVMFSocketNodeMemPool* memPool = NULL;
status = AllocatePortMemPool(tag, memPool);
if (status == PVMFSuccess)
sockConfig->iMemPool = memPool;
}
if (status != PVMFSuccess)
{
//put the port config back in the allocated port vector
iAllocatedPortVector.push_back(sockConfig);
return status;
}
//move the sock config to the PVMF port vector and link the config
//back to the port.
aPort->iConfig = sockConfig;
sockConfig->iPVMFPort = aPort;
return status;
}
break;
default:
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: ERROR - Invalid protocol for port request"));
return PVMFFailure;
}
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: ERROR - Invalid protocol for port request"));
return PVMFFailure;
}
//Release ports is a do-nothing for this node.
PVMFStatus PVMFSocketNode::DoReleasePort(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
return PVMFSuccess;
}
PVMFStatus PVMFSocketNode::DoInit(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoInit() In"));
if (iInterfaceState != EPVMFNodeIdle)
{
return PVMFErrInvalidState;
}
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.Init();
#endif
PVMFStatus status = PVMFSuccess;
//Create socket server session
if (NULL == iSockServ)
{
int32 err;
OSCL_TRY(err, iSockServ = OsclSocketServ::NewL(iAlloc););
if (err || (iSockServ == NULL))
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_ERR,
(0, "PVMFSocketNode::DoInit: ERROR. OsclSocketServ::NewL() fail Ln %d", __LINE__));
iCommandErrorCode = PVMFSocketNodeErrorSocketServerCreateError;
status = PVMFErrNoResources;
}
else
{
if (iSockServ->Connect() == OsclErrNone)
{
status = PVMFSuccess;
}
else
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_ERR,
(0, "PVMFSocketNode::DoInit: ERROR. OsclSocketServ::Connect() fail Ln %d", __LINE__));
iCommandErrorCode = PVMFSocketNodeErrorSocketServConnectError;
status = PVMFErrResource;
}
}
}
return status;
}
//Prepare is a do-nothing for this node.
PVMFStatus PVMFSocketNode::DoPrepare(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::DoPrepare() In"));
if (iInterfaceState != EPVMFNodeInitialized)
{
return PVMFErrInvalidState;
}
return PVMFSuccess;
}
PVMFStatus PVMFSocketNode::DoStart(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::DoStart() In"));
/*
* If the node is already started just return success - multiple starts can happen with
* flow control (auto-pause / auto-resume) scenarios
*/
if (iInterfaceState == EPVMFNodeStarted)
{
return PVMFSuccess;//already started.
}
if (iInterfaceState != EPVMFNodePrepared &&
iInterfaceState != EPVMFNodePaused)
{
return PVMFErrInvalidState;
}
//We need to update the node state before
//we can get things going.
TPVMFNodeInterfaceState curState = iInterfaceState;
ChangeExternalState(EPVMFNodeStarted);
PVMFStatus status = PVMFSuccess;
for (uint32 i = 0; i < iPortVector.size(); i++)
{
SocketPortConfig* sockConfig = iPortVector[i]->iConfig;
if (sockConfig)
{
//Start handling incoming messages
if (CanProcessIncomingMsg(*sockConfig))
ProcessIncomingMsg(*sockConfig);
//Start the receives.
if (CanReceive(*sockConfig))
{
status = StartRecvOperation(*sockConfig);
//receive may be pending, but we don't block
//the start command for it.
if (status == PVMFPending)
{
status = PVMFSuccess;
}
else if (status != PVMFSuccess)
{
//fail the command if any individual startups fail.
//restore the old node state.
ChangeExternalState(curState);
break;//out of for loop
}
}
}
}
return status;
}
PVMFStatus PVMFSocketNode::DoStop(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoStop() In"));
if ((iInterfaceState != EPVMFNodeStarted) && (iInterfaceState != EPVMFNodePaused))
{
return PVMFErrInvalidState;
}
//Stop will essentially just stop the receive operations and input port processing,
//since those handlers check the node state. Current operations will not be
//interrupted.
return PVMFSuccess;
}
//Stop all current socket/DNS activity and clear port messages.
//Return PVMFSuccess if everything is stopped, or else PVMFPending if some
//ports still have operations pending.
//
//Also update the global "iNumStopNodeActivityPending"
//with the number of ports that still have operations pending
PVMFStatus PVMFSocketNode::DoStopNodeActivity()
{
//note that "iNumStopNodeActivityPending" uses the value (-1) to indicate idle state,
//or N...0 to countdown how many ports we're waiting on.
//Note: in case there's already a "stop node activity" going on, just
//return the status.
if (iNumStopPortActivityPending > 0)
return PVMFPending; //keep waiting on completion (in SequenceComplete)
//Notify all PVMF ports to suspend their input, and discard any current messages.
for (uint32 i = 0; i < iPortVector.size(); i++)
{
iPortVector[i]->SuspendInput();
iPortVector[i]->ClearMsgQueues();
}
//Stop socket activity on all ports.
{
uint32 nPortsPending = 0;
for (uint32 i = 0; i < iPortVector.size(); i++)
{
SocketPortConfig* portConfig = iPortVector[i]->iConfig;
if (portConfig)
{
// discard any saved socket activity events
portConfig->iSocketRecvActivity.iValid = false;
// start a shutdown sequence on the port
// look at what else might be happening currently.
//if a request port is going on, be sure to complete the command,
//although we will interrupt the current processing.
if (portConfig->iState.iSequence == EPVSocketPortSequence_RequestPort
&& iCurrentCmdQueue.size()
&& iCurrentCmdQueue.front().iCmd == PVMF_GENERIC_NODE_REQUESTPORT)
{
//@TODO would be nice to cleanup the allocated port in case of a failure.
CommandComplete(iCurrentCmdQueue, iCurrentCmdQueue.front(), PVMFErrCancelled, NULL);
}
//for any other sequence, we will just interrupt it with the shutdown.
//if a cleanup is already underway, just keep waiting for it to complete,
//else start a new sequence
if (portConfig->iState.iSequence == EPVSocketPortSequence_SocketCleanup)
{
nPortsPending++;
}
else if (StartSequence(*portConfig, EPVSocketPortSequence_SocketCleanup) == PVMFPending)
{
nPortsPending++;
}
}
}
if (nPortsPending > 0)
iNumStopPortActivityPending = nPortsPending;
}
if (iNumStopPortActivityPending > 0)
return PVMFPending; //wait on completion in SequenceComplete.
return PVMFSuccess;
}
//Flush is implemented for this node, but hasn't been tested.
PVMFStatus PVMFSocketNode::DoFlush(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoFlush() In"));
if ((iInterfaceState != EPVMFNodeStarted) && (iInterfaceState != EPVMFNodePaused))
{
return PVMFErrInvalidState;
}
//Notify all ports to suspend their input, then
//wait on completion of processing of current input.
PVMFStatus status = PVMFSuccess;
{
for (uint32 i = 0; i < iPortVector.size(); i++)
{
iPortVector[i]->SuspendInput();
if (status != PVMFPending
&& iPortVector[i]->IncomingMsgQueueSize() > 0)
{
status = PVMFPending;//Wait on this queue to empty.
//Completion is detected in SequenceComplete.
}
}
}
return status;
}
//Pause is a do-nothing for this node.
PVMFStatus PVMFSocketNode::DoPause(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoPause() In"));
/*
* If the node is already paused just return success - multiple pause can happen with
* flow control (auto-pause / auto-resume) scenarios
*/
if (iInterfaceState == EPVMFNodePaused)
{
return PVMFSuccess;
}
if (iInterfaceState != EPVMFNodeStarted)
{
return PVMFErrInvalidState;
}
return PVMFSuccess;
}
PVMFStatus PVMFSocketNode::DoCancelCommand(PVMFSocketNodeCommand& aCmd)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoCancelCommand() Called"));
// extract the command ID from the parameters.
PVMFCommandId cmdId;
aCmd.PVMFSocketNodeCommandBase::Parse(cmdId);
// first check "current" command if any
// it's safe to assume the "current cmd queue" is only 1-deep.
if (!iCurrentCmdQueue.empty()
&& iCurrentCmdQueue.front().iId == cmdId)
{
return DoCancelCurrentCommand(iCurrentCmdQueue, iCurrentCmdQueue.front());
//wait on current command to complete. The cancel command
//will ultimately be completed in the "CommandComplete" for the current command.
}
// next see if the command is still queued in the "pending command queue"
{
// start at element 1 since this cancel command is element 0
PVMFSocketNodeCommand* cmd = iPendingCmdQueue.FindById(cmdId, 1);
if (cmd)
{
// cancel the queued command. Note this will complete out-of-order.
CommandComplete(iPendingCmdQueue, *cmd, PVMFErrCancelled);
// no further action is required.
return PVMFSuccess;
}
}
//this command fails if the given command is not queued or in progress.
return PVMFErrArgument;
}
PVMFStatus PVMFSocketNode::DoCancelAllCommands(PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmd);
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoCancelCommand() IN"));
//This is not a conventional "cancel all" implementation.
//Streaming manager currently expects this call to shutdown all activity.
//It would really be more correct to use node Reset for this.
PVMFStatus status = DoStopNodeActivity();
//Since node activity is suspended, just go ahead and forcibly cancel any other
//pending commands now. There could be a Reset, a Flush, or a RequestPort
//happening.
if (!iCurrentCmdQueue.empty())
CommandComplete(iCurrentCmdQueue, iCurrentCmdQueue.front(), PVMFErrCancelled);
//Cancel all other pending commands, except for this one which is
//element 0 in the queue
for (uint32 i = 1; i < iPendingCmdQueue.size(); i++)
CommandComplete(iPendingCmdQueue, iPendingCmdQueue[i], PVMFErrCancelled);
//May need to wait on completion of StopNodeActivity.
return status;
}
//This routine must cancel the currently executing node command.
//It only needs to handle those commands that may have asynchronous completion.
PVMFStatus PVMFSocketNode::DoCancelCurrentCommand(PVMFSocketNodeCmdQ& aCmdQ, PVMFSocketNodeCommand& aCmd)
{
OSCL_UNUSED_ARG(aCmdQ);
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::DoCancelCurrentCommand()"));
switch (aCmd.iCmd)
{
case PVMF_GENERIC_NODE_REQUESTPORT:
//there may be a connect operation pending-- cancel it.
{
if (iRequestedPort)
{
SocketPortConfig* sockConfig = iRequestedPort->iConfig;
if (sockConfig
&& sockConfig->iState.iConnectOperation != EPVSocketPortConnectOperation_None)
{
CancelConnectOperation(*sockConfig);
return PVMFPending;//wait on the operation to complete
//in HandleSocketEvent or HandleDNSEvent
}
}
//shouldn't get here...
return PVMFFailure;
}
break;
case PVMF_GENERIC_NODE_RESET:
//it's too complicated to cancel a reset, so just wait on completion
return PVMFPending;
case PVMF_GENERIC_NODE_FLUSH:
//to cancel a flush, just discard all remaining port messages
//and keep waiting on completion of current message.
{
for (uint32 i = 0; i < iPortVector.size(); i++)
iPortVector[i]->ClearMsgQueues();
}
return PVMFPending;//keep waiting on flush completion in SequenceComplete.
case PVMF_GENERIC_NODE_CANCELCOMMAND:
case PVMF_GENERIC_NODE_CANCELALLCOMMANDS:
//these have asynchronous completion, but the command handling prevents
//processing a cancel during a cancel, so we shouldn't get here.
OSCL_ASSERT(0);
return PVMFFailure;
default:
//no other node commands have asynchronous completion, so this is unexpected.
OSCL_ASSERT(0);
return PVMFFailure;
}
}
//////////////////////////////////////
// End Command Handlers
//////////////////////////////////////
//////////////////////////////////////
// DNS & Socket Callbacks
//////////////////////////////////////
OSCL_EXPORT_REF void PVMFSocketNode::HandleDNSEvent(int32 aId, TPVDNSFxn aFxn, TPVDNSEvent aEvent, int32 aError)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::HandleDNSEvent() In aId=%d, aFxn=%d, aEvent=%d, aError=%d", aId, aFxn, aEvent, aError));
iInSocketCallback = true;
if (aEvent != EPVDNSSuccess)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleDNSEvent: Error - Failure"));
}
//Note: since the DNS object ID is identical to the socket ID, we can lookup the
//socket port config using the DNS ID.
SocketPortConfig* sockConfigPtr = FindSocketPortConfig((uint32)aId);
if (!sockConfigPtr)
{
//it's possible the DNS object was already delete-- so just fail gracefully.
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleDNSEvent: Error - Invalid Socket ID"));
}
else
{
#if(ENABLE_SOCKET_NODE_STATS)
sockConfigPtr->iPortStats.iNumDNSCallback++;
#endif
//create socket activity
PVMFSocketActivity activity((aEvent == EPVDNSSuccess) ? PVMFSuccess : PVMFFailure, aId, aFxn, aEvent, aError);
//Call the appropriate handler
switch (aFxn)
{
case EPVDNSGetHostByName:
OSCL_ASSERT(sockConfigPtr->iState.iConnectOperation == EPVSocketPortConnectOperation_GetHostByName);
#if(ENABLE_SOCKET_NODE_STATS)
if (aEvent == EPVDNSSuccess)
sockConfigPtr->iPortStats.EndConnectTime(sockConfigPtr->iState.iConnectOperation);
#endif
ConnectOperationComplete(*sockConfigPtr, activity.iStatus, &activity);
break;
default:
OSCL_ASSERT(0);//unexpected
break;
}
}
iInSocketCallback = false;
}
//This is the callback from Oscl Sockets for socket operation completion.
OSCL_EXPORT_REF void PVMFSocketNode::HandleSocketEvent(int32 aId, TPVSocketFxn aFxn, TPVSocketEvent aEvent, int32 aError)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_STACK_TRACE, (0, "PVMFSocketNode::HandleSocketEvent() In aId=%d, aFxn=%d, aEvent=%d, aError=%d", aId, aFxn, aEvent, aError));
iInSocketCallback = true;
SocketPortConfig* sockConfigPtr = FindSocketPortConfig((uint32)aId);
if (!sockConfigPtr)
{
//it's possible the socket was already deleted-- so just fail gracefully.
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleSocketEvent: Error - Invalid Socket ID"));
}
else
{
if (aEvent != EPVSocketSuccess)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::HandleSocketEvent: Error - Event=%d, SockId=%d, Mime=%s",
aEvent, aId, sockConfigPtr->iMime.get_cstr()));
}
#if(ENABLE_SOCKET_NODE_STATS)
sockConfigPtr->iPortStats.iNumSocketCallback++;
#endif
//create socket activity
PVMFSocketActivity activity((aEvent == EPVSocketSuccess) ? PVMFSuccess : PVMFFailure, aId, aFxn, aEvent, aError);
//Call the appropriate handler
switch (aFxn)
{
case EPVSocketRecvFrom:
case EPVSocketRecv:
OSCL_ASSERT(sockConfigPtr->iState.iRecvOperation == EPVSocketPortRecvOperation_Recv
|| sockConfigPtr->iState.iRecvOperation == EPVSocketPortRecvOperation_RecvFrom);
#if(ENABLE_SOCKET_NODE_STATS)
if (aEvent == EPVSocketSuccess)
sockConfigPtr->iPortStats.EndRecvTime(sockConfigPtr->iState.iRecvOperation);
#endif
RecvOperationComplete(*sockConfigPtr, activity.iStatus, &activity);
break;
case EPVSocketSendTo:
case EPVSocketSend:
OSCL_ASSERT(sockConfigPtr->iState.iSendOperation == EPVSocketPortSendOperation_Send
|| sockConfigPtr->iState.iSendOperation == EPVSocketPortSendOperation_SendTo);
#if(ENABLE_SOCKET_NODE_STATS)
if (aEvent == EPVSocketSuccess)
sockConfigPtr->iPortStats.EndSendTime(sockConfigPtr->iState.iSendOperation);
#endif
SendOperationComplete(*sockConfigPtr, activity.iStatus, &activity);
break;
case EPVSocketConnect:
OSCL_ASSERT(sockConfigPtr->iState.iConnectOperation == EPVSocketPortConnectOperation_Connect);
#if(ENABLE_SOCKET_NODE_STATS)
if (aEvent == EPVSocketSuccess)
sockConfigPtr->iPortStats.EndConnectTime(sockConfigPtr->iState.iConnectOperation);
#endif
ConnectOperationComplete(*sockConfigPtr, activity.iStatus, &activity);
break;
case EPVSocketShutdown:
OSCL_ASSERT(sockConfigPtr->iState.iConnectOperation == EPVSocketPortConnectOperation_Shutdown);
#if(ENABLE_SOCKET_NODE_STATS)
if (aEvent == EPVSocketSuccess)
sockConfigPtr->iPortStats.EndConnectTime(sockConfigPtr->iState.iConnectOperation);
#endif
ConnectOperationComplete(*sockConfigPtr, activity.iStatus, &activity);
break;
default:
OSCL_ASSERT(0);//unexpected
break;
}
}
iInSocketCallback = false;
}
//////////////////////////////////////
// End DNS & Socket Callbacks
//////////////////////////////////////
///////////////////////
// Low-level port processing
///////////////////////
//Allocate a node port and add it to the port vector.
PVMFStatus PVMFSocketNode::AddPort(int32 tag, PVMFSocketPort* &port)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::AddPort() In"));
int32 err;
OsclAny *ptr = NULL;
OSCL_TRY(err, ptr = iPortVector.Allocate(););
if (err != OsclErrNone || !ptr)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::AddPort: ERROR - err=%d Ln %d", err, __LINE__));
return PVMFErrNoMemory;
}
port = new(ptr) PVMFSocketPort((int32)tag,
this,
DEFAULT_DATA_QUEUE_CAPACITY,
DEFAULT_DATA_QUEUE_CAPACITY,
DEFAULT_READY_TO_RECEIVE_THRESHOLD_PERCENT,
//the output queue is empty because
//this node pushes data directly to the
//connecte port's input.
0, 0, 0);
//Add the port to the port vector.
OSCL_TRY(err, iPortVector.AddL(port););
if (err != OsclErrNone)
{
iPortVector.DestructAndDealloc(port);
return PVMFErrNoMemory;
}
return PVMFSuccess;
}
//Allocate a memory pool for the input port config.
//When finished, the port config is updated with the
//mempool and the port tag.
PVMFStatus PVMFSocketNode::AllocatePortMemPool(int32 tag, PVMFSocketNodeMemPool* & aMemPool)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::AllocatePortMemPool() In"));
aMemPool = NULL;
OsclAny *MemPtr = NULL;
int32 err;
OSCL_TRY(err, MemPtr = iAlloc.ALLOCATE(sizeof(PVMFSocketNodeMemPool)););
if ((err != OsclErrNone) || (MemPtr == NULL))
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: PVMFSocketNodeMemPool Allocate Failed Ln %d", err, __LINE__));
return PVMFErrNoMemory;
}
switch (tag)
{
case PVMF_SOCKET_NODE_PORT_TYPE_SOURCE:
//source tag is used for RTP
{
int32 errcode = 0;
errcode = SocketPlacementNew(aMemPool, MemPtr, DEFAULT_NUM_MEDIA_MSGS_IN_JITTER_BUFFER);
if (errcode != OsclErrNone)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: PVMFSocketNodeMemPool Construct Failed Ln %d", err, __LINE__));
iAlloc.deallocate(MemPtr);
return PVMFErrNoMemory;
}
}
break;
case PVMF_SOCKET_NODE_PORT_TYPE_SINK:
case PVMF_SOCKET_NODE_PORT_TYPE_PASSTHRU:
//sink tag is used for RTCP
{
int32 errcode = 0;
errcode = SocketPlacementNew(aMemPool, MemPtr, SNODE_DEFAULT_NUMBER_MEDIADATA_IN_MEMPOOL);
if (errcode != OsclErrNone)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::DoRequestPort: PVMFSocketNodeMemPool Construct Failed Ln %d", err, __LINE__));
iAlloc.deallocate(MemPtr);
return PVMFErrNoMemory;
}
}
break;
default://unexpected-- caller already validated the tag.
iAlloc.deallocate(MemPtr);
return PVMFFailure;
break;
}
aMemPool->iPortTag = tag;
return PVMFSuccess;
}
//Lookup the socket port config entry, given a socket address.
//If the entry is in the "allocated port" vector, it will be erased, so
//it's up to the caller to make sure it gets deleted or moved.
SocketPortConfig* PVMFSocketNode::FindSocketPortConfig(SOCKET_ADDR& aSockAddr)
{
uint32 i;
//search the allocated port vector
for (i = 0; i < iAllocatedPortVector.size(); i++)
{
if (iAllocatedPortVector[i]
&& MatchSocketAddr(aSockAddr, *iAllocatedPortVector[i]))
{
//erase it from the vector before returning to the caller.
SocketPortConfig* elem = iAllocatedPortVector[i];
iAllocatedPortVector.erase(&iAllocatedPortVector[i]);
return elem;
}
}
//search the port vector
for (i = 0; i < iPortVector.size(); i++)
{
if (iPortVector[i]->iConfig
&& MatchSocketAddr(aSockAddr, *iPortVector[i]->iConfig))
{
return iPortVector[i]->iConfig;
}
}
return NULL;
}
//return "true" if input socket address matches port config
bool PVMFSocketNode::MatchSocketAddr(SOCKET_ADDR& aSockAddr, SocketPortConfig& aSockConfig)
{
return (aSockConfig.iAddr.iProtocol == aSockAddr.iProtocol
&& aSockConfig.iAddr.iRemoteAdd.port == aSockAddr.iRemoteAdd.port
&& aSockConfig.iAddr.iRemoteHost == aSockAddr.iRemoteHost
&& aSockConfig.iTag == aSockAddr.iTag
//local address may or may not be specified-- ignore if not.
&& (!aSockAddr.iLocalAdd.port
|| (aSockConfig.iAddr.iLocalAdd.port == aSockAddr.iLocalAdd.port)));
}
//Lookup the socket port config entry, given a socket ID.
SocketPortConfig* PVMFSocketNode::FindSocketPortConfig(uint32 aId)
{
for (uint32 i = 0; i < iPortVector.size(); i++)
{
if (iPortVector[i]->iConfig
&& iPortVector[i]->iConfig->iSockId == aId)
{
return iPortVector[i]->iConfig;
}
}
return NULL;
}
//Create a socket using the given socket ID and protocol.
//For UDP, this will also attempt to bind, incrementing port numbers until
//either success or maximum attempts is reached. This has the side effect
//of updating the port number in the local address.
//It returns a pointer to the socket cast as OsclAny*.
OsclAny* PVMFSocketNode::CreateOsclSocketAndBind(SOCKET_ADDR &aSockAdd, uint32 aSockId)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::CreateOsclSocketAndBind() In"));
//create the socket server session if it doesn't exist yet.
if (iSockServ == NULL)
{
int32 err;
OSCL_TRY(err, iSockServ = OsclSocketServ::NewL(iAlloc););
if (err || (iSockServ == NULL))
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_ERR, (0, "PVMFSocketNode::CreateSocket() ERROR- iSockServ=0x%x, err=%d, Ln %d", iSockServ, err, __LINE__));
return NULL;
}
if (iSockServ->Connect() != OsclErrNone)
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_ERR, (0, "PVMFSocketNode::CreateSocket() ERROR Connect(), Ln %d", __LINE__));
return NULL;
}
}
switch (aSockAdd.iProtocol)
{
case INET_UDP:
{
int32 err;
OsclUDPSocket *UDPSocket = NULL;
OSCL_TRY(err, UDPSocket = OsclUDPSocket::NewL(iAlloc, *iSockServ, this, aSockId););
if (err || (UDPSocket == NULL))
{
PVLOGGER_LOGMSG(PVLOGMSG_INST_LLDBG, iLogger, PVLOGMSG_ERR, (0, "PVMFSocketNode::CreateSocket() ERROR- UDPSocket=0x%x, err=%d, Ln %d", UDPSocket, err, __LINE__));
return NULL;
}
//Bind this socket to the address
bool bBindOK = false;
int maxNumOfBind = UDP_PORT_RANGE;
while (maxNumOfBind-- > 0)
{
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.iNumBind++;
#endif
if (UDPSocket->Bind(aSockAdd.iLocalAdd) == EPVSocketSuccess)
{
bBindOK = true;
break;
}
aSockAdd.iLocalAdd.port++;
};
//If bind failed, delete the socket.
if (!bBindOK)
{
UDPSocket->~OsclUDPSocket();
iAlloc.deallocate(UDPSocket);
UDPSocket = NULL;
}
return UDPSocket;
}
case INET_TCP:
{
int32 err;
OsclTCPSocket *TCPSocket = NULL;
OSCL_TRY(err, TCPSocket = OsclTCPSocket::NewL(iAlloc, *iSockServ, this, aSockId););
if (err || (TCPSocket == NULL))
{
return NULL;
}
return TCPSocket;
}
default:
return NULL;
}
}
//This finishes up a TCP socket cleanup, after the socket shutdown
//is complete.
void PVMFSocketNode::CleanupTCP(SocketPortConfig& aSockConfig)
{
if (aSockConfig.iTCPSocket)
{
//Close the socket. This will free the OS-level resources. This
//socket object cannot be re-used.
aSockConfig.iTCPSocket->Close();
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CleanupTCP - Socket Closed - SockId=%d, Mime=%s, Port=0x%x", aSockConfig.iSockId, aSockConfig.iMime.get_str(), aSockConfig.iPVMFPort));
//Send EOS. To avoid multiple EOS we only do this when the
//socket is still open.
if (aSockConfig.iPVMFPort
&& aSockConfig.iPVMFPort->IsConnected())
{
PVMFSharedMediaCmdPtr sharedMediaCmdPtr = PVMFMediaCmd::createMediaCmd();
sharedMediaCmdPtr->setFormatID(PVMF_MEDIA_CMD_EOS_FORMAT_ID);
PVMFSharedMediaMsgPtr mediaMsgOut;
convertToPVMFMediaCmdMsg(mediaMsgOut, sharedMediaCmdPtr);
#if(ENABLE_SOCKET_NODE_STATS)
aSockConfig.iPortStats.iNumQueueOutgoingEOSMsg++;
#endif
PVMFStatus status = aSockConfig.iPVMFPort->QueueOutgoingMsg(mediaMsgOut);
//There's no recovery from a queue busy condition here.
//We assume the caller already waited on the connected port before
//calling this routine.
if (status != PVMFSuccess)
{
PVMF_SOCKETNODE_LOGERROR((0, "PVMFSocketNode::SendPVMFEOS() ERROR:%d, Outgoing queue size=%d. Data discarded!", status, aSockConfig.iPVMFPort->OutgoingMsgQueueSize()));
ReportErrorEvent(PVMFErrPortProcessing);
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode::CleanupTCP - EOS Send FAILED - SockId=%d, Mime=%s, Port=0x%x", aSockConfig.iSockId, aSockConfig.iMime.get_str(), aSockConfig.iPVMFPort));
}
else
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CleanupTCP - EOS Sent - SockId=%d, Mime=%s, Port=0x%x", aSockConfig.iSockId, aSockConfig.iMime.get_str(), aSockConfig.iPVMFPort));
}
}
if (iInSocketCallback)
{
//prevent deleting an object from within its own Run.
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CleanupTCP - queueing old socket for deletion - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
iClosedTCPSocketVector.push_back(aSockConfig.iTCPSocket);
aSockConfig.iTCPSocket = NULL;
RunIfNotReady();//schedule the AO to delete the socket.
}
else
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CleanupTCP - Deleting socket - SockId=%d, Mime=%s, Port=0x%x", aSockConfig.iSockId, aSockConfig.iMime.get_str(), aSockConfig.iPVMFPort));
aSockConfig.iTCPSocket->~OsclTCPSocket();
iAlloc.deallocate(aSockConfig.iTCPSocket);
aSockConfig.iTCPSocket = NULL;
}
}
//cleanup media messages
//(this must happen *after* the optional EOS)
if (aSockConfig.iPendingRecvMediaData.GetRep() != NULL)
aSockConfig.iPendingRecvMediaData.Unbind();
if (aSockConfig.iPendingSendMediaData.GetRep() != NULL)
aSockConfig.iPendingSendMediaData.Unbind();
}
//Cancel and/or delete UDP socket
void PVMFSocketNode::CleanupUDP(SocketPortConfig& aSockConfig)
{
//delete
if (aSockConfig.iUDPSocket)
{
if (iInSocketCallback)
{
//prevent deleting an object from within its own Run.
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CleanupTCP - queueing old socket for deletion - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
iClosedUDPSocketVector.push_back(aSockConfig.iUDPSocket);
aSockConfig.iUDPSocket = NULL;
RunIfNotReady();//schedule the AO to delete the socket.
}
else
{
aSockConfig.iUDPSocket->~OsclUDPSocket();
iAlloc.deallocate(aSockConfig.iUDPSocket);
aSockConfig.iUDPSocket = NULL;
}
}
//cleanup media messages when socket is deleted
if (aSockConfig.iPendingRecvMediaData.GetRep() != NULL)
aSockConfig.iPendingRecvMediaData.Unbind();
if (aSockConfig.iPendingSendMediaData.GetRep() != NULL)
aSockConfig.iPendingSendMediaData.Unbind();
}
void PVMFSocketNode::CleanupDNS(SocketPortConfig& aSockConfig)
{
if (aSockConfig.iDNS)
{
if (iInSocketCallback)
{
//prevent deleting an object from within its own Run.
PVMF_SOCKETNODE_LOGDATATRAFFIC_I((0, "PVMFSocketNode::CleanupDNS - queueing old DNS for deletion - SockId=%d, Mime=%s", aSockConfig.iSockId, aSockConfig.iMime.get_str()));
iClosedDNSVector.push_back(aSockConfig.iDNS);
aSockConfig.iDNS = NULL;
RunIfNotReady();//schedule the AO to delete the object.
}
else
{
aSockConfig.iDNS->~OsclDNS();
iAlloc.deallocate(aSockConfig.iDNS);
aSockConfig.iDNS = NULL;
}
}
}
//A synchronous cleanup routine for all ports.
//It is assumed that all sockets are idle when this is called.
//This is used in node destructor, and in completion of
//Reset and Cancel All commands.
void PVMFSocketNode::CleanupPorts()
{
// Cleanup port vector
while (!iPortVector.empty())
{
SocketPortConfig* it = iPortVector.front()->iConfig;
if (it)
{
//unlink the PVMFPort so we won't try to send any EOS msg
//during the TCP cleanup.
it->iPVMFPort = NULL;
CleanupTCP(*it);
CleanupUDP(*it);
CleanupDNS(*it);
it->CleanupMemPools(iAlloc);
OSCL_DELETE(it);
iPortVector.front()->iConfig = NULL;
}
iPortVector.Erase(&iPortVector.front());
}
// Cleanup allocated ports (these may have bound UDP sockets
// but don't have any PVMF ports).
while (!iAllocatedPortVector.empty())
{
SocketPortConfig* it = iAllocatedPortVector.front();
CleanupTCP(*it);
CleanupUDP(*it);
CleanupDNS(*it);
it->CleanupMemPools(iAlloc);
OSCL_DELETE(it);
iAllocatedPortVector.erase(&iAllocatedPortVector.front());
}
}
void PVMFSocketNode::CleanupClosedTCPSockets()
{
while (!iClosedTCPSocketVector.empty())
{
OsclTCPSocket* obj = iClosedTCPSocketVector.front();
obj->~OsclTCPSocket();
iAlloc.deallocate(obj);
iClosedTCPSocketVector.erase(&iClosedTCPSocketVector.front());
}
}
void PVMFSocketNode::CleanupClosedUDPSockets()
{
while (!iClosedUDPSocketVector.empty())
{
OsclUDPSocket* obj = iClosedUDPSocketVector.front();
obj->~OsclUDPSocket();
iAlloc.deallocate(obj);
iClosedUDPSocketVector.erase(&iClosedUDPSocketVector.front());
}
}
void PVMFSocketNode::CleanupClosedDNS()
{
while (!iClosedDNSVector.empty())
{
OsclDNS* obj = iClosedDNSVector.front();
obj->~OsclDNS();
iAlloc.deallocate(obj);
iClosedDNSVector.erase(&iClosedDNSVector.front());
}
}
bool PVMFSocketNode::ParseTransportConfig(OSCL_String *aPortConfig,
SOCKET_ADDR &aSockConfig,
OSCL_String& aMime)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::ParseTransportConfig() In"));
char* head = aPortConfig->get_str();
int32 len = aPortConfig->get_size();
return ParseTransportConfig(head, len, aSockConfig, aMime);
}
bool PVMFSocketNode::ParseTransportConfig(char *aPortConfig,
int32 aLen,
SOCKET_ADDR &aSockConfig,
OSCL_String& aMime)
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::ParseTransportConfig() In"));
char* head = aPortConfig;
int32 len = aLen;
char *tail = head + len;
if (oscl_strstr(head, "TCP"))
aSockConfig.iProtocol = INET_TCP;
else if (oscl_strstr(head, "UDP"))
aSockConfig.iProtocol = INET_UDP;
else
return false;//invalid protocol.
//for future extension
aSockConfig.iLocalAdd.ipAddr.Set("0.0.0.0"); //INADDR_ANY
aSockConfig.iRemoteAdd.ipAddr.Set("");
OSCL_StackString<128> address("remote_address=");
char *tmpHead = OSCL_CONST_CAST(char*, oscl_strstr(head, address.get_cstr()));
if (tmpHead)
{
tmpHead += address.get_size();
char *tmpTail = OSCL_CONST_CAST(char*, oscl_strstr(tmpHead, ";"));
if (tmpTail == NULL)
{
tmpTail = tail;
}
char tmpChar = *tmpTail;
*tmpTail = '\0';
aSockConfig.iRemoteAdd.ipAddr.Set(tmpHead);
OSCL_HeapString<PVMFSocketNodeAllocator> tmpServerName(tmpHead, (tmpTail - tmpHead));
aSockConfig.iRemoteHost = tmpServerName;
*tmpTail = tmpChar;
}
{
TimeValue current_time;
current_time.set_to_current_time();
uint32 my_seed = current_time.get_sec();
OsclRand random_num;
random_num.Seed(my_seed);
int32 first = random_num.Rand();
uint32 myport = (first & 0x1FFF) + 0x2000; //start from 8192
aSockConfig.iLocalAdd.port = (myport >> 1) << 1; //start from even;
}
OSCL_StackString<128> client_port("client_port=");
tmpHead = OSCL_CONST_CAST(char*, oscl_strstr(head, client_port.get_cstr()));
if (tmpHead)
{
tmpHead += client_port.get_size();
char *tmpTail = OSCL_CONST_CAST(char*, oscl_strstr(tmpHead, ";"));
if (tmpTail == NULL)
{
tmpTail = tail;
}
char tmpChar = *tmpTail;
*tmpTail = '\0';
uint32 atoi_tmp;
if (PV_atoi(tmpHead, 'd', atoi_tmp))
{
aSockConfig.iLocalAdd.port = atoi_tmp;
}
*tmpTail = tmpChar;
}
aSockConfig.iRemoteAdd.port = 0;
OSCL_StackString<128> server_port("remote_port=");
tmpHead = OSCL_CONST_CAST(char*, oscl_strstr(head, server_port.get_cstr()));
if (tmpHead)
{
tmpHead += server_port.get_size();
char *tmpTail = OSCL_CONST_CAST(char*, oscl_strstr(tmpHead, ";"));
if (tmpTail == NULL)
{
tmpTail = tail;
}
char tmpChar = *tmpTail;
*tmpTail = '\0';
uint32 atoi_tmp;
if (PV_atoi(tmpHead, 'd', atoi_tmp))
{
aSockConfig.iRemoteAdd.port = atoi_tmp;
}
*tmpTail = tmpChar;
}
OSCL_StackString<128> mime("mime=");
tmpHead = OSCL_CONST_CAST(char*, oscl_strstr(head, mime.get_cstr()));
if (tmpHead)
{
tmpHead += mime.get_size();
char *tmpTail = OSCL_CONST_CAST(char*, oscl_strstr(tmpHead, ";"));
if (tmpTail == NULL)
{
tmpTail = tail;
}
aMime += tmpHead;
}
return true;
}
///////////////////////
// End Low-level port processing
///////////////////////
//////////////////////////////
// Logging & Diagnostics
//////////////////////////////
void PVMFSocketNode::LogRTPHeaderFields(SocketPortConfig& aSockConfig,
OsclRefCounterMemFrag& memFragIn)
{
OSCL_UNUSED_ARG(aSockConfig);
uint8* rtpHeader = NULL;
uint32 rtpPacketLen = 0;
/* Get start of RTP packet */
rtpHeader = (uint8*)(memFragIn.getMemFrag().ptr);
rtpPacketLen = memFragIn.getMemFrag().len;
if ((rtpHeader != NULL) && (rtpPacketLen > 0))
{
/* is this a legal data packet? */
if (rtpPacketLen <= 12)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode - illegal RTP pkt size"));
}
/* Parse RTP version */
uint8 rtpVersion = (((*rtpHeader) & 0xC0) >> 6);
if (rtpVersion != 2)
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_E((0, "PVMFSocketNode illegal rtp version"));
}
/* Check for padding */
uint8 pbit = (((*rtpHeader) & 0x20) >> 5);
uint8 numPaddingOctets = 0;
if (pbit)
{
numPaddingOctets = *(rtpHeader + (rtpPacketLen - 1));
}
// skip over xbit, csrc and mbit
rtpHeader++;
rtpHeader++;
/* Parse sequence number */
uint16 seqNum16 = 0;
oscl_memcpy((char *)&seqNum16, rtpHeader, sizeof(seqNum16));
big_endian_to_host((char *)&seqNum16, sizeof(seqNum16));
rtpHeader += 2;
/* Parse rtp time stamp */
uint32 ts32 = 0;
oscl_memcpy((char *)&ts32, rtpHeader, sizeof(ts32));
big_endian_to_host((char *)&ts32, sizeof(ts32));
rtpHeader += 4;
/* Parse SSRC */
uint32 ssrc32 = 0;
oscl_memcpy((char *)&ssrc32, rtpHeader, sizeof(ssrc32));
big_endian_to_host((char *)&ssrc32, sizeof(ssrc32));
rtpHeader += 4;
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTP((0, "SN-RTP - Mime=%s, ssrc=%d, seq=%d, ts=%d, len=%d",
aSockConfig.iMime.get_cstr(), ssrc32, (uint32)seqNum16, (PVMFTimestamp)ts32, rtpPacketLen));
}
}
void PVMFSocketNode::LogRTCPHeaderFields(SocketPortConfig& aSockConfig,
OsclRefCounterMemFrag& memFragIn)
{
OSCL_UNUSED_ARG(aSockConfig);
const uint8 SR_PACKET_TYPE = 200;
const uint8 RR_PACKET_TYPE = 201;
const uint8 SDES_PACKET_TYPE = 202;
const uint8 BYE_PACKET_TYPE = 203;
const uint8 APP_PACKET_TYPE = 204;
uint8* rtcpHeader = NULL;
int32 rtcpPacketLen = 0;
/* Get start of RTP packet */
rtcpHeader = (uint8*)(memFragIn.getMemFrag().ptr);
rtcpPacketLen = (int32)(memFragIn.getMemFrag().len);
if ((rtcpHeader != NULL) && (rtcpPacketLen > 0))
{
while (rtcpPacketLen > 0)
{
OsclBinIStreamBigEndian inStream;
inStream.Attach(rtcpHeader, rtcpPacketLen);
uint8 tempChar;
inStream >> tempChar;
if (inStream.fail())
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, Invalid RTCP Packet",
aSockConfig.iMime.get_cstr()));
return;
}
// read the type
uint8 payloadType;
uint16 rtcpLength;
inStream >> payloadType;
inStream >> rtcpLength;
if (inStream.fail())
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, Invalid RTCP Packet Headers",
aSockConfig.iMime.get_cstr()));
return;
}
switch (payloadType)
{
case SR_PACKET_TYPE:
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, RTCP_SR RECVD",
aSockConfig.iMime.get_cstr()));
}
break;
case RR_PACKET_TYPE:
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, RTCP_RR RECVD",
aSockConfig.iMime.get_cstr()));
}
break;
case SDES_PACKET_TYPE:
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, RTCP_SDES_PKT RECVD",
aSockConfig.iMime.get_cstr()));
}
break;
case BYE_PACKET_TYPE:
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, RTCP_BYE RECVD",
aSockConfig.iMime.get_cstr()));
}
break;
case APP_PACKET_TYPE:
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, RTCP_APP_PKT RECVD",
aSockConfig.iMime.get_cstr()));
}
break;
default:
break;
}
rtcpPacketLen -= (rtcpLength + 1) * 4;
rtcpHeader += (rtcpLength + 1) * 4;
if (rtcpLength)
{
inStream.seekFromCurrentPosition((rtcpLength*4));
if (inStream.fail())
{
PVMF_SOCKETNODE_LOGDATATRAFFIC_RTCP((0, "SN-RTCP - Mime=%s, RTCP Packet Read Failed",
aSockConfig.iMime.get_cstr()));
return;
}
}
} //end of while loop
}
}
//////////////////////////////
// End Logging & Diagnostics
//////////////////////////////
///////////////////////////////////
// AO
///////////////////////////////////
void PVMFSocketNode::Run()
{
PVMF_SOCKETNODE_LOGSTACKTRACE((0, "PVMFSocketNode::Run() In"));
#if(ENABLE_SOCKET_NODE_STATS)
iSocketNodeStats.iNumRun++;
#endif
if (!iClosedTCPSocketVector.empty())
CleanupClosedTCPSockets();
if (!iClosedUDPSocketVector.empty())
CleanupClosedUDPSockets();
if (!iClosedDNSVector.empty())
CleanupClosedDNS();
//Process node commands
while (CanProcessCommand())
ProcessCommand(iPendingCmdQueue, iPendingCmdQueue.front());
}
int32 PVMFSocketNode::SocketPlacementNew(PVMFSocketNodeMemPool *& aMemPool, OsclAny *aMemPtr, int32 aMemPoolNumBufs)
{
int32 errcode = 0;
OSCL_TRY(errcode, aMemPool = OSCL_PLACEMENT_NEW(aMemPtr, PVMFSocketNodeMemPool(aMemPoolNumBufs);));
return errcode;
}
int32 PVMFSocketNode::CreateMediaData(SocketPortConfig& aSockConfig, OsclSharedPtr< PVMFMediaDataImpl > &aMediaptr)
{
int32 err = 0;
OSCL_TRY_NO_TLS(iOsclErrorTrapImp, err,
aSockConfig.iPendingRecvMediaData = PVMFMediaData::createMediaData(aMediaptr, aSockConfig.iMemPool->iMediaDataMemPool););
return err;
}
int32 PVMFSocketNode::Allocate(SocketPortConfig &aSockConfig, OsclSharedPtr< PVMFMediaDataImpl > &aMediaFragGroup)
{
#if SNODE_ENABLE_UDP_MULTI_PACKET
int32 err = 0;
OSCL_TRY_NO_TLS(iOsclErrorTrapImp, err,
aMediaFragGroup = aSockConfig.iMemPool->iMediaFragGroupAlloc->allocate(););
return err;
#else
return OsclErrNotSupported;
#endif
}
int32 PVMFSocketNode::GetMediaDataImpl(SocketPortConfig& aSockConfig, OsclSharedPtr<PVMFMediaDataImpl> &aMediaDataImpl, int32 aMemSize)
{
int32 err = 0;
aMediaDataImpl = aSockConfig.iMemPool->getMediaDataImpl(aMemSize);
if (NULL == aMediaDataImpl.GetRep())
{
err = OsclErrNoMemory;
}
return err;
}