protocol.c

Open DMT Client C Source code

    ProtocolVars_t *pv = _protoGetVars(LOGSRC,protoNdx);
    return pv->xFtns->xportIsOpen()? pv->speakFreely : utFalse;
}

/* return transport 'open' state */
utBool protocolIsOpen(int protoNdx)
{
    ProtocolVars_t *pv = _protoGetVars(LOGSRC,protoNdx);
    return pv->xFtns->xportIsOpen();
}

/* open connection to server */
static utBool _protocolOpen(ProtocolVars_t *pv, TransportType_t type)
{
    utBool didOpen = pv->xFtns->xportOpen(type);
        
    if (didOpen) {
        // opened, reset session

        /* clear volatile queue on initial connection */
        pqueResetQueue(&(pv->volatileQueue));

        _protocolEnableOverwrite(pv, utFalse); // disable overwrites while connected
        if (pv->isPrimary) { // data stats
            // data stats only recorded for 'primary' transport
            pv->totalReadBytes      = propGetUInt32(PROP_COMM_BYTES_READ   , 0L); // primary only
            pv->totalWriteBytes     = propGetUInt32(PROP_COMM_BYTES_WRITTEN, 0L); // primary only
        } else {
            // data stats not recorded for 'secondary' transport
            pv->totalReadBytes      = 0L;
            pv->totalWriteBytes     = 0L;
        }
        pv->sessionReadBytes        = 0L;
        pv->sessionWriteBytes       = 0L;

#if defined(TRANSPORT_MEDIA_SERIAL)
        // send account/device for serial transport
        pv->sendIdentification      = SEND_ID_ACCOUNT;
#else
        // try unique-id first for everything else
        pv->sendIdentification      = SEND_ID_UNIQUE;
#endif

        pv->severeErrorCount        = 0;
        pv->checkSumErrorCount      = 0;
        pv->invalidAcctErrorCount   = 0;

        if (pv->isSerial) {
            motionResetMovingMessageTimer();
            // TODO: generate a connection message?
        }

    }
    return didOpen;
}

/* close connection to server */
static utBool _protocolClose(ProtocolVars_t *pv, TransportType_t xportType, utBool sendUDP)
{

    /* close transport */
    // If the connection is via Simplex, the data will be sent now.
    utBool doSend = (xportType == TRANSPORT_SIMPLEX)? sendUDP : utFalse;
    utBool didClose = pv->xFtns->xportClose(doSend);
        
    if (didClose && pv->isPrimary) { // data stats
        // save read/write byte counts if 'close' was successful (primary only)
        propSetUInt32(PROP_COMM_BYTES_READ   , pv->totalReadBytes );
        propSetUInt32(PROP_COMM_BYTES_WRITTEN, pv->totalWriteBytes);
    }
    
    /* reset volitile queue */
    pqueResetQueue(&(pv->volatileQueue));
    
    /* re-enable event queue overwrites while not connected */
    _protocolEnableOverwrite(pv,EVENT_QUEUE_OVERWRITE); // enabled only while not connected
    
    /* check for severe errors */
    if (xportType == TRANSPORT_DUPLEX) {
        if (pv->severeErrorCount) {
            // this helps prevent runnaway clients from abusing the server
            pv->totalSevereErrorCount += pv->severeErrorCount;
            logWARNING(LOGSRC,"Severe errors encountered --> %d", pv->totalSevereErrorCount);
            if (pv->isSerial) {
                // severe errors are ignored for bluetooth connections
            } else
            if (pv->isPrimary) { // severe errors
                if (pv->totalSevereErrorCount >= MAX_SEVERE_ERRORS) {
                    // Slow down minimum connection interval
                    UInt32 minXmitRate = propGetUInt32(PROP_COMM_MIN_XMIT_RATE, 0L); // primary only
                    if (minXmitRate < HOUR_SECONDS(12)) {
                        //if (minXmitRate < 1) { minXmitRate = 1; }
                        if (minXmitRate < MIN_XMIT_RATE) { minXmitRate = MIN_XMIT_RATE; }
                        propAddUInt32(PROP_COMM_MIN_XMIT_RATE, minXmitRate); // doubles rate (primary only)
                    }
                    UInt32 minXmitDelay = propGetUInt32(PROP_COMM_MIN_XMIT_DELAY, 0L); // primary only
                    if (minXmitDelay < HOUR_SECONDS(12)) {
                        //if (minXmitDelay < 1) { minXmitDelay = 1; }
                        if (minXmitDelay < MIN_XMIT_DELAY) { minXmitDelay = MIN_XMIT_DELAY; }
                        propAddUInt32(PROP_COMM_MIN_XMIT_DELAY, minXmitDelay); // doubles delay (primary only)
                    }
                }
                if (pv->totalSevereErrorCount >= EXCESSIVE_SEVERE_ERRORS) {
                    // Turn off periodic messaging
                    logERROR(LOGSRC,"Excessive severe errors! Disabling periodic events!");
                    propSetUInt32(PROP_MOTION_START, 0L); // primary only
                    propSetUInt32(PROP_MOTION_IN_MOTION, 0L); // primary only
                    propSetUInt32(PROP_MOTION_DORMANT_INTRVL, 0L); // primary only
                }
            } else {
                // severe errors are ignored for secondary transports
            }
        } else
        if (pv->totalSevereErrorCount > 0) {
            // a session without any severe errors will reduce this count
            pv->totalSevereErrorCount--;
        }
    }
    
    return didClose;
}

