//
//  Copyright © 2004, RESON Inc. All Rights Reserved.
//
//  No part of this file may be reproduced or transmitted in any form or by
//  any means, electronic or mechanical, including photocopy, recording, or
//  information storage or retrieval system, without permission in writing
//  from RESON Inc.
//
//  Filename:   
//
//  Project:    6046
//
//  Author(s):  W. Arcus
//
//  Purpose:    
//
//  Notes:      
//

#include "StdAfx.h"
#include "MessageProcessor.h"

#include "..\..\..\Utils\NetUtils\7kProtocol.h"
#include "..\..\..\Utils\NetUtils\PLCSensorData.h"

#pragma warning( disable : 4663 )
#pragma warning( disable : 4018 )
#include <map>
#pragma warning( default : 4018 )
#pragma warning( default : 4663 )

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

namespace
{

    template<typename tCountType>
    class CCounter                                                                                          // Helper to do instance couting.
    {
    private:

        tCountType  m_tCount;

    public:

        CCounter( void )
        {
            m_tCount = 0;
        }

        CCounter( const CCounter<tCountType> &rRhs )
        {
            m_tCount = rRhs.m_tCount;
        }

        CCounter<tCountType> & operator = ( const CCounter<tCountType> &rRhs )
        {
            if ( this != &rRhs )
            {
                m_tCount = rRhs.m_tCount;
            }

            return *this;
        }

        virtual ~CCounter ( void )
        {
            m_tCount = 0;
        }

        CCounter<tCountType> & operator ++ ( int )                                                          // Postincrement operator.
        {
            m_tCount++;
            return *this;
        }

        CCounter<tCountType> & operator ++ ( void )                                                         // Preincrement  operator.
        {
            ++m_tCount;
        }

        operator tCountType( void ) const                                                                   // Conversion operator to extract the counter.
        {
            return m_tCount;
        }
    };

}

//////////////////////////////////////////////////////////////////////
// CMessageProcessor class implementation.

CMessageProcessor::CMessageProcessor( void )
                  :CPacketAssembler ()
{
    Reset();

    m_bFirstFragment        = true;

    m_iBytes                = 0;

    m_ulTimeStamp           = 0UL;
    m_ulTimeOfFirstFragment = 0UL;
}

CMessageProcessor::~CMessageProcessor( void )
{
    m_bFirstFragment        = true;

    m_iBytes                = 0;

    m_ulTimeStamp           = 0UL;
    m_ulTimeOfFirstFragment = 0UL;
}

bool CMessageProcessor::ValidateAndQueueRecord( const BYTE          *pby7kNetworkRecord,
                                                const unsigned long &rulPacketBytes,
                                                const unsigned long &rulTimestamp,
                                                const int           &riSensorIndex )
{
    bool bSuccess = false;                                                                      // Assume failure for now.

    if ( rulPacketBytes == 0UL )                                                                // No data means nothing to do.
    {
        bSuccess = true;
    }
    else if ( pby7kNetworkRecord == NULL )                                                      // Check for a valid input data pointer.
    {
        TRACE( _T( "CMessageProcessor::ValidateAndQueueRecord(), pby7kNetworkRecord is NULL.\n" ) );
        bSuccess = false;
    }
    else if ( ! IsProtocolVersionCorrect( pby7kNetworkRecord, rulPacketBytes ) )                // Verify external host has sending the expected protocol version.
    {
        TRACE( _T( "CMessageProcessor::ValidateAndQueueRecord(), Incorrect protocol version.\n" ) );
        bSuccess = false;
    }
    else if ( ! m_SensorDataPool.IsInitialized() )                                              // Ensure the data pool is available.
    {
        TRACE( _T( "CMessageProcessor::ValidateAndQueueRecord(), m_SensorDataPool.IsInitialized() failed.\n" ) );
        bSuccess = false;
    }
    else if ( IsPacketACompleteRecord( pby7kNetworkRecord, rulPacketBytes ) )                   // Is it an entire record ?... e.g., from a TCP socket.
    {
        bSuccess = WriteSingleRecord( pby7kNetworkRecord, rulPacketBytes, rulTimestamp, riSensorIndex );
    }
    else if ( Add( pby7kNetworkRecord, rulPacketBytes, rulTimestamp, riSensorIndex ) )          // It must be a record fragment so queue it.
    {
        bSuccess = WriteCompleteRecords();                                                      // Check to see if a complete record is available then write to the PLC's sensor data pool.
    }
    else
    {
        TRACE( _T( "CMessageProcessor::ValidateAndQueueRecord(), Can't add packet to assember.\n" ) );
        bSuccess = false;
    }

    return bSuccess;
}

inline
bool CMessageProcessor::IsValid7kRecord(    const BYTE          *pby7kRecord,
                                            const unsigned long &rulBytes )
{
    bool bValid = false;

    try
    {
        if ( C7kProtocol::IsValid7kRecord( pby7kRecord, rulBytes ) )
        {
            // Validate the record here. Check the basic frame layout, version match and checksum if flag is set.

            DATARECORDFRAME const *pDRF = reinterpret_cast<DATARECORDFRAME *>( const_cast<BYTE *>( pby7kRecord ) );

            if ( pDRF->m_unVersion == un7kDataProtocolVersion_c )
            {
                if ( ( pDRF->m_unFlags & unBitFlagCheckSum_c ) == 0 )
                {
                    bValid = true;            // Checksum flag inndicated checksum field in record is not used so assume ok for now.
                }
                else
                {
                    const unsigned long ulComputedChecksum = ComputeChecksum( pby7kRecord, rulBytes - Checksum7kSize_m() );
                    const unsigned long ulEmbeddedChecksum = *reinterpret_cast<unsigned long *>( const_cast<BYTE *>( &pby7kRecord[ rulBytes - Checksum7kSize_m() ] ) );
                    
                    bValid = ( ulComputedChecksum == ulEmbeddedChecksum );
                }
            }
        }
    }
    catch ( ... )
    {
        bValid = false;
    }

    return bValid;
}

inline
unsigned long CMessageProcessor::ComputeChecksum(   const unsigned char *pucStream,
                                                    const unsigned long &rulStreamBytes )
{
    ULARGE_INTEGER ui64ComputedChecksum = { 0 };

    for ( unsigned long ulByte = 0UL; ulByte < rulStreamBytes; ulByte++ )
    {
        ui64ComputedChecksum.QuadPart += pucStream[ ulByte ];
    }

    return ui64ComputedChecksum.LowPart;
}

inline
bool CMessageProcessor::IsProtocolVersionCorrect(   const BYTE          *pby7kNetworkRecord,
                                                    const unsigned long &rulPacketBytes )
{
    return ( ( rulPacketBytes >= NETWORKFRAMEHEADER::Size() ) &&
             ( *reinterpret_cast<const unsigned short *>( pby7kNetworkRecord ) == un7kSocketProtocolVersion_c ) );
}

inline
bool CMessageProcessor::IsPacketACompleteRecord(    const BYTE          *pby7kNetworkRecord,
                                                    const unsigned long &rulPacketBytes )
{
    // Here we check to see whether there is a single valid and complete 7k record embedded within this
    // network packet; if so, return true else false.

    const NETWORKFRAMEHEADER *  psNetworkHeader     = reinterpret_cast<const NETWORKFRAMEHEADER *>( pby7kNetworkRecord );
    const unsigned long         ulNetworkHeaderSize = tagNETWORKFRAMEHEADER::Size();

    if ( ( psNetworkHeader != NULL                  )  &&
         ( rulPacketBytes >= ulNetworkHeaderSize    )  &&
         ( psNetworkHeader->IsValid()               )  &&
         ( psNetworkHeader->m_ulTotalPackets == 1UL )   )
    {
        return IsValid7kRecord( pby7kNetworkRecord + ulNetworkHeaderSize, rulPacketBytes - ulNetworkHeaderSize );
    }

    return false;
}

inline
bool CMessageProcessor::WriteSingleRecord(  const BYTE          *pby7kNetworkRecord,
                                            const unsigned long &rulPacketBytes,
                                            const unsigned long &rulTimestamp,
                                            const int           &riSensorIndex )
{
    const unsigned long  ulSize         = static_cast<unsigned long>( tagNETWORKFRAMEHEADER::Size() );
    const BYTE          *pby7kRecord    = pby7kNetworkRecord + ulSize;
    const unsigned long  ulRecordLength = rulPacketBytes     - ulSize;
    RECORDHEADER         sRecordHeader  = { 0 };

    sRecordHeader.m_iSensorIndex            = riSensorIndex;
    sRecordHeader.m_ulMillisecondTimestamp  = rulTimestamp;
    sRecordHeader.m_ulRecordCounter         = 0UL;
    sRecordHeader.m_ulRecordSizeInBytes     = ulRecordLength;

    return WriteRecord( sRecordHeader, pby7kRecord, ulRecordLength );
}

inline
bool CMessageProcessor::WriteRecord( const RECORDHEADER    &rsRecordHeader,
                                     const BYTE            *pby7kRecord,
                                     const unsigned long   &rulRecordLength )
{
    if ( m_SensorDataPool.IsInitialized() && IsValid7kRecord( pby7kRecord, rulRecordLength ) )
    {
        return m_SensorDataPool.Write( &rsRecordHeader, pby7kRecord, rulRecordLength );
    }

    return false;
}

inline
bool CMessageProcessor::WriteCompleteRecords( void )
{
    bool            bSuccess         = true;
    unsigned long   ulTransmissionId = 0UL;

    while ( IsRecordComplete( ulTransmissionId ) && bSuccess )
    {
        unsigned long   ulRecordBytes = 0UL;
        RECORDHEADER    sRecordHeader = { 0 };
        BYTE           *pby7kRecord   = NULL;

        __TRY
        {
            Enter();

            if ( RetrieveRecord( ulTransmissionId, sRecordHeader, pby7kRecord, ulRecordBytes ) )
            {
                if ( ! WriteRecord( sRecordHeader, pby7kRecord, ulRecordBytes ) )
                {
                    bSuccess = false;
                    TRACE( _T( "CMessageProcessor::WriteCompleteRecords(), WriteRecord() failed.\n" ) );
                }
            }

            RemoveRecord( ulTransmissionId );
        }
        __FINALLY
        {
            Leave();
        }
        __ENDFINALLY
    }

    return bSuccess;
}


///////////////////////////////////////////////////////////////////////////////
// CPacketAssembler class implementation.

CPacketAssembler::CPacketAssembler  ( const unsigned long &rulPacketSize,
                                      const unsigned long &rulMaxPackets )

                 :m_ulPacketSize    ( rulPacketSize ),
                  m_ulMaxPackets    ( rulMaxPackets ),
                  m_ulPoolSize      ( rulMaxPackets * ( rulPacketSize + tagPACKETINFO::Size() ) )
{
    m_ulWriteSlotNumber = 0UL;
    m_pbyDataPool = new BYTE[ m_ulPoolSize ];
    ASSERT( m_pbyDataPool != NULL );
    m_PacketIndexTable.clear();
    m_7kRecord.Reset();
}

CPacketAssembler::~CPacketAssembler( void )
{
    m_7kRecord.Reset();

    m_ulWriteSlotNumber = 0UL;

    if ( m_pbyDataPool != NULL )
    {
        delete [] m_pbyDataPool;
        m_pbyDataPool = NULL;
    }

    m_PacketIndexTable.clear();
}

void CPacketAssembler::Reset( void )
{
    m_ulWriteSlotNumber = 0UL;
    m_PacketIndexTable.clear();
    m_7kRecord.Reset();

    for ( unsigned long ulPacket = 0UL; ulPacket < m_ulMaxPackets; ulPacket++ )
    {
        PACKETINFO *psPacketInfo = reinterpret_cast<PACKETINFO *>( GetSlot( ulPacket ) );

        psPacketInfo->m_ulUsed                   = 0UL;
        psPacketInfo->m_ulTransmissionIdentifier = 0UL;
        psPacketInfo->m_ulSequenceNumber         = 0UL;
        psPacketInfo->m_ulSlotNumber             = 0UL;
        psPacketInfo->m_ulTimestamp              = 0UL;
        psPacketInfo->m_iSensorIndex             = 0;
    }
}

void CPacketAssembler::RemoveRecord( const unsigned long &rulTransmissionId )
{
    for ( unsigned long ulPacket = 0UL; ulPacket < m_ulMaxPackets; ulPacket++ )
    {
        PACKETINFO *psPacketInfo = reinterpret_cast<PACKETINFO *>( GetSlot( ulPacket ) );

        if ( ( psPacketInfo->m_ulUsed ) && ( psPacketInfo->m_ulTransmissionIdentifier == rulTransmissionId ) )
        {
            psPacketInfo->m_ulUsed = 0UL;
        }
    }

    RebuildPacketIndexInfo();
}

bool CPacketAssembler::Add( const BYTE          *pby7kNetworkRecord, 
                            const unsigned long &rulPacketBytes,
                            const unsigned long &rulTimestamp,
                            const int           &riSensorIndex )
{
    bool bSuccess = false;

    __TRY
    {
        Enter();

        BYTE *pbyWritePoint = GetWriteSlot();
        ASSERT( pbyWritePoint != NULL );

        if ( pbyWritePoint != NULL )
        {
            NETWORKFRAMEHEADER *psNetworkHeader      = reinterpret_cast<NETWORKFRAMEHEADER *>( const_cast<BYTE *>( pby7kNetworkRecord ) );
            PACKETINFO         *psPacketInfo         = reinterpret_cast<PACKETINFO *>( pbyWritePoint );

            psPacketInfo->m_ulUsed                   = 1UL;
            psPacketInfo->m_ulSequenceNumber         = psNetworkHeader->m_ulSequenceNumber;
            psPacketInfo->m_ulTransmissionIdentifier = psNetworkHeader->m_unTransmissionIdentifier;
            psPacketInfo->m_ulSlotNumber             = m_ulWriteSlotNumber;
            psPacketInfo->m_ulTimestamp              = rulTimestamp;
            psPacketInfo->m_iSensorIndex             = riSensorIndex;

            memcpy( pbyWritePoint + tagPACKETINFO::Size(), pby7kNetworkRecord, rulPacketBytes );
        }

        IncrementWritePoint();

        RebuildPacketIndexInfo();

        bSuccess = true;
    }
    __FINALLY
    {
        Leave();
    }
    __ENDFINALLY

    return bSuccess;
}

bool CPacketAssembler::IsRecordComplete( unsigned long &rulTransmissionId )
{
    // Given the list of queued packets, determine whether there are any complete sequence(s). If so,
    // return the transmission id of the first detected sequence (with lowest transmission id) so that 
    // the sequence may be extracted as a correctly formatted 7k record.

    bool bComplete = false;

    __TRY
    {
        Enter();

        if ( ! m_PacketIndexTable.empty() )
        {
            // Define a map<> container to be used to do our instance counting of transmission id's. Here, the transmission id
            // is the key and the count the associated data item.

            typedef unsigned long                                        TransmissionId_t;
            typedef std::map<unsigned long, CCounter<TransmissionId_t> > TransmissionIdTable_t;
            TransmissionIdTable_t                                        TransmissionIdTable;

            for ( PacketListIterator_t pItem = m_PacketIndexTable.begin(); pItem != m_PacketIndexTable.end(); pItem++ )
            {
                TransmissionIdTable[ pItem->m_ulTransmissionIdentifier ]++;
            }

            PacketList_t UniqueList( m_PacketIndexTable );

            UniqueList.unique();

            if ( ! UniqueList.empty() )
            {
                UniqueList.sort();

                for ( PacketListIterator_t pUnique = UniqueList.begin(); pUnique != UniqueList.end(); pUnique++ )
                {
                    BYTE                *pbyPacket       = GetSlot( pUnique->m_ulSlotNumber );
                    NETWORKFRAMEHEADER  *psNetworkHeader = reinterpret_cast<NETWORKFRAMEHEADER *>( const_cast<BYTE *>( pbyPacket + tagPACKETINFO::Size() ) );

                    if ( TransmissionIdTable[ pUnique->m_ulTransmissionIdentifier ] == psNetworkHeader->m_ulTotalPackets )
                    {
                        bComplete = true;
                        rulTransmissionId = pUnique->m_ulTransmissionIdentifier;
                        break;
                    }
                }
            }
        }
    }
    __FINALLY
    {
        Leave();
    }
    __ENDFINALLY

    return bComplete;
}

bool CPacketAssembler::RetrieveRecord(  const unsigned long &rulTransmissionId,
                                        RECORDHEADER        &rsRecordHeader,
                                        BYTE const *        &rpbyRecord,
                                        unsigned long       &rulRecordBytes )
{
    // Detect a complete record, extract all relevant component packets, format the completed 7k record and
    // return a pointer to it along with its relevant size information.

    // Clear the outputs initlally.

    rpbyRecord      = NULL;
    rulRecordBytes  = 0UL;

    memset( &rsRecordHeader, 0x00, sizeof( RECORDHEADER ) );

    m_7kRecord.Reset();

    bool          bFirst     = true;
    unsigned long ulExpected = 0UL;
    unsigned long ulCounter  = 0UL;

    RECORDHEADER  sHeader    = { 0 };

    for ( PacketListIterator_t pItem = m_PacketIndexTable.begin(); pItem != m_PacketIndexTable.end(); pItem++ )
    {
        if ( ( pItem->m_ulUsed ) && ( rulTransmissionId == pItem->m_ulTransmissionIdentifier ) )
        {
            BYTE               *pbyPacket       = GetSlot( pItem->m_ulSlotNumber );
            NETWORKFRAMEHEADER *psNetworkHeader = reinterpret_cast<NETWORKFRAMEHEADER *>( const_cast<BYTE *>( pbyPacket + tagPACKETINFO::Size() ) );
            BYTE               *pbyRecordData   = pbyPacket + tagPACKETINFO::Size() + sizeof( NETWORKFRAMEHEADER );

            if ( bFirst )
            {
                bFirst     = false;
                ulExpected = psNetworkHeader->m_ulTotalPackets;

                sHeader.m_ulRecordCounter        = 0UL;
                sHeader.m_ulMillisecondTimestamp = pItem->m_ulTimestamp;
                sHeader.m_iSensorIndex           = pItem->m_iSensorIndex;
                sHeader.m_ulRecordSizeInBytes    = 0UL;
            }

            if ( ulCounter++ < ulExpected )
            {
                m_7kRecord.Add( pbyRecordData, psNetworkHeader->m_ulPacketSize - sizeof( NETWORKFRAMEHEADER ) );
            }
            else
            {
                ASSERT( false );
            }
        }
    }

    if ( m_7kRecord.Size() > 0UL )
    {
        rsRecordHeader = sHeader;
        rsRecordHeader.m_ulRecordSizeInBytes = m_7kRecord.Size() * sizeof( BYTE );

        rpbyRecord      = m_7kRecord.GetAt( 0 );
        rulRecordBytes  = rsRecordHeader.m_ulRecordSizeInBytes;

        return true;
    }

    return false;
}

///////////////////////////////////////////////////////////////////////////////
// Private internal helpers.

inline
BYTE * CPacketAssembler::GetWriteSlot( void )
{
    return GetSlot( m_ulWriteSlotNumber );
}

inline
BYTE * CPacketAssembler::GetSlot( const unsigned long &rulSlotNumber )
{
    ASSERT( rulSlotNumber < m_ulMaxPackets );
    return ( &m_pbyDataPool[ rulSlotNumber * ( m_ulPacketSize + tagPACKETINFO::Size()) ] );
}

inline
void CPacketAssembler::IncrementWritePoint( void )
{
    m_ulWriteSlotNumber = ( m_ulWriteSlotNumber + 1UL ) % m_ulMaxPackets;
}

inline
void CPacketAssembler::RebuildPacketIndexInfo( void )
{
    if ( ! m_PacketIndexTable.empty() )
    {
        m_PacketIndexTable.clear();
    }

    for ( unsigned long ulPacket = 0UL; ulPacket < m_ulMaxPackets; ulPacket++ )
    {
        PPACKETINFO psPacketInfo = reinterpret_cast<PPACKETINFO>( GetSlot( ulPacket ) );

        if ( psPacketInfo->m_ulUsed )
        {
            m_PacketIndexTable.push_back( *psPacketInfo );
        }
    }

    if ( ! m_PacketIndexTable.empty() )
    {
        m_PacketIndexTable.sort();
    }
}