socket/XmlBlasterConnectionUnparsed.c

Go to the documentation of this file.

/*----------------------------------------------------------------------------
Name:      XmlBlasterConnectionUnparsed.c
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Wraps raw socket connection to xmlBlaster
           for complete synchronous xmlBlaster access,
           without callbacks and not threading necessary
Author:    "Marcel Ruff" <xmlBlaster@marcelruff.info>
See:       http://www.xmlblaster.org/xmlBlaster/doc/requirements/protocol.socket.html
-----------------------------------------------------------------------------*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h> /* isalpha() */
#if defined(WINCE)
#  if defined(XB_USE_PTHREADS)
#     include <pthreads/pthread.h>
#  else
      /*#include <pthreads/need_errno.h> */
      static int errno=0; /* single threaded workaround*/
#  endif
#else
#  include <errno.h>
#  include <sys/types.h>
#endif
#include <socket/xmlBlasterSocket.h>
#include <socket/xmlBlasterZlib.h>
#include <XmlBlasterConnectionUnparsed.h>
#define SOCKET_TCP false

static bool initConnection(XmlBlasterConnectionUnparsed *xb, XmlBlasterException *exception);
static bool xmlBlasterInitQueue(XmlBlasterConnectionUnparsed *xb, QueueProperties *queueProperties, XmlBlasterException *exception);
static bool getResponse(XmlBlasterConnectionUnparsed *xb, SocketDataHolder *responseSocketDataHolder, XmlBlasterException *exception, bool udp);
static char *xmlBlasterConnect(XmlBlasterConnectionUnparsed *xb, const char * const qos, XmlBlasterException *exception);
static bool xmlBlasterDisconnect(XmlBlasterConnectionUnparsed *xb, const char * const qos, XmlBlasterException *exception);
static char *xmlBlasterPublish(XmlBlasterConnectionUnparsed *xb, MsgUnit *msgUnit, XmlBlasterException *exception);
static QosArr *xmlBlasterPublishArr(XmlBlasterConnectionUnparsed *xb, MsgUnitArr *msgUnitArr, XmlBlasterException *exception);
static void xmlBlasterPublishOneway(XmlBlasterConnectionUnparsed *xb, MsgUnitArr *msgUnitArr, XmlBlasterException *exception);
static char *xmlBlasterSubscribe(XmlBlasterConnectionUnparsed *xb, const char * const key, const char * qos, XmlBlasterException *exception);
static QosArr *xmlBlasterUnSubscribe(XmlBlasterConnectionUnparsed *xb, const char * const key, const char * qos, XmlBlasterException *exception);
static QosArr *xmlBlasterErase(XmlBlasterConnectionUnparsed *xb, const char * const key, const char * qos, XmlBlasterException *exception);
static MsgUnitArr *xmlBlasterGet(XmlBlasterConnectionUnparsed *xb, const char * const key, const char * qos, XmlBlasterException *exception);
static char *xmlBlasterPing(XmlBlasterConnectionUnparsed *xb, const char * const qos, XmlBlasterException *exception);
static bool isConnected(XmlBlasterConnectionUnparsed *xb);
static void xmlBlasterConnectionShutdown(XmlBlasterConnectionUnparsed *xb);
static ssize_t writenPlain(void *xb, const int fd, const char *ptr, const size_t nbytes);
static ssize_t writenCompressed(void *xb, const int fd, const char *ptr, const size_t nbytes);
static ssize_t readnPlain(void *xb, const int fd, char *ptr, const size_t nbytes, XmlBlasterNumReadFunc fpNumRead, void *userP2);
static ssize_t readnCompressed(void *userP, const int fd, char *ptr, const size_t nbytes, XmlBlasterNumReadFunc fpNumRead, void *userP2);
static bool checkArgs(XmlBlasterConnectionUnparsed *xb, const char *methodName, bool checkIsConnected, XmlBlasterException *exception);

XmlBlasterConnectionUnparsed *getXmlBlasterConnectionUnparsed(int argc, const char* const* argv) {
   XmlBlasterConnectionUnparsed *xb = (XmlBlasterConnectionUnparsed *)calloc(1, sizeof(XmlBlasterConnectionUnparsed));
   if (xb == 0) return xb;
   xb->argc = argc;
   xb->argv = argv;
   xb->props = createProperties(xb->argc, xb->argv);
   if (xb->props == 0) {
      freeXmlBlasterConnectionUnparsed(&xb);
      return (XmlBlasterConnectionUnparsed *)0;
   }
   xb->socketToXmlBlaster = -1;
   xb->socketToXmlBlasterUdp = -1;
   xb->isInitialized = false;
   xb->requestId = 0;
   *xb->secretSessionId = 0;
   xb->initConnection = initConnection;
   xb->initQueue = xmlBlasterInitQueue;
   xb->connect = xmlBlasterConnect;
   xb->disconnect = xmlBlasterDisconnect;
   xb->publish = xmlBlasterPublish;
   xb->publishArr = xmlBlasterPublishArr;
   xb->publishOneway = xmlBlasterPublishOneway;
   xb->subscribe = xmlBlasterSubscribe;
   xb->unSubscribe = xmlBlasterUnSubscribe;
   xb->erase = xmlBlasterErase;
   xb->get = xmlBlasterGet;
   xb->ping = xmlBlasterPing;
   xb->isConnected = isConnected;
   xb->shutdown = xmlBlasterConnectionShutdown;
   xb->preSendEvent = 0;
   xb->preSendEvent_userP = 0;
   xb->postSendEvent = 0;
   xb->postSendEvent_userP = 0;
   xb->queueP = 0;
   xb->logLevel = parseLogLevel(xb->props->getString(xb->props, "logLevel", "WARN"));
   xb->log = xmlBlasterDefaultLogging;
   xb->logUserP = 0;
   xb->useUdpForOneway = false;
   xb->writeToSocket.writeToSocketFuncP = 0;
   xb->writeToSocket.userP = xb;
   xb->zlibWriteBuf = 0;
   xb->readFromSocket.readFromSocketFuncP = 0;
   xb->readFromSocket.userP = xb;
   xb->zlibReadBuf = 0;
   return xb;
}

void freeXmlBlasterConnectionUnparsed(XmlBlasterConnectionUnparsed **xb_)
{
   XmlBlasterConnectionUnparsed *xb = *xb_;
   if (xb != 0) {
      if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "freeXmlBlasterConnectionUnparsed 0x%x", xb);
      freeProperties(xb->props);
      if (xb->zlibWriteBuf) {
         xmlBlaster_endZlibWriter(xb->zlibWriteBuf);
         free(xb->zlibWriteBuf);
         xb->zlibWriteBuf = 0;
      }
      if (xb->zlibReadBuf) {
         xmlBlaster_endZlibReader(xb->zlibReadBuf);
         free(xb->zlibReadBuf);
         xb->zlibReadBuf = 0;
      }
      xmlBlasterConnectionShutdown(xb);
      free(xb);
      *xb_ = 0;
   }
}

static bool initConnection(XmlBlasterConnectionUnparsed *xb, XmlBlasterException *exception)
{
   const char *servTcpPort = 0;

   struct sockaddr_in xmlBlasterAddr;
   struct hostent hostbuf, *hostP = 0;
   struct servent *portP = 0;

   size_t hstbuflen=0;

   char serverHostName[256];
   char errP[MAX_ERRNO_LEN];

#if defined(_WINDOWS)
   WORD wVersionRequested;
   WSADATA wsaData;
   int err;
   wVersionRequested = MAKEWORD( 2, 2 );
   err = WSAStartup( wVersionRequested, &wsaData );
   if ( err != 0 ) {
      strncpy0(exception->errorCode, "resource.unavailable", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN, "[%.100s:%d] Couldn't find a usable WinSock DLL", __FILE__, __LINE__);
      if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
      return false;
   }

   if ( LOBYTE( wsaData.wVersion ) != 2 ||
   HIBYTE( wsaData.wVersion ) != 2 ) {
      WSACleanup( );
      strncpy0(exception->errorCode, "resource.unavailable", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN, "[%.100s:%d] Couldn't find a usable WinSock DLL which supports version 2.2", __FILE__, __LINE__);
      if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
      return false; 
   }
# endif
   *errP = 0;

   if (xb->isInitialized) {
      return true;
   }

   {  /* Switch on compression? */
      const char *compressType = xb->props->getString(xb->props, "plugin/socket/compress/type", "");
      compressType = xb->props->getString(xb->props, "dispatch/connection/plugin/socket/compress/type", compressType);

      if (!strcmp(compressType, "zlib:stream")) {
         
         xb->zlibWriteBuf = (XmlBlasterZlibWriteBuffers *)malloc(sizeof(struct XmlBlasterZlibWriteBuffers));
         xb->zlibReadBuf = (XmlBlasterZlibReadBuffers *)malloc(sizeof(struct XmlBlasterZlibReadBuffers));

         if (xmlBlaster_initZlibWriter(xb->zlibWriteBuf) != 0/*Z_OK*/) {
            if (xb->logLevel>=XMLBLASTER_LOG_ERROR) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_ERROR, __FILE__,
                  "Failed switching on 'plugin/socket/compress/type=%s'", compressType);
            strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
            SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                     "[%.100s:%d] Failed switching on 'plugin/socket/compress/type=%s'",
                     __FILE__, __LINE__, compressType);
            free(xb->zlibWriteBuf);
            xb->zlibWriteBuf = 0;
            free(xb->zlibReadBuf);
            xb->zlibReadBuf = 0;
            return false;
         }

         if (xmlBlaster_initZlibReader(xb->zlibReadBuf) != 0/*Z_OK*/) {
            if (xb->logLevel>=XMLBLASTER_LOG_ERROR) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_ERROR, __FILE__,
                  "Failed switching on 'plugin/socket/compress/type=%s'", compressType);
            strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
            SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                     "[%.100s:%d] Failed switching on 'plugin/socket/compress/type=%s'",
                     __FILE__, __LINE__, compressType);
            free(xb->zlibWriteBuf);
            xb->zlibWriteBuf = 0;
            free(xb->zlibReadBuf);
            xb->zlibReadBuf = 0;
            return false;
         }

         if (xb->logLevel>=XMLBLASTER_LOG_DUMP) {
            xb->zlibWriteBuf->debug = true;
            xb->zlibReadBuf->debug = true;
         }

         if (!xb->writeToSocket.writeToSocketFuncP) {  /* Accept setting from XmlBlasterAccessUnparsed */
            xb->writeToSocket.writeToSocketFuncP = writenCompressed;
            xb->readFromSocket.readFromSocketFuncP = readnCompressed;
         }
      }
      else {
         if (strcmp(compressType, "")) {
            xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__, "Unsupported compression type 'plugin/socket/compress/type=%s', falling back to plain mode.", compressType);
         }
         if (!xb->writeToSocket.writeToSocketFuncP) {  /* Accept setting from XmlBlasterAccessUnparsed */
            xb->writeToSocket.writeToSocketFuncP = writenPlain;
            xb->readFromSocket.readFromSocketFuncP = readnPlain;
         }
      }
   }


   servTcpPort = xb->props->getString(xb->props, "plugin/socket/port", "7607");
   servTcpPort = xb->props->getString(xb->props, "dispatch/connection/plugin/socket/port", servTcpPort);

   strncpy0(serverHostName, "localhost", 250);
   gethostname(serverHostName, 250);
   {
      const char *hn = xb->props->getString(xb->props, "plugin/socket/hostname", serverHostName);
      memmove(serverHostName, hn, strlen(hn)+1);  /* including '\0' */
      hn = xb->props->getString(xb->props, "dispatch/connection/plugin/socket/hostname", serverHostName);
      memmove(serverHostName, hn, strlen(hn)+1);
   }

   if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
      "Lookup xmlBlaster on -dispatch/connection/plugin/socket/hostname %s -dispatch/connection/plugin/socket/port %s ...",
      serverHostName, servTcpPort);

   *xb->secretSessionId = 0;
   memset((char *)&xmlBlasterAddr, 0, sizeof(xmlBlasterAddr));
   xmlBlasterAddr.sin_family=AF_INET;

# if _WINDOWS_NOT_YET_PORTED /* Windows gethostbyname is deprecated */
   const struct addrinfo hints;
   struct addrinfo** res;
   int getaddrinfo(serverHostName, servTcpPort, &hints, res);
   res->ai_next : ai_family, ai_socktype, and ai_protocol

   ...

   void freeaddrinfo(*res);
# endif
        if (isalpha(serverHostName[0]) || strchr(serverHostName,':') != 0) { /* look for dns name or ipv6 */
           char *tmphstbuf=0;
           memset((char *)&hostbuf, 0, sizeof(struct hostent));
           xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_INFO, __FILE__, "Server DNS lookup of hostname '%s'", serverHostName);
           hostP = gethostbyname_re(serverHostName, &hostbuf, &tmphstbuf, &hstbuflen, errP);
           if (hostP == 0) {
              if (*errP != 0) {
                 strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
                 SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                    "[%.100s:%d] Connecting to xmlBlaster failed, can't determine hostname (hostP=0), -dispatch/connection/plugin/socket/hostname %s -dispatch/connection/plugin/socket/port %.10s: %s",
                          __FILE__, __LINE__, serverHostName, servTcpPort, errP);
                 xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__, exception->message);
                 *errP = 0;
              }
              else {
                 strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
                 SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                          "[%.100s:%d] Connecting to xmlBlaster failed, can't determine hostname (hostP=0), -dispatch/connection/plugin/socket/hostname %s -dispatch/connection/plugin/socket/port %.10s, errno=%d",
                          __FILE__, __LINE__, serverHostName, servTcpPort, errno);
                 xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__, exception->message);
              }
              return false;
           }
           xmlBlasterAddr.sin_addr.s_addr = ((struct in_addr *)(hostP->h_addr))->s_addr; /* inet_addr("192.168.1.2"); */
           free(tmphstbuf);
        }
        else {
           xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_INFO, __FILE__, "Server IP4 usage (without DNS lookup) of IP '%s'", serverHostName);
           /* use ip4 addr directly to avoid dns lookup */
           xmlBlasterAddr.sin_addr.s_addr = inet_addr(serverHostName);
        }

   portP = getservbyname(servTcpPort, "tcp");
   if (portP != 0)
                xmlBlasterAddr.sin_port = (u_short)portP->s_port;
   else
      xmlBlasterAddr.sin_port = htons((u_short)atoi(servTcpPort));

   xb->socketToXmlBlaster = (int)socket(AF_INET, SOCK_STREAM, 0);
   if (xb->socketToXmlBlaster != -1) {
      int ret=0;
      const char *localHostName = xb->props->getString(xb->props, "plugin/socket/localHostname", 0);
      int localPort = xb->props->getInt(xb->props, "plugin/socket/localPort", 0);
      localHostName = xb->props->getString(xb->props, "dispatch/connection/plugin/socket/localHostname", localHostName);
      localPort = xb->props->getInt(xb->props, "dispatch/connection/plugin/socket/localPort", localPort);

      /* Sometimes a user may whish to force the local host/port setting (e.g. for firewall tunneling
         and on multi homed hosts */
      if (localHostName != 0 || localPort > 0) {
         struct sockaddr_in localAddr;
         struct hostent localHostbuf, *localHostP = 0;
         char *tmpLocalHostbuf=0;
         size_t localHostbuflen=0;
         memset(&localAddr, 0, sizeof(localAddr));
         localAddr.sin_family = AF_INET;
         if (localHostName) {
            if (isalpha(localHostName[0]) || strchr(localHostName,':') != 0) { /* look for dns name or ipv6 */
               if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "Local hostname DNS lookup of hostname '%s'", localHostName);
               *errP = 0;
               localHostP = gethostbyname_re(localHostName, &localHostbuf, &tmpLocalHostbuf, &localHostbuflen, errP);
               if (*errP != 0) {
                  strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
                  SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                           "[%.100s:%d] Lookup of local IP failed, %s",
                           __FILE__, __LINE__, errP);
                  xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__, exception->message);
                  *errP = 0;
               }
               if (localHostP != 0) {
                  localAddr.sin_addr.s_addr = ((struct in_addr *)(localHostP->h_addr))->s_addr; /* inet_addr("192.168.1.2"); */
                  free(tmpLocalHostbuf);
               }
            }
            else {
               if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "Local IP4 usage (without DNS lookup) of IP '%s'", localHostName);
               /* use ip4 addr directly to avoid dns lookup */
               localAddr.sin_addr.s_addr = inet_addr(localHostName);
            }
         }
         if (localPort > 0) {
            localAddr.sin_port = htons((unsigned short)localPort);
         }
         if (bind(xb->socketToXmlBlaster, (struct sockaddr *)&localAddr, sizeof(localAddr)) < 0) {
            if (xb->logLevel>=XMLBLASTER_LOG_WARN) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__,
               "Failed binding local port -dispatch/connection/plugin/socket/localHostname %s -dispatch/connection/plugin/socket/localPort %d",
                  localHostName, localPort);
         }
         else {
            xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_INFO, __FILE__,
               "Bound local port -dispatch/connection/plugin/socket/localHostname %s -dispatch/connection/plugin/socket/localPort %d",
                  localHostName, localPort);
         }
      }

      /* int retval = fcntl(xb->socketToXmlBlaster, F_SETFL, O_NONBLOCK); */ /* Switch on none blocking mode: we then should use select() to be notified when the kernel succeeded with connect() */

      if ((ret=connect(xb->socketToXmlBlaster, (struct sockaddr *)&xmlBlasterAddr, sizeof(xmlBlasterAddr))) != -1) {
         if (xb->logLevel>=XMLBLASTER_LOG_INFO) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_INFO, __FILE__, "Connected to xmlBlaster");
         xb->useUdpForOneway = xb->props->getBool(xb->props, "plugin/socket/useUdpForOneway", xb->useUdpForOneway);
         xb->useUdpForOneway = xb->props->getBool(xb->props, "dispatch/connection/plugin/socket/useUdpForOneway", xb->useUdpForOneway);

         if (xb->useUdpForOneway) {
            struct sockaddr_in localAddr;
            socklen_t size = (socklen_t)sizeof(localAddr);
            xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_INFO, __FILE__,
               "Using UDP connection for oneway calls, see -dispatch/connection/plugin/socket/useUdpForOneway true");

            xb->socketToXmlBlasterUdp = (int)socket(AF_INET, SOCK_DGRAM, 0);

            if (xb->socketToXmlBlasterUdp != -1) {
               if (getsockname(xb->socketToXmlBlaster, (struct sockaddr *)&localAddr, &size) == -1) {
                  if (xb->logLevel>=XMLBLASTER_LOG_WARN) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__,
                     "Can't determine the local socket host and port (in UDP), errno=%d", errno);
                  return false;
               }

               if (bind(xb->socketToXmlBlasterUdp, (struct sockaddr *)&localAddr, sizeof(localAddr)) < 0) {
                  if (xb->logLevel>=XMLBLASTER_LOG_WARN) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__,
                     "Failed binding local port (in UDP) -dispatch/connection/plugin/socket/localHostname %s -dispatch/connection/plugin/socket/localPort %d",
                     localHostName, localPort);
                  return false;
               }
               if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
                  "Bound local UDP port -dispatch/connection/plugin/socket/localHostname %s -dispatch/connection/plugin/socket/localPort %d",
                  localHostName, localPort);

               if ((ret=connect(xb->socketToXmlBlasterUdp, (struct sockaddr *)&xmlBlasterAddr, sizeof(xmlBlasterAddr))) == -1) {
                  char errnoStr[MAX_ERRNO_LEN];
                  xb_strerror(errnoStr, MAX_ERRNO_LEN, errno);
                  strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
                  SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                           "[%.100s:%d] Connecting to xmlBlaster -dispatch/connection/plugin/socket/hostname %s -dispatch/connection/plugin/socket/port %.10s failed (in UDP), ret=%d, %s",
                           __FILE__, __LINE__, serverHostName, servTcpPort, ret, errnoStr);
                  if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
                  return false;
               }
               if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "Connected to xmlBlaster with UDP");
            } /* if (xb->socketToXmlBlasterUdp != -1) */
            else {
               strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
               SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                        "[%.100s:%d] Connecting to xmlBlaster (socket=-1) -dispatch/connection/plugin/socket/hostname %s -dispatch/connection/plugin/socket/port %.10s failed (in UDP) errno=%d",
                        __FILE__, __LINE__, serverHostName, servTcpPort, errno);
               return false;
            }
         } /* if (xb->useUdpForOneway) */

      }
      else { /* connect(...) == -1 */
         char errnoStr[MAX_ERRNO_LEN];
         xb_strerror(errnoStr, MAX_ERRNO_LEN, errno);
         strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
         SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
                  "[%.100s:%d] Connecting to xmlBlaster -dispatch/connection/plugin/socket/hostname %s -dispatch/connection/plugin/socket/port %.10s failed, ret=%d, %s",
                  __FILE__, __LINE__, serverHostName, servTcpPort, ret, errnoStr);
         if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
         return false;
      }
   }
   else {
      strncpy0(exception->errorCode, "user.configuration", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
               "[%.100s:%d] Connecting to xmlBlaster (socket=-1) -dispatch/connection/plugin/socket/hostname %s -dispatch/connection/plugin/socket/port %.10s failed errno=%d",
               __FILE__, __LINE__, serverHostName, servTcpPort, errno);
      return false;
   }
   xb->isInitialized = true;
   return true;
}


static bool xmlBlasterInitQueue(XmlBlasterConnectionUnparsed *xb, QueueProperties *queueProperties, XmlBlasterException *exception)
{
#ifdef XMLBLASTER_PERSISTENT_QUEUE_TEST
   if (checkArgs(xb, "initQueue", false, exception) == false ) return false;
   if (xb->queueP) {
      char message[XMLBLASTEREXCEPTION_MESSAGE_LEN];
      SNPRINTF(message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
               "[%.100s:%d] The queue is initialized already, call to initQueue() is ignored", __FILE__, __LINE__);
      embedException(exception, "user.illegalArgument", message, exception);
      xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__, exception->message);
      return false;
   }

   {
      QueueProperties tmp;
      memset(&tmp, 0, sizeof(QueueProperties));

      if (queueProperties == 0)
         queueProperties = &tmp;

      if (*queueProperties->dbName == 0) {
         strncpy0(queueProperties->dbName, xb->props->getString(xb->props, "queue/connection/dbName", "xmlBlasterClient.db"), QUEUE_DBNAME_MAX);
      }
      if (*queueProperties->nodeId == 0) {
         strncpy0(queueProperties->nodeId, xb->props->getString(xb->props, "queue/connection/nodeId", "client"), QUEUE_ID_MAX);
      }
      if (*queueProperties->queueName == 0) {
         strncpy0(queueProperties->queueName, xb->props->getString(xb->props, "queue/connection/queueName", "connection_client"), QUEUE_ID_MAX);
      }
      if (*queueProperties->tablePrefix == 0) {
         strncpy0(queueProperties->tablePrefix, xb->props->getString(xb->props, "queue/connection/tablePrefix", "XB_"), QUEUE_PREFIX_MAX);
      }
      if (queueProperties->maxNumOfEntries == 0) {
         queueProperties->maxNumOfEntries = xb->props->getInt(xb->props, "queue/connection/maxEntries", 10000000);
      }
      if (queueProperties->maxNumOfBytes == 0) {
         queueProperties->maxNumOfBytes = xb->props->getInt64(xb->props, "queue/connection/maxBytes", 10000000LL);
      }
      if (queueProperties->logFp == 0) queueProperties->logFp = xb->log;
      if (queueProperties->logLevel == 0) queueProperties->logLevel = xb->logLevel;
      if (queueProperties->userObject == 0) queueProperties->userObject = xb->userObject;

      xb->queueP = createQueue(queueProperties, exception);
      if (*exception->errorCode != 0) {
         xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "Queue initializeation failed: [%s] %s\n", exception->errorCode, exception->message);
         return false;
      }
      xb->queueP->userObject = xb;
   }
   return true;
#else
   if (queueProperties) {} /* To suppress compiler warning that not used */
   strncpy0(exception->errorCode, "user.illegalArgument", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
   SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
            "[%.100s:%d] Queue support is not compiled into the library, please recompile with '-DXMLBLASTER_PERSISTENT_QUEUE=1 and -DXMLBLASTER_PERSISTENT_QUEUE_TEST=1", __FILE__, __LINE__);
   xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_WARN, __FILE__, exception->message);
   return false;
#endif /* XMLBLASTER_PERSISTENT_QUEUE_TEST */
}

static bool isConnected(XmlBlasterConnectionUnparsed *xb)
{
   return (xb->socketToXmlBlaster > -1) ? true : false;
}

const char *xmlBlasterConnectionUnparsedUsage()
{
   /* To prevent compiler warning */
   /*   "string length `596' is greater than the length `509' ISO C89 compilers are required to support" */
   /* we have a static variable */
   enum { SIZE=2048 };
   static char usage[SIZE];
   strncpy0(usage, 
      "\n   -dispatch/connection/plugin/socket/hostname [localhost]"
      "\n                       Where to find xmlBlaster."
      "\n   -dispatch/connection/plugin/socket/port [7607]"
      "\n                       The port where xmlBlaster listens."
      "\n   -dispatch/connection/plugin/socket/localHostname [NULL]", SIZE/2);
   strncat0(usage,
      "\n                       Force the local IP, useful on multi homed computers."
      "\n   -dispatch/connection/plugin/socket/localPort [0]"
      "\n                       Force the local port, useful to tunnel firewalls."
      "\n   -dispatch/connection/plugin/socket/compress/type []"
#if XMLBLASTER_ZLIB==1
      "\n                       Switch on compression with 'zlib:stream'."
#else
      "\n                       No compression support. Try recompiling with with '-DXMLBLASTER_ZLIB=1'."
#endif
      "\n   -dispatch/connection/plugin/socket/useUdpForOneway [false]"
      "\n                       Use UDP for publishOneway() calls.", SIZE/2);
   return usage;
}

static void xmlBlasterConnectionShutdown(XmlBlasterConnectionUnparsed *xb)
{
   if (xb != 0 && xb->isConnected(xb)) {
#     if defined(_WINDOWS)
      int how = SD_BOTH;   /* SD_BOTH requires Winsock2.h */
#     else
      int how = SHUT_RDWR; /* enum SHUT_RDWR = 2 */
#     endif
      if (xb->logLevel>=XMLBLASTER_LOG_TRACE) xb->log(xb->logUserP, xb->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
            "shutdown() socketToXmlBlaster=%d socketToXmlBlasterUdp=%d", xb->