dnxTransport.c

Go to the documentation of this file.

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

   Copyright (c) 2006-2007, 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 "dnxTransport.h"
#include "dnxTSPI.h"

#include "dnxDebug.h"
#include "dnxError.h"
#include "dnxLogging.h"
#include "dnxProtocol.h"

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <syslog.h>
#include <errno.h>
#include <assert.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#define elemcount(x) (sizeof(x)/sizeof(*(x)))

#define DNX_MAX_CHAN_MAP   1000

typedef struct DnxChanMap_
{
   char * name;         
   char * url;          
   int (*txAlloc)(char * url, iDnxChannel ** icpp);  
} DnxChanMap;

typedef struct DnxTransport
{
   char * scheme;
   char * libpath;
   int (*txAlloc)(char * url, iDnxChannel ** icpp);
   int (*txInit)(int (**ptxAlloc)(char * url, iDnxChannel ** icpp));
   void (*txExit)(void);
} DnxTransport;

// ---------------------------------------------------------------------------
// This structure is temporary till we start loading from disk, at which point
// URL schemes may then be added at runtime via the configuration file, eg.,
// transport = dnx+transport : /path/to/transport/service/provider/library

#include "dnxTcp.h"
#include "dnxUdp.h"
#include "dnxMsgQ.h"

static DnxTransport gTMList[] =
{
   { "tcp",  0, 0, dnxTcpInit,  dnxTcpDeInit  },
   { "udp",  0, 0, dnxUdpInit,  dnxUdpDeInit  },
   { "msgq", 0, 0, dnxMsgQInit, dnxMsgQDeInit },
};

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

static int dnxInit = 0;             
static pthread_mutex_t chanMutex;   
static DnxChanMap gChannelMap[DNX_MAX_CHAN_MAP]; 

/*--------------------------------------------------------------------------
                              IMPLEMENTATION
  --------------------------------------------------------------------------*/

char * dnxNtop(const void * sastr, char * buf, size_t bufsz)
{
   const struct sockaddr * sa = (const struct sockaddr *)sastr;

   if (!sa)
   {
      strncpy(buf, "<Unknown Address>", bufsz);
      buf[bufsz - 1] = 0;
      return buf;
   }
   switch(sa->sa_family)
   {
      case AF_INET:
         inet_ntop(AF_INET, &((struct sockaddr_in *)sa)->sin_addr, buf, bufsz);
         break;

      case AF_INET6:
         inet_ntop(AF_INET6, &(((struct sockaddr_in6 *)sa)->sin6_addr),buf, INET6_ADDRSTRLEN);
         break;

      default:
         strncpy(buf, "<Unknown AF>", bufsz);
         buf[bufsz - 1] = 0;
         break;
   }
   return buf;
}

static int dnxChanMapUrlParse(DnxChanMap * chanMap, char * url)
{
   char * ep;
   int i;

   assert(chanMap && url && strlen(url) <= DNX_MAX_URL);

   // locate end of scheme
   if ((ep = strstr(url, "://")) == 0)
      return DNX_ERR_BADURL;

   // set allocator for this transport based on the scheme string
   for (i = 0; i < elemcount(gTMList); i++)
      if (!strncmp(url, gTMList[i].scheme, ep - url))
      {
         chanMap->txAlloc = gTMList[i].txAlloc;
         break;
      }

   return i < elemcount(gTMList) ? DNX_OK : DNX_ERR_BADURL;
}

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

static int dnxChanMapFindSlot(DnxChanMap ** chanMap)
{
   int i;

   assert(chanMap);

   // see if we can find an empty slot in the global channel map
   for (i = 0; i < DNX_MAX_CHAN_MAP && gChannelMap[i].name; i++)
      ;

   *chanMap = (DnxChanMap *)((i < DNX_MAX_CHAN_MAP) ? &gChannelMap[i] : 0);

   return *chanMap ? DNX_OK : DNX_ERR_CAPACITY;
}

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

static int dnxChanMapFindName(char * name, DnxChanMap ** chanMap)
{
   int i;

   assert(name && *name && chanMap);

   // see if this name exists in the global channel map
   for (i = 0; i < DNX_MAX_CHAN_MAP; i++)
      if (gChannelMap[i].name && !strcmp(name, gChannelMap[i].name))
         break;

   *chanMap = (DnxChanMap *)((i < DNX_MAX_CHAN_MAP) ? &gChannelMap[i] : 0);

   return *chanMap ? DNX_OK : DNX_ERR_NOTFOUND;
}

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

static int dnxChanMapAllocChannel(char * name, iDnxChannel ** icpp)
{
   DnxChanMap * chanMap;
   int ret;

   assert(name && *name && icpp);

   DNX_PT_MUTEX_LOCK(&chanMutex);

   if ((ret = dnxChanMapFindName(name, &chanMap)) == DNX_OK)
      ret = chanMap->txAlloc(chanMap->url, icpp);

   DNX_PT_MUTEX_UNLOCK(&chanMutex);

   return ret;
}

/*--------------------------------------------------------------------------
                                 INTERFACE
  --------------------------------------------------------------------------*/

int dnxChanMapAdd(char * name, char * url)
{
   DnxChanMap tmp, * chanMap;
   int ret;

   assert(name && *name && url && strlen(url) < DNX_MAX_URL);

   // parse and validate the URL
   if ((ret = dnxChanMapUrlParse(&tmp, url)) != DNX_OK)
      return ret;

   // set the name, unless we are overriding an existing channel
   if ((tmp.name = xstrdup(name)) == 0 || (tmp.url = xstrdup(url)) == 0)
   {
      xfree(tmp.name);
      return DNX_ERR_MEMORY;
   }

   DNX_PT_MUTEX_LOCK(&chanMutex);

   // see if this name already exists, otherwise grab an empty channel slot
   if ((ret = dnxChanMapFindName(name, &chanMap)) == DNX_OK
         || (ret = dnxChanMapFindSlot(&chanMap)) == DNX_OK)
   {
      xfree(chanMap->name);
      xfree(chanMap->url);
      memcpy(chanMap, &tmp, sizeof *chanMap);
   }

   DNX_PT_MUTEX_UNLOCK(&chanMutex);

   // on error, release previously allocated memory
   if (ret != DNX_OK)
   {
      xfree(tmp.name);
      xfree(tmp.url);
   }
   return ret;
}

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

void dnxChanMapDelete(char * name)
{
   DnxChanMap * chanMap;

   assert(name && *name);

   DNX_PT_MUTEX_LOCK(&chanMutex);

   // locate resource by name
   if (dnxChanMapFindName(name, &chanMap) == DNX_OK)
   {
      // release allocated variables, clear object
      xfree(chanMap->name);
      xfree(chanMap->url);
      memset(chanMap, 0, sizeof *chanMap);
   }

   DNX_PT_MUTEX_UNLOCK(&chanMutex);
}

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

int dnxConnect(char * name, int mode, DnxChannel ** channel)
{
   DnxChanMap * chanMap;
   iDnxChannel * icp;
   int ret;

   assert(name && *name && channel);

   // locate channel resource by name, return a new unopened channel
   if ((ret = dnxChanMapAllocChannel(name, &icp)) == DNX_OK)
   {
      if ((ret = icp->txOpen(icp, mode)) != DNX_OK)
         icp->txDelete(icp);
   }
   if (ret == DNX_OK)
      *channel = (DnxChannel *)icp;
   return ret;
}

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

void dnxDisconnect(DnxChannel * channel)
{
   iDnxChannel * icp = (iDnxChannel *)channel;

   assert(channel);

   if (icp->txClose(icp) == DNX_OK)
      icp->txDelete(icp);
}

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

int dnxGet(DnxChannel * channel, char * buf, int * size, int timeout, char * src)
{
   iDnxChannel * icp = (iDnxChannel *)channel;
   assert(channel && buf && size && *size > 0);
   return icp->txRead(icp, buf, size, timeout, src);
}

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

int dnxPut(DnxChannel * channel, char * buf, int size, int timeout, char * dst)
{
   iDnxChannel * icp = (iDnxChannel *)channel;
   assert(channel && buf && size > 0 && size < DNX_MAX_MSG);
   return icp->txWrite(icp, buf, size, timeout, dst);
}

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

void dnxGetStats(DnxChannel * channel, DnxTransStats * tsp)
{
   iDnxChannel * icp = (iDnxChannel *)channel;
   assert(channel && tsp);
   if (icp->txGetStats)
      icp->txGetStats(icp, tsp);
}

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

void dnxResetStats(DnxChannel * channel)
{
   iDnxChannel * icp = (iDnxChannel *)channel;
   assert(channel);
   if (icp->txResetStats)
      icp->txResetStats(icp);
}

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

int dnxChanMapInit(char * fileName)
{
   int i;

   assert(!dnxInit);

   memset(gChannelMap, 0, sizeof gChannelMap);

   DNX_PT_MUTEX_INIT(&chanMutex);

   // initialize transport module table
   for (i = 0; i < elemcount(gTMList); i++)
   {
      int ret;
      if ((ret = gTMList[i].txInit(&gTMList[i].txAlloc)) != DNX_OK)
      {
         while (i--) gTMList[i].txExit();
         DNX_PT_MUTEX_DESTROY(&chanMutex);
         return ret;
      }
   }

   dnxInit = 1;

   return DNX_OK;
}

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

void dnxChanMapRelease(void)
{
   if (dnxInit)
   {
      int i;

      DNX_PT_MUTEX_LOCK(&chanMutex);

      for (i = 0; i < DNX_MAX_CHAN_MAP; i++)
      {
         xfree(gChannelMap[i].name);
         xfree(gChannelMap[i].url);
      }

      memset(gChannelMap, 0, sizeof gChannelMap);

      DNX_PT_MUTEX_UNLOCK(&chanMutex);
      DNX_PT_MUTEX_DESTROY(&chanMutex);

      // de-initialize transport module table
      i = elemcount(gTMList);
      while (i--) gTMList[i].txExit();

      dnxInit = 0;
   }
}

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