inet::SctpNatPeer Class Reference

Accepts any number of incoming connections, and sends back whatever arrives on them. More...

#include <SctpNatPeer.h>

Inheritance diagram for inet::SctpNatPeer:

Classes
struct	pathStatus

Public Types
typedef std::map< L3Address, pathStatus >	SctpPathStatus

Public Member Functions
	SctpNatPeer ()
virtual	~SctpNatPeer ()
virtual int	numInitStages () const override
void	initialize (int stage) override
void	handleMessage (cMessage *msg) override
void	finish () override
void	handleTimer (cMessage *msg)
void	generateAndSend ()
void	connect (L3Address connectAddress, int32_t connectPort)
void	connectx (AddressVector connectAddressList, int32_t connectPort)
virtual void	socketAvailable (SctpSocket *socket, Indication *indication) override
void	socketEstablished (SctpSocket *socket, unsigned long int buffer) override
	Does nothing but update statistics/status. More...
void	socketDataArrived (SctpSocket *socket, Packet *msg, bool urgent) override
	Does nothing but update statistics/status. More...
void	socketDataNotificationArrived (SctpSocket *socket, Message *msg) override
void	socketPeerClosed (SctpSocket *socket) override
	Since remote SCTP closed, invokes close(). More...
void	socketClosed (SctpSocket *socket) override
	Does nothing but update statistics/status. More...
void	socketFailure (SctpSocket *socket, int32_t code) override
	Does nothing but update statistics/status. More...
void	socketStatusArrived (SctpSocket *socket, SctpStatusReq *status) override
	Redefine to handle incoming SctpStatusInfo. More...
void	msgAbandonedArrived (SctpSocket *socket) override
void	setPrimaryPath ()
void	sendStreamResetNotification ()
void	sendRequestArrived (SctpSocket *socket) override
void	sendQueueRequest ()
void	shutdownReceivedArrived (SctpSocket *socket) override
void	sendqueueFullArrived (SctpSocket *socket) override
void	addressAddedArrived (SctpSocket *socket, L3Address localAddr, L3Address remoteAddr) override
void	setStatusString (const char *s)
Public Member Functions inherited from inet::SctpSocket::ICallback
virtual	~ICallback ()
virtual void	socketOptionsArrived (SctpSocket *socket, Indication *indication)
virtual void	socketFailure (SctpSocket *socket, int code)
virtual void	socketDeleted (SctpSocket *socket)
virtual void	sendqueueAbatedArrived (SctpSocket *socket, unsigned long int buffer)
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 Attributes
SctpPathStatus	sctpPathStatus

Protected Types
typedef std::map< int32_t, int64_t >	RcvdPacketsPerAssoc
typedef std::map< int32_t, int64_t >	SentPacketsPerAssoc
typedef std::map< int32_t, int64_t >	RcvdBytesPerAssoc
typedef std::map< int32_t, cOutVector * >	BytesPerAssoc
typedef std::map< int32_t, cHistogram * >	HistEndToEndDelay
typedef std::map< int32_t, cOutVector * >	EndToEndDelay

Protected Member Functions
void	sendOrSchedule (cMessage *msg)
void	sendRequest (bool last=true)

Protected Attributes
int32_t	notifications
int32_t	serverAssocId
SctpSocket	clientSocket
SctpSocket	peerSocket
SctpSocket	rendezvousSocket
double	delay
bool	echo
bool	schedule
bool	shutdownReceived
int64_t	bytesSent
int32_t	packetsSent
int32_t	packetsRcvd
int32_t	numSessions
int32_t	numRequestsToSend
bool	ordered
int32_t	queueSize
cMessage *	timeoutMsg
cMessage *	timeMsg
int32_t	outboundStreams
int32_t	inboundStreams
int32_t	bytesRcvd
int32_t	echoedBytesSent
int32_t	lastStream
bool	sendAllowed
int32_t	chunksAbandoned
int32_t	numPacketsToReceive
bool	rendezvous
L3Address	peerAddress
int32_t	peerPort
AddressVector	peerAddressList
AddressVector	localAddressList
RcvdPacketsPerAssoc	rcvdPacketsPerAssoc
SentPacketsPerAssoc	sentPacketsPerAssoc
RcvdBytesPerAssoc	rcvdBytesPerAssoc
BytesPerAssoc	bytesPerAssoc
HistEndToEndDelay	histEndToEndDelay
EndToEndDelay	endToEndDelay

Detailed Description

Accepts any number of incoming connections, and sends back whatever arrives on them.

Member Typedef Documentation

BytesPerAssoc

typedef std::map<int32_t, cOutVector *> inet::SctpNatPeer::BytesPerAssoc

protected

EndToEndDelay

typedef std::map<int32_t, cOutVector *> inet::SctpNatPeer::EndToEndDelay

protected

HistEndToEndDelay

typedef std::map<int32_t, cHistogram *> inet::SctpNatPeer::HistEndToEndDelay

protected

RcvdBytesPerAssoc

typedef std::map<int32_t, int64_t> inet::SctpNatPeer::RcvdBytesPerAssoc

protected

RcvdPacketsPerAssoc

typedef std::map<int32_t, int64_t> inet::SctpNatPeer::RcvdPacketsPerAssoc

protected

SctpPathStatus

typedef std::map<L3Address, pathStatus> inet::SctpNatPeer::SctpPathStatus

SentPacketsPerAssoc

typedef std::map<int32_t, int64_t> inet::SctpNatPeer::SentPacketsPerAssoc

protected

Constructor & Destructor Documentation

SctpNatPeer()

inet::SctpNatPeer::SctpNatPeer

(

)

