inet::VoipStreamReceiver Class Reference

#include <VoipStreamReceiver.h>

Inheritance diagram for inet::VoipStreamReceiver:

Classes
class	Connection

Public Member Functions
	VoipStreamReceiver ()
	~VoipStreamReceiver ()
Public Member Functions inherited from inet::LifecycleUnsupported
virtual bool	handleOperationStage (LifecycleOperation *operation, IDoneCallback *doneCallback) override
	Perform one stage of a lifecycle operation. More...
Public Member Functions inherited from inet::ILifecycle
virtual	~ILifecycle ()
Public Member Functions inherited from inet::UdpSocket::ICallback
virtual	~ICallback ()

Protected Member Functions
virtual void	initialize (int stage) override
virtual int	numInitStages () const override
virtual void	handleMessage (cMessage *msg) override
virtual void	finish () override
virtual void	createConnection (Packet *vp)
virtual void	checkSourceAndParameters (Packet *vp)
virtual void	closeConnection ()
virtual void	decodePacket (Packet *vp)
virtual void	socketDataArrived (UdpSocket *socket, Packet *packet) override
	Notifies about data arrival, packet ownership is transferred to the callee. More...
virtual void	socketErrorArrived (UdpSocket *socket, Indication *indication) override
	Notifies about error indication arrival, indication ownership is transferred to the callee. More...
virtual void	socketClosed (UdpSocket *socket) override
	Notifies about socket closed, indication ownership is transferred to the callee. More...

Protected Attributes
int	localPort = -1
simtime_t	playoutDelay
const char *	resultFile = nullptr
UdpSocket	socket
Connection	curConn

Static Protected Attributes
static simsignal_t	lostSamplesSignal = registerSignal("lostSamples")
static simsignal_t	lostPacketsSignal = registerSignal("lostPackets")
static simsignal_t	packetHasVoiceSignal = registerSignal("packetHasVoice")
static simsignal_t	connStateSignal = registerSignal("connState")
static simsignal_t	delaySignal = registerSignal("delay")

Constructor & Destructor Documentation

VoipStreamReceiver()

inet::VoipStreamReceiver::VoipStreamReceiver ( )

inline

{}

~VoipStreamReceiver()

inet::VoipStreamReceiver::~VoipStreamReceiver

(

)

{
    closeConnection();
}

Member Function Documentation

checkSourceAndParameters()

void inet::VoipStreamReceiver::checkSourceAndParameters ( Packet *  vp )

protectedvirtual

{
    ASSERT(!curConn.offline);
 
    const auto& vp = pk->peekAtFront<VoipStreamPacket>();
    auto l3Addresses = pk->getTag<L3AddressInd>();
    auto ports = pk->getTag<L4PortInd>();
    L3Address srcAddr = l3Addresses->getSrcAddress();
    L3Address destAddr = l3Addresses->getDestAddress();
 
    if (curConn.srcAddr != srcAddr
        || curConn.srcPort != ports->getSrcPort()
        || curConn.destAddr != destAddr
        || curConn.destPort != ports->getDestPort()
        || vp->getSsrc() != curConn.ssrc)
        throw cRuntimeError("Voice packet received from third party during a voice session (concurrent voice sessions not supported)");
 
    if (vp->getCodec() != curConn.codec
        || vp->getSampleBits() != curConn.sampleBits
        || vp->getSampleRate() != curConn.sampleRate
        || vp->getSamplesPerPacket() != curConn.samplesPerPacket
        || vp->getTransmitBitrate() != curConn.transmitBitrate)
        throw cRuntimeError("Cannot change voice encoding parameters a during session");
}

Referenced by socketDataArrived().

closeConnection()

void inet::VoipStreamReceiver::closeConnection ( )

protectedvirtual

{
    if (!curConn.offline) {
        curConn.offline = true;
        avcodec_close(curConn.decCtx);
        avcodec_free_context(&curConn.decCtx);
        curConn.outFile.close();
        emit(connStateSignal, -1L); // so that sum() yields the number of active sessions
    }
}

Referenced by finish(), and ~VoipStreamReceiver().

createConnection()

void inet::VoipStreamReceiver::createConnection ( Packet *  vp )

protectedvirtual

