redundancymanager.cxx

SIP（Session Initiation Protocol）是由IETF定义

/* ====================================================================
 * The Vovida Software License, Version 1.0 
 * 
 * Copyright (c) 2000 Vovida Networks, Inc.  All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 * 
 * 3. The names "VOCAL", "Vovida Open Communication Application Library",
 *    and "Vovida Open Communication Application Library (VOCAL)" must
 *    not be used to endorse or promote products derived from this
 *    software without prior written permission. For written
 *    permission, please contact [e-mail].
 *
 * 4. Products derived from this software may not be called "VOCAL", nor
 *    may "VOCAL" appear in their name, without prior written
 *    permission of Vovida Networks, Inc.
 * 
 * THIS SOFTWARE IS PROVIDED "AS IS" AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, TITLE AND
 * NON-INFRINGEMENT ARE DISCLAIMED.  IN NO EVENT SHALL VOVIDA
 * NETWORKS, INC. OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT DAMAGES
 * IN EXCESS OF $1,000, NOR FOR ANY INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
 * OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 * USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
 * DAMAGE.
 * 
 * ====================================================================
 * 
 * This software consists of voluntary contributions made by Vovida
 * Networks, Inc. and many individuals on behalf of Vovida Networks,
 * Inc.  For more information on Vovida Networks, Inc., please see
 * <http://www.vovida.org/>.
 *
 */

static const char* const RedundancyManager_cxx_Version =
    "$Id: RedundancyManager.cxx,v 1.7 2002/09/26 01:14:03 bko Exp $";

#include <string.h>
#include <stdio.h>
#include <map>

#include "NetworkAddress.h"
#include "Tcp_ClientSocket.hxx"
#include "Vpp_def.h"
#include "cpLog.h"
#include "Sptr.hxx"
#include "VNetworkException.hxx"
#include "LocalScopeAllocator.hxx"
#include "RedundancyManager.hxx"
#include "PSNetworkClient.hxx"
#include "PSSecret.hxx"

static bool block = false;

RedundancyManager::RedundancyManager()
{
    _state = DNSYNC;
    _use = false;

    _host = "";
    _port = 0;
}


void
RedundancyManager::initialize(const string &host, int port, bool useTls)
{

    LockHelper lockAid( this->_aMutex );

    _use = true;

    _host = host;
    _port = port;
    myUseTls = useTls;

    _networkAddress = new NetworkAddress(_host);
    _networkAddress->setPort(_port);

    cpLog(LOG_DEBUG, "Using redundant server at %s:%i", _host.c_str(), _port);

}


// All of these are state transition functions
int
RedundancyManager::heartBeatGone()
{

    LockHelper lockAid( this->_aMutex );

    assert(checkState());

    _state = DNSYNC;  // now dead

    return NMSG;

}


int
RedundancyManager::heartBeat()
{

    LockHelper lockAid( this->_aMutex );

    if (_state != DNSYNC)
    {
        assert(checkState());
        return NMSG;
    }
    _state = ANSYNC;  // now alive, not synced

    return OK;

}


int
RedundancyManager::connFailed()
{

    LockHelper lockAid( this->_aMutex );

    assert(checkState());

    if (_state == DNSYNC)
    {
        return ERR;
    }
    _state = DNSYNC;

    return NMSG;

}


int
RedundancyManager::abortSync()
{

    LockHelper lockAid( this->_aMutex );

    assert(checkState());

    _state = ANSYNC;

    return NMSG;

}


int
RedundancyManager::timedoutSync()
{

    LockHelper lockAid( this->_aMutex );

    assert(checkState());

    // this is the abort
    if (_state == ASLAVE)
    {
        _state = ANSYNC;
        return FAIL;
    }

    return NMSG;

}

int
RedundancyManager::badReturn()
{

    LockHelper lockAid( this->_aMutex );

    assert(checkState());

    if (_state == DNSYNC)
    {
        return ERR;
    }
    _state = ANSYNC;

    return NMSG;

}


int
RedundancyManager::syncStart()
{

    LockHelper lockAid( this->_aMutex );

    if (_state == ANSYNC)
    {
        _state = ASLAVE;  // switch to slave and start sync
        return OK;
    }
    if ((_state == AMASTER) || (_state == ASLAVE) || (_state == ASYNC))
    {
        _state = ANSYNC;  // something weird, abort and resync
        return FAIL;
    }
    if (_state == DNSYNC)
    {
        return FAIL;  // stay dead, send fail (this is a really strange state)
    }

    _state = DNSYNC;

    return ERR;

}


int
RedundancyManager::startSync()
{

    LockHelper lockAid( this->_aMutex );

    // if we are alive and unsynched, we want to initiate a sync,
    // else do nothing
    if (_state == ANSYNC)
    {
        _state = AMASTER;  // switch to master and start sync
        return OK;
    }
    return FAIL;

}


int
RedundancyManager::syncGood()
{

    LockHelper lockAid( this->_aMutex );

    if (_state == ASLAVE)
    {
        _state = ASYNC;  // all is good -- go to synched
        return OK;
    }
    if ((_state == AMASTER) || (_state == ANSYNC) || (_state == ASYNC))
    {
        _state = ANSYNC;  // fail, start sync process over
        return FAIL;
    }
    if (_state == DNSYNC)
    {
        return FAIL;  // message from a "dead" partner -- ignore
    }

    _state = DNSYNC;

    return ERR;

}



int
RedundancyManager::syncExchange()
{

    LockHelper lockAid( this->_aMutex );

    // we got a message from the other server
    if ((_state == AMASTER) || (_state == ASYNC) || (_state == ASLAVE))
    {
        return OK;
    }

    if ((_state == ANSYNC))
    {
        // we should not get messages for sync until we are syncing or
        // we are synched
        return FAIL;
    }

    if (_state == DNSYNC)
    {
        return FAIL;  // the dead speak!
    }


    _state = DNSYNC;
    return ERR;

}


int
RedundancyManager::syncRegList()
{

    LockHelper lockAid( this->_aMutex );

    assert(checkState());

    if (_state == DNSYNC)
    {
        _state = DNSYNC;
        return ERR;
    }

    // we got a message from the other server
    if (_state == ASLAVE)
    {
        return OK;
    }

    _state = ANSYNC;
    return ERR;

}


bool
RedundancyManager::isAlive ()
{

    LockHelper lockAid( this->_aMutex );

    return (_state & ALIVE);

}


bool
RedundancyManager::isSync ()
{

    LockHelper lockAid( this->_aMutex );

    return (_state & SYNC);

}


bool
RedundancyManager::isMaster ()
{

    LockHelper lockAid( this->_aMutex );

    return (_state & MASTER);

}


bool
RedundancyManager::isSlave ()
{

    LockHelper lockAid( this->_aMutex );

    return (_state & SLAVE);

}


bool
RedundancyManager::checkState()
{

    return (_state == ASYNC ||
            _state == DNSYNC ||
            _state == ANSYNC ||
            _state == ASLAVE ||
            _state == AMASTER);

}


int
RedundancyManager::getRemoteMap(map < string, TimeStamp > &remote)
{

    char buf[256], msg[256];
    string sendData;
    int status, len, nread = 0;

    LockHelper lockAid( this->_aMutex );

    // want it empty, 'cause if the other server returns nothing we
    // want an empty list
    remote.clear();

    try
    {

#if 0
        // send request for the informaion
        Sptr < TcpClientSocket > clientSocket = 0;
        clientSocket = new TcpClientSocket(*_networkAddress);
        clientSocket->connect();
#endif
        PSNetworkClient clientSocket(*_networkAddress, true, block, myUseTls);
        Connection& conn = clientSocket.getConn(PSSecret::writeSecret());


        sprintf(buf, "SYNCSTART %s\n", VPP_VERSION);

        Data result;
        int len;
        Data data;

        sendCommand(conn, buf, result, len, data, false);

        if(result == "200 OK\n")
        {
            if (len == 0)
            {
                // no data on remote server, so just return OK
                return OK;
            }

            char *token;
            char file[256];
            TimeStamp timestamp;

            char* bigbuf;
            LocalScopeAllocator bufbuf(&bigbuf, data.length() + 2);
            LocalScopeAllocator lo;
            strncpy(bigbuf, data.getData(lo), data.length() + 2);

            // return list is filename,timestamp,..., (ends with ,)
            // should ALWAYS be in pairs.
            token = strtok(bigbuf, ",");
            sscanf(token, "%s", file);
            token = strtok(NULL, ",");
            if (token == NULL)
            {
                cpLog(LOG_ERR, "Parse error on socket - closing");
                return ERR;
            }
            sscanf(token, "%lu", &timestamp);
            
            // add timestamp to map indexed by filename
            remote[string(file)] = timestamp;
            
            // got first pair, no do all the others...
            while ((token = strtok(NULL, ",")) != NULL)
            {
                sscanf(token, "%s", file);
                cpLog(LOG_DEBUG, "File is %s", file);
                token = strtok(NULL, ",");
                if (token == NULL)
                {
                    cpLog(LOG_ERR, "Parse error on socket - closing");
                    return ERR;
                }
                sscanf(token, "%lu", &timestamp);
                remote[file] = timestamp;
                
            }

            return OK;
        }
        else
        {
            cpLog(LOG_DEBUG, "SYNCSTART File returned FAILED");
            return FAIL;
        }
    }
    catch (VNetworkException &e)
    {
        cpLog(LOG_DEBUG, "Failed to send SYNCSTART to %s:%d.",
              _host.c_str(), _port);
        return ERR;
    }

}


int
RedundancyManager::writeItem(const string &group, const string &name,
                           const string &contents,
                           TimeStamp timestamp)
{

    char buf[256], msg[256];
    string sendData;
    int status, nread = 0;

    LockHelper lockAid( this->_aMutex );

    try
    {
#if 0
        Sptr < TcpClientSocket > clientSocket = 0;
        clientSocket = new TcpClientSocket(*_networkAddress);
        clientSocket->connect();
#endif
        PSNetworkClient clientSocket(*_networkAddress, true, block, myUseTls);
        Connection& conn = clientSocket.getConn(PSSecret::writeSecret());

        // format and send message
        sprintf(buf, "SYNCPUT %s %s %s\n", group.c_str(), name.c_str(),
                VPP_VERSION);
        sendData += buf;
        sprintf(buf, "Timestamp: %lu\n", timestamp);
        sendData += buf;
        sprintf(buf, "%s %d\n", CONT_LENGTH, contents.size());
        sendData += buf;
        sendData += contents;

        Data result;
        int len;
        Data data;

        sendCommand(conn, sendData.c_str(), result, len, data, false);

        if(result == "200 OK\n")
        {
            // if ok, just return ok
            return OK;
        }
        else
        {
            cpLog(LOG_DEBUG, "SYNCPUT File returned FAILED");
            return FAIL;
        }
    }
    catch (VNetworkException &e)
    {
        cpLog(LOG_DEBUG, "Failed to write to redundant server %s:%d.",
              _host.c_str(), _port);