ISISDS.cpp

Go to the documentation of this file.

#include "stdafx.h"
#include "isisds_command.h"
#include "ISISDS.h"

//#define NO_CDAE
//#include "..\isisicp\dae.h"
//#include "..\isisicp\isisicp_i.c"

#include "..\newicp\isiscrpt.h"
#include "..\newicp\isisraw.h"

ISISDS* m_ds = NULL;

ISISDS::ISISDS() : ThreadedClass(), ISIS::Base()
{
        setLoggerName("ISISDS");
}

ISISDS::~ISISDS()
{
// call stop and delete on m_requests
}

// where the running m_thread ends up
void ISISDS::run()
{
    HRESULT hr;
        // allow things elsewhere to start
        Sleep(2000);
        int setkeepalive = 1;
    // wait for ICP global section
        if (m_crpt.mapCRPT("isis_current", "isis_data", true, true) == DAEstatus::Failure)
        {
                return;
        }
        WSADATA WsaDat;
        if (WSAStartup(MAKEWORD(2, 2), &WsaDat) != 0)
        {
                return;
        }
        SOCKET s = socket(PF_INET, SOCK_STREAM, 0);
        if (s == INVALID_SOCKET)
        {
                perror("socket");
                return;
        }
    hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
    if (FAILED(hr))
    {
           return;
    }
        try 
        {
            hr = m_dae.CoCreateInstance(isisicpLib::CLSID_dae, NULL, CLSCTX_INPROC_SERVER|CLSCTX_INPROC_HANDLER|CLSCTX_LOCAL_SERVER);
        }
        catch(...)
        {
                return;
        }
        if (FAILED(hr))
        {
                return;
        }
        struct sockaddr_in sin, sad_from;
        memset(&sin, 0, sizeof(sin));
        struct in_addr the_addr;
        the_addr.S_un.S_addr = INADDR_ANY;
        sin.sin_family = PF_INET;
        sin.sin_addr = the_addr;
        sin.sin_port = htons(ISISDS_PORT);
        if (bind(s, (struct sockaddr*)&sin, sizeof(sin)) == -1)
        {
                perror("bind");
                closesocket(s);
                return;
        }
        if (listen(s, 5) == -1)
        {
                perror("listen");
                closesocket(s);
                return;
        }
        // may need to disable SIGPIPE - might get this on external connection close
        int n, len_sadfrom;
        const char* address;
        fd_set readfds;
        SOCKET sd;
        std::list<ISISDSRequest*>::iterator iter;
        while(1)
        {
                for(iter=m_requests.begin(); iter != m_requests.end(); iter++)
                {
                        if ( !((*iter)->isRunning()) )
                        {
//                              std::cerr << "Terminating from " << (*iter)->address() << std::endl;
                                delete (*iter);
                                m_requests.erase(iter);
                                break; // loop seems to die unless we exit after erase()
                        }
                }
                FD_ZERO(&readfds);
                FD_SET(s, &readfds);
                struct timeval select_timeout = { 5, 0 }; 
                n = select(FD_SETSIZE, &readfds, NULL, NULL, &select_timeout);
                if (n < 0)
                {
                        perror("select");
                        return; // select failed, socket invalid?
                }
                else if (n == 0) // select timeout
                {
                        try 
                        {
                            hr = m_dae->areYouThere();
                        }
                        catch(...)
                        {
                                return;
                        }
                        if (FAILED(hr))
                        {
                                return;
                        }
                        continue;
                }
                if (FD_ISSET(s, &readfds))
                {
                        len_sadfrom = sizeof(sad_from);
                        sd = accept(s, (struct sockaddr*)&sad_from, &len_sadfrom);
                        if (sd == INVALID_SOCKET)
                        {
                                Poco::Thread::sleep(2000);
                                continue;
                        }
                        setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (char*)&setkeepalive, sizeof(setkeepalive));
                        address = inet_ntoa(sad_from.sin_addr);
                        LOGSTR_DEBUG("Connection from " << address);
                        ISISDSRequest* req = new ISISDSRequest(sd, address, m_crpt);
                        m_requests.push_back(req);
                        req->start();
                }
        }
}