{
    ASSERT(curConn.offline);
 
    const auto& vp = pk->peekAtFront<VoipStreamPacket>();
    auto l3Addresses = pk->getTag<L3AddressInd>();
    auto ports = pk->getTag<L4PortInd>();
 
    curConn.srcAddr = l3Addresses->getSrcAddress();
    curConn.srcPort = ports->getSrcPort();
    curConn.destAddr = l3Addresses->getDestAddress();
    curConn.destPort = ports->getDestPort();
    curConn.seqNo = vp->getSeqNo() - 1;
    curConn.timeStamp = vp->getTimeStamp();
    curConn.ssrc = vp->getSsrc();
    curConn.codec = (enum AVCodecID)(vp->getCodec());
    curConn.sampleBits = vp->getSampleBits();
    curConn.sampleRate = vp->getSampleRate();
    curConn.transmitBitrate = vp->getTransmitBitrate();
    curConn.samplesPerPacket = vp->getSamplesPerPacket();
    curConn.lastPacketFinish = simTime() + playoutDelay;
 
    curConn.pCodecDec = avcodec_find_decoder(curConn.codec);
    if (curConn.pCodecDec == nullptr)
        throw cRuntimeError("Codec %d not found", curConn.codec);
 
    curConn.decCtx = avcodec_alloc_context3(curConn.pCodecDec);
    curConn.decCtx->bit_rate = curConn.transmitBitrate;
    curConn.decCtx->sample_rate = curConn.sampleRate;
    curConn.decCtx->channels = 1;
    curConn.decCtx->bits_per_coded_sample = curConn.sampleBits;
 
    int ret = avcodec_open2(curConn.decCtx, curConn.pCodecDec, nullptr);
    if (ret < 0)
        throw cRuntimeError("could not open decoding codec %d (%s): err=%d", curConn.codec, curConn.pCodecDec->name, ret);
 
    curConn.openAudio(resultFile);
    curConn.offline = false;
    emit(connStateSignal, 1);
}

Referenced by socketDataArrived().

decodePacket()

void inet::VoipStreamReceiver::decodePacket ( Packet *  vp )

protectedvirtual

{
    const auto& vp = pk->popAtFront<VoipStreamPacket>();
    uint16_t newSeqNo = vp->getSeqNo();
    if (newSeqNo > curConn.seqNo + 1)
        emit(lostPacketsSignal, newSeqNo - (curConn.seqNo + 1));
 
    if (simTime() > curConn.lastPacketFinish) {
        int lostSamples = ceil(SIMTIME_DBL((simTime() - curConn.lastPacketFinish) * curConn.sampleRate));
        ASSERT(lostSamples > 0);
        EV_INFO << "Lost " << lostSamples << " samples\n";
        emit(lostSamplesSignal, lostSamples);
        curConn.writeLostSamples(lostSamples);
        curConn.lastPacketFinish += lostSamples * (1.0 / curConn.sampleRate);
        FINGERPRINT_ADD_EXTRA_DATA(lostSamples);
    }
    emit(delaySignal, curConn.lastPacketFinish - pk->getCreationTime());
    curConn.seqNo = newSeqNo;
 
    if (vp->getType() == VOICE) {
        emit(packetHasVoiceSignal, 1);
        uint16_t len = vp->getDataLength();
        auto bb = pk->peekDataAt<BytesChunk>(b(0), B(len));
        auto buff = bb->getBytes();
        curConn.writeAudioFrame(&buff.front(), buff.size());
        FINGERPRINT_ADD_EXTRA_DATA2((const char *)(&buff.front()), buff.size());
    }
    else if (vp->getType() == SILENCE) {
        emit(packetHasVoiceSignal, 0);
        int silenceSamples = vp->getSamplesPerPacket();
        curConn.writeLostSamples(silenceSamples);
        curConn.lastPacketFinish += silenceSamples * (1.0 / curConn.sampleRate);
        FINGERPRINT_ADD_EXTRA_DATA(silenceSamples);
    }
    else
        throw cRuntimeError("The received VoipStreamPacket has unknown type %d", vp->getType());
}

Referenced by socketDataArrived().

finish()

void inet::VoipStreamReceiver::finish ( )

overrideprotectedvirtual

{
    EV_TRACE << "Sink finish()" << endl;
    closeConnection();
}

handleMessage()

void inet::VoipStreamReceiver::handleMessage ( cMessage *  msg )

overrideprotectedvirtual

{
    if (msg->arrivedOn("socketIn")) {
        socket.processMessage(msg);
    }
    else
        throw cRuntimeError("Unknown incoming gate: '%s'", msg->getArrivalGate()->getFullName());
}

initialize()

void inet::VoipStreamReceiver::initialize ( int  stage )

overrideprotectedvirtual

