phNetSource.cpp

Go to the documentation of this file.

/* ---------------------------------------------------------------------------
    Phission : 
        Realtime Vision Processing System
    
    Copyright (C) 2003-2006 Philip D.S. Thoren (pthoren@cs.uml.edu)
    University of Massachusetts at Lowell,
    Laboratory for Artificial Intelligence and Robotics
    
    This file is part of Phission.

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

    Phission 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 Lesser General Public License for more details.

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

 ---------------------------------------------------------------------------*/
#ifdef HAVE_CONFIG_H
    #include <phissionconfig.h>
#endif

#define phNetSource_PRINT_IMAGE_SIZE() 0
#define phNetSource_USE_JITTER_SYNC() 0

#define VERBOSE() 0

/* ---------------------------------------------------------------------- */
#include <phStandard.h>
#include <ImageConversions.h>

#include <phObject.h>
#include <phMutex.h>
#include <phSemaphore.h>
#include <phCondition.h>
#include <phRWLock.h>
#include <phThread.h>
#include <phList.h>
#include <phTimeStamp.h>
#include <phTimeInterval.h>
#include <phLiveObject.h>
#include <phDataObject.h>
#include <phImage.h>
#include <phCaptureInterface.h>
#include <phImageCapture.h>

#include <phTimeInterval.h>
#include <phSocket.h>
#include <phNetSource.h>

#include <phError.h>
#include <phMemory.h>
#include <phPrint.h>

/* ---------------------------------------------------------------------- */
/* TODO:  */
struct phNetSourceInfo_t
{
    phSocket    *m_comms;
    char        *m_host;
    uint32_t    m_port;
    /* The width,height, format and title are retreived from the server.  */
    uint32_t    m_width;
    uint32_t    m_height;
    uint32_t    m_format;
    char        *m_title;

    int         m_do_resize;
    int         m_do_convert;
};

/* ---------------------------------------------------------------------- */
phNetSource::phNetSource(char *path)
{
    phFUNCTION("phNetSource::phNetSource")
    
    rc = this->lock();
    phPRINT_RC(rc,NULL,"this->lock()");
   
    this->setName("phNetSource");
   
    this->m_info = (struct phNetSourceInfo_t *)phCalloc(1,
                        sizeof(struct phNetSourceInfo_t));
    phPRINT_PTR(this->m_info,"phCalloc",
              "phCalloc failed to alloc phNetSourceInfo");
    this->m_info->m_host        = NULL;
    this->m_info->m_port        = 0;
    this->m_info->m_comms       = NULL;
    this->m_info->m_width       = 0;
    this->m_info->m_height      = 0;
    this->m_info->m_format      = phImageNOFORMAT;
    this->m_info->m_title       = NULL;
    this->m_info->m_do_resize   = 0;
    this->m_info->m_do_convert  = 0;
    
    this->m_orig_format = phImageNOFORMAT;
    this->m_orig_height = 0;
    this->m_orig_width  = 0;
    this->m_orig_size   = 0;

    this->m_is_open     = 0;
    
    this->m_type        = phNetTypeTCP;
    this->m_temp_path   = NULL;
   
    this->m_connected_type = phNetTypeUNKNOWN;
    
    this->m_reconnect   = 1;
    this->m_server_wait = 1;
        
    if (path != NULL)
    {
        this->setPath(path);
    }
    else
    {
        this->setPath("tcp://localhost:22345");
    }

    /* This applies the default (1) picture/image settings[w,h,format,etc.]
     * (2) channel settings  */
    this->applySettings();
    
    /* opening the device must be explicitly done */
    /* capturing must be explicitly done. */
#if 0
error:
#endif
    rc = this->unlock();
    phPRINT_RC(rc,NULL,"this->unlock()");
    
    return;
}

/* ---------------------------------------------------------------------- */
phNetSource::~phNetSource ()
{
    phFUNCTION("phNetSource::~phNetSource")
  
    rc = this->lock();
    phPRINT_RC(rc,NULL,"this->lock()");

    if (this->isRunning())
    {
        rc = this->stop();
        phPRINT_RC(rc,NULL,"this->stop()");
    }
   
    if (this->isOpen())
    {
        rc = this->close();
        phPRINT_RC(rc,NULL,"this->close()");
    }
 
    phFree(this->m_info);
    
    rc = this->unlock();
    phPRINT_RC(rc,NULL,"this->unlock()");
}

/* ---------------------------------------------------------------------- */
int phNetSource::parsePath( const char  *path,
                            char        **host_ptr,
                            uint32_t    *port_ptr,
                            uint32_t    *type_ptr   )
{
    phFUNCTION("phNetSource::parsePath")
    
    uint32_t i                  = 0;
    uint32_t type               = phNetTypeTCP;
    uint32_t port               = 22345;
    char    *host               = NULL;
    int32_t  host_start         = 0;
    uint32_t host_length        = 0;
    uint32_t host_ptr_length    = 0;
    uint32_t length             = (uint32_t)((path != NULL) ? strlen(path) : 0);
    
    /* Find the first ':' */
    for (i = 0; ((i < length) && (path[i] != ':')); i++ );

    if (i >= length)
    {
        phCHECK_RC(-1,NULL,"Invalid path string: %s", path);
    }

    if (strncmp("tcp",&(path[0]),i) == 0)
    {
        type = phNetTypeTCP;
    }
    else
    {
        type = phNetTypeUNKNOWN;
    }

    /* Go past the "//" to the host */
    for (i++; ((i < length) && (path[i] != '/')); i++ );
    for (i++; ((i < length) && (path[i] != '/')); i++ );
    
    if (i >= length)
    {
        phCHECK_RC(-1,NULL,"Invalid path string: %s", path);
    }

    /* Parse out the host */
    for (i++, host_start = i, host_length = 0;
         ((i < length) && (path[(i > 0) ? i - 1 : 0] != ':')); 
         i++, host_length++ );

    if ((i >= length) || (host_length <= 0))
    {
        phCHECK_RC(-1,NULL,"Invalid path string: %s", path);
    }

    host = (char *)phCalloc((host_length + 1),sizeof(char));
    phCHECK_PTR(host,"phCalloc",
              "phCalloc failed to allocate host string buffer of length %d",
              host_length );
    
#if defined(HAVE_SNPRINTF)
    snprintf(host,host_length,"%s",&(path[host_start]));
#else
    sprintf(host,"%s",&(path[host_start]));
#endif

    /* Get the port from the end of the line */
    port = atoi(&(path[i]));
    
    if (host_ptr != NULL)
    {
        /* Figure out the path length */
        host_ptr_length = (uint32_t)((*host_ptr) != NULL ? strlen(*host_ptr) : 0);
    
        /* allocate enough space for the path */
        phDALLOC((*host_ptr),host_ptr_length,(host_length+1),char);

#if defined(HAVE_SNPRINTF)
        snprintf(*host_ptr,host_length+1,"%s",host);
#else
        sprintf(*host_ptr,"%s",host);
#endif
    }

    /* Assign the values through the return pointers */
    if (type_ptr != NULL)
        *type_ptr = type;
    if (port_ptr != NULL)
        *port_ptr = port;
    
    phFree(host);

    //phPRINT("-%s:%d-\n",(*host_ptr),(*port_ptr));

    return phSUCCESS;
error:
    phFree(host);
    
    return phFAIL;
}

/* ---------------------------------------------------------------------- */
/* makePath : 
 *
 * This function allocates space to *path, be sure to free it */
/* ---------------------------------------------------------------------- */
int phNetSource::makePath( const char  *host, 
                           uint32_t    port,
                           uint32_t    type,
                           char        **path )
{
    phFUNCTION("phNetSource::makePath")

    uint32_t pathLength     = 0;
    uint32_t length         = 0;
    uint32_t host_length    = 0;
    uint32_t type_length    = 0;
    char *   type_string    = NULL;

    if (path == NULL) return phFAIL;
  
    /* Default to the local machine */
    if (host == NULL)
    {
        host = "localhost";
    }
    host_length = (uint32_t)strlen(host);
    
    /* make sure the path is valid */
    port = (port != 0) ? ((port < 65535) ? port : 22345) : 22345;
   
    /* Set up the type string */
    switch (type)
    {
        case phNetTypeTCP:
        default:
            type_string = "tcp://";
            break;
    }
    type_length = (uint32_t)strlen(type_string);

    length = type_length + host_length + 1 + 7;
    
    /* Figure out the path length */
    pathLength = (uint32_t)((*path) != NULL ? strlen(*path) : 0);
    
    /* allocate enough space for the path */
    phDALLOC((*path),pathLength,length,char);

    sprintf(*path,"%s%s:%u",
            type_string,
            host,
            port );
    return phSUCCESS;
error:
    return phFAIL;
}
/* ---------------------------------------------------------------------- */
int phNetSource::setHost( const char *host )
{
    phFUNCTION("phNetSource::setHost")
    int locked = 0;
   
    phTHIS_LOCK(locked);

    rc = phNetSource::makePath( host,
                                this->m_info->m_port,
                                this->m_type,
                                &this->m_temp_path);
    phCHECK_RC(rc,NULL,"phNetSource::makePath");

    rc = this->setPath(this->m_temp_path);
    phCHECK_RC(rc,NULL,"this->setPath(%s)",this->m_temp_path);

    phTHIS_UNLOCK_RET(locked,phSUCCESS,phFAIL);
}

/* ---------------------------------------------------------------------- */
const char *phNetSource::getHost( )
{
    phFUNCTION("phNetSource::getHost")
    int locked = 0;
    char *retval = NULL;
   
    phTHIS_LOCK(locked);

    retval = this->m_info->m_host;
    
    phTHIS_UNLOCK_RET(locked,((const char *)retval),NULL);
}

/* ---------------------------------------------------------------------- */
int phNetSource::setPort( uint32_t port )
{
    phFUNCTION("phNetSource::setPort")
    int locked = 0;
   
    phTHIS_LOCK(locked);

    rc = phNetSource::makePath( this->m_info->m_host,
                                port,
                                this->m_type,
                                &this->m_temp_path);
    phCHECK_RC(rc,NULL,"phNetSource::makePath");

    rc = this->setPath(this->m_temp_path);
    phCHECK_RC(rc,NULL,"this->setPath(%s)",this->m_temp_path);

    phTHIS_UNLOCK_RET(locked,phSUCCESS,phFAIL);
}

/* ---------------------------------------------------------------------- */
uint32_t phNetSource::getPort( )
{
    phFUNCTION("phNetSource::getPort")
    int locked = 0;
    uint32_t retval = 0;
   
    phTHIS_LOCK(locked);

    retval = this->m_info->m_port;
    
    phTHIS_UNLOCK_RET(locked,retval,0);
}

/* ---------------------------------------------------------------------- */
int phNetSource::setReconnect( int reconnect )
{
    phFUNCTION("phNetSource::setReconnect")
    int locked = 0;
   
    phTHIS_LOCK(locked);

    this->m_reconnect = reconnect;
    
    phTHIS_UNLOCK_RET(locked,phSUCCESS,phFAIL);
}

/* ---------------------------------------------------------------------- */
int phNetSource::getReconnect( )
{
    phFUNCTION("phNetSource::getReconnect")
    int locked = 0;
    int retval = 0;
   
    phTHIS_LOCK(locked);

    retval = this->m_reconnect;
    
    phTHIS_UNLOCK_RET(locked,retval,0);
}

/* ---------------------------------------------------------------------- */
int phNetSource::setServerWait( int wait )
{
    phFUNCTION("phNetSource::setServerWait")
    int locked = 0;
   
    phTHIS_LOCK(locked);

    this->m_server_wait = wait;
    
    phTHIS_UNLOCK_RET(locked,phSUCCESS,phFAIL);
}

/* ---------------------------------------------------------------------- */
int phNetSource::getServerWait( )
{
    phFUNCTION("phNetSource::getServerWait")
    int locked = 0;
    int retval = 0;
   
    phTHIS_LOCK(locked);

    retval = this->m_server_wait;
    
    phTHIS_UNLOCK_RET(locked,retval,0);
}

/* ---------------------------------------------------------------------- */
int phNetSource::setOriginalSize( )
{
    phFUNCTION("phNetSource::setOriginalSize")

    rc = this->m_settings_lock.lock();
    phPRINT_RC(rc,NULL,"this->m_settings_lock.lock() failed");
   
    this->m_info->m_do_resize = 0;

    rc = this->m_settings_lock.unlock();
    phPRINT_RC(rc,NULL,"this->m_settings_lock.unlock() failed");
 
    return phSUCCESS;
}

/* ---------------------------------------------------------------------- */
int phNetSource::setOriginalFormat( )
{
    phFUNCTION("phNetSource::setOriginalFormat")

    rc = this->m_settings_lock.lock();
    phPRINT_RC(rc,NULL,"this->m_settings_lock.lock() failed");
   
    this->m_info->m_do_convert = 0;

    rc = this->m_settings_lock.unlock();
    phPRINT_RC(rc,NULL,"this->m_settings_lock.unlock() failed");
 
    return phSUCCESS;
}

/* ---------------------------------------------------------------------- */
int phNetSource::setOriginal( )
{
    phFUNCTION("phNetSource::setOriginal")

    rc = this->m_settings_lock.lock();
    phPRINT_RC(rc,NULL,"this->m_settings_lock.lock() failed");
   
    this->m_info->m_do_convert = 0;
    this->m_info->m_do_resize  = 0;

    rc = this->m_settings_lock.unlock();
    phPRINT_RC(rc,NULL,"this->m_settings_lock.unlock() failed");
 
    return phSUCCESS;
}

/* ---------------------------------------------------------------------- */
int phNetSource::getOriginal(   uint32_t *format, 
                                uint32_t *height,
                                uint32_t *width,
                                uint32_t *size )
{
    phFUNCTION("phNetSource::getOriginal")
    int locked = 0;
   
    phMUTEX_LOCK(this->m_orig_lock,locked);

    if (format != NULL)
        *format = this->m_orig_format;
    if (height != NULL)
        *height = this->m_orig_height;
    if (width != NULL)
        *width = this->m_orig_width;
    if (size != NULL)
        *size = this->m_orig_size;
    
    phMUTEX_UNLOCK_RET(this->m_orig_lock,locked,phSUCCESS,phFAIL);
}

/* ---------------------------------------------------------------------- */
uint32_t phNetSource::getOriginalFormat()
{
    phFUNCTION("phNetSource::getOriginalFormat")
    int locked = 0;
    uint32_t retval = 0;
   
    phMUTEX_LOCK(this->m_orig_lock,locked);

    retval = this->m_orig_format;
    
    phMUTEX_UNLOCK_RET(this->m_orig_lock,locked,retval,0);
}

/* ---------------------------------------------------------------------- */
uint32_t phNetSource::getOriginalHeight()
{
    phFUNCTION("phNetSource::getOriginalHeight")
    int locked = 0;
    uint32_t retval = 0;
   
    phMUTEX_LOCK(this->m_orig_lock,locked);

    retval = this->m_orig_height;
    
    phMUTEX_UNLOCK_RET(this->m_orig_lock,locked,retval,0);
}

/* ---------------------------------------------------------------------- */
uint32_t phNetSource::getOriginalWidth()
{
    phFUNCTION("phNetSource::getOriginalWidth")
    int locked = 0;
    uint32_t retval = 0;
   
    phMUTEX_LOCK(this->m_orig_lock,locked);

    retval = this->m_orig_width;
    
    phMUTEX_UNLOCK_RET(this->m_orig_lock,locked,retval,0);
}

/* ---------------------------------------------------------------------- */
uint32_t phNetSource::getOriginalSize()
{
    phFUNCTION("phNetSource::getOriginalSize")
    int locked = 0;
    uint32_t retval = 0;
   
    phMUTEX_LOCK(this->m_orig_lock,locked);

    retval = this->m_orig_size;
    
    phMUTEX_UNLOCK_RET(this->m_orig_lock,locked,retval,0);
}

/* ---------------------------------------------------------------------- */
const char *phNetSource::getTitle()
{
    return (const char *)this->m_info->m_title;
}

/* ---------------------------------------------------------------------- */
int phNetSource::resetDimensions( )
{
    phFUNCTION("phNetSource::resetDimensions")
    int locked  = 0;
    int mlocked = 0;
   
    phTHIS_LOCK(locked);
    phMUTEX_LOCK(this->m_orig_lock,mlocked);

    rc = this->set( this->m_orig_width,
                    this->m_orig_height,
                    NULL,-1,-1,-1,-1,-1,-1,
                    this->m_orig_format);
    phCHECK_RC(rc,NULL,"this->set(%d,%d,NULL,-1,-1,-1,-1,-1,-1,%s)",
             this->m_orig_width,this->m_orig_height,
             phImageFormatToString(this->m_orig_format));

    phMUTEX_UNLOCK(this->m_orig_lock,mlocked);
    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phMUTEX_ERROR_UNLOCK(this->m_orig_lock,mlocked);
    phTHIS_ERROR_UNLOCK(locked);

    return phFAIL;
}

/* ---------------------------------------------------------------------- */
int phNetSource::onApplySettings()
{
    phFUNCTION("phNetSource::onApplySettings")
    int retrc = phSUCCESS;

    uint32_t i = 0;

    int ch_init = 0;
    int ch_b = 0;
    int ch_hue = 0;
    int ch_color = 0;
    int ch_con = 0;
    int ch_white = 0;
    int ch_channel = 0;
    int ch_h = 0;
    int ch_w = 0;
    int ch_f = 0;
    int ch_p = 0;
    
    ch_init = ph_int32_array_find(this->m_changed_array, phCHANGE_INIT,NULL);
    /* 
     * this->m_changed_array contains the list of changes that are to 
     *  be applied.
     *
     * This method will call phImageCapture::onApplySettings which will empty
     * the array so that settings are applied only once.
     */
    
#if 0
    /* Copy the image settings into the V4L structure */
    ch_b = ph_int32_array_find(this->m_changed_array,       phCHANGE_BRIGHTNESS,
                            NULL);
    ch_hue = ph_int32_array_find(this->m_changed_array,     phCHANGE_HUE,
                            NULL);
    ch_color = ph_int32_array_find(this->m_changed_array,   phCHANGE_COLOR,
                            NULL);
    ch_con = ph_int32_array_find(this->m_changed_array,     phCHANGE_CONTRAST,
                            NULL);
    ch_white = ph_int32_array_find(this->m_changed_array,   phCHANGE_WHITENESS,
                            NULL);
    if (ch_init > 0) ch_b = ch_hue = ch_color = ch_con = ch_white = 1;
    if ((ch_b > 0) || (ch_hue > 0) || (ch_color > 0) || 
        (ch_con > 0) || (ch_white > 0))
    {
    }
    
    /*--------------------------------------------------------------------*
     * From linux/Documentation/video4linux/API.html:
     *
     * "Each channel can be queries with the VIDIOCGCHAN ioctl call. 
     * Before invoking this function the caller must set the channel 
     * field to the channel that is being queried. On return the 
     * struct video_channel is filled in with information about the 
     * nature of the channel itself."
     *--------------------------------------------------------------------*/
    ch_channel = ph_int32_array_find(this->m_changed_array,phCHANGE_CHANNEL,
                                    NULL);
    if ((ch_channel > 0) || (ch_init > 0))
    {
    }
    /* else, the channel is < 0 or the current chn # */
#endif
    
    /* settings that require the capture to not be in progress for the 
     * phNetSource settings changes to be applied */
    ch_h = ph_int32_array_find(this->m_changed_array, phCHANGE_HEIGHT, NULL);
    ch_w = ph_int32_array_find(this->m_changed_array, phCHANGE_WIDTH, NULL);
    ch_f = ph_int32_array_find(this->m_changed_array, phCHANGE_FORMAT, NULL);

    /* We need to check the width and height settings before we force them to be
     * true.
     *
     * The image coming from the server can't explicitly have it's size set. 
     * 
     * Instead we resize the image that is received from the server using the 
     * class's width and height settings. 
     * 
     * If the setSize/setHeight/setWidth method is called, then the image will
     * be resized from the original width and height to the new width and 
     * height. 
     */
    if (ch_w || ch_h)
    {
        phPROGRESS("\n");
        this->m_info->m_do_resize = 1;
    }
    if (ch_f)
    {
        this->m_info->m_do_convert = 1;
    }
        
    if (ch_init > 0) ch_h = ch_w = ch_f = 1;

    /* update internal information when the path is given */
    /* Only do this when the thread isn't running and the stream isn't open */
    ch_p = ph_int32_array_find(this->m_changed_array, phCHANGE_PATH, NULL);
    /*
    i = ph_int32_array_get_length(this->m_changed_array);
    phPROGRESS("ch_p:%d ch_init:%d running:%d open:%d length:%u\n",
            ch_p,ch_init,this->isRunning(),this->m_is_open,i);
    */
    /*
    if ((ch_init) && (!this->isRunning()))
    {
        this->m_user_dimensions = 0;
    }
    */

    /* These checks prevent changing the path while the thread is active */
    if (((ch_p > 0) || (ch_init > 0)) && 
        ((!(this->isRunning())) && (!(this->m_is_open))))
    {
        rc = phNetSource::parsePath(this->getPath(),
                                    &this->m_info->m_host,
                                    &this->m_info->m_port,
                                    &this->m_type);
        phPRINT_RC(rc,NULL,"phNetSource::parsePath");
        retrc = 1;
    }

    rc = this->phImageCapture::onApplySettings();
    phCHECK_RC(rc,NULL,"this->phImageCapture::onApplySettings() failed.");
 
    return retrc;
error:

    return phFAIL;
}

/* ---------------------------------------------------------------------- */
int phNetSource::open(char *path)
{
    phFUNCTION("phNetSource::open")
    
    int locked = 0;

    phTHIS_LOCK(locked);
   
    /* device is already open, close it */
    if (this->isOpen())
    {
        rc = this->close();
        phPRINT_RC(rc,NULL,"this->close()");
    }
  
    if (path != NULL)
    {
        rc = this->setPath( path );
        phPRINT_RC(rc,NULL,"this->setPath( path:%s )",path);
    }
 
    if (!this->isOpen())
    {
        rc = phNetSource::parsePath(path == NULL ? this->getPath() : path,
                                    &this->m_info->m_host,
                                    &this->m_info->m_port,
                                    &this->m_type);
        phPRINT_RC(rc,NULL,"phNetSource::parsePath");
               /* Try to apply all the settings to the V4L device */
        ph_int32_array_append(this->m_changed_array,phCHANGE_INIT);
        this->applySettings();
        
        this->m_is_open = 1;

        this->connect( false );
    }

    phTHIS_UNLOCK(locked);
    
    return phSUCCESS;
error:
    
    phTHIS_ERROR_UNLOCK(locked);
    
    return phFAIL;
}

/* ---------------------------------------------------------------------- */
int phNetSource::isOpen() 
{ 
    return ((this->m_is_open > 0) ? 1 : 0); 
}    

/* ---------------------------------------------------------------------- */
int phNetSource::isConnected() 
{ 
    phFUNCTION("phNetSource::isConnected")
    int locked = 0;
    int retrc = 0;

    phTHIS_LOCK(locked);

    if (this->m_info != NULL)
    {
        if (this->m_connected_type == phNetTypeTCP)
        {
            if (this->m_info->m_comms->isConnected()) retrc = 1;
        }
    }
    
    phTHIS_UNLOCK_RET(locked,retrc,phFAIL);
}    


/* ---------------------------------------------------------------------- */
int phNetSource::close()
{
    phFUNCTION("phNetSource::close")
    int locked = 0;

    phTHIS_LOCK(locked);

    if (this->isRunning())
    {
        rc = this->stop();
        phPRINT_RC(rc,NULL,"this->stop()");
    }

    if (this->m_is_open)
    {
        rc = this->disconnect();
        phCHECK_RC(rc,NULL,"this->disconnect");

        this->m_is_open = 0;
    }

    phTHIS_UNLOCK_RET(locked,phSUCCESS,phFAIL);
}

/* ---------------------------------------------------------------------- */
int phNetSource::connect( bool poll )
{
    phFUNCTION("phNetSource::connect")
    int locked = 0;
    int retrc = phSUCCESS;

    phTHIS_LOCK(locked);
 
    if (this->m_is_open && (!this->isConnected()))
    {
        if (this->m_type == phNetTypeTCP)
        {
            this->m_connected_type = phNetTypeTCP;
            
            if (this->m_info->m_comms != NULL)
            {
                phDelete(this->m_info->m_comms);
            }
            this->m_info->m_comms = new phSocket();
            this->m_info->m_comms->setWarningLevel(0);

            rc = this->tcp_connect(this->m_info->m_comms,
                                   poll,
                                   this->m_info->m_host,
                                   this->m_info->m_port,
                                   &(this->m_info->m_width),
                                   &(this->m_info->m_height),
                                   &(this->m_info->m_format),
                                   &(this->m_info->m_title)
                                   );
            retrc = rc;
            /*phCHECK_RC(rc,NULL,"this->tcp_connect()");*/
        }
    }

    phTHIS_UNLOCK_RET(locked,retrc,phFAIL);
}

/* ---------------------------------------------------------------------- */
int phNetSource::getNextFrame(phImage  *image,
                              uint8_t  **image_buf,
                              uint32_t *image_size )
{
    phFUNCTION("phNetSource::getNextFrame")
    int locked = 0;

    phTHIS_LOCK(locked);

    if (this->m_is_open && (this->isConnected()))
    {
        if (this->m_connected_type == phNetTypeTCP)
        {
            /* Get the next buffer */
            /* Get next frame */
            rc = this->tcp_recv(this->m_info->m_comms,
                                image,
                                &this->m_info->m_width,
                                &this->m_info->m_height,
                                &this->m_info->m_format,
                                image_buf,
                                image_size);
            phSocket_CHECK_RC(rc,NULL,"this->tcp_recv");
        }
    }
    phTHIS_UNLOCK_RET(locked,phSUCCESS,rc);
}
/* ---------------------------------------------------------------------- */
int phNetSource::disconnect( )
{
    phFUNCTION("phNetSource::disconnect")
    int locked = 0;
    int retrc = phSUCCESS;

    phTHIS_LOCK(locked);
 
    if (this->m_is_open && (this->isConnected()))
    {
        if (this->m_connected_type == phNetTypeTCP)
        {
            rc = this->tcp_disconnect( this->m_info->m_comms );
            phCHECK_RC(rc,NULL,"this->tcp_disconnect()");
        }

        this->m_connected_type = phNetTypeUNKNOWN;
    }

    phTHIS_UNLOCK_RET(locked,retrc,phFAIL);
}

/* ------------------------------------------------------------------------- */
int phNetSource::tcp_connect( phSocket  *comms,
                              bool      poll,
                              char      *host,
                              uint32_t  port,
                              uint32_t  *w,
                              uint32_t  *h,
                              uint32_t  *f,
                              char      **title )
{
    phFUNCTION("phNetSource::tcp_connect")
    int failrc = phFAIL;
  
    char *serverTitle = NULL;
    
    /*
    phPROGRESS("Attempting to connect to server - %s:%d\n",
             host,port );
    */
    /* connect to the server */
    rc = comms->connect(host, port, poll);
    if (!comms->isConnected() || (rc != 0))
    {
        failrc = rc;
        goto error;
    }
    /*phCHECK_RC(rc,NULL,"Failed to connect to %s\n\n", host);*/
    
    /*
    phPROGRESS("Connected to server (rc:%d connected:%d)- %s:%d\n",
             rc,comms->isConnected(),host,port );
     */
    rc = this->tcp_init(comms,&serverTitle,w,h,f);
    phCHECK_RC(rc,NULL,"tcp_init");
    
    phFree(*title);
    
    (*title) = (char *)phCalloc(strlen(serverTitle) + 
                                strlen(host) +
                                255,
                                sizeof(char));
    phCHECK_NULLPTR(title,"phCalloc","phCalloc failed to allocate title.");

    sprintf((*title),"%s:%d - %s",
            host,
            port,
            serverTitle );
    
    phFree(serverTitle);

    return phSUCCESS;
error:
    phFree(serverTitle);
    return failrc;
}

/* ------------------------------------------------------------------------- */
int phNetSource::tcp_init( phSocket     *comms,
                           char         **title, 
                           uint32_t     *w, 
                           uint32_t     *h, 
                           uint32_t     *f )
{
    phFUNCTION("phNetSource::tcp_init")

    uint32_t width = 0;
    uint32_t height = 0;
    uint32_t format = phImageNOFORMAT;
    uint32_t titleSize = 0;
    uint32_t next_image  = 0xaaaa;
    
    rc = comms->recv_uint32(&titleSize);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive title size\n\n");

    phFree(*title);
    
    *title = (char *)phMalloc(titleSize+5);
    phCHECK_NULLPTR(*title,"phMalloc","phMalloc failed to allocate the title space.");
    phMemset(*title,0,titleSize+5);
    
    rc = comms->recv(*title, titleSize);
    DEBUG_PRINT("title:%s\n",*title);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive title size\n\n");

    /* receive image dimensions from server */
    rc = comms->recv_uint32(&width);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive image dimensions(width)\n\n");

    rc = comms->recv_uint32(&height);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive image dimensions(height)\n\n");

    rc = comms->recv_uint32(&format);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive image format \n\n");
    
    *w = width;
    *h = height;
    *f = format;
    
    return phSUCCESS;
    
error:
    return rc;
}

/* ------------------------------------------------------------------------- */
int phNetSource::tcp_recv(  phSocket     *comms,
                            phImage      *inputImage,
                            uint32_t     *w, 
                            uint32_t     *h, 
                            uint32_t     *f,
                            uint8_t      **data,
                            uint32_t     *data_size
                          )
{
    phFUNCTION("phNetSource::tcp_recv")

    uint32_t    width           = 0;
    uint32_t    height          = 0;
    uint32_t    format          = (uint32_t)phImageNOFORMAT;
    uint32_t    size            = 0;

    phTimeStamp time_start;
    phTimeStamp time_stop;
    uint32_t    host_send_time  = 0;
    int32_t     packet_id       = 0xaaaa;

    time_start.stamp();

    /* receive image dimensions from server */
    rc = comms->recv_uint32(&width);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive image dimensions(width)\n\n");
    //phPRINT("w:%u\n",width);

    rc = comms->recv_uint32(&height);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive image dimensions(height)\n\n");
    //phPRINT("h:%u\n",height);

    rc = comms->recv_uint32(&format);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive image format \n\n");
    //phPRINT("f:%u\n",format);

    rc = comms->recv_uint32(&host_send_time);
    phCHECK_RC(rc,NULL,"Failed to receive image host_send_time \n\n");
    //phPRINT("hst:%d\n",host_send_time);

    /* 0.)  packet_id */
    rc = comms->send_int32(packet_id);
    phSocket_CHECK_RC(rc,NULL,"Failed to send packet_id image message \n\n");
    /* 1.) host_send_time: time the server started sending the data */
    rc = comms->send_int32(host_send_time);
    phSocket_CHECK_RC(rc,NULL,"Failed to send host_send_time image message \n\n");

    rc = comms->recv_uint32(&size);
    phSocket_CHECK_RC(rc,NULL,"Failed to receive image size \n\n");
    //phPRINT("s:%u\n",size);

    /* receive a image from the server */
    DEBUG_PRINT("receiving image data\n");

    if (size == 0) /* Try to calculate it */
    {
        size = phIMAGE_SIZE(width,height,format);
    }
    
    /* allocate data if it hasn't been initialized yet */
    if (((*data_size) == 0) || (*data == NULL))
    {
        (*data) = (uint8_t *)phMalloc(size);
        phCHECK_NULLPTR(*data,"phMalloc","phMalloc failed to allocate data.");
    }
    /* reallocate the data if the size changes */
    else if (((*data_size) != size) && (*data != NULL))
    {
        (*data) = (uint8_t *)phRealloc((*data),size);
        phCHECK_NULLPTR(*data,"phRealloc","phRealloc failed to allocate data.");
    }
       
    /* Always set this */
    *data_size = size;
    
    /* Get the data from the server */
    rc = comms->recv(*data, *data_size);
    phSocket_CHECK_RC(rc,NULL,"recv for image data failed.");

    time_stop.stamp();
    
    /* Send stats back to the server */
    /* 2.) */
    rc = comms->send_uint32(*data_size);
    phSocket_CHECK_RC(rc,NULL,"Failed to send message \n\n");
    /* 3.) */
    rc = comms->send_uint32(time_start.getMicroseconds());
    phSocket_CHECK_RC(rc,NULL,"Failed to send message\n\n");
    /* 4.) */
    rc = comms->send_uint32(time_stop.getMicroseconds());
    phSocket_CHECK_RC(rc,NULL,"Failed to send message\n\n");
    
#if 0
    phPRINT("0:%8u:%12.lf 1:%8u:%12.lf\n",
            time_start.getMicroseconds(),
            time_start.getStampValue(),
            time_stop.getMicroseconds(),
            time_stop.getStampValue());
#endif
#if phNetSource_PRINT_IMAGE_SIZE()
    phPROGRESS("format:%s\n",
               phImageFormatToString(format));
    phPROGRESS("recv:%lu\tactual:%lu\n",
               (long unsigned)*data_size,
               (long unsigned)phIMAGE_SIZE(width,height,format));
#endif

    /* allocate space for the image */
    inputImage->setImage(width,height,format,
                        *data_size,
                        *data);
    
    /* keep track */
    *w = width;
    *h = height;
    *f = format;
    
    return phSUCCESS;
error:
    /* might as well... */
    phFree(*data);
    
    *data_size = 0;
    
    return rc;
}

/* ------------------------------------------------------------------------- */
int phNetSource::tcp_disconnect( phSocket  *comms )
{
    phFUNCTION("phNetSource::tcp_disconnect")
  
    /* connect to the server */
    comms->disconnect();
    
    return phSUCCESS;
error:
    return phFAIL;
}

/* ---------------------------------------------------------------------- */
int phNetSource::wakeup()
{
    phFUNCTION("phNetSource::wakeup")

    if (this->m_info != NULL)
    {
        if (this->m_info->m_comms != NULL)
        {
            this->m_info->m_comms->wakeup();
        }
    }

    rc = phImageCapture::wakeup();
    phPRINT_RC(rc,NULL,"phImageCapture::wakeup");

    return phSUCCESS;
}

/* ---------------------------------------------------------------------- */
int phNetSource::cleanup()
{
    phFUNCTION("phNetSource::cleanup")
    int locked = 0;
   
    phTHIS_LOOSE_LOCK(locked);
   
    rc = this->close();
    phPRINT_RC(rc,NULL,"this->close()");
    
    rc = this->phImageCapture::cleanup();
    phPRINT_RC(rc,NULL,"this->phImageCapture::cleanup()");
    
    phTHIS_LOOSE_UNLOCK(locked);
    
    return phSUCCESS;
}

/* ---------------------------------------------------------------------- */
int phNetSource::run()
{
    phFUNCTION("phNetSource::run")
    int i = 0;
    int foo = 0;
    int orig_locked = 0;
        
    /* Returned value from getNextFrame and a loop count */
    phImage    *output_buffer   = this->getImage();
    
    /* local copies of phNetSource object's member vars */
    phImage    *image           = new phImage();
    uint8_t    *image_buf       = NULL;
    uint32_t    image_size      = 0;
    int32_t     local_format    = 0;
    uint32_t    local_width     = 0;
    uint32_t    local_height    = 0;
    int         local_do_resize = 0;
    int         local_do_convert= 0;

    int         slept               = 0;
#if phNetSource_USE_JITTER_SYNC()
    int32_t     sleeptime           = 0;
    int32_t     client_jitter_avg   = -1;
    int32_t     notify_jitter_avg   = -1;
#endif
    
    /* Setup */
    local_format      = this->m_format;
    local_width       = this->m_width;
    local_height      = this->m_height;
    local_do_resize   = this->m_info->m_do_resize;
    local_do_convert  = this->m_info->m_do_convert;
   
    /* Release the thread that spawned this one */
    /* Everything has been copied and everything is ready to start */
    rc = this->signal_running();
    phPRINT_RC(rc,NULL,"this->signal_running");

    /* if we haven't connected by now and we want to wait, then wait */
    while (this->m_server_wait && 
           (!this->isConnected()) &&
           (this->isRunning()))
    {
        rc = this->connect( false );
        
        if ((rc != -5) && (rc < 0)) this->setRunning(0);

        if (!(foo = this->isConnected()))
        {
            phUSleep(10);
            //phYield();
        }
    }
    
    DEBUG_PRINT("Thread Started\n");
    DEBUG_PRINT("output_buffer: %p\n",output_buffer);
    
#if phNetSource_USE_JITTER_SYNC()
    /* Reset all the jitter values in the object:
       notify, client and client avg*/
    rc = output_buffer->resetJitter();
    phPRINT_RC(rc,NULL,"output_buffer->resetJitter()");
    
    /* We want a moving average over 10 samples */
    rc = output_buffer->setNotifyJitterAvgSamples(10);
    phPRINT_RC(rc,NULL,"output_buffer->setNotifyJitterAvgSamples()");
    
    /* We want a moving average over 10 samples */
    rc = output_buffer->setClientJitterAvgSamples(10);
    phPRINT_RC(rc,NULL,"output_buffer->setClientJitterAvgSamples()");
#endif
    
    while (this->isRunning() && 
           (this->isConnected() || (this->m_reconnect)))
    {
        /* Get the current width/height/format settings to resize the 
         * input to for output to connected objects */
        rc = this->m_settings_lock.trylock();
        phPRINT_RC(rc,NULL,"this->m_settings_lock.trylock();");

        /* only copy the settings if we have exclusize access */
        if (rc == phSUCCESS)
        {
            local_do_resize   = this->m_info->m_do_resize;
            local_do_convert  = this->m_info->m_do_convert;
            local_width       = this->m_width;
            local_height      = this->m_height;
            local_format      = this->m_format;

            rc = this->m_settings_lock.unlock();
            phPRINT_RC(rc,NULL,"this->m_settings_lock.unlock()");
        }
                
        if (!this->isConnected())
        {
            if (this->m_reconnect)
            {
                rc = this->connect(false);

                if (!this->isConnected())
                {
                    phUSleep(10);
                    //phYield();
                }
                
                continue;
            }
            else 
            {
                continue;
            }
        }
        
        if (!this->isRunning()) 
        {
            continue;
        }

        rc = this->getNextFrame(image,&image_buf,&image_size);
        phPRINT_RC(rc,NULL,"this->recv()");

        /* If the recv failed, then disconnect;
         * this->m_reconnect above will try to reconnect*/
        if (rc < 0)
        {
            this->disconnect();
            continue;
        }
        
        /* If there's an error, then go to the beginning
         * of the loop. There's a potential we can reset ourselves */
        if (rc != phSUCCESS) continue;

        /* Convert to the output format */
        /* Setup the image according to desired output format */
        if (local_do_resize)
        {
            rc = image->resize(local_width,local_height);
            phPRINT_RC(rc,NULL,"image->resize(%d,%d)",
                    local_width,local_height);
        }
        if (local_do_convert)
        {
            rc = image->convert(local_format);
            phPRINT_RC(rc,NULL,"image->convert(%s)",
                       phImageFormatToString(local_format));
        }
        
        /* if (output_buffer->tryWritelock() == 0) */
        if (output_buffer->writeLock() == 0)
        {
            rc = output_buffer->swapImage(*image);
            phCHECK_RC(rc,NULL,"output_buffer->swapImage(*image)");
     
            /* Save the current frame's original settings */
            phMUTEX_LOOSE_LOCK(this->m_orig_lock,orig_locked);
            /* \/ ORIGINAL CODE \/ */
            this->m_orig_width  = this->m_info->m_width;
            this->m_orig_height = this->m_info->m_height;
            this->m_orig_format = this->m_info->m_format;
            this->m_orig_size   = image_size;
            phMUTEX_LOOSE_UNLOCK(this->m_orig_lock,orig_locked);
       
#if phNetSource_USE_JITTER_SYNC()
            client_jitter_avg = output_buffer->getClientJitterAvg();
            notify_jitter_avg = output_buffer->getNotifyJitterAvg();
#endif
        
            output_buffer->rwUnlock();

#if phNetSource_USE_JITTER_SYNC()
#if VERBOSE()
            /*phPROGRESS("(%p) <---\n",output_buffer);*/
            phPRINT("(%s) client:%ld notify:%ld\n",
                            function,
                            client_jitter_avg,
                            notify_jitter_avg);
#endif
            /* We check to see if the client jitter is greater than the notify
             * jitter so we slow down the sending of images from the other side
             * to synch. up with how fast images are being processed on this
             * side */
            if (notify_jitter_avg < client_jitter_avg)
            {
                sleeptime = (client_jitter_avg - notify_jitter_avg);
#if VERBOSE()
                phPRINT("(%s)sleeptime:%ld client:%d notify:%d\n",
                            function,
                            sleeptime,
                            client_jitter_avg,
                            notify_jitter_avg);
#endif           
                slept = 1;
                phUSleep(sleeptime);
            }
#endif
        }
        if (!slept)
        {
            if (1)
            {
                phUSleep(0);
            }
            else
            {
                phYield();
            }
        }
        slept = 0;
    }
        
    phDelete(image);
    phFree(image_buf);
    image_size = 0;
    
    DEBUG_PRINT("Thread returning cleanly\n");
   
    return phSUCCESS;
    
error:
    phPROGRESS("Thread returning with error\n");
    
    phDelete(image);
    phFree(image_buf);
    image_size = 0;

    rc = this->signal_error();
    phPRINT_RC(rc,NULL,"this->signal_error");
    
    return phFAIL;
}

Copyright (C) 2002 - 2007

Philip D.S. Thoren ( pthoren@users.sourceforge.net ) University Of Massachusetts at Lowell Robotics Lab

---