void ISISDSRequest::run()
{
        int dims_array_out[10], ndims_out;
    HRESULT hr;
        ISISDSDataType type_out = ISISDSUnknown;
    hr = CoInitializeEx(NULL, COINIT_MULTITHREADED);
    if (FAILED(hr))
    {
           return;
    }
    char *command_in, *comm_data_in, *comm_data_out;
        bool free_out;
        int comm_data_len_out, istat;
        ISISDSAccessMode access_type;
        if (isisds_recv_open(m_socket, &access_type) <= 0)
        {
                return;
        }
        try
        {
            hr = m_dae.CoCreateInstance(isisicpLib::CLSID_dae, NULL, CLSCTX_INPROC_SERVER|CLSCTX_INPROC_HANDLER|CLSCTX_LOCAL_SERVER);
        }
        catch(...)
        {
            return;
    }
        if (FAILED(hr))
        {
                return;
        }
        int ndims_in, data_dims_in[10];
        ISISDSDataType data_type_in;
        while(true)
        {
                if (isisds_recv_command_alloc(m_socket, &command_in, (void**)&comm_data_in, &data_type_in, data_dims_in, &ndims_in) <= 0)
                {
                        break;
                }
                if (processCommand(access_type, command_in, comm_data_in, 0,
                                                        comm_data_out, type_out, dims_array_out, ndims_out, free_out))
                {
                        istat = isisds_send_command(m_socket, "OK", comm_data_out, type_out, dims_array_out, ndims_out);
                }
                else
                {
                        istat = isisds_send_command(m_socket, "ERROR", comm_data_out, type_out, dims_array_out, ndims_out);
                }
                free(command_in);
                if (comm_data_in != NULL)
                {
                        free(comm_data_in);
                }
                if (free_out)
                {
                        free(comm_data_out);
                }
                if (istat <= 0)
                {
                        break;
                }
        }
        return;
}

// return true on OK
// access_type = 0 for dae, 1 for crpt
// need to use malloc/free for comm_data_out
bool ISISDSRequest::processCommand(ISISDSAccessMode access_type, const char* command_in, const char* comm_data_in, int comm_data_len_in,
                                                                   char*& comm_data_out, ISISDSDataType& type_out, int* dims_array_out, int& ndims_out, bool& free_out)
{
        DAEstatus status;
        std::string cvalue;
        long lvalue;
        double dvalue, *darray;
        float *farray;
        long *larray;
        HRESULT hr;
//      VARIANT result;
//      BSTR item_name, messages;
//      hr = m_dae->getValue(item_name, &messages, &result);
        free_out = false;
        comm_data_out = NULL;
        int i, total_size;
        ndims_out = 1;
        dims_array_out[0] = 0;
        type_out = ISISDSUnknown;
        if ( !strcmp(command_in, "GETPARC08") && 
                                (m_crpt->getCharItem(comm_data_in, cvalue, status) == DAEstatus::Success) )
        {
                comm_data_out = strdup(cvalue.c_str());
                dims_array_out[0] = strlen(comm_data_out);
                free_out = true;
                type_out = ISISDSChar;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" for item \"" << comm_data_in << "\" returning \"" << comm_data_out << "\"");
                return true;
        }
        else if ( !strcmp(command_in, "GETPARI32") && 
                           (m_crpt->getIntItem(comm_data_in, lvalue, status) == DAEstatus::Success) )
        {
                dims_array_out[0] = 1;
                comm_data_out = (char*)malloc(sizeof(int));
                *((int*)comm_data_out) = lvalue;
                free_out = true;
                type_out = ISISDSInt32;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" for item \"" << comm_data_in << "\" returning " << lvalue);
                return true;
        }
        else if (!strcmp(command_in, "GETPARR32") && 
                                        (m_crpt->getRealItem(comm_data_in, dvalue, status) == DAEstatus::Success) )
        {
                dims_array_out[0] = 1;
                comm_data_out = (char*)malloc(sizeof(float));
                *((float*)comm_data_out) = dvalue;
                free_out = true;
                type_out = ISISDSReal32;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" for item \"" << comm_data_in << "\" returning " << dvalue);
                return true;
        }
        else if (!strcmp(command_in, "GETPARR64") && 
                                        (m_crpt->getRealItem(comm_data_in, dvalue, status) == DAEstatus::Success) )
        {
                dims_array_out[0] = 1;
                comm_data_out = (char*)malloc(sizeof(double));
                *((double*)comm_data_out) = dvalue;
                free_out = true;
                type_out = ISISDSReal64;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" for item \"" << comm_data_in << "\" returning " << dvalue);
                return true;
        }
        else if (!strcmp(command_in, "GETPARR64") && 
                        (m_crpt->getRealArrayItemSize(comm_data_in, dims_array_out, ndims_out, status) == DAEstatus::Success) )
        {
                total_size = 1;
                for(i=0; i<ndims_out; i++)
                {
                        total_size *= dims_array_out[i];
                }
                darray = (double*)malloc(total_size * sizeof(double));
                m_crpt.getRealArrayItem(comm_data_in, darray, status);
                comm_data_out = (char*)darray;
                free_out = true;
                type_out = ISISDSReal64;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" for item \"" << comm_data_in << "\" returning array length " << total_size);
                return true;
        }
        else if (!strcmp(command_in, "GETPARR32") && 
                        (m_crpt->getRealArrayItemSize(comm_data_in, dims_array_out, ndims_out, status) == DAEstatus::Success) )
        {
                total_size = 1;
                for(i=0; i<ndims_out; i++)
                {
                        total_size *= dims_array_out[i];
                }
                farray = (float*)malloc(total_size * sizeof(float));
                m_crpt.getRealArrayItem(comm_data_in, farray, status);
                comm_data_out = (char*)farray;
                free_out = true;
                type_out = ISISDSReal32;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" for item \"" << comm_data_in << "\" returning array length " << total_size);
                return true;
        }
        else if ( !strcmp(command_in, "GETPARI32") &&
                                (m_crpt->getIntArrayItemSize(comm_data_in, dims_array_out, ndims_out, status) == DAEstatus::Success) )
        {
                total_size = 1;
                for(i=0; i<ndims_out; i++)
                {
                        total_size *= dims_array_out[i];
                }
                larray = (long*)malloc(total_size * sizeof(long));
                m_crpt.getIntArrayItem(comm_data_in, larray, status);
                comm_data_out = (char*)larray;
                free_out = true;
                type_out = ISISDSInt32;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" for item \"" << comm_data_in << "\" returning array length " << total_size);
                return true;
        }
        else if ( !strcmp(command_in, "GETDAT") )
        {
                int dims_temp[5], ndims_temp;
                int ifsn = ((int*)comm_data_in)[0];
                int nos = ((int*)comm_data_in)[1];
                long* spec_array = new long[nos];
                for(i=0; i<nos; i++)
                {
                        spec_array[i] = ifsn + i;
                }
                BSTR messages = NULL;
                try
                {
                        hr = m_dae->updateCRPTSpectra(0, ifsn, nos, &messages);    // update dae for current period, but (ifsn > nsp1) will do second and subsequent periods
                }
                catch( ... )
                {
                        LOGSTR_ERROR("updateCRPTSpectra failed");
                        status.addInfo(FAC_DAE, "updateCRPTSpectra failed");
                }
                if ( FAILED(hr) )
                {
                        LOGSTR_ERROR("updateCRPTSpectra failed");
                        status.addInfo(FAC_DAE, "updateCRPTSpectra failed");
                }
                m_crpt->getIntArrayItemSize("SPECTRUMT0", spec_array, nos, dims_temp, ndims_temp, status);
                total_size = dims_temp[0] * nos;
//              ndims_out = 2;
//              dims_array_out[0] = nos;
//              dims_array_out[1] = dims_temp[0]; // spectrum size
                ndims_out = 1;
                dims_array_out[0] = total_size;
                larray = (long*)malloc(total_size * sizeof(long));
                m_crpt.getIntArrayItem("SPECTRUMT0", spec_array, nos, larray, status);
                comm_data_out = (char*)larray;
                delete[] spec_array;
                free_out = true;
                type_out = ISISDSInt32;
                LOGSTR_DEBUG("Processing command \"" << command_in << "\" ifsn " << ifsn << " nos " << nos << " returning array length " << total_size);
                return true;
        }
        LOGSTR_ERROR("Invalid command " << command_in);
        return false;
}