arzmq-sys 0.5.0

// This is a test application for experimenting
// with the NORM API implementation during its
// development.  A better-documented and complete
// example of the NORM API usage will be provided
// when the NORM API is more complete.

#include "normApi.h"
#include "protokit.h"  // for protolib debug, stuff, etc

#include <stdio.h>
#include <stdlib.h>  // for srand()

#ifdef UNIX
#include <unistd.h>  // for "sleep()"
#endif // UNIX

int main(int argc, char* argv[])
{
    bool send = false;
    bool recv = false;
    bool cc = false;
    bool trace = false;
    int debugLevel = 2;
    double loss = 0.0;
    int sendMax = 3000;//-1;    // -1 means unlimited
    const char* cmd = NULL;
    
    
    // 1) Parse command-line options
    for (int i = 1; i < argc; i++)
    {
        if (!strcmp(argv[i], "send"))
        {
            send = true; 
        }
        else if (!strcmp(argv[i], "recv"))
        {
            recv = true;   
        } 
        else if (!strcmp(argv[i], "cc"))
        {
            cc = true;
        }
        else if (!strcmp(argv[i], "trace"))
        {
            trace = true;
        }
        else if (!strcmp(argv[i], "debug"))
        {
            debugLevel = atoi(argv[++i]);
        }
        else if (!strcmp(argv[i], "loss"))
        {
            loss = (double)atoi(argv[++i]);
        }
        else if (!strcmp(argv[i], "smax"))
        {
            sendMax = atoi(argv[++i]);
        }
        else if (!strcmp(argv[i], "cmd"))
        {
            cmd = argv[++i];
        }
    }
    
    TRACE("loss = %lf\n", loss);
    
    int sendMid = sendMax / 2;
    
    if (!send && !recv)
    {
        TRACE("normTest: neither 'send' or 'recv' operation was specified?!\n");
        TRACE("Usage: normTest {send, recv} [cc]\n");
        return -1;   
    }
    
    bool loopback = (send && recv);
    
    NormInstanceHandle instance = NormCreateInstance();

#ifdef WIN32
    const char* cachePath = "C:\\Adamson\\Temp\\cache\\";
#else
    const char* cachePath =  "/tmp/";
#endif // if/else WIN32/UNIX

    NormSetCacheDirectory(instance, cachePath);
    
    // Here's a trick to generate a _hopefully_ unique NormNodeId
    // based on an XOR of the system's IP address and the process id.
    // (We use ProtoAddress::GetEndIdentifier() to get the local
    //  "default" IP address for the system)
    // (Note that passing "NORM_NODE_ANY" to the last arg of 
    //  NormCreateSession() does a similar thing but without
    //  the processId XOR ... perhaps we should add the processId
    //  hack to that default "NORM_NODE_ANY" local NormNodeId picker???
    ProtoAddress localAddr;
    if (!localAddr.ResolveLocalAddress())
    {
        fprintf(stderr, "normTest: error resolving local IP address\n");
        NormDestroyInstance(instance);
        return -1;
    }
    NormNodeId localId = localAddr.EndIdentifier();
#ifdef WIN32
    DWORD processId = GetCurrentProcessId();
#else
    pid_t processId = getpid();
#endif // if/else WIN32/UNIX
    localId ^= (NormNodeId)processId;
    
    // If needed, permutate to a valid, random NormNodeId
    while ((NORM_NODE_ANY == localId) ||
           (NORM_NODE_NONE == localId))
    {
        localId ^= (NormNodeId)rand();
    }
    //localId = 15;  // for testing purposes

    // Create a NORM session instance
    NormSessionHandle session = NormCreateSession(instance,
                                                  "224.1.1.1", 
                                                   6001,
                                                   localId);
    
    //NormSetTOS(session, 0x20);
    
    // NOTE: These are debugging routines available (not for normal app use)
    // (IMPORTANT NOTE: On Win32 builds with Norm.dll, this "SetDebugLevel()" has no
    //                  effect on the Protolib instance in the NORM DLL !!!
    //                  (TBD - provide a "NormSetDebugLevel()" function for this purpose!)
    SetDebugLevel(debugLevel);
    // Option to turn on debug NORM message tracing
    NormSetMessageTrace(session, trace);
    
    // Uncomment to turn on some random packet loss
    NormSetTxLoss(session, loss);    
    //NormSetRxLoss(session, 20.0);
    struct timeval currentTime;
    ProtoSystemTime(currentTime);
    // Uncomment to get different packet loss patterns from run to run
    // (and a different sender sessionId)
    srand(currentTime.tv_sec);  // seed random number generator
    
    NormSetGrttEstimate(session, 0.25);  // 1 msec initial grtt
    
    NormSetTxRate(session, 5.0e+06);  // in bits/second
    
    // Option to enable TCP-friendly congestion control (overrides NormSetTxRate())
    if (cc) NormSetCongestionControl(session, true);
    NormSetTxRateBounds(session, 5.0e+06, 6.0e+06);
    
    //NormSetDefaultRepairBoundary(session, NORM_BOUNDARY_BLOCK); 
    
    // Uncomment to use a _specific_ transmit port number
    // (Can be the same as session port (rx port), but this
    //  is _not_ recommended when unicast feedback may be
    //  possible!)
    //NormSetTxPort(session, 6001); 
    
    //NormSetDefaultUnicastNack(session, true);
        
    // Uncomment to allow reuse of rx port
    // (This allows multiple "normTest" instances on the same machine
    //  for the same NormSession - note those instances must use
    //  unique local NormNodeIds (see NormCreateSession() above).
    //NormSetRxPortReuse(session, true);
    
    // Uncomment to receive your own traffic
    if (loopback)
        NormSetLoopback(session, true);     
    
    //NormSetSilentReceiver(session, true);
    
    // Uncomment this line to participate as a receiver
    if (recv) NormStartReceiver(session, 1024*1024);
    
    // We use a random "sessionId"
    NormSessionId sessionId = (NormSessionId)rand();
    
    if (send) NormStartSender(session, sessionId, 2097152, 1420, 16, 4);
    
    // Uncomment to set large rx socket buffer size
    // (might be needed for high rate sessions)
    //NormSetRxSocketBuffer(session, 512000);
    
    //NormSetAutoParity(session, 6);

    // Uncomment to set large tx socket buffer size
    // (might be needed for high rate sessions)
    //NormSetTxSocketBuffer(session, 512000);
    
    NormAddAckingNode(session, NORM_NODE_NONE); //15); //NormGetLocalNodeId(session));
    
    NormObjectHandle stream = NORM_OBJECT_INVALID;
#ifdef WIN32
    const char* filePath = "C:\\Adamson\\Images\\Art\\giger205.jpg";
#else
    const char* filePath = "/home/adamson/images/art/giger/giger205.jpg";
#endif
    //const char* filePath = "/home/adamson/pkgs/rh73.tgz";
    const char* fileName = "file1.jpg";
    const char* fileName2 = "file2.jpg";
    
    
    // Uncomment this line to send a stream instead of the file
    stream = NormStreamOpen(session, 2*1024*1024);
       
    // NORM_FLUSH_PASSIVE automatically flushes full writes to
    // the stream.
    NormStreamSetAutoFlush(stream, NORM_FLUSH_PASSIVE);   
    
    //NormStreamWrite(stream, 0, 0);//txBuffer+txIndex, want);
    
    // Some variable for stream input/output
    char txBuffer[8192], rxBuffer[8192];
    int txIndex = 0;
    int txLen = 0;
    int rxIndex = 0;
    
#define STREAM_MSG_PREFIX_SIZE 1000  // we pad with 'a' character value (the msgPrefix is a prefix)
    char msgPrefix[STREAM_MSG_PREFIX_SIZE];
    memset(msgPrefix, 'a', STREAM_MSG_PREFIX_SIZE);
    bool msgSync = false;
    
    int msgCount = 0;
    int recvCount = -1;  // used to monitor reliable stream reception
    int sendCount = 0;
    
    
    int fileMax = 1;
    NormObjectHandle txFile = NORM_OBJECT_INVALID;
    
    NormEvent theEvent;
    
    bool rxActive = false;
    
    if (NULL != cmd)
        NormSendCommand(session, cmd, strlen(cmd) + 1, false);
                    
    
    while (NormGetNextEvent(instance, &theEvent))
    {
        switch (theEvent.type)
        {
            case NORM_CC_ACTIVE:
                PLOG(PL_INFO, "normTest: NORM_CC_ACTIVE event ...\n");
                if (rxActive)
                {
                    break;
                }
                else
                {
                    rxActive = true;
                    // assume there is vacancy
                }
            
            case NORM_TX_QUEUE_VACANCY:
            case NORM_TX_QUEUE_EMPTY:
                //if (!rxActive) break;
                /*
                if (NORM_TX_QUEUE_VACANCY == theEvent.type)
                    TRACE("NORM_TX_QUEUE_VACANCY ...\n");
                else if (NORM_TX_QUEUE_EMPTY == theEvent.type)
                    TRACE("NORM_TX_QUEUE_EMPTY ...\n");
                else
                    TRACE("writing to stream after CC_ACTIVE\n");
                */
                        
                if ((sendMax > 0) && (sendCount >= sendMax)) 
                {
                    // send "cmd" again at end
                    if (NULL != cmd)
                        NormSendCommand(session, cmd, strlen(cmd) + 1, false);
                    break;
                }
                //if (NORM_OBJECT_INVALID != theEvent.object)
                if (true)
                {
                    // We loop here to keep stream buffer full ....
                    bool keepWriting = true;
                    while (keepWriting)
                    {
                        if (0 == txLen)
                        {
                            memset(txBuffer, 'a', STREAM_MSG_PREFIX_SIZE);
                            // Write a message into the "txBuffer" for transmission
                            sprintf(txBuffer+STREAM_MSG_PREFIX_SIZE, " normTest says hello %d ...\n", sendCount);
                            txLen = strlen(txBuffer);
                        }
                        unsigned int want = txLen - txIndex;
                        unsigned int put = NormStreamWrite(stream, txBuffer+txIndex, want);
                        if (put != want) keepWriting = false;
                        txIndex += put;
                        if (txIndex == txLen)
                        {
                            // _Instead_ of "NormStreamSetAutoFlush(stream, NORM_FLUSH_PASSIVE)" above
                            // _and_ "NormStreamMarkEom()" here, I could have used 
                            // "NormStreamFlush(stream, true)" here to perform explicit flushing 
                            // and EOM marking in one fell swoop.  That would be a simpler approach 
                            // for apps where very big stream messages might need to be written with 
                            // multiple calls to "NormStreamWrite()"
                            NormStreamMarkEom(stream);
                            txLen = txIndex = 0;
                            sendCount++;
                            
                            if ((NULL != cmd) && (sendCount == sendMid))
                                NormSendCommand(session, cmd, strlen(cmd) + 1, true);        
                                
                
                            
                            if (0 == (sendCount % 1000))
                                TRACE("normTest: sender sent %d messages ...\n", sendCount);
                            if (sendCount == 15)
                            {
                                //NormSetWatermark(session, stream);   
                            }
                            if ((sendMax > 0) && (sendCount >= sendMax))
                            {
                                // Uncomment to gracefully shut down the stream
                                // after "sendMax" messages
                                TRACE("normTest: sender closing stream ...\n");
                                NormStreamClose(stream, true);  
                                keepWriting = false; 
                            }                            
                        }
                    }  // end while(keepWriting)
                }
                else if (NORM_OBJECT_INVALID == stream)  // we aren't sending a stream 
                {
                    if ((fileMax < 0) || (sendCount < fileMax))
                    {
                        const char* namePtr = (0 == (sendCount & 0x01)) ? fileName : fileName2;
                        txFile = 
                            NormFileEnqueue(session,
                                            filePath,
                                            namePtr,
                                            strlen(fileName));
                        // Repeatedly queue our file for sending
                        if (NORM_OBJECT_INVALID == txFile)
                        {
                            PLOG(PL_ERROR, "normTest: error queuing file: %s\n", filePath);
                            break;   
                        }
                        else
                        {
                            TRACE("QUEUED FILE ...\n");
                            sendCount++;
                            if (sendCount < 2) NormSetWatermark(session, txFile);
                        }
                    }
                }
                break;   

            case NORM_TX_FLUSH_COMPLETED:
                PLOG(PL_INFO, "normTest: NORM_TX_FLUSH_COMPLETED event ...\n");
                // This line of code "requeues" the last file sent
                // (Good for use with silent receivers ... can repeat transmit object this way)
                NormRequeueObject(session, txFile);
                break;  

            case NORM_TX_WATERMARK_COMPLETED:
                PLOG(PL_INFO, "normTest: NORM_TX_WATERMARK_COMPLETED event ...\n");
                break;
                
            case NORM_TX_CMD_SENT:
                PLOG(PL_INFO, "normTest: NORM_TX_CMD_SENT event ...\n");
                break;
                
            case NORM_TX_OBJECT_SENT:
                PLOG(PL_INFO, "normTest: NORM_TX_WATERMARK_COMPLETED event ...\n");
                break;

            case NORM_TX_OBJECT_PURGED:
                PLOG(PL_INFO, "normTest: NORM_TX_OBJECT_PURGED event ...\n");
                break;  
                
            case NORM_CC_INACTIVE:
                PLOG(PL_INFO, "normTest: NORM_CC_INACTIVE event ...\n");
                rxActive = false;
                // (TBD) add APIs to delete remote sender's state entirely
                // (TBD) add APIs to automate buffer freeing, remote sender managment somewhat
                break;
                
            case NORM_REMOTE_SENDER_ACTIVE:
                break;  
                
            case NORM_REMOTE_SENDER_INACTIVE:
                NormNodeFreeBuffers(theEvent.sender);  // frees up some memory allocated for this remote sender's state
                break;    
                
            case NORM_RX_CMD_NEW:
            {
                char cmdBuffer[8192];
                unsigned int cmdLength = 8192;
                if (NormNodeGetCommand(theEvent.sender, cmdBuffer, &cmdLength))
                {
                    char addrBuffer[16];
                    unsigned int addrLength = 16;
                    if (!NormNodeGetAddress(theEvent.sender, addrBuffer, &addrLength))
                        PLOG(PL_ERROR, "normTest: error getting sender addr!\n");
                    ProtoAddress senderAddr;
                    if (4 == addrLength)
                        senderAddr.SetRawHostAddress(ProtoAddress::IPv4, addrBuffer, 4);
                    else if (16 == addrLength)
                        senderAddr.SetRawHostAddress(ProtoAddress::IPv6, addrBuffer, 16);
                    cmdBuffer[cmdLength] = '\0';
                    PLOG(PL_ERROR, "normTest: recvd cmd \"%s\" from sender %s\n",
                            cmdBuffer, senderAddr.GetHostString());
                }
                PLOG(PL_INFO, "normTest: NORM_RX_CMD_NEW event ...\n");
                break;
            }

            case NORM_RX_OBJECT_NEW:
                PLOG(PL_INFO, "normTest: NORM_RX_OBJECT_NEW event ...\n");
                break;

            case NORM_RX_OBJECT_INFO:
                // Assume info contains '/' delimited <path/fileName> string
                if (NORM_OBJECT_FILE == NormObjectGetType(theEvent.object))
                {
                    char fileName[PATH_MAX];
                    strcpy(fileName, cachePath);
                    int pathLen = strlen(fileName);
                    unsigned short nameLen = PATH_MAX - pathLen;
                    nameLen = NormObjectGetInfo(theEvent.object, fileName+pathLen, nameLen);
                    fileName[nameLen + pathLen] = '\0';
                    char* ptr = fileName + 5;
                    while ('\0' != *ptr)
                    {
                        if ('/' == *ptr) *ptr = PROTO_PATH_DELIMITER;
                        ptr++;   
                    }
                    if (!NormFileRename(theEvent.object, fileName))
                        TRACE("normTest: NormSetFileName(%s) error\n", fileName);
                    PLOG(PL_DEBUG, "normTest: recv'd info for file: %s\n", fileName);   
                }
                break;

            case NORM_RX_OBJECT_UPDATED:
            {
                //TRACE("normTest: NORM_RX_OBJECT_UPDATE event ...\n");
                if (NORM_OBJECT_STREAM != NormObjectGetType(theEvent.object))
                    break;
                unsigned int len;
                do
                {
                    if (!msgSync) 
                        msgSync = NormStreamSeekMsgStart(theEvent.object);
                    if (!msgSync) break;
                    len = 8191 - rxIndex;
                    if (NormStreamRead(theEvent.object, rxBuffer+rxIndex, &len))
                    {
                        rxIndex += len;
                        rxBuffer[rxIndex] = '\0';
                        if (rxIndex > 0)
                        {
                            // The '\n' indicates we have received an entire message
                            int parsedLen = 0;
                            bool searching = true;
                            char* startPtr = rxBuffer;
                            while (searching)
                            {
                                char* endPtr = strchr(startPtr, '\n');
                                if (endPtr)
                                {
                                    // Save sub-string message length
                                    parsedLen += (endPtr - startPtr + 1);
                                    // Validate received message string
                                    if (0 == memcmp(rxBuffer, msgPrefix, STREAM_MSG_PREFIX_SIZE))
                                    {
                                        char* ptr = strstr(startPtr, "hello");
                                        if (ptr)
                                        {
                                            int value;
                                            if (1 == sscanf(ptr, "hello %d", &value))
                                            {
                                                msgCount++; // successful recv
                                                if (recvCount >= 0)
                                                {
                                                    if (0 != (value - recvCount))
                                                        TRACE("WARNING! possible break? value:%d recvCount:%d\n",
                                                                value, recvCount);
                                                }
                                                //else
                                                //    TRACE("validated recv msg len:%d\n", len);
                                                recvCount = value+1;   
                                                if (0 == msgCount % 1000)
                                                {
                                                    TRACE("normTest: recv status> msgCount:%d of total:%d (%lf)\n",
                                                          msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount));
                                                }
                                            }
                                            else
                                            {
                                                TRACE("couldn't find index!? len:%d in %s\n", len, ptr);
                                                ASSERT(0);
                                            }
                                        }
                                        else
                                        {
                                            TRACE("couldn't find \"hello\"!? len:%d in %s\n", len, rxBuffer);
                                            ASSERT(0);
                                        }
                                    }
                                    else
                                    {
                                        TRACE("invalid received message!?\n");
                                        ASSERT(0);
                                    }
                                    if (parsedLen >= rxIndex) 
                                    {
                                        rxIndex = 0;
                                        searching = false;
                                    }
                                    else
                                    {
                                        startPtr = rxBuffer + parsedLen;
                                    }
                                }  // end if (endPtr)
                                else if (parsedLen > 0)
                                {
                                    memmove(rxBuffer, rxBuffer+parsedLen, rxIndex - parsedLen);
                                    rxIndex -= parsedLen;
                                    searching = false;
                                }
                                else
                                {
                                    searching = false;
                                }
                            }  // end while(searching)
                        }
                    }
                    else
                    {
                        TRACE("normTest: error reading stream\n"); 
                        TRACE("normTest: recv status> msgCount:%d of total:%d (%lf)\n",
                              msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount));
                        msgSync = false;
                        rxIndex = 0;
                        break; 
                    }  
                } while (0 != len);
                break;                 
            }

            case NORM_RX_OBJECT_COMPLETED:
                TRACE("normTest: NORM_RX_OBJECT_COMPLETED event ...\n");
                TRACE("normTest: recv status> msgCount:%d of total:%d (%lf)\n",
                      msgCount, recvCount, 100.0*((double)msgCount)/((double)recvCount));
                //NormStopReceiver(session);
                break;

            case NORM_RX_OBJECT_ABORTED:
                TRACE("normTest: NORM_RX_OBJECT_ABORTED event ...\n");
                break;
                
            case NORM_GRTT_UPDATED:
                //TRACE("normTest: NORM_GRTT_UPDATED event ...\n");
                break;

            default:
                TRACE("Got event type: %d\n", theEvent.type); 
        }  // end switch(theEvent.type)
        
        // Uncomment to exit program after sending "sendMax" messages or files
        //if ((sendMax > 0) && (sendCount >= sendMax)) break;
        
    }  // end while (NormGetNextEvent())
   
    
#ifdef WIN32
    Sleep(30000);
#else
    sleep(30);  // allows time for cleanup if we're sending to someone else
#endif // if/else WIN32/UNIX

    //NormStreamClose(stream);  // stream is already closed in loop above 
    NormStopReceiver(session);
    NormStopSender(session);
    NormDestroySession(session);
    NormDestroyInstance(instance);
    
    fprintf(stderr, "normTest: Done.\n");
    return 0;
}  // end main()