dnxSrvProt.c

Go to the documentation of this file.

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

   Copyright (c) 2006-2010, Intellectual Reserve, Inc. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License version 2 as
   published by the Free Software Foundation.

   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

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

#include "dnxSrvProt.h"
#include "dnxXml.h"
#include "dnxError.h"
#include "dnxDebug.h"

#include <assert.h>
#include <string.h>

//----------------------------------------------------------------------------

int dnxSendJob(DnxChannel * channel, DnxJob * pJob, char * address)
{
   DnxXmlBuf xbuf;

   assert(channel && pJob && pJob->cmd && *pJob->cmd);

   // create the XML message
   dnxXmlOpen (&xbuf, "Job");
   dnxXmlAdd  (&xbuf, "XID",      DNX_XML_XID,  &pJob->xid);
   dnxXmlAdd  (&xbuf, "GUID",     DNX_XML_XID,  &pJob->xid); // old format - for bc
   dnxXmlAdd  (&xbuf, "State",    DNX_XML_INT,  &pJob->state);
   dnxXmlAdd  (&xbuf, "Priority", DNX_XML_INT,  &pJob->priority);
   dnxXmlAdd  (&xbuf, "Timeout",  DNX_XML_INT,  &pJob->timeout);
   dnxXmlAdd  (&xbuf, "Command",  DNX_XML_STR,   pJob->cmd);
   dnxXmlClose(&xbuf);

   dnxDebug(3, "dnxSendJob: XML msg(%d bytes)=%s.", xbuf.size, xbuf.buf);

   // send it on the specified channel
   return dnxPut(channel, xbuf.buf, xbuf.size, 0, address);
}

//----------------------------------------------------------------------------

int dnxWaitForNodeRequest(DnxChannel * channel, DnxNodeRequest * pReg, 
      char * address, int timeout)
{
   DnxXmlBuf xbuf;
   int ret;

   assert(channel && pReg);

   memset(pReg, 0, sizeof *pReg);

   // await a message from the specified channel
   xbuf.size = sizeof xbuf.buf - 1;
   if ((ret = dnxGet(channel, xbuf.buf, &xbuf.size, timeout, address)) != DNX_OK)
      return ret;

   // decode the XML message:
   xbuf.buf[xbuf.size] = 0;
   dnxDebug(3, "dnxWaitForNodeRequest: XML msg(%d bytes)=%s.", xbuf.size, xbuf.buf);

   // verify this is a "NodeRequest" message
   if ((ret = dnxXmlCmpStr(&xbuf, "Request", "NodeRequest")) != DNX_OK)
      return ret;

   // decode the worker node's XID (support older GUID format as well)
   if ((ret = dnxXmlGet(&xbuf, "XID", DNX_XML_XID, &pReg->xid)) != DNX_OK
         && (ret = dnxXmlGet(&xbuf, "GUID", DNX_XML_XID, &pReg->xid)) != DNX_OK)
      return ret;

   // decode request type
   if ((ret = dnxXmlGet(&xbuf, "ReqType", DNX_XML_INT, &pReg->reqType)) != DNX_OK)
      return ret;

   // decode job capacity (support strange mixture of JobCap and Capacity)
   if ((ret = dnxXmlGet(&xbuf, "JobCap", DNX_XML_INT, &pReg->jobCap)) != DNX_OK
         && (ret = dnxXmlGet(&xbuf, "Capacity", DNX_XML_INT, &pReg->jobCap)) != DNX_OK)
      return ret;

   // decode job expiration (Time-To-Live in seconds)
   if ((ret = dnxXmlGet(&xbuf, "TTL", DNX_XML_INT, &pReg->ttl)) != DNX_OK)
      return ret;

   // msg okay - translate address now to save time in logging later
   dnxNtop(address, pReg->addrstr, sizeof pReg->addrstr);

   return DNX_OK;
}

//----------------------------------------------------------------------------

int dnxWaitForResult(DnxChannel * channel, DnxResult * pResult, 
      char * address, int timeout)
{
   DnxXmlBuf xbuf;
   int ret;

   assert(channel && pResult);

   memset(pResult, 0, sizeof *pResult);

   // await a message from the specified channel
   xbuf.size = sizeof xbuf.buf - 1;
   if ((ret = dnxGet(channel, xbuf.buf, &xbuf.size, timeout, address)) != DNX_OK)
      return ret;

   // decode the XML message
   xbuf.buf[xbuf.size] = 0;
   dnxDebug(3, "dnxWaitForResult: XML msg(%d bytes)=%s.", xbuf.size, xbuf.buf);

   // verify this is a "Result" message
   if ((ret = dnxXmlCmpStr(&xbuf, "Request", "Result")) == DNX_OK)
   {

       // decode the result's XID (support older GUID format as well)
       if ((ret = dnxXmlGet(&xbuf, "XID", DNX_XML_XID, &pResult->xid)) != DNX_OK
             && (ret = dnxXmlGet(&xbuf, "GUID", DNX_XML_XID, &pResult->xid)) != DNX_OK)
          return ret;

       // decode the result's state
       if ((ret = dnxXmlGet(&xbuf, "State", DNX_XML_INT, &pResult->state)) != DNX_OK)
          return ret;

       // decode the result's execution time delta
       if ((ret = dnxXmlGet(&xbuf, "Delta", DNX_XML_UINT, &pResult->delta)) != DNX_OK)
          return ret;

       // decode the result's result code
       if ((ret = dnxXmlGet(&xbuf, "ResultCode", DNX_XML_INT, &pResult->resCode)) != DNX_OK)
          return ret;

       // decode the result's result data
       return dnxXmlGet(&xbuf, "ResultData", DNX_XML_STR, &pResult->resData);
   }

}

//----------------------------------------------------------------------------