{
    cSimpleModule::initialize(stage);
 
    if (stage == INITSTAGE_LOCAL) {
        // KLUDGE hack to create results folder (doesn't work when record-scalars = false)
        recordScalar("hackForCreateResultsFolder", 0);
 
        // Say Hello to the world
        EV_TRACE << "VoIPSinkApp initialize()" << endl;
 
        // read parameters
        localPort = par("localPort");
        resultFile = par("resultFile");
        playoutDelay = par("playoutDelay");
 
        // initialize avcodec library
        av_register_all();
    }
    else if (stage == INITSTAGE_APPLICATION_LAYER) {
        cModule *node = findContainingNode(this);
        NodeStatus *nodeStatus = node ? check_and_cast_nullable<NodeStatus *>(node->getSubmodule("status")) : nullptr;
        bool isOperational = (!nodeStatus) || nodeStatus->getState() == NodeStatus::UP;
        if (!isOperational)
            throw cRuntimeError("This module doesn't support starting in node DOWN state");
 
        socket.setOutputGate(gate("socketOut"));
        socket.bind(localPort);
        socket.setCallback(this);
    }
}

numInitStages()

virtual int inet::VoipStreamReceiver::numInitStages ( ) const

inlineoverrideprotectedvirtual

{ return NUM_INIT_STAGES; }

socketClosed()

virtual void inet::VoipStreamReceiver::socketClosed ( UdpSocket *  socket )

inlineoverrideprotectedvirtual

Notifies about socket closed, indication ownership is transferred to the callee.

Implements inet::UdpSocket::ICallback.

{}

socketDataArrived()

void inet::VoipStreamReceiver::socketDataArrived ( UdpSocket *  socket, Packet *  packet  )

overrideprotectedvirtual

Notifies about data arrival, packet ownership is transferred to the callee.

Implements inet::UdpSocket::ICallback.

{
    // process incoming packet
 
    const auto& vp = pk->peekAtFront<VoipStreamPacket>();
    bool ok = true;
    if (curConn.offline)
        createConnection(pk);
    else {
        checkSourceAndParameters(pk);
        ok = vp->getSeqNo() > curConn.seqNo && vp->getTimeStamp() > curConn.timeStamp;
    }
 
    if (ok) {
        emit(packetReceivedSignal, pk);
        decodePacket(pk);
    }
    else {
        PacketDropDetails details;
        details.setReason(CONGESTION);
        emit(packetDroppedSignal, pk, &details);
    }
 
    delete pk;
}

socketErrorArrived()

void inet::VoipStreamReceiver::socketErrorArrived ( UdpSocket *  socket, Indication *  indication  )

overrideprotectedvirtual

Notifies about error indication arrival, indication ownership is transferred to the callee.

Implements inet::UdpSocket::ICallback.

{
    EV_WARN << "Unknown message '" << indication->getName() << "', kind = " << indication->getKind() << ", discarding it." << endl;
    delete indication;
}

Member Data Documentation

connStateSignal

simsignal_t inet::VoipStreamReceiver::connStateSignal = registerSignal("connState")

staticprotected

Referenced by closeConnection(), and createConnection().

curConn

Connection inet::VoipStreamReceiver::curConn

protected

Referenced by checkSourceAndParameters(), closeConnection(), createConnection(), decodePacket(), and socketDataArrived().

delaySignal

simsignal_t inet::VoipStreamReceiver::delaySignal = registerSignal("delay")

staticprotected

Referenced by decodePacket().

localPort

int inet::VoipStreamReceiver::localPort = -1

protected

Referenced by initialize().

lostPacketsSignal

simsignal_t inet::VoipStreamReceiver::lostPacketsSignal = registerSignal("lostPackets")

staticprotected

Referenced by decodePacket().

lostSamplesSignal

simsignal_t inet::VoipStreamReceiver::lostSamplesSignal = registerSignal("lostSamples")

staticprotected

Referenced by decodePacket().

packetHasVoiceSignal

simsignal_t inet::VoipStreamReceiver::packetHasVoiceSignal = registerSignal("packetHasVoice")

staticprotected

Referenced by decodePacket().

playoutDelay

simtime_t inet::VoipStreamReceiver::playoutDelay

protected

Referenced by createConnection(), and initialize().

resultFile

const char* inet::VoipStreamReceiver::resultFile = nullptr

protected

Referenced by createConnection(), and initialize().

socket

UdpSocket inet::VoipStreamReceiver::socket

protected

Referenced by handleMessage(), and initialize().

---

The documentation for this class was generated from the following files:

• VoipStreamReceiver.h
• VoipStreamReceiver.cc