{
    timeMsg = nullptr;
    timeoutMsg = nullptr;
    numSessions = 0;
    packetsSent = 0;
    packetsRcvd = 0;
    bytesSent = 0;
    notifications = 0;
    serverAssocId = 0;
    delay = 0;
    echo = false;
    schedule = false;
    shutdownReceived = false;
    sendAllowed = true;
    numRequestsToSend = 0;
    ordered = true;
    queueSize = 0;
    outboundStreams = 1;
    inboundStreams = 1;
    bytesRcvd = 0;
    echoedBytesSent = 0;
    lastStream = 0;
    chunksAbandoned = 0;
    numPacketsToReceive = 1;
    rendezvous = false;
    peerPort = 0;
}

~SctpNatPeer()

inet::SctpNatPeer::~SctpNatPeer ( )

virtual

{
    cancelAndDelete(timeMsg);
    cancelAndDelete(timeoutMsg);
}

Member Function Documentation

addressAddedArrived()

void inet::SctpNatPeer::addressAddedArrived ( SctpSocket *  socket, L3Address  localAddr, L3Address  remoteAddr  )

overridevirtual

Reimplemented from inet::SctpSocket::ICallback.

{
    EV << getFullPath() << ": addressAddedArrived for remoteAddr " << remoteAddr << "\n";
    localAddressList.push_back(localAddr);
    clientSocket.addAddress(localAddr);
    if (rendezvous) {
        uint8_t buffer[100];
        int buflen = 16;
        struct nat_message *nat = (struct nat_message *)(buffer);
        nat->peer1 = par("ownName");
        nat->peer2 = par("peerName");
        nat->portPeer1 = par("localPort");
        nat->portPeer2 = 0;
        nat->numAddrPeer1 = 2;
        nat->numAddrPeer2 = 2;
        bool mul = par("multi");
        if (mul == true) {
            nat->multi = 1;
        }
        else {
            nat->multi = 0;
        }
        nat->peer1Addresses[0] = L3Address().toIpv4().getInt();
        nat->peer2Addresses[0] = L3Address().toIpv4().getInt();
        nat->peer1Addresses[1] = L3Address().toIpv4().getInt();
        nat->peer2Addresses[1] = L3Address().toIpv4().getInt();
        buflen = ADD_PADDING(buflen + 4 * (nat->numAddrPeer1 + nat->numAddrPeer2));
 
        auto applicationData = makeShared<BytesChunk>(buffer, buflen);
        applicationData->addTag<CreationTimeTag>()->setCreationTime(simTime());
        auto applicationPacket = new Packet("ApplicationPacket", SCTP_C_SEND_ORDERED);
        applicationPacket->insertAtBack(applicationData);
        auto sctpSendReq = applicationPacket->addTag<SctpSendReq>();
        sctpSendReq->setLast(true);
        sctpSendReq->setPrMethod(0);
        sctpSendReq->setPrValue(0);
        sctpSendReq->setSid(0);
        clientSocket.send(applicationPacket);
    }
}

connect()

void inet::SctpNatPeer::connect	(	L3Address	connectAddress,
		int32_t	connectPort
	)

{
    clientSocket.setOutboundStreams(outboundStreams);
    clientSocket.setInboundStreams(inboundStreams);
 
    EV << "issuing OPEN command\n";
    EV << "Assoc " << clientSocket.getSocketId() << "::connect to address " << connectAddress << ", port " << connectPort << "\n";
    bool streamReset = par("streamReset");
    clientSocket.connect(connectAddress, connectPort, streamReset, static_cast<int32_t>(par("prMethod")), static_cast<uint32_t>(par("numRequestsPerSession")));
    numSessions++;
 
    if (!streamReset)
        streamReset = false;
    else if (streamReset == true) {
        cMessage *cmsg = new cMessage("StreamReset", MSGKIND_RESET);
        EV << "StreamReset Timer scheduled at " << simTime() << "\n";
        scheduleAfter(par("streamRequestTime").doubleValue(), cmsg);
    }
    uint32_t streamNum = 0;
    cStringTokenizer tokenizer(par("streamPriorities"));
    while (tokenizer.hasMoreTokens()) {
        const char *token = tokenizer.nextToken();
        clientSocket.setStreamPriority(streamNum, static_cast<uint32_t>(atoi(token)));
 
        streamNum++;
    }
}

connectx()

void inet::SctpNatPeer::connectx	(	AddressVector	connectAddressList,
		int32_t	connectPort
	)

{
    uint32_t outStreams = par("outboundStreams");
    clientSocket.setOutboundStreams(outStreams);
    uint32_t inStreams = par("inboundStreams");
    clientSocket.setInboundStreams(inStreams);
 
    EV << "issuing OPEN command\n";
    EV << "Assoc " << clientSocket.getSocketId() << "::connect to  port " << connectPort << "\n";
    bool streamReset = par("streamReset");
    clientSocket.connectx(connectAddressList, connectPort, streamReset, static_cast<uint32_t>(par("prMethod")), static_cast<uint32_t>(par("numRequestsPerSession")));
    numSessions++;
 
    if (!streamReset)
        streamReset = false;
    else if (streamReset == true) {
        cMessage *cmsg = new cMessage("StreamReset", MSGKIND_RESET);
        EV << "StreamReset Timer scheduled at " << simTime() << "\n";
        scheduleAfter(par("streamRequestTime").doubleValue(), cmsg);
    }
    uint32_t streamNum = 0;
    cStringTokenizer tokenizer(par("streamPriorities"));
    while (tokenizer.hasMoreTokens()) {
        const char *token = tokenizer.nextToken();
        clientSocket.setStreamPriority(streamNum, static_cast<uint32_t>(atoi(token)));
 
        streamNum++;
    }
}

finish()

void inet::SctpNatPeer::finish ( )

override

