protocol.c

Open DMT Client C Source code

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

static protDataCallbackFtn_t  ftnErrorHandler = 0;

void protocolSetErrorHandler(protDataCallbackFtn_t ftn)
{
    ftnErrorHandler = ftn;
}

static void protocolHandleError(UInt16 errKey, const UInt8 *errData, UInt16 errDataLen)
{
    if (ftnErrorHandler) {
        (*ftnErrorHandler)(errKey, errData, errDataLen);
    }
}

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

static protClientInitCallbackFtn_t  ftnClientInitHandler = 0;

void protocolSetClientInitHandler(protClientInitCallbackFtn_t ftn)
{
    ftnClientInitHandler = ftn;
}

static void protocolHandleClientInit()
{
    
    /* external client init */
    if (ftnClientInitHandler) {
        (*ftnClientInitHandler)();
    }
    
    /* send pending packets */
    if (protHasPendingPackets()) {
        logINFO(LOGSRC,"Sending pending packets during client initialization ...");
        Packet_t pkt;
        while (protGetPendingPacket(&pkt)) {
            serverWritePacket(&pkt);
        }
    }
    
}

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

/* protocol loop entry point */
void protocolLoop(
    const char *portName, utBool portLog,
    utBool cliKeepAlive, 
    utBool cliSpeaksFirst)
{
    TimerSec_t revokeSpeakFreelyTimer = 0L;
    UInt32 haveEvents = 0L;
    Packet_t packet, *pkt = &packet;
    utBool clientNeedsInit = utFalse;
    
    /* global states */
    clientKeepAlive = cliKeepAlive; // utFalse
    clientSpeaksFirst = cliSpeaksFirst; // utTrue;
    
    /* pending packet mutex init */
    threadMutexInit(&pendingMutex);
        
    /* init */
    serverInitialize();
    lastEventSequence = 0L;
    lastEventSeqLen   = 0;
    lastEventTimer    = 0L;

    /* loop forever */
    for (;;) {
        
        /* open port */
        if (!serverIsOpen()) {
            logINFO(LOGSRC,"Openning server port: %s", portName);
            if (!serverOpen(portName, portLog)) {
                // Unable to open/accept client (message already displayed)
                /* sleep and try again */
                threadSleepMS(3000L);
                continue;
            }
            logINFO(LOGSRC,"Server port opened: %s", portName);
            *clientAccountID = 0;
            *clientDeviceID = 0;
            revokeSpeakFreelyTimer = utcGetTimer();
            if (!clientSpeaksFirst) {
                // If the client does not speak first, then we must nudge the client
                // to let him know to speak now.
                serverWritePacketFmt(PKT_SERVER_EOB_DONE,"%1u",(UInt32)0L);
                isSpeakFreelyMode = utFalse;
            }
            // Send any client initialization at the start of each new connection
            clientNeedsInit = utTrue; // clientSpeaksFirst;
            serverNeedsMoreInfo = utFalse;
        }
        
        /* send pending packets, etc. */
        if (cliKeepAlive) {
#if defined(INCLUDE_UPLOAD)
            if (pendingFileUpload) {
                utBool rtn = utFalse;
                if (pendingFileClient && *pendingFileClient) {
                    logINFO(LOGSRC,"Uploading file: %s", pendingFileUpload);
                    rtn = uploadSendFile(pendingFileUpload, pendingFileClient);
                } else {
                    logINFO(LOGSRC,"Uploading encoded file: %s", pendingFileUpload);
                    rtn = uploadSendEncodedFile(pendingFileUpload);
                }
                if (!rtn) {
                    logERROR(LOGSRC,"Upload failed!");
                }
                pendingFileUpload = (char*)0;
                pendingFileClient = (char*)0;
            } else
#endif
            if (protHasPendingPackets()) {
                logINFO(LOGSRC,"Sending pending packet during client keep-alive ...");
                Packet_t pkt;
                while (protGetPendingPacket(&pkt)) {
                    serverWritePacket(&pkt);
                }
            } else 
            if (stopSpeakFreely) {
                if (isSpeakFreelyMode) {
                    serverWritePacketFmt(PKT_SERVER_EOB_DONE,"%1u",(UInt32)0L);
                    isSpeakFreelyMode = utFalse;
                }
                stopSpeakFreely = utFalse;
            } else
            if (startSpeakFreely) {
                if (!isSpeakFreelyMode) {
                    if (speakFreelyMaxEvents >= 0) {
                        serverWritePacketFmt(PKT_SERVER_EOB_SPEAK_FREELY,"%1u",(UInt32)speakFreelyMaxEvents);
                    } else {
                        serverWritePacketFmt(PKT_SERVER_EOB_SPEAK_FREELY,"");
                    }
                    isSpeakFreelyMode = utTrue;
                }
                startSpeakFreely = utFalse;
            }
        }

        /* read client packet */
        int err = serverReadPacket(pkt);
        //logINFO(LOGSRC,"Client Packet HEADER=%04X", pkt->hdrType);
        if (err == SRVERR_TRANSPORT_ERROR) {
            logERROR(LOGSRC,"Read error (EOF?)");
            serverClose();
            continue;
        } else
        if (err == SRVERR_TIMEOUT) {
            // read timeout
            if (!clientKeepAlive) {
                serverWritePacketFmt(PKT_SERVER_EOT,"");
                logINFO(LOGSRC,"End-Of-Transmission\n\n");
                serverClose();
            } else
            if (clientNeedsInit) {
                // we haven't heard from the client, thus we haven't initialized it either
                // nudge the client again to get him to speak
                logINFO(LOGSRC,"Init Client");
                serverWritePacketFmt(PKT_SERVER_EOB_DONE,"%1u",(UInt32)0L);
                isSpeakFreelyMode = utFalse;
            } else
            if ((lastEventTimer != 0L) && utcIsTimerExpired(lastEventTimer,3L)) {
                // Acknowledge the events that we've received
                // [Note: Only needed if client never releases 'speakFreely' rights]
                logWARNING(LOGSRC,"Read timeout, ackowledging received events");
                Packet_t pkt;
                if (lastEventSeqLen > 0) {
                    pktInit(&pkt, PKT_SERVER_ACK, "%*x", lastEventSeqLen, lastEventSequence);
                } else {
                    pktInit(&pkt, PKT_SERVER_ACK, "");
                }
                serverWritePacket(&pkt);
                haveEvents = 0;
                lastEventTimer = 0L;
            } else
            if (utcIsTimerExpired(revokeSpeakFreelyTimer,REVOKE_SPEAK_FREELY_INTERVAL)) {
                // read timeout, revoke speak-freely (ping client)
                //logDEBUG(LOGSRC,"Client Timeout");
                serverWritePacketFmt(PKT_SERVER_EOB_DONE,"%1u",(UInt32)0L);
                isSpeakFreelyMode = utFalse;
                revokeSpeakFreelyTimer = utcGetTimer();
            }
            continue;
        } else
        if (err != SRVERR_OK) {
            // SRVERR_CHECKSUM_FAILED
            // SRVERR_PARSE_ERROR
            // SRVERR_PACKET_LENGTH
            logERROR(LOGSRC,"Checksum error");
            // The remainder of the session is suspect, flush the rest
            serverWritePacketFmt(PKT_SERVER_EOT,"");
            logINFO(LOGSRC,"End-Of-Transmission\n\n");
            serverClose();
            continue;
        }
        
        /* reset "revoke speak-freely" timer */
        // reset timer when we hear from the client
        revokeSpeakFreelyTimer = utcGetTimer();

        /* print received packet */
        pktPrintPacket(pkt, "[RX]", ENCODING_CSV);
        
        /* handle event packet */
        ClientPacketType_t pht = pkt->hdrType;
        if (((pht >= PKT_CLIENT_FIXED_FMT_STD  ) && (pht <= PKT_CLIENT_FIXED_FORMAT_F )) ||
            ((pht >= PKT_CLIENT_DMTSP_FORMAT_0 ) && (pht <= PKT_CLIENT_DMTSP_FORMAT_F )) ||
            ((pht >= PKT_CLIENT_CUSTOM_FORMAT_0) && (pht <= PKT_CLIENT_CUSTOM_FORMAT_F))   ) {
            Event_t ev;
            if (evParseEventPacket(pkt, &ev)) {
                if ((lastEventSequence > 0L) && ((lastEventSequence + 1L) != ev.sequence)) {
                    logERROR(LOGSRC,"********************************************************");
                    logERROR(LOGSRC,"Possible Event Data Loss");
                    logERROR(LOGSRC,"Expected sequence: 0x%04X", (lastEventSequence + 1L));
                    logERROR(LOGSRC,"Found sequence...: 0x%04X", ev.sequence);
                    logERROR(LOGSRC,"********************************************************");
                }
                lastEventSequence = ev.sequence;
                lastEventSeqLen   = ev.seqLen;
                protocolHandleEvent(pkt, &ev);
            }
            haveEvents++;
            lastEventTimer = utcGetTimer();
            continue;
        } 
        /* handle other */
        switch ((UInt16)pkt->hdrType) {
            
            case PKT_CLIENT_UNIQUE_ID:
                // ignore
                break;

            case PKT_CLIENT_ACCOUNT_ID:
                memset(clientAccountID, 0, sizeof(clientAccountID));
                binScanf(pkt->data, pkt->dataLen, "%*s", MAX_ID_SIZE, clientAccountID);
                strTrimTrailing(clientAccountID);
                // protocolHandleAccountID(clientAccountID);
                break;

            case PKT_CLIENT_DEVICE_ID:
                memset(clientDeviceID, 0, sizeof(clientDeviceID));
                binScanf(pkt->data, pkt->dataLen, "%*s", MAX_ID_SIZE, clientDeviceID);
                strTrimTrailing(clientDeviceID);
                logINFO(LOGSRC,"Client Account/Device: %s/%s\n", clientAccountID, clientDeviceID);
                // protocolHandleDeviceID(clientDeviceID);
                break;
                
            case PKT_CLIENT_PROPERTY_VALUE:
                if (pkt->dataLen >= 2) {
                    UInt32 propKey = 0L;
                    binScanf(pkt->data, pkt->dataLen, "%2x", &propKey);
                    protocolHandleProperty((UInt16)propKey, pkt->data + 2, (UInt16)(pkt->dataLen - 2));
                } else {
                    // invalid property packet (just ignore)
                }
                break;
                
            case PKT_CLIENT_DIAGNOSTIC:
                if (pkt->dataLen >= 2) {
                    UInt32 diagKey = 0L;
                    binScanf(pkt->data, pkt->dataLen, "%2x", &diagKey);
                    protocolHandleDiag((UInt16)diagKey, pkt->data + 2, (UInt16)(pkt->dataLen - 2));
                } else {
                    // invalid diagnostic packet (just ignore)
                }
                break;
                
            case PKT_CLIENT_ERROR:
                if (pkt->dataLen >= 2) {
                    UInt32 errKey = 0L;
                    binScanf(pkt->data, pkt->dataLen, "%2x", &errKey);
                    protocolHandleError((UInt16)errKey, pkt->data + 2, (UInt16)(pkt->dataLen - 2));
                } else {
                    // invalid error packet (just ignore)
                }
                break;
                
            case PKT_CLIENT_EOB_DONE:
            case PKT_CLIENT_EOB_MORE:
                // NOTE: any client supplied Fletcher checksum is ignored here.
                serverReadFlush(); // flush any remaining byte in the client input queue
                if (haveEvents > 0) {
                    // Acknowledge the events that we've received
                    Packet_t pkt;
                    if (lastEventSeqLen > 0) {
                        pktInit(&pkt, PKT_SERVER_ACK, "%*x", lastEventSeqLen, lastEventSequence);
                    } else {
                        pktInit(&pkt, PKT_SERVER_ACK, "");
                    }
                    serverWritePacket(&pkt);
                    haveEvents = 0;
                    lastEventTimer = 0L;
                }
                if (clientNeedsInit) {
                    // send any desired client initialization
                    clientNeedsInit = utFalse;
                    protocolHandleClientInit();
                }
                if (clientKeepAlive) {
                    if (serverNeedsMoreInfo) {
                        serverWritePacketFmt(PKT_SERVER_EOB_DONE,"%1x",(UInt32)0L);
                        isSpeakFreelyMode = utFalse;
                        serverNeedsMoreInfo = utFalse;
                    } else
                    if (isSpeakFreelyMode) {
                        // tell client to speak when he wishes
                        if (speakFreelyMaxEvents >= 0) {
                            serverWritePacketFmt(PKT_SERVER_EOB_SPEAK_FREELY,"%1u",(UInt32)speakFreelyMaxEvents);
                        } else {
                            serverWritePacketFmt(PKT_SERVER_EOB_SPEAK_FREELY,"");
                        }
                        isSpeakFreelyMode = utTrue;
                    }
                } else
                if (serverNeedsMoreInfo) {
                    // we need more info from the client
                    serverWritePacketFmt(PKT_SERVER_EOB_DONE,"%1x",(UInt32)0L);
                    isSpeakFreelyMode = utFalse;
                    serverNeedsMoreInfo = utFalse;
                } else
                if ((UInt16)pkt->hdrType == PKT_CLIENT_EOB_DONE) {
                    // client is finished, close socket
                    serverWritePacketFmt(PKT_SERVER_EOT,"");
                    logINFO(LOGSRC,"End-Of-Transmission\n\n");
                    serverClose();
                } else {
                    // client isn't done yet, tell client to continue
                    serverWritePacketFmt(PKT_SERVER_EOB_DONE,"");
                    isSpeakFreelyMode = utFalse;
                }
                break;
                
            default:
                serverWriteError("%2x%2x", (UInt32)NAK_PACKET_TYPE, (UInt32)pkt->hdrType);
                break;
                
        }
        
    } // for (;;) 
    
    /* we only get here if there is an error */
    
}