/* write data to server */
static int _protocolWrite(ProtocolVars_t *pv, const UInt8 *buf, int bufLen, utBool calcChksum)
{
    // guaranteed to contain only a single packet
    
    /* print */
    if (bufLen > 0) {
        if (*buf == PACKET_ASCII_ENCODING_CHAR) {
            logINFO(LOGSRC,"Tx%d]%.*s", pv->protoNdx, bufLen - 1, buf);
        } else {
            UInt8 hex[PACKET_MAX_ENCODED_LENGTH];
            logINFO(LOGSRC,"Tx%d]0x%s", pv->protoNdx, strEncodeHex((char*)hex, sizeof(hex), buf, bufLen)); 
        }
    }
    
    /* write */
    int len = pv->xFtns->xportWritePacket(buf, bufLen);
        
    if (len >= 0) {
        if (calcChksum) { 
            cksumCalcFletcher(buf, bufLen); 
        }
        pv->totalWriteBytes   += len;
        pv->sessionWriteBytes += len;
    } else {
        // error 
    }
    return len;
    
}

/* write packet to server */
static int _protocolWritePacket(ProtocolVars_t *pv, Packet_t *pkt)
{
    UInt8 buf[PACKET_MAX_ENCODED_LENGTH];
    Buffer_t bb, *dest = binBuffer(&bb, buf, sizeof(buf), BUFFER_DESTINATION);
    binResetBuffer(dest);
    pktEncodePacket(dest, pkt, pv->sessionFirstEncoding); // we ignore any internal errors
    int rtnWriteLen = _protocolWrite(pv, BUFFER_PTR(dest), BUFFER_DATA_LENGTH(dest), utTrue);
    pv->sessionFirstEncoding = pv->sessionEncoding;
    return rtnWriteLen;
}

// ----------------------------------------------------------------------------

/* queue specified packet for transmission */
static utBool _protocolQueuePacket(ProtocolVars_t *pv, Packet_t *pkt)
{
    // Notes:
    // - By default packets are created with 'PRIORITY_NORMAL', and will be placed
    // into the volatile queue which is reset at the start of each session.  Thus,
    // the volatile queue should only be used within a server connected session. 
    // - If a packet is important enough to be retained until it is transmitted to
    // the server, then its priority should be set to 'PRIORITY_HIGH'. It will then
    // be added to the pending queue and will be retained until it is successfully 
    // transmitted to the server.  This is also true for important packets that are
    // queued while NOT within a server connected session.
    if (!pkt) {
        return utFalse;
    } else
    if (pkt->priority >= PRIORITY_HIGH) {
        // Place high priority packets in the pending queue
        // This queue persists across sessions
        return pqueAddPacket(&(pv->pendingQueue), pkt);
    } else {
        // place normal/low priority packets in the volatile queue
        // This queue is cleared before/after each session
        return pqueAddPacket(&(pv->volatileQueue), pkt);
    }
}
utBool protocolQueuePacket(int protoNdx, Packet_t *pkt)
{
    ProtocolVars_t *pv = _protoGetVars(LOGSRC,protoNdx);
    return _protocolQueuePacket(pv, pkt);
}

/* queue specified error for transmission */
static utBool _protocolQueueError(ProtocolVars_t *pv, const char *fmt, ...)
{
    Packet_t pkt;
    va_list ap;
    va_start(ap, fmt);
    pktVInit(&pkt, PKT_CLIENT_ERROR, fmt, ap); // default PRIORITY_NORMAL
    va_end(ap);
    return _protocolQueuePacket(pv,&pkt);
}
utBool protocolQueueError(int protoNdx, const char *fmt, ...)
{
    ProtocolVars_t *pv = _protoGetVars(LOGSRC,protoNdx);
    Packet_t pkt;
    va_list ap;
    va_start(ap, fmt);
    pktVInit(&pkt, PKT_CLIENT_ERROR, fmt, ap); // default PRIORITY_NORMAL
    va_end(ap);
    return _protocolQueuePacket(pv,&pkt);
}

/* queue specified diagnostic for transmission */
utBool protocolQueueDiagnostic(int protoNdx, const char *fmt, ...)
{
    ProtocolVars_t *pv = _protoGetVars(LOGSRC,protoNdx);
    Packet_t pkt;
    va_list ap;
    va_start(ap, fmt);
    pktVInit(&pkt, PKT_CLIENT_DIAGNOSTIC, fmt, ap); // default PRIORITY_NORMAL
    va_end(ap);
    return _protocolQueuePacket(pv,&pkt);
}

// ----------------------------------------------------------------------------

/* parse packet received from server */
static Packet_t *_protocolParseServerPacket(ProtocolVars_t *pv, Packet_t *pkt, const UInt8 *pktBuf)
{
    
    /* clear packet */
    memset(pkt, 0, sizeof(Packet_t));
    
    /* parse header/data */
    if (*pktBuf == PACKET_ASCII_ENCODING_CHAR) {
        // The packet is assumed to be null-terminated (ie. '\r' was replaced with '0')
        
        /* print packet */
        logDEBUG(LOGSRC,"Rx%d]%s\n", pv->protoNdx, pktBuf); 
        
        /* parse */
        int pktBufLen;
        if (!cksumIsValidCharXOR((char*)pktBuf, &pktBufLen)) {
            // checksum failed: ERROR_PACKET_CHECKSUM
            _protocolQueueError(pv,"%2x%2x", (UInt32)ERROR_PACKET_CHECKSUM, (UInt32)0);
            return (Packet_t*)0;
        } else
        if (pktBufLen < 5) {
            // invalid length: ERROR_PACKET_LENGTH
            _protocolQueueError(pv,"%2x%2x", (UInt32)ERROR_PACKET_LENGTH, (UInt32)0);
            return (Packet_t*)0;
        } else {
            UInt8 buf[2];
            int hlen = strParseHex((char*)(pktBuf + 1), 4, buf, sizeof(buf));
            if (hlen != 2) {
                // header was not parsable: ERROR_PACKET_HEADER
                _protocolQueueError(pv,"%2x%2x", (UInt32)ERROR_PACKET_HEADER, (UInt32)0);
                return (Packet_t*)0;
            } else {
                pkt->hdrType = CLIENT_HEADER_TYPE(buf[0],buf[1]);
                if (pktBufLen > 6) { // $E0FF:XXXX...
                    // encoded character, plus data
                    if (pktBuf[5] == ENCODING_BASE64_CHAR) {
                        int len = (UInt8)base64Decode((char*)(pktBuf + 6), pktBufLen - 6, pkt->data, sizeof(pkt->data));
                        pkt->dataLen = (len >= 0)? (UInt8)len : 0;
                    } else
                    if (pktBuf[5] == ENCODING_HEX_CHAR) {
                        int len = (UInt8)strParseHex((char*)(pktBuf + 6), pktBufLen - 6, pkt->data, sizeof(pkt->data));
                        pkt->dataLen = (len >= 0)? (UInt8)len : 0;
                    } else
                    if (pktBuf[5] == ENCODING_CSV_CHAR) {
                        // unsupported encoding: ERROR_PACKET_ENCODING
                        // parsing CSV encoded packets is not supported in this reference implementation
                        logWARNING(LOGSRC,"CSV parsing is not supported.\n");
                        _protocolQueueError(pv,"%2x%2x", (UInt32)ERROR_PACKET_ENCODING, (UInt32)pkt->hdrType);
                        return (Packet_t*)0;
                    } else {
                        // unrecognized encoding: ERROR_PACKET_ENCODING
                        logWARNING(LOGSRC,"Unrecognized encoding: %d\n", pktBuf[5]);
                        _protocolQueueError(pv,"%2x%2x", (UInt32)ERROR_PACKET_ENCODING, (UInt32)pkt->hdrType);
                        return (Packet_t*)0;
                    }
                }
            }
        }
        
    } else
    if (*pktBuf == PACKET_HEADER_BASIC) {
        // The packet header is assumed to be a valid length.
        
        /* print packet */
        UInt8 hex[PACKET_MAX_ENCODED_LENGTH];
        UInt16 len = (UInt16)pktBuf[2] + 3;
        logINFO(LOGSRC,"Rx%d]0x%s\n", pv->protoNdx, strEncodeHex((char*)hex, sizeof(hex), pktBuf, len)); 
        
        /* parse into packet */
        pkt->hdrType = CLIENT_HEADER_TYPE(pktBuf[0],pktBuf[1]);