{
    EV << getFullPath() << ": opened " << numSessions << " sessions\n";
    EV << getFullPath() << ": sent " << bytesSent << " bytes in " << packetsSent << " packets\n";
    for (auto& elem : rcvdBytesPerAssoc) {
        EV << getFullPath() << ": received " << elem.second << " bytes in assoc " << elem.first << "\n";
    }
    EV << getFullPath() << "Over all " << packetsRcvd << " packets received\n ";
    EV << getFullPath() << "Over all " << notifications << " notifications received\n ";
    for (auto& elem : bytesPerAssoc) {
        delete elem.second;
    }
    bytesPerAssoc.clear();
    for (auto& elem : endToEndDelay) {
        delete elem.second;
    }
    endToEndDelay.clear();
    for (auto& elem : histEndToEndDelay) {
        delete elem.second;
    }
    histEndToEndDelay.clear();
    rcvdPacketsPerAssoc.clear();
    sentPacketsPerAssoc.clear();
    rcvdBytesPerAssoc.clear();
}

generateAndSend()

void inet::SctpNatPeer::generateAndSend

(

)

{
    auto applicationPacket = new Packet("ApplicationPacket");
    auto applicationData = makeShared<BytesChunk>();
    int numBytes = par("requestLength");
    EV_INFO << "Send " << numBytes << " bytes of data, bytesSent = " << bytesSent << endl;
    std::vector<uint8_t> vec;
    vec.resize(numBytes);
    for (int i = 0; i < numBytes; i++)
        vec[i] = (bytesSent + i) & 0xFF;
    applicationData->setBytes(vec);
    applicationData->addTag<CreationTimeTag>()->setCreationTime(simTime());
    applicationPacket->insertAtBack(applicationData);
    auto sctpSendReq = applicationPacket->addTag<SctpSendReq>();
    sctpSendReq->setLast(true);
    sctpSendReq->setPrMethod(par("prMethod"));
    sctpSendReq->setPrValue(par("prValue"));
    lastStream = (lastStream + 1) % outboundStreams;
    sctpSendReq->setSid(lastStream);
    sctpSendReq->setSocketId(serverAssocId);
    applicationPacket->setKind(ordered ? SCTP_C_SEND_ORDERED : SCTP_C_SEND_UNORDERED);
    applicationPacket->addTag<SocketReq>()->setSocketId(serverAssocId);
    applicationPacket->addTag<DispatchProtocolReq>()->setProtocol(&Protocol::sctp);
    bytesSent += numBytes;
    packetsSent++;
    clientSocket.send(applicationPacket);
}

handleMessage()

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

override

{
    int32_t id;
 
    if (msg->isSelfMessage()) {
        handleTimer(msg);
    }
    else {
        EV << "SctpNatPeer::handleMessage kind=" << SctpAssociation::indicationName(msg->getKind()) << " (" << msg->getKind() << ")\n";
        switch (msg->getKind()) {
            case SCTP_I_ADDRESS_ADDED:
                if (rendezvous)
                    clientSocket.processMessage(msg);
                // else
                delete msg;
                break;
 
            case SCTP_I_PEER_CLOSED:
            case SCTP_I_ABORT: {
                if (rendezvous)
                    clientSocket.processMessage(msg);
                else {
                    Message *message = check_and_cast<Message *>(msg);
                    auto& msgtags = message->getTags();
                    auto& command = msgtags.findTag<SctpCommandReq>();
                    Request *cmsg = new Request("SCTP_C_ABORT", SCTP_C_ABORT);
                    auto cmd = cmsg->addTag<SctpSendReq>();
                    int assocId = command->getSocketId();
                    cmsg->addTag<SocketReq>()->setSocketId(assocId);
                    cmsg->addTag<DispatchProtocolReq>()->setProtocol(&Protocol::sctp);
                    cmd->setSocketId(assocId);
                    cmd->setSid(command->getSid());
                    cmd->setNumMsgs(command->getNumMsgs());
                    delete msg;
                    sendOrSchedule(cmsg);
                }
                break;
            }
 
            case SCTP_I_ESTABLISHED: {
                if (clientSocket.getState() == SctpSocket::CONNECTING) {
                    clientSocket.processMessage(msg);
                }
                else {
                    int32_t count = 0;
                    Message *message = check_and_cast<Message *>(msg);
                    auto& tags = message->getTags();
                    auto& connectInfo = tags.findTag<SctpConnectReq>();
                    numSessions++;
                    serverAssocId = connectInfo->getSocketId();
                    id = serverAssocId;
                    outboundStreams = connectInfo->getOutboundStreams();
                    inboundStreams = connectInfo->getInboundStreams();
                    rcvdPacketsPerAssoc[serverAssocId] = static_cast<int64_t>(par("numPacketsToReceivePerClient"));
                    sentPacketsPerAssoc[serverAssocId] = static_cast<int64_t>(par("numPacketsToSendPerClient"));
                    char text[128];
                    sprintf(text, "App: Received Bytes of assoc %d", serverAssocId);
                    bytesPerAssoc[serverAssocId] = new cOutVector(text);
                    rcvdBytesPerAssoc[serverAssocId] = 0;
                    sprintf(text, "App: EndToEndDelay of assoc %d", serverAssocId);
                    endToEndDelay[serverAssocId] = new cOutVector(text);
                    sprintf(text, "Hist: EndToEndDelay of assoc %d", serverAssocId);
                    histEndToEndDelay[serverAssocId] = new cHistogram(text);
 
                    delete msg;
 
                    if (static_cast<int64_t>(par("numPacketsToSendPerClient")) > 0) {
                        auto i = sentPacketsPerAssoc.find(serverAssocId);
                        numRequestsToSend = i->second;
                        if (par("thinkTime").doubleValue() > 0.0) {
                            generateAndSend();
                            timeoutMsg->setKind(SCTP_C_SEND);
                            scheduleAfter(par("thinkTime"), timeoutMsg);
                            numRequestsToSend--;
                            i->second = numRequestsToSend;
                        }
                        else {
                            if (queueSize == 0) {
                                while (numRequestsToSend > 0) {
                                    generateAndSend();
                                    numRequestsToSend--;
                                    i->second = numRequestsToSend;
                                }
                            }
                            else if (queueSize > 0) {
                                while (numRequestsToSend > 0 && count++ < queueSize * 2) {
                                    generateAndSend();
                                    numRequestsToSend--;
                                    i->second = numRequestsToSend;
                                }
 
                                Request *cmsg = new Request("SCTP_C_QUEUE_MSGS_LIMIT", SCTP_C_QUEUE_MSGS_LIMIT);
                                auto qinfo = cmsg->addTag<SctpInfoReq>();
                                qinfo->setText(queueSize);
                                qinfo->setSocketId(id);
                                sendOrSchedule(cmsg);
                            }
 
                            EV << "!!!!!!!!!!!!!!!All data sent from Peer !!!!!!!!!!\n";
 
                            auto j = rcvdPacketsPerAssoc.find(serverAssocId);
                            if (j->second == 0 && par("waitToClose").doubleValue() > 0.0) {
                                char as[5];
                                sprintf(as, "%d", serverAssocId);
                                cMessage *abortMsg = new cMessage(as, SCTP_I_ABORT);
                                scheduleAfter(par("waitToClose"), abortMsg);
                            }
                            else {
                                EV << "no more packets to send, call shutdown for assoc " << serverAssocId << "\n";
                                Request *cmsg = new Request("ShutdownRequest", SCTP_C_SHUTDOWN);
                                auto& cmd = cmsg->addTag<SctpCommandReq>();
                                cmd->setSocketId(serverAssocId);
                                sendOrSchedule(cmsg);
                            }
                        }
                    }
                }
                break;
            }
 
            case SCTP_I_DATA_NOTIFICATION: {
                EV_DETAIL << "NatPeer: SCTP_I_DATA_NOTIFICATION arrived\n";
                notifications++;
                Message *message = check_and_cast<Message *>(msg);
                auto& intags = message->getTags();
                auto& ind = intags.findTag<SctpCommandReq>();
                id = ind->getSocketId();
                Request *cmsg = new Request("ReceiveRequest", SCTP_C_RECEIVE);
                auto cmd = cmsg->addTag<SctpSendReq>();
                cmsg->addTag<SocketReq>()->setSocketId(id);
                cmsg->addTag<DispatchProtocolReq>()->setProtocol(&Protocol::sctp);
                cmd->setSocketId(id);
                cmd->setSid(ind->getSid());
                cmd->setNumMsgs(ind->getNumMsgs());
                delete msg;
                if (!cmsg->isScheduled() && schedule == false) {
                    scheduleAfter(par("delayFirstRead"), cmsg);
                }
                else if (schedule == true)
                    sendOrSchedule(cmsg);
                break;
            }
 
            case SCTP_I_DATA: {
                Packet *message = check_and_cast<Packet *>(msg);
                auto& ind = message->findTag<SctpRcvReq>();
                id = ind->getSocketId();
                if (rendezvous) {
                    const auto& smsg = message->peekDataAsBytes();
                    int bufferlen = B(smsg->getChunkLength()).get();
                    uint8_t buffer[bufferlen];
                    std::vector<uint8_t> vec = smsg->getBytes();
                    for (int i = 0; i < bufferlen; i++) {
                        buffer[i] = vec[i];
                    }
                    struct nat_message *nat = (struct nat_message *)buffer;
                    peerAddressList.clear();
                    if (nat->multi) {
                        peerAddressList.push_back(Ipv4Address(nat->peer2Addresses[0]));
                        EV << "address 0: " << Ipv4Address(nat->peer2Addresses[0]).str() << endl;
                        peerAddressList.push_back(Ipv4Address(nat->peer2Addresses[1]));
                        EV << "address 1: " << Ipv4Address(nat->peer2Addresses[1]).str() << endl;
                    }
                    else {
                        peerAddress = Ipv4Address(nat->peer2Addresses[0]);
                    }
                    peerPort = nat->portPeer2;
                    delete msg;
                }
                else {
                    auto j = rcvdBytesPerAssoc.find(id);
                    if (j == rcvdBytesPerAssoc.end() && (clientSocket.getState() == SctpSocket::CONNECTED))
                        clientSocket.processMessage(PK(msg));
                    else if (j != rcvdBytesPerAssoc.end()) {
                        j->second += PK(msg)->getBitLength() / 8;
                        auto k = bytesPerAssoc.find(id);
                        k->second->record(j->second);
                        packetsRcvd++;
                        if (!echo) {
                            if (par("numPacketsToReceivePerClient").intValue() > 0) {
                                auto i = rcvdPacketsPerAssoc.find(id);
                                i->second--;
                                SctpSimpleMessage *smsg = check_and_cast<SctpSimpleMessage *>(msg);
                                auto j = endToEndDelay.find(id);
                                j->second->record(simTime() - smsg->getCreationTime());
                                auto k = histEndToEndDelay.find(id);
                                k->second->collect(simTime() - smsg->getCreationTime());
 
                                if (i->second == 0) {
                                    Request *cmsg = new Request("SCTP_C_NO_OUTSTANDING", SCTP_C_NO_OUTSTANDING);
                                    auto qinfo = cmsg->addTag<SctpCommandReq>();
                                    qinfo->setSocketId(id);
                                    sendOrSchedule(cmsg);
                                }
                            }
                            delete msg;
                        }
                        else {
                            SctpSimpleMessage *smsg = check_and_cast<SctpSimpleMessage *>(msg->dup());
                            auto j = endToEndDelay.find(id);
                            j->second->record(simTime() - smsg->getCreationTime());
                            auto k = histEndToEndDelay.find(id);
                            k->second->collect(simTime() - smsg->getCreationTime());
                            Packet *cmsg = new Packet("SCTP_C_SEND", SCTP_C_SEND);
                            bytesSent += smsg->getByteLength();
                            auto cmd = cmsg->addTag<SctpSendReq>();
                            cmd->setSendUnordered(cmd->getSendUnordered());
                            lastStream = (lastStream + 1) % outboundStreams;
                            cmd->setSocketId(id);
                            cmd->setPrValue(0);
                            cmd->setSid(lastStream);
                            cmd->setLast(true);
                            cmsg->encapsulate(smsg);
                            cmsg->setControlInfo(cmd.get()); // KLUDGE
                            packetsSent++;
                            delete msg;
                            sendOrSchedule(cmsg);
                        }
                    }
                    else {
                        delete msg;
                    }
                }
                break;
            }
 
            case SCTP_I_SHUTDOWN_RECEIVED: {
                Message *message = check_and_cast<Message *>(msg);
                id = message->getTag<SocketInd>()->getSocketId();
                EV << "peer: SCTP_I_SHUTDOWN_RECEIVED for assoc " << id << "\n";
                auto i = rcvdPacketsPerAssoc.find(id);
                if (i == rcvdPacketsPerAssoc.end() && (clientSocket.getState() == SctpSocket::CONNECTED)) {
                    clientSocket.processMessage(msg);
                }
                else if (i != rcvdPacketsPerAssoc.end()) {
                    if (i->second == 0) {
                        Request *cmsg = new Request("SCTP_C_NO_OUTSTANDING", SCTP_C_NO_OUTSTANDING);
                        auto qinfo = cmsg->addTag<SctpCommandReq>();
                        qinfo->setSocketId(id);
                        sendOrSchedule(cmsg);
                    }
 
                    shutdownReceived = true;
                    delete msg;
                }
                else {
                    delete msg;
                }
                delete msg;
            }
 
            case SCTP_I_SEND_STREAMS_RESETTED:
            case SCTP_I_RCV_STREAMS_RESETTED: {
                EV << "Streams have been resetted\n";
                break;
            }
 
            case SCTP_I_CLOSED:
            case SCTP_I_SENDSOCKETOPTIONS:
                delete msg;
                break;
        }
    }
 
    if (hasGUI()) {
        char buf[32];
        auto l = rcvdBytesPerAssoc.find(id);
        sprintf(buf, "rcvd: %lld bytes\nsent: %lld bytes", (long long int)l->second, (long long int)bytesSent);
        getDisplayString().setTagArg("t", 0, buf);
    }
}

handleTimer()

void inet::SctpNatPeer::handleTimer

(

cMessage *

msg

)

{
    int32_t id;
 
    EV << "SctpNatPeer::handleTimer\n";
 
    switch (msg->getKind()) {
        case MSGKIND_CONNECT:
            EV << "starting session call connect\n";
            connect(L3AddressResolver().resolve(par("connectAddress"), 1), par("connectPort"));
            delete msg;
            break;
 
        case SCTP_C_SEND:
 
            EV << "SctpNatPeer:MSGKIND_SEND\n";
 
            if (numRequestsToSend > 0) {
                generateAndSend();
                if (par("thinkTime").doubleValue() > 0.0)
                    scheduleAfter(par("thinkTime"), timeoutMsg);
                numRequestsToSend--;
            }
            break;
 
        case SCTP_I_ABORT: {
            EV << "SctpNatPeer:MsgKIND_ABORT for assoc " << atoi(msg->getName()) << "\n";
 
            Request *cmsg = new Request("SCTP_C_CLOSE", SCTP_C_CLOSE);
            auto& cmd = cmsg->addTag<SctpCommandReq>();
            id = atoi(msg->getName());
            cmd->setSocketId(id);
            cmsg->setControlInfo(cmd.get()); // KLUDGE
            sendOrSchedule(cmsg);
            break;
        }
 
        case SCTP_C_RECEIVE:
 
            EV << "SctpNatPeer:SCTP_C_RECEIVE\n";
            schedule = true;
            sendOrSchedule(msg);
            break;
 
        default:
 
            EV << "MsgKind =" << msg->getKind() << " unknown\n";
 
            break;
    }
}

initialize()

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

override

{
    cSimpleModule::initialize(stage);
    EV_DEBUG << "initialize SCTP NAT Peer stage " << stage << endl;
    if (stage == INITSTAGE_LOCAL) {
        WATCH(numSessions);
        WATCH(packetsSent);
        WATCH(packetsRcvd);
        WATCH(bytesSent);
        WATCH(numRequestsToSend);
        timeoutMsg = new cMessage("SrvAppTimer");
        queueSize = par("queueSize");
    }
    else if (stage == INITSTAGE_APPLICATION_LAYER) {
        // parameters
        const char *addressesString = par("localAddress");
        AddressVector addresses = L3AddressResolver().resolve(cStringTokenizer(addressesString).asVector());
        int32_t port = par("localPort");
        echo = par("echo");
        delay = par("echoDelay");
        outboundStreams = par("outboundStreams");
        inboundStreams = par("inboundStreams");
        ordered = par("ordered");
        clientSocket.setOutputGate(gate("socketOut"));
        clientSocket.setCallback(this);
        if (addresses.size() == 0) {
            clientSocket.bind(port);
        }
        else {
            clientSocket.bindx(addresses, port);
        }
 
        rendezvous = par("rendezvous");
        if (par("startTime").doubleValue() > 0.0) {
            cMessage *msg = new cMessage("ConnectTimer", MSGKIND_CONNECT);
            scheduleAt(par("startTime"), msg);
        }
    }
}

msgAbandonedArrived()

void inet::SctpNatPeer::msgAbandonedArrived ( SctpSocket *  socket )

overridevirtual

Reimplemented from inet::SctpSocket::ICallback.

{
    chunksAbandoned++;
}

numInitStages()

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

inlineoverridevirtual

{ return NUM_INIT_STAGES; }

sendOrSchedule()

void inet::SctpNatPeer::sendOrSchedule ( cMessage *  msg )

protected

{
    if (delay == 0) {
        send(msg, "socketOut");
    }
    else {
        scheduleAfter(delay, msg);
    }
}

sendqueueFullArrived()

void inet::SctpNatPeer::sendqueueFullArrived ( SctpSocket *  socket )

overridevirtual

Reimplemented from inet::SctpSocket::ICallback.

{
    sendAllowed = false;
}

sendQueueRequest()

void inet::SctpNatPeer::sendQueueRequest

(

)

{
    Request *cmsg = new Request("SCTP_C_QUEUE_MSGS_LIMIT", SCTP_C_QUEUE_MSGS_LIMIT);
    auto& qinfo = cmsg->addTag<SctpInfoReq>();
    qinfo->setText(queueSize);
    qinfo->setSocketId(clientSocket.getSocketId());
    clientSocket.sendRequest(cmsg);
}

sendRequest()

void inet::SctpNatPeer::sendRequest ( bool  last = true )

protected

{
    EV << "sending request, " << numRequestsToSend - 1 << " more to go\n";
    int64_t numBytes = par("requestLength");
    if (numBytes < 1)
        numBytes = 1;
 
    EV << "SctpNatPeer: sending " << numBytes << " data bytes\n";
 
    auto cmsg = new Packet("ApplicationPacket");
    cmsg->setKind(ordered ? SCTP_C_SEND_ORDERED : SCTP_C_SEND_UNORDERED);
    auto msg = makeShared<BytesChunk>();
    std::vector<uint8_t> vec;
    vec.resize(numBytes);
    for (int i = 0; i < numBytes; i++)
        vec[i] = (bytesSent + i) & 0xFF;
    msg->setBytes(vec);
    msg->addTag<CreationTimeTag>()->setCreationTime(simTime());
    cmsg->insertAtBack(msg);
    auto sendCommand = cmsg->addTag<SctpSendReq>();
    sendCommand->setLast(true);
    // send SctpMessage with SctpSimpleMessage enclosed
    clientSocket.send(cmsg);
    bytesSent += numBytes;
}

sendRequestArrived()

void inet::SctpNatPeer::sendRequestArrived ( SctpSocket *  socket )

overridevirtual

Reimplemented from inet::SctpSocket::ICallback.

{
    int32_t count = 0;
 
    EV << "sendRequestArrived numRequestsToSend=" << numRequestsToSend << "\n";
    while (numRequestsToSend > 0 && count++ < queueSize && sendAllowed) {
        numRequestsToSend--;
        if (count == queueSize || numRequestsToSend == 0)
            sendRequest();
        else
            sendRequest(false);
 
        if (numRequestsToSend == 0) {
            EV << "no more packets to send, call shutdown\n";
            clientSocket.shutdown();
        }
    }
}

sendStreamResetNotification()

void inet::SctpNatPeer::sendStreamResetNotification

(

)

setPrimaryPath()

void inet::SctpNatPeer::setPrimaryPath

(

)

setStatusString()

void inet::SctpNatPeer::setStatusString

(

const char *

s

)

{
    if (hasGUI())
        getDisplayString().setTagArg("t", 0, s);
}

shutdownReceivedArrived()

void inet::SctpNatPeer::shutdownReceivedArrived ( SctpSocket *  socket )

overridevirtual

Reimplemented from inet::SctpSocket::ICallback.

{
    if (numRequestsToSend == 0 || rendezvous) {
        Message *cmsg = new Message("SCTP_C_NO_OUTSTANDING", SCTP_C_NO_OUTSTANDING);
        auto& qinfo = cmsg->addTag<SctpCommandReq>();
        qinfo->setSocketId(socket->getSocketId());
        clientSocket.sendNotification(cmsg);
    }
}

socketAvailable()

virtual void inet::SctpNatPeer::socketAvailable ( SctpSocket *  socket, Indication *  indication  )

inlineoverridevirtual

Implements inet::SctpSocket::ICallback.

                                                                                      {
        throw cRuntimeError("Model error, this module doesn't use any listener SCTP sockets");
    }

socketClosed()

void inet::SctpNatPeer::socketClosed ( SctpSocket *  socket )

overridevirtual

Does nothing but update statistics/status.

Redefine if you want to do something else, such as opening a new connection.

Reimplemented from inet::SctpSocket::ICallback.

{
    // *redefine* to start another session etc.
 
    EV << "connection closed\n";
    setStatusString("closed");
    clientSocket.close();
    if (rendezvous) {
        const char *addressesString = par("localAddress");
        AddressVector addresses = L3AddressResolver().resolve(cStringTokenizer(addressesString).asVector());
        int32_t port = par("localPort");
        rendezvousSocket.setOutputGate(gate("socketOut"));
        rendezvousSocket.setOutboundStreams(outboundStreams);
        rendezvousSocket.setInboundStreams(inboundStreams);
        if (addresses.size() == 0) {
            rendezvousSocket.bind(port);
            clientSocket.bind(port);
        }
        else {
            clientSocket.bindx(addresses, port);
            rendezvousSocket.bindx(addresses, port);
        }
        rendezvousSocket.listen(true, par("streamReset").boolValue(), par("numPacketsToSendPerClient"));
        if (par("multi").boolValue())
            connectx(peerAddressList, peerPort);
        else
            connect(peerAddress, peerPort);
        rendezvous = false;
    }
}

socketDataArrived()

void inet::SctpNatPeer::socketDataArrived ( SctpSocket *  socket, Packet *  msg, bool  urgent  )

overridevirtual

Does nothing but update statistics/status.

Redefine to perform or schedule next sending. Beware: this funcion deletes the incoming message, which might not be what you want.

Implements inet::SctpSocket::ICallback.

{
    // *redefine* to perform or schedule next sending
    packetsRcvd++;
 
    EV << "Client received packet Nr " << packetsRcvd << " from SCTP\n";
 
    auto& tags = msg->getTags();
    const auto& ind = tags.findTag<SctpRcvReq>();
 
    bytesRcvd += msg->getByteLength();
 
    if (echo) {
        const auto& smsg = msg->peekData();
        auto cmsg = new Packet("ApplicationPacket");
        cmsg->setKind(ind->getSendUnordered() ? SCTP_C_SEND_UNORDERED : SCTP_C_SEND_ORDERED);
        cmsg->insertAtBack(smsg);
        auto cmd = cmsg->addTag<SctpSendReq>();
        cmd->setLast(true);
        cmd->setSocketId(ind->getSocketId());
        cmd->setPrValue(0);
        cmd->setSid(ind->getSid());
        packetsSent++;
        clientSocket.send(cmsg);
    }
    if (par("numPacketsToReceive").intValue() > 0) {
        numPacketsToReceive--;
        delete msg;
        if (numPacketsToReceive == 0) {
            EV << "Peer: all packets received\n";
        }
    }
}

socketDataNotificationArrived()

void inet::SctpNatPeer::socketDataNotificationArrived ( SctpSocket *  socket, Message *  msg  )

overridevirtual

Implements inet::SctpSocket::ICallback.

{
    Message *message = check_and_cast<Message *>(msg);
    auto& intags = message->getTags();
    auto& ind = intags.findTag<SctpCommandReq>();
    Request *cmesg = new Request("SCTP_C_RECEIVE", SCTP_C_RECEIVE);
    auto& cmd = cmesg->addTag<SctpSendReq>();
    cmd->setSocketId(ind->getSocketId());
    cmd->setSid(ind->getSid());
    cmd->setNumMsgs(ind->getNumMsgs());
    clientSocket.sendNotification(cmesg);
}

socketEstablished()

void inet::SctpNatPeer::socketEstablished ( SctpSocket *  socket, unsigned long int  buffer  )

overridevirtual

Does nothing but update statistics/status.

Redefine to perform or schedule first sending.

Reimplemented from inet::SctpSocket::ICallback.

{
    int32_t count = 0;
    // *redefine* to perform or schedule first sending
    EV << "SctpNatPeer: socketEstablished\n";
    setStatusString("connected");
    if (rendezvous) {
        uint8_t buffer[100];
        int buflen = 14;
        struct nat_message *nat = (struct nat_message *)(buffer);
 
        nat->peer1 = par("ownName");
        nat->peer2 = par("peerName");
        nat->portPeer1 = par("localPort");
        nat->portPeer2 = 0;
        nat->peer1Addresses[0] = L3Address().toIpv4().getInt();
        nat->peer2Addresses[0] = L3Address().toIpv4().getInt();
        nat->numAddrPeer1 = 1;
        nat->numAddrPeer2 = 1;
        buflen = ADD_PADDING(buflen + 4 * (nat->numAddrPeer1 + nat->numAddrPeer2));
        bool mul = par("multi");
        if (mul == true) {
            nat->multi = 1;
        }
        else {
            nat->multi = 0;
        }
 
        auto applicationData = makeShared<BytesChunk>(buffer, buflen);
        applicationData->addTag<CreationTimeTag>()->setCreationTime(simTime());
        auto applicationPacket = new Packet("ApplicationPacket", SCTP_C_SEND_ORDERED);
        applicationPacket->insertAtBack(applicationData);
        auto sctpSendReq = applicationPacket->addTag<SctpSendReq>();
        sctpSendReq->setLast(true);
        sctpSendReq->setPrMethod(0);
        sctpSendReq->setPrValue(0);
        sctpSendReq->setSid(0);
        clientSocket.send(applicationPacket);
 
        if (par("multi").boolValue()) {
            Request *cmesg = new Request("SCTP_C_SEND_ASCONF", SCTP_C_SEND_ASCONF);
            auto& cmd = cmesg->addTag<SctpCommandReq>();
            cmd->setSocketId(clientSocket.getSocketId());
            clientSocket.sendNotification(cmesg);
        }
    }
    else {
        EV << " determine number of requests in this session\n";
        numRequestsToSend = par("numRequestsPerSession");
        numPacketsToReceive = par("numPacketsToReceive");
        if (numRequestsToSend < 1)
            numRequestsToSend = 0;
        // perform first request (next one will be sent when reply arrives)
        if (numRequestsToSend > 0) {
            if (par("thinkTime").doubleValue() > 0.0) {
                if (sendAllowed) {
                    sendRequest();
                    numRequestsToSend--;
                }
                timeMsg->setKind(MSGKIND_SEND);
                scheduleAfter(par("thinkTime"), timeMsg);
            }
            else {
                if (queueSize > 0) {
                    while (numRequestsToSend > 0 && count++ < queueSize * 2 && sendAllowed) {
                        if (count == queueSize * 2)
                            sendRequest();
                        else
                            sendRequest(false);
                        numRequestsToSend--;
                    }
                    if (numRequestsToSend > 0 && sendAllowed)
                        sendQueueRequest();
                }
                else {
                    while (numRequestsToSend > 0 && sendAllowed) {
                        sendRequest();
                        numRequestsToSend--;
                    }
                }
 
                if (numPacketsToReceive == 0 && par("waitToClose").doubleValue() > 0.0) {
                    timeMsg->setKind(MSGKIND_ABORT);
                    scheduleAfter(par("waitToClose"), timeMsg);
                }
                if (numRequestsToSend == 0 && par("waitToClose").doubleValue() == 0.0) {
                    EV << "socketEstablished:no more packets to send, call shutdown\n";
                    clientSocket.shutdown();
                }
            }
        }
    }
}

socketFailure()

void inet::SctpNatPeer::socketFailure ( SctpSocket *  socket, int32_t  code  )

override

Does nothing but update statistics/status.

Redefine if you want to try reconnecting after a delay.

{
    // subclasses may override this function, and add code try to reconnect after a delay.
    EV << "connection broken\n";
    setStatusString("broken");
 
//    numBroken++;
 
    // reconnect after a delay
    timeMsg->setKind(MSGKIND_CONNECT);
    scheduleAfter(par("reconnectInterval"), timeMsg);
}

socketPeerClosed()

void inet::SctpNatPeer::socketPeerClosed ( SctpSocket *  socket )

overridevirtual

Since remote SCTP closed, invokes close().

Redefine if you want to do something else.

Reimplemented from inet::SctpSocket::ICallback.

{
    // close the connection (if not already closed)
    if (clientSocket.getState() == SctpSocket::PEER_CLOSED) {
        EV << "remote Sctp closed, closing here as well\n";
        setStatusString("closed");
        clientSocket.close();
        if (rendezvous) {
            const char *addressesString = par("localAddress");
            AddressVector addresses = L3AddressResolver().resolve(cStringTokenizer(addressesString).asVector());
            int32_t port = par("localPort");
            rendezvousSocket.setOutputGate(gate("socketOut"));
            rendezvousSocket.setOutboundStreams(outboundStreams);
            rendezvousSocket.setInboundStreams(inboundStreams);
            if (addresses.size() == 0) {
                rendezvousSocket.bind(port);
                clientSocket.bind(port);
            }
            else {
                clientSocket.bindx(addresses, port);
                rendezvousSocket.bindx(addresses, port);
            }
            rendezvousSocket.listen(true, par("streamReset").boolValue(), par("numPacketsToSendPerClient"));
            if (par("multi").boolValue())
                connectx(peerAddressList, peerPort);
            else
                connect(peerAddress, peerPort);
            rendezvous = false;
        }
    }
}

socketStatusArrived()

void inet::SctpNatPeer::socketStatusArrived ( SctpSocket *  socket, SctpStatusReq *  status  )

overridevirtual

Redefine to handle incoming SctpStatusInfo.

Reimplemented from inet::SctpSocket::ICallback.

{
    struct pathStatus ps;
    auto i = sctpPathStatus.find(status->getPathId());
    if (i != sctpPathStatus.end()) {
        ps = i->second;
        ps.active = status->getActive();
    }
    else {
        ps.active = status->getActive();
//        ps.pid = status->pathId();  FIXME
        ps.primaryPath = false;
        sctpPathStatus[ps.pid] = ps;
    }
}

Member Data Documentation

bytesPerAssoc

BytesPerAssoc inet::SctpNatPeer::bytesPerAssoc

protected

bytesRcvd

int32_t inet::SctpNatPeer::bytesRcvd

protected

bytesSent

int64_t inet::SctpNatPeer::bytesSent

protected

chunksAbandoned

int32_t inet::SctpNatPeer::chunksAbandoned

protected

clientSocket

SctpSocket inet::SctpNatPeer::clientSocket

protected

delay

double inet::SctpNatPeer::delay

protected

echo

bool inet::SctpNatPeer::echo

protected

echoedBytesSent

int32_t inet::SctpNatPeer::echoedBytesSent

protected

endToEndDelay

EndToEndDelay inet::SctpNatPeer::endToEndDelay

protected

histEndToEndDelay

HistEndToEndDelay inet::SctpNatPeer::histEndToEndDelay

protected

inboundStreams

int32_t inet::SctpNatPeer::inboundStreams

protected

lastStream

int32_t inet::SctpNatPeer::lastStream

protected

localAddressList

AddressVector inet::SctpNatPeer::localAddressList

protected

notifications

int32_t inet::SctpNatPeer::notifications

protected

numPacketsToReceive

int32_t inet::SctpNatPeer::numPacketsToReceive

protected

numRequestsToSend

int32_t inet::SctpNatPeer::numRequestsToSend

protected

numSessions

int32_t inet::SctpNatPeer::numSessions

protected

ordered

bool inet::SctpNatPeer::ordered

protected

outboundStreams

int32_t inet::SctpNatPeer::outboundStreams

protected

packetsRcvd

int32_t inet::SctpNatPeer::packetsRcvd

protected

packetsSent

int32_t inet::SctpNatPeer::packetsSent

protected

peerAddress

L3Address inet::SctpNatPeer::peerAddress

protected

peerAddressList

AddressVector inet::SctpNatPeer::peerAddressList

protected

peerPort

int32_t inet::SctpNatPeer::peerPort

protected

peerSocket

SctpSocket inet::SctpNatPeer::peerSocket

protected

queueSize

int32_t inet::SctpNatPeer::queueSize

protected

rcvdBytesPerAssoc

RcvdBytesPerAssoc inet::SctpNatPeer::rcvdBytesPerAssoc

protected

rcvdPacketsPerAssoc

RcvdPacketsPerAssoc inet::SctpNatPeer::rcvdPacketsPerAssoc

protected

rendezvous

bool inet::SctpNatPeer::rendezvous

protected

rendezvousSocket

SctpSocket inet::SctpNatPeer::rendezvousSocket

protected

schedule

bool inet::SctpNatPeer::schedule

protected

sctpPathStatus

SctpPathStatus inet::SctpNatPeer::sctpPathStatus

sendAllowed

bool inet::SctpNatPeer::sendAllowed

protected

sentPacketsPerAssoc

SentPacketsPerAssoc inet::SctpNatPeer::sentPacketsPerAssoc

protected

serverAssocId

int32_t inet::SctpNatPeer::serverAssocId

protected

shutdownReceived

bool inet::SctpNatPeer::shutdownReceived

protected

timeMsg

cMessage* inet::SctpNatPeer::timeMsg

protected

timeoutMsg

cMessage* inet::SctpNatPeer::timeoutMsg

protected

---

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

• SctpNatPeer.h
• SctpNatPeer.cc