/****************************************************************************

  module      : OMS_Stream.cpp

  -------------------------------------------------------------------------

  responsible : ThomasA

  special area: OMS 
  description : OMS streams

  last changed: 2000-07-11  13:54
  see also    : example.html ...
  first created:2000-05-26  19:32

  -------------------------------------------------------------------------





    ========== licence begin  GPL
    Copyright (c) 2000-2005 SAP AG

    This program is free software; you can redistribute it and/or
    modify it under the terms of the GNU General Public License
    as published by the Free Software Foundation; either version 2
    of the License, or (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
    ========== licence end




*****************************************************************************/

#include "Oms/OMS_Stream.hpp"
#include "Oms/OMS_Defines.h"
#include "Oms/OMS_Types.hpp"
#include "Oms/OMS_Globals.hpp"
#include "Oms/OMS_Handle.hpp"
#include "Oms/OMS_Session.hpp"
#include "Oms/OMS_ObjectContainer.hpp"
#include "SAPDBCommon/SAPDB_MemCopyMove.hpp"
#include "hsp20.h"
#include <new.h>

class OMS_StreamBufferChunk
{
public :
    int Capacity()
    {
        return m_capacity;
    }
    static OMS_StreamBufferChunk* Create(OMS_Session *pSession, int rowSize)
    {
        int bufferSize = (rowSize > DEFAULT_BUFFER_SIZE) ? rowSize : DEFAULT_BUFFER_SIZE;
        return new(pSession->allocate(sizeof(OMS_StreamBufferChunk) - DEFAULT_BUFFER_SIZE + bufferSize)) 
            OMS_StreamBufferChunk(bufferSize / rowSize);
    }
    void DecrementCount()
    {
        --m_cnt;
    }
    SAPDB_Byte* FirstPos()
    {
        return &m_buffer[0];
    }
    int GetCount() const
    {
        return m_cnt;
    }
    OMS_StreamBufferChunk* GetNext() const
    {
        return m_next;
    }
    OMS_StreamBufferChunk** GetNextAddr()
    {
        return &m_next;
    }
    void SetCount(int count)
    {
        m_cnt = count;
    }
private :
    enum { DEFAULT_BUFFER_SIZE = 8000 };
    OMS_StreamBufferChunk(int capacity)
        : m_next(NULL)
        , m_cnt(0)
        , m_capacity(capacity)
    {
    }
    OMS_StreamBufferChunk* m_next;
    SAPDB_Int4             m_cnt;
    SAPDB_Int4             m_capacity;
    SAPDB_Byte             m_buffer[DEFAULT_BUFFER_SIZE]; // may be greater in reality
};

/*----------------------------------------------------------------------*/

class OMS_StreamBody
{
public:
    OMS_StreamBody(OmsHandle& h, OmsTypeABAPTabHandle& tabHandle, int rowSize, bool isInStream);
    ~OMS_StreamBody();
    void               AllocOutStreamBuffer();
    inline static void CopyAndSwapI2(const SAPDB_Byte* pSource, SAPDB_Byte* pDest);
    inline static void CopyAndSwapI4(const SAPDB_Byte* pSource, SAPDB_Byte* pDest);
    inline static void CopyAndSwapI8(const SAPDB_Byte* pSource, SAPDB_Byte* pDest);
    void               Flush();
    SAPDB_Byte*        GetNext();
    void               IncopyStream(const SAPDB_Byte* buffer);
    void               Write(SAPDB_Byte* pObj);

    const OmsHandle    &getHandle() const
    {
      return m_handle;
    }
private:
    OmsHandle&             m_handle;
    SAPDB_Int4             m_rowSize;
    SAPDB_Int4             m_clientRowSize;
    SAPDB_Int4             m_rowCount;
    SAPDB_Int4             m_maxOutCount;
    int                    m_lowByte;
    int                    m_highByte;
    bool                   m_eot;
    bool                   m_isInStream;
    bool                   m_isAbapStream;
    OmsTypeABAPTabHandle&  m_streamDescriptor;   
    OMS_StreamBufferChunk* m_firstChunk;
    OMS_StreamBufferChunk* m_lastChunk;
    SAPDB_Byte*            m_curr;
    SAPDB_Byte*            m_outBuffer;
};

/*----------------------------------------------------------------------*/

inline void OMS_StreamBody::CopyAndSwapI2(const SAPDB_Byte* pSource, SAPDB_Byte* pDest)
{
    *pDest     = *(pSource+1);
    *(pDest+1) = *pSource;
}

/*----------------------------------------------------------------------*/

inline void OMS_StreamBody::CopyAndSwapI4(const SAPDB_Byte* pSource, SAPDB_Byte* pDest)
{
    for (int ix = 0; ix < 4; ++ix)
    {
        pDest[ix] = pSource[3-ix];
    }
}

/*----------------------------------------------------------------------*/

inline void OMS_StreamBody::CopyAndSwapI8(const SAPDB_Byte* pSource, SAPDB_Byte* pDest)
{
    for (int ix = 0; ix < 8; ++ix)
    {
        pDest[ix] = pSource[7-ix];
    }
}

/*----------------------------------------------------------------------*/

OMS_StreamBody::OMS_StreamBody(OmsHandle& h, OmsTypeABAPTabHandle& tabHandle, int rowSize, bool isInStream)
: m_handle(h)
, m_rowSize(rowSize)
, m_clientRowSize(tabHandle.rowSize)
, m_rowCount(0)
, m_maxOutCount(0)
, m_lowByte(0)
, m_highByte(0)
, m_eot(isInStream && (-1 == tabHandle.rowCount))
, m_isInStream(isInStream)
, m_isAbapStream(tabHandle.rowCount != 0)
, m_streamDescriptor(tabHandle)
, m_firstChunk(NULL)
, m_lastChunk(NULL)
, m_curr(NULL)
, m_outBuffer(NULL)
{
    if (tabHandle.colCount > 0)
    {
        // find out byte order for swapping
        for (int colIdx = 0; colIdx < m_streamDescriptor.colCount; ++colIdx)
        {
            if (m_streamDescriptor.colDesc[colIdx].abap_type >= OMS_SwapAndModifyInfo::asciiClientWydeCpp)
            {
                if (OMS_SwapAndModifyInfo::unicodeClientAsciiCpp == m_streamDescriptor.colDesc[colIdx].abap_type)
                {
                    m_lowByte  = 1;
                    break;
                }
                else
                {
                    if (OMS_SwapAndModifyInfo::swappedUnicodeClientAsciiCpp == m_streamDescriptor.colDesc[colIdx].abap_type)
                    {
                        m_highByte = 1;
                        break;
                    }
                }
            }
        }
    }
    if (!isInStream)
    {
        this->AllocOutStreamBuffer();
    }
}

/*----------------------------------------------------------------------*/

OMS_StreamBody::~OMS_StreamBody()
{
    if (m_isInStream)
    {
        while (NULL != m_firstChunk)
        {
            void* toFree = m_firstChunk;
            m_firstChunk = m_firstChunk->GetNext();
            m_handle.m_pSession->deallocate(toFree);
        }
    }
    else
    {
        m_handle.m_pSession->deallocate(m_outBuffer);
    }
}

/*----------------------------------------------------------------------*/

void OMS_StreamBody::AllocOutStreamBuffer()
{
    SAPDB_Int4 PacketSize = OMS_Globals::GetKernelInterface()->GetPacketMaxUsableArea();
    m_outBuffer           = REINTERPRET_CAST(SAPDB_Byte*, m_handle.m_pSession->allocate(PacketSize));
    m_maxOutCount         = PacketSize / m_clientRowSize; 
    m_rowCount            = 0;
    m_curr                = m_outBuffer;
}

/*----------------------------------------------------------------------*/

SAPDB_Byte* OMS_StreamBody::GetNext()
{
    if (!m_handle.m_pSession->m_stream_io) 
    {
        return NULL;
    }
    m_handle.m_pSession->IncReadStreamRow();
    if (0 == m_rowCount)
    {
        if (m_eot) 
        {
            while (NULL != m_firstChunk)
            {
                void* toFree = m_firstChunk;
                m_firstChunk = m_firstChunk->GetNext();
                m_handle.m_pSession->deallocate(toFree);
            }
            return NULL;
        }
        SAPDB_Int2  DBError;
        int         bufSize;
        SAPDB_Byte* pBuf;
        m_handle.m_pSession->m_lcSink->ABAPRead (m_streamDescriptor.ABAPTabId, 
            m_rowCount, bufSize, pBuf, DBError);
        m_handle.m_pSession->IncReadStreamBuffer();
        if ( DBError != 0 )
        {
            if ( e_no_next_record == DBError )
            {
                m_eot = true;
            }
            else {
                m_handle.m_pSession->m_stream_io = false;
                m_handle.m_pSession->ThrowDBError (DBError, "omsNxt", __MY_FILE__, __LINE__);
            }
        }
        if (0 == m_rowCount) 
        {
            m_eot = true;
            return NULL;
        }
        this->IncopyStream(pBuf);
    }
    else
    {
        if (0 == m_firstChunk->GetCount())
        {
            // current chunk has been traversed completely, move to next one
            // and release current chunk
            void* toFree = m_firstChunk;
            m_firstChunk = m_firstChunk->GetNext();
            m_handle.m_pSession->deallocate(toFree);
            m_curr = m_firstChunk->FirstPos();
        }
        else
        {
            m_curr += m_rowSize;
        }
    }
    m_firstChunk->DecrementCount();
    --m_rowCount;
    return m_curr;
}

/*----------------------------------------------------------------------*/

void OMS_StreamBody::IncopyStream(const SAPDB_Byte* buffer)
{
    register const SAPDB_Byte* pBuf  = buffer;
    OMS_StreamBufferChunk**    pPrev = &m_firstChunk;
    int ix = 0; 
    while (ix < m_rowCount)
    {
        if (0 == *pPrev)
        {
            *pPrev = OMS_StreamBufferChunk::Create(m_handle.m_pSession,m_rowSize);
        }
        m_lastChunk   = *pPrev;
        pPrev         = m_lastChunk->GetNextAddr();
        int copyCount = m_rowCount - ix;
        if (copyCount > m_lastChunk->Capacity())
        {
            copyCount = m_lastChunk->Capacity();
        }
        SAPDB_Byte* pChunk = m_lastChunk->FirstPos(); 
        for (int copyIdx = 0; copyIdx < copyCount; ++copyIdx)
        {
            // translate struct from client layout into c++ layout
            for (int descIdx = 0; descIdx < m_streamDescriptor.colCount; ++descIdx)
            {
                OmsTypeABAPColDesc& abapDesc = m_streamDescriptor.colDesc[descIdx];
                register const SAPDB_Byte* pSource = pBuf   + abapDesc.offset;
                register       SAPDB_Byte* pDest   = pChunk + abapDesc.dec;
                // copy and transform one struct member
                switch (abapDesc.abap_type) {
                case OMS_SwapAndModifyInfo::swap_2 :
                    {
                        // swap 2 bytes integers
                        // PTS 1132754, PG
                        int intCnt = 0;
                        if (abapDesc.abap_type == ABTYPINT2)
                        {
                            intCnt = 1; // == abapDesc.length/2
                        }
                        else // ABTYPWYDE
                        {
                            intCnt = abapDesc.length;
                        }
                        for (int l = 0; l < intCnt; ++l)
                        {
                            CopyAndSwapI2(pSource, pDest);
                            pSource += 2;
                            pDest   += 2;
                        }
                        break;
                    }
                case OMS_SwapAndModifyInfo::swap_4 :
                    {
                        // swap 4 byte integer
                        CopyAndSwapI4(pSource, pDest);
                        break;
                    }
                case OMS_SwapAndModifyInfo::swap_8 :
                    {
                        // swap 8 byte datatypes (int8, double)
                        CopyAndSwapI8(pSource, pDest);
                        break;
                    }
                case OMS_SwapAndModifyInfo::asciiClientWydeCpp :
                    {
                        // expand 1 byte character to wyde
                        OmsTypeWyde* pWyde = REINTERPRET_CAST(OmsTypeWyde*, pDest);
                        for (int l = 0; l < abapDesc.length; ++l)
                        {
                            *pWyde = *pSource;
                            ++pWyde;
                            ++pSource;
                        }
                        break;
                    }
                case OMS_SwapAndModifyInfo::unicodeClientAsciiCpp       :
                case OMS_SwapAndModifyInfo::swappedUnicodeClientAsciiCpp :
                    {
                        // translate UCS2 to ASCII7
                        for (int l = 0; l < abapDesc.length; ++l)
                        {
                            if ((0 != *(pSource + m_highByte)) || (*(pSource + m_lowByte) > 127))
                            {
                                // unicode char does not fit into ascii7
                                DbpBase base(m_handle.m_pSession->m_lcSink);
                                base.dbpOpError("unicode not translatable to ascii7 : %d", *(pSource + m_highByte) * 256 + *(pSource + m_lowByte)); 
                                // PTS 1128956
                                base.dbpOpError("offset of member is %d bytes", abapDesc.offset);
                                base.dbpOpError("index  in member is %d", l);
                                tsp00_Int4 bufLength = sizeof(m_streamDescriptor.ABAPTabId);
                                tsp00_Int2 dummyError;
                                m_handle.m_pSession->m_lcSink->MultiPurpose (m_return_error, mm_nil, 
                                       &bufLength, reinterpret_cast<unsigned char*>(&m_streamDescriptor.ABAPTabId), &dummyError);
                                // end PTS 1128956
                                m_handle.m_pSession->ThrowDBError (e_not_translatable, "OMS_StreamBody::IncopyStream", __MY_FILE__, __LINE__);
                            }
                            *pDest = *(pSource + m_lowByte);
                            ++pDest;
                            pSource += 2;
                        }
                        break;
                    }
                default :
                    {
                        // just copy without any translation 
                        SAPDB_MemCopyNoCheck(pDest, pSource, abapDesc.length);
                    }
                } // switch
            } // for every member
            pBuf   += m_clientRowSize;
            pChunk += m_rowSize; 
        } // for every row
        m_lastChunk->SetCount(copyCount);  
        ix += copyCount;
    } // while every row
    m_curr = m_firstChunk->FirstPos();
}

/*----------------------------------------------------------------------*/

void OMS_StreamBody::Flush()
{
    if ((!m_handle.m_pSession->m_stream_io) || (0 == m_rowCount))
    {
        return;
    }
    SAPDB_Int2 DBError;
    m_handle.m_pSession->m_lcSink->ABAPWrite (&m_streamDescriptor.ABAPTabId,
        m_clientRowSize, m_rowCount, &m_outBuffer[0], &DBError);
    m_rowCount = 0;
    m_curr     = m_outBuffer;
    m_handle.m_pSession->IncWriteStreamBuffer();
    if ( DBError != 0 ) {
        m_handle.m_pSession->m_stream_io = false; 
        m_handle.m_pSession->ThrowDBError (DBError, "omsFlush (stream)", __MY_FILE__, __LINE__);
    }
}

/*----------------------------------------------------------------------*/


void OMS_StreamBody::Write(SAPDB_Byte* pObj)
{
    // write one row into the output buffer. The row is transferred to the client layout
    // if necessary
    if (!m_handle.m_pSession->m_stream_io) 
    {
        return;
    }
    if (m_maxOutCount == m_rowCount)
    {
        this->Flush();
    }
    for (int descIdx = 0; descIdx < m_streamDescriptor.colCount; ++descIdx)
    {
        OmsTypeABAPColDesc& abapDesc = m_streamDescriptor.colDesc[descIdx];
        register const SAPDB_Byte* pSource = pObj   + abapDesc.dec;
        register       SAPDB_Byte* pDest   = m_curr + abapDesc.offset;
        // copy and transform one struct member
        switch (abapDesc.abap_type) {
        case OMS_SwapAndModifyInfo::swap_2 :
            {
                // swap 2 bytes integers
                // PTS 1132754, PG
                int intCnt = 0;
                if (abapDesc.abap_type == ABTYPINT2) 
                {
                    intCnt = 1; // == abapDesc.length/2
                }
                else // ABTYPWYDE
                {
                    intCnt = abapDesc.length;
                }
                for (int l = 0; l < intCnt; ++l)
                {
                    CopyAndSwapI2(pSource, pDest);
                    pSource += 2;
                    pDest   += 2;
                }
                break;
            }
        case OMS_SwapAndModifyInfo::swap_4 :
            {
                // swap 4 byte integer
                CopyAndSwapI4(pSource, pDest);
                break;
            }
        case OMS_SwapAndModifyInfo::swap_8 :
            {
                // swap 8 byte datatypes (int8, double)
                CopyAndSwapI8(pSource, pDest);
                break;
            }
        case OMS_SwapAndModifyInfo::asciiClientWydeCpp :
            {
                // truncate wyde character to ascii character
                const OmsTypeWyde* pWyde = REINTERPRET_CAST(const OmsTypeWyde*, pSource);
                for (int l = 0; l < abapDesc.length; ++l)
                {
                    if (*pWyde > 255)
                    {
                        // unicode char does not fit into ascii
                        DbpBase base(m_handle.m_pSession->m_lcSink);
                        base.dbpOpError("unicode not translatable to ascii : %d", *pWyde);
                        // PTS 1128956
                        base.dbpOpError("offset of member is %d bytes", abapDesc.offset);
                        base.dbpOpError("index  in member is %d", l);
                        tsp00_Int4 bufLength = sizeof(m_streamDescriptor.ABAPTabId);
                        tsp00_Int2 dummyError;
                        m_handle.m_pSession->m_lcSink->MultiPurpose (m_return_error, mm_nil, 
                            &bufLength, reinterpret_cast<unsigned char*>(&m_streamDescriptor.ABAPTabId), &dummyError);
                        // end PTS 1128956                       
                        m_handle.m_pSession->ThrowDBError (e_not_translatable, "OMS_StreamBody::Write", __MY_FILE__, __LINE__);
                    }
                    *pDest = (SAPDB_Byte) *pWyde;
                    ++pWyde;
                    ++pDest;
                }
                break;
            }
        case OMS_SwapAndModifyInfo::unicodeClientAsciiCpp       :
        case OMS_SwapAndModifyInfo::swappedUnicodeClientAsciiCpp :
            {
                // expand ascii to unicode
                for (int l = 0; l < abapDesc.length; ++l)
                {
                    *(pDest + m_highByte) = 0;
                    *(pDest + m_lowByte ) = *pSource;
                    pDest   += 2;
                    ++pSource;
                }
                break;
            }
        default :
            {
                // just copy without any translation 
                SAPDB_MemCopyNoCheck(pDest, pSource, abapDesc.length);
            }
        } // case
    }
    ++m_rowCount;
    m_handle.m_pSession->IncWriteStreamRow();
    m_curr += m_clientRowSize;
}

/*----------------------------------------------------------------------*/

OmsStream::OmsStream(OmsHandle* h, OmsTypeABAPTabHandle& sh, long rsz, long sz, bool istream):
m_pSession(h->m_pSession)
{
    if (rsz == 0 || sh.rowSize == 0 || sh.rowSize > OMS_Globals::GetKernelInterface()->
        GetPacketMaxUsableArea()) {
        // invalid row size
        if (rsz == 0 || sh.rowSize == 0) {
            h->dbpOpError("OmsStream::OmsStream: stream row size must be > 0");
        } else {
            h->dbpOpError("OmsStream::OmsStream: stream row size %d must be < %d",
              sh.rowSize, OMS_Globals::GetKernelInterface()->
              GetPacketMaxUsableArea());
        }
        throw DbpError(DbpError::DB_ERROR, e_invalid_rowsize, __MY_FILE__, __LINE__);
    }
    m_body = new (m_pSession->allocate(sizeof(OMS_StreamBody))) OMS_StreamBody (*h, sh, rsz, istream);
}

/*----------------------------------------------------------------------*/

OmsStream::OmsStream(OmsHandle* h, OmsTypeStreamHandle& sh, long rsz, long sz, bool istream):
m_pSession(h->m_pSession)
{
    if (rsz == 0 || sh.size == 0 || sh.size > OMS_Globals::GetKernelInterface()->
        GetPacketMaxUsableArea()) {
        // invalid row size
        if (rsz == 0 || sh.size == 0) {
            h->dbpOpError("OmsStream::OmsStream: stream row size must be > 0");
        } else {
            h->dbpOpError("OmsStream::OmsStream: stream row size %d must be < %d",
              sh.size, OMS_Globals::GetKernelInterface()->
              GetPacketMaxUsableArea());
        }
        throw DbpError(DbpError::DB_ERROR, e_invalid_rowsize, __MY_FILE__, __LINE__);
    }
    m_body = new (m_pSession->allocate(sizeof(OMS_StreamBody))) OMS_StreamBody (*h, 
        *REINTERPRET_CAST(OmsTypeABAPTabHandle*, &sh), rsz, istream);
}

/*----------------------------------------------------------------------*/

OmsStream::~OmsStream() 
{ 
    if (m_body)
    {
        m_body->~OMS_StreamBody();
        m_pSession->deallocate(m_body);
        m_body = NULL;
    }
}

/*----------------------------------------------------------------------*/

void OmsStream::omsClose() 
{
    m_body->Flush();
}

/*----------------------------------------------------------------------*/

void OmsStream::omsFlush() {
    m_body->Flush();  
}

/*----------------------------------------------------------------------*/

unsigned char* OmsStream::omsNxt() 
{
    _TRACE_METHOD_ENTRY(m_body->getHandle(),"OmsStream::omsNxt");
    OMS_CHECK_EXCEPTION(m_body->getHandle());
    return m_body->GetNext();
}

/*----------------------------------------------------------------------*/

void OmsStream::omsWrite(unsigned char* obj) {
    
    _TRACE_METHOD_ENTRY(m_handle,"OmsStream::omsWrite");
    m_body->Write(obj);
}

/*----------------------------------------------------------------------*/