bufferedsocketdevice.cpp

Go to the documentation of this file.

// Program Name: bufferedsocketdevice.cpp
// Created     : Oct. 1, 2005
//
// Purpose     : 
//                                                                            
// Copyright (c) 2005 David Blain <dblain@mythtv.org>
//                                          
// Licensed under the GPL v2 or later, see COPYING for details                    
//

#include <algorithm>
#include <chrono> // for milliseconds
#include <thread> // for sleep_for

#include "mythtimer.h"
#include "bufferedsocketdevice.h"
#include "upnputil.h"
#include "mythlogging.h"

//

BufferedSocketDevice::BufferedSocketDevice( int nSocket  )
{
    m_pSocket = new MSocketDevice();

    m_pSocket->setSocket         ( nSocket, MSocketDevice::Stream );
    m_pSocket->setBlocking       ( false );
    m_pSocket->setAddressReusable( true );

    struct linger ling = {1, 1};

    if ( setsockopt(socket(), SOL_SOCKET, SO_LINGER, (const char *)&ling,
                    sizeof(ling)) < 0) 
        LOG(VB_GENERAL, LOG_ERR, 
            "BufferedSocketDevice: setsockopt - SO_LINGER: " + ENO);

    m_nDestPort          = 0;

    m_nMaxReadBufferSize = 0; 
    m_nWriteSize         = 0;
    m_nWriteIndex        = 0;
    m_bHandleSocketDelete= true;
}

//

BufferedSocketDevice::BufferedSocketDevice( MSocketDevice *pSocket /* = nullptr*/,
                                            bool bTakeOwnership /* = false */ )
{
    m_pSocket            = pSocket;

    m_nDestPort          = 0;

    m_nMaxReadBufferSize = 0; 
    m_nWriteSize         = 0;
    m_nWriteIndex        = 0;
    m_bHandleSocketDelete= bTakeOwnership;

}

//

BufferedSocketDevice::~BufferedSocketDevice()
{
    Close();
}

//

void BufferedSocketDevice::Close()
{
    Flush();
    ReadBytes();

    m_bufRead.clear();
    ClearPendingData();

    if (m_pSocket != nullptr)
    {
        if (m_pSocket->isValid())
            m_pSocket->close();

        if (m_bHandleSocketDelete)
            delete m_pSocket;

        m_pSocket = nullptr;
    }

}

//

bool BufferedSocketDevice::Connect( const QHostAddress &addr, quint16 port )
{
    if (m_pSocket == nullptr)
        return false;

    return m_pSocket->connect( addr, port );
}

//

MSocketDevice *BufferedSocketDevice::SocketDevice()
{
    return( m_pSocket );
}

//

void BufferedSocketDevice::SetSocketDevice( MSocketDevice *pSocket )
{
    if ((m_bHandleSocketDelete) && (m_pSocket != nullptr))
        delete m_pSocket;
    
    m_bHandleSocketDelete = false;

    m_pSocket = pSocket;
}

//

void BufferedSocketDevice::SetDestAddress(
    QHostAddress hostAddress, quint16 nPort)
{
    m_DestHostAddress = hostAddress;
    m_nDestPort       = nPort;
}

//

void BufferedSocketDevice::SetReadBufferSize( qulonglong bufSize )
{
    m_nMaxReadBufferSize = bufSize;
}

//

qulonglong BufferedSocketDevice::ReadBufferSize(void) const
{
    return m_nMaxReadBufferSize;
}

//

int BufferedSocketDevice::ReadBytes()
{
    if (m_pSocket == nullptr)
        return m_bufRead.size();

    qlonglong maxToRead = 0;

    if ( m_nMaxReadBufferSize > 0 ) 
    {
        maxToRead = m_nMaxReadBufferSize - m_bufRead.size();

        if ( maxToRead <= 0 ) 
            return m_bufRead.size();
    }

    qlonglong nbytes = m_pSocket->bytesAvailable();
    qlonglong nread;

    QByteArray *a = nullptr;

    if ( nbytes > 0 )
    {
        a = new QByteArray();
        a->resize(nbytes);

        nread = m_pSocket->readBlock(
            a->data(), maxToRead ? std::min(nbytes, maxToRead) : nbytes);

        if (( nread > 0 ) && ( nread != a->size() ))
        {
            // unexpected
            a->resize( nread );
        }
    }

    if (a)
    {
#if 0
        QString msg;
        for( long n = 0; n < a->count(); n++ )
            msg += QString("%1").arg(a->at(n));
        LOG(VB_GENERAL, LOG_DEBUG, msg);
#endif

        m_bufRead.append( a );
    }

    return m_bufRead.size();
}

//

bool BufferedSocketDevice::ConsumeWriteBuf( qulonglong nbytes )
{
    if ( !nbytes || ((qlonglong)nbytes > m_nWriteSize) )
        return false;

    m_nWriteSize -= nbytes;

    for ( ;; ) 
    {
        QByteArray *a = m_bufWrite.front();

        if ( m_nWriteIndex + nbytes >= (qulonglong)a->size() ) 
        {
            nbytes -= a->size() - m_nWriteIndex;
            m_bufWrite.pop_front();
            delete a;

            m_nWriteIndex = 0;

            if ( nbytes == 0 )
                break;
        } 
        else 
        {
            m_nWriteIndex += nbytes;
            break;
        }
    }

    return true;
}

//

void BufferedSocketDevice::Flush()
{

    if ((m_pSocket == nullptr) || !m_pSocket->isValid())
        return;

    bool osBufferFull = false;
    //int  consumed     = 0;

    while ( !osBufferFull && ( m_nWriteSize > 0 ) && m_pSocket->isValid())
    {
        deque<QByteArray*>::iterator it = m_bufWrite.begin();
        QByteArray *a = *it;

        int nwritten = 0;
        int i = 0;

        if ( a->size() - m_nWriteIndex < 1460 )
        {
            QByteArray out;
            out.resize(65536);

            int j = m_nWriteIndex;
            int s = a->size() - j;

            while ( a && i+s < out.size() )
            {
                memcpy( out.data()+i, a->data()+j, s );
                j = 0;
                i += s;
                ++it;
                a = *it;
                s = a ? a->size() : 0;
            }

            if (m_nDestPort != 0)
                nwritten = m_pSocket->writeBlock( out.data(), i, m_DestHostAddress, m_nDestPort );
            else
                nwritten = m_pSocket->writeBlock( out.data(), i );
        } 
        else 
        {
            // Big block, write it immediately
            i = a->size() - m_nWriteIndex;

            if (m_nDestPort != 0)
                nwritten = m_pSocket->writeBlock( a->data() + m_nWriteIndex, i, m_DestHostAddress, m_nDestPort );
            else
                nwritten = m_pSocket->writeBlock( a->data() + m_nWriteIndex, i );
        }

        if ( nwritten > 0 ) 
        {
            if ( ConsumeWriteBuf( nwritten ) )
            {
                //consumed += nwritten;
            }
        }

        if ( nwritten < i )
            osBufferFull = true;
    }
}


//

qlonglong BufferedSocketDevice::Size()
{
    return (qlonglong)BytesAvailable();
}

//

qlonglong BufferedSocketDevice::At() const
{
    return( 0 );
}

//

bool BufferedSocketDevice::At( qlonglong index )
{
    ReadBytes();

    if ( index > m_bufRead.size() )
        return false;

    // throw away data 0..index-1
    m_bufRead.consumeBytes( (qulonglong)index, nullptr );

    return true;
}

//

bool BufferedSocketDevice::AtEnd()
{
    if ( !m_pSocket->isValid() )
        return true;

    ReadBytes();

    return m_bufRead.size() == 0;

}

//

qulonglong BufferedSocketDevice::BytesAvailable(void)
{
    if ( !m_pSocket->isValid() )
        return 0;

    return ReadBytes();
}

//

qulonglong BufferedSocketDevice::WaitForMore(
    int msecs, bool *pTimeout /* = nullptr*/ ) 
{
    bool bTimeout = false;

    if ( !m_pSocket->isValid() )
        return 0;

    qlonglong nBytes = BytesAvailable();

    if (nBytes == 0)
    {
/*
        The following code is a possible workaround to the lost request problem
        I just hate looping too much to put it in.  I believe there is something
        I'm missing that is causing the lost packets... Just need to find it.

        bTimeout      = true;
        int    nCount = 0;
        int    msWait = msecs / 100;
        
        while (((nBytes = ReadBytes()) == 0 ) && 
               (nCount++              < 100 ) &&
                bTimeout                      && 
                m_pSocket->isValid()         )
        {
            // give up control

            // should be some multiple of msWait.
            std::this_thread::sleep_for(std::chrono::milliseconds(1));

        }
    }
*/
        // -=>TODO: Override the timeout to 1 second... Closes connection sooner
        //          to help recover from lost requests.  (hack until better fix found)

        msecs  = 1000;

        nBytes = m_pSocket->waitForMore( msecs, &bTimeout );

        if (pTimeout != nullptr)
            *pTimeout = bTimeout;
    }
            
    return nBytes; // nBytes  //m_bufRead.size();
}

//

qulonglong BufferedSocketDevice::BytesToWrite(void) const
{
    return m_nWriteSize;

}

//

void BufferedSocketDevice::ClearPendingData()
{
    while (!m_bufWrite.empty())
    {
        delete m_bufWrite.back();
        m_bufWrite.pop_back();
    }
    m_nWriteIndex = m_nWriteSize = 0;
}

//

void BufferedSocketDevice::ClearReadBuffer()
{
    m_bufRead.clear();
}

//

qlonglong BufferedSocketDevice::ReadBlock( char *data, qulonglong maxlen )
{
    if ( data == nullptr && maxlen != 0 ) 
        return -1;

    if ( !m_pSocket->isOpen() ) 
        return -1;

    ReadBytes();

    if ( maxlen >= (qulonglong)m_bufRead.size() )
        maxlen = m_bufRead.size();

    m_bufRead.consumeBytes( maxlen, data );

    return maxlen;

}

//

qlonglong BufferedSocketDevice::WriteBlock( const char *data, qulonglong len )
{
    if ( len == 0 )
        return 0;

    QByteArray *a = m_bufWrite.back();

    bool writeNow = ( (m_nWriteSize + len >= 1400) || (len > 512) );

    if ( a && (a->size() + len < 128) ) 
    {
        // small buffer, resize
        int i = a->size();

        a->resize( i+len );
        memcpy( a->data()+i, data, len );
    } 
    else 
    {
        // append new buffer
        m_bufWrite.push_back(new QByteArray(data, len));
    }

    m_nWriteSize += len;

    if ( writeNow )
        Flush();

    return len;
}

//

qlonglong BufferedSocketDevice::WriteBlockDirect(
    const char *data, qulonglong len)
{
    qlonglong nWritten = 0;
        
    // must Flush data just in case caller is mixing buffered & un-buffered calls

    Flush();

    if (m_nDestPort != 0)
        nWritten = m_pSocket->writeBlock( data, len, m_DestHostAddress, m_nDestPort );
    else
        nWritten = m_pSocket->writeBlock( data, len );

    return nWritten;
}

//

int BufferedSocketDevice::Getch()
{
    if ( m_pSocket->isOpen() )
    {
        ReadBytes();

        if (m_bufRead.size() > 0 ) 
        {
            uchar c;

            m_bufRead.consumeBytes( 1, (char*)&c );
        
            return c;
        }
    }

    return -1;
}

//

int BufferedSocketDevice::Putch( int ch )
{
    char buf[2];

    buf[0] = ch;

    return WriteBlock(buf, 1) == 1 ? ch : -1;
}

//

int BufferedSocketDevice::Ungetch(int ch )
{
    return m_bufRead.ungetch( ch );
}

//

bool BufferedSocketDevice::CanReadLine()
{
    ReadBytes();

    if (( BytesAvailable() > 0 ) && m_bufRead.scanNewline( nullptr ) )
        return true;

    return false;
}
                               
//

QString BufferedSocketDevice::ReadLine()
{
    QByteArray a;
    a.resize(256);

    ReadBytes();

    bool nl = m_bufRead.scanNewline( &a );

    QString s;

    if ( nl ) 
    {
        At( a.size() );     // skips the data read

        s = QString( a );
    }

    return s;
}

//

QString BufferedSocketDevice::ReadLine( int msecs )
{
    MythTimer timer;
    QString   sLine;

    if ( CanReadLine() )
        return( ReadLine() );
        
    // ----------------------------------------------------------------------
    // If the user supplied a timeout, lets loop until we can read a line 
    // or timeout.
    // ----------------------------------------------------------------------

    if ( msecs > 0)
    {
        bool bTimeout = false;

        timer.start();

        while ( !CanReadLine() && !bTimeout )
        {
#if 0
            LOG(VB_HTTP, LOG_DEBUG, "Can't Read Line... Waiting for more." );
#endif

            WaitForMore( msecs, &bTimeout );

            if ( timer.elapsed() >= msecs ) 
            {
                bTimeout = true;
                LOG(VB_HTTP, LOG_INFO, "Exceeded Total Elapsed Wait Time." );
            }
        }
            
        if (CanReadLine())
            sLine = ReadLine();
    }

    return( sLine );
}

//

quint16 BufferedSocketDevice::Port(void) const
{
    if (m_pSocket)
        return( m_pSocket->port() );

    return 0;
}

//

quint16 BufferedSocketDevice::PeerPort(void) const
{
    if (m_pSocket)
        return( m_pSocket->peerPort() );

    return 0;
}

//

QHostAddress BufferedSocketDevice::Address() const
{
    if (m_pSocket)
        return( m_pSocket->address() );

    QHostAddress tmp;

    return tmp;
}

//

QHostAddress BufferedSocketDevice::PeerAddress() const
{
    if (m_pSocket)
        return( m_pSocket->peerAddress() );

    QHostAddress tmp;

    return tmp;
}