socket/XmlBlasterAccessUnparsed.c

Go to the documentation of this file.

/*----------------------------------------------------------------------------
Name:      XmlBlasterAccessUnparsed.c
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Wraps raw socket connection to xmlBlaster
           Implements sync connection and async callback
           Needs pthread to compile (multi threading).
Author:    "Marcel Ruff" <xmlBlaster@marcelruff.info>
Compile:
  LINUX:   gcc -DXmlBlasterAccessUnparsedMain -D_ENABLE_STACK_TRACE_ -rdynamic -export-dynamic -Wall -pedantic -g -D_REENTRANT -I.. -o XmlBlasterAccessUnparsedMain XmlBlasterAccessUnparsed.c ../util/msgUtil.c ../util/Properties.c xmlBlasterSocket.c XmlBlasterConnectionUnparsed.c CallbackServerUnparsed.c -lpthread
           g++ -DXmlBlasterAccessUnparsedMain -DXMLBLASTER_C_COMPILE_AS_CPP -Wall -pedantic -g -D_REENTRANT -I.. -o XmlBlasterAccessUnparsedMain XmlBlasterAccessUnparsed.c ../util/msgUtil.c ../util/Properties.c xmlBlasterSocket.c XmlBlasterConnectionUnparsed.c CallbackServerUnparsed.c -lpthread
           icc -DXmlBlasterAccessUnparsedMain -D_ENABLE_STACK_TRACE_ -rdynamic -g -D_REENTRANT -I.. -o XmlBlasterAccessUnparsedMain XmlBlasterAccessUnparsed.c ../util/msgUtil.c ../util/Properties.c xmlBlasterSocket.c XmlBlasterConnectionUnparsed.c CallbackServerUnparsed.c -lpthread
  WIN:     cl /MT /W4 -DXmlBlasterAccessUnparsedMain -D_WINDOWS -I.. -I../pthreads /FeXmlBlasterAccessUnparsedMain.exe  XmlBlasterAccessUnparsed.c ..\util\msgUtil.c ..\util\Properties.c xmlBlasterSocket.c XmlBlasterConnectionUnparsed.c CallbackServerUnparsed.c ws2_32.lib pthreadVC2.lib
           (download pthread for Windows and WinCE from http://sources.redhat.com/pthreads-win32)
  Solaris: cc  -DXmlBlasterAccessUnparsedMain -v -Xc -g -D_REENTRANT -I.. -o XmlBlasterAccessUnparsedMain XmlBlasterAccessUnparsed.c ../util/msgUtil.c ../util/Properties.c xmlBlasterSocket.c XmlBlasterConnectionUnparsed.c CallbackServerUnparsed.c -lpthread -lsocket -lnsl
           CC  -DXmlBlasterAccessUnparsedMain -DXMLBLASTER_C_COMPILE_AS_CPP -g -D_REENTRANT -I.. -o XmlBlasterAccessUnparsedMain XmlBlasterAccessUnparsed.c ../util/msgUtil.c ../util/Properties.c xmlBlasterSocket.c XmlBlasterConnectionUnparsed.c CallbackServerUnparsed.c -lpthread -lsocket -lnsl

  Linux with libxmlBlasterC.so:
           gcc -DXmlBlasterAccessUnparsedMain -o XmlBlasterAccessUnparsedMain XmlBlasterAccessUnparsed.c  -L../../../lib -lxmlBlasterClientC -I.. -Wl,-rpath=../../../lib -D_REENTRANT  -lpthread
See:       http://www.xmlblaster.org/xmlBlaster/doc/requirements/protocol.socket.html
-----------------------------------------------------------------------------*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#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 <XmlBlasterAccessUnparsed.h>

typedef struct Dll_Export UpdateContainer {
   XmlBlasterAccessUnparsed *xa;
   MsgUnitArr *msgUnitArrP;
   void *userData;
   XmlBlasterException exception;     /* Holding a clone from the original as the callback thread may use it for another message */
   SocketDataHolder socketDataHolder; /* Holding a clone from the original */
} UpdateContainer;

static bool initialize(XmlBlasterAccessUnparsed *xa, UpdateFp update, XmlBlasterException *exception);
static char *xmlBlasterConnect(XmlBlasterAccessUnparsed *xa, const char * const qos, UpdateFp update, XmlBlasterException *exception);
static bool xmlBlasterDisconnect(XmlBlasterAccessUnparsed *xa, const char * const qos, XmlBlasterException *exception);
static char *xmlBlasterPublish(XmlBlasterAccessUnparsed *xa, MsgUnit *msgUnit, XmlBlasterException *exception);
static QosArr *xmlBlasterPublishArr(XmlBlasterAccessUnparsed *xa, MsgUnitArr *msgUnitArr, XmlBlasterException *exception);
static void xmlBlasterPublishOneway(XmlBlasterAccessUnparsed *xa, MsgUnitArr *msgUnitArr, XmlBlasterException *exception);
static char *xmlBlasterSubscribe(XmlBlasterAccessUnparsed *xa, const char * const key, const char * qos, XmlBlasterException *exception);
static QosArr *xmlBlasterUnSubscribe(XmlBlasterAccessUnparsed *xa, const char * const key, const char * qos, XmlBlasterException *exception);
static QosArr *xmlBlasterErase(XmlBlasterAccessUnparsed *xa, const char * const key, const char * qos, XmlBlasterException *exception);
static MsgUnitArr *xmlBlasterGet(XmlBlasterAccessUnparsed *xa, const char * const key, const char * qos, XmlBlasterException *exception);
static char *xmlBlasterPing(XmlBlasterAccessUnparsed *xa, const char * const qos, XmlBlasterException *exception);
static bool isConnected(XmlBlasterAccessUnparsed *xa);
static void responseEvent(MsgRequestInfo *msgRequestInfoP, void /*SocketDataHolder*/ *socketDataHolder);
static MsgRequestInfo *preSendEvent(MsgRequestInfo *msgRequestInfo, XmlBlasterException *exception);
static MsgRequestInfo *postSendEvent(MsgRequestInfo *msgRequestInfo, XmlBlasterException *exception);
static bool checkArgs(XmlBlasterAccessUnparsed *xa, const char *methodName, bool checkIsConnected, XmlBlasterException *exception);
static void interceptUpdate(MsgUnitArr *msgUnitArr, void *userData, XmlBlasterException *xmlBlasterException, void/*SocketDataHolder*/ *socketDataHolder);
static bool mutexUnlock(MsgRequestInfo *msgRequestInfoP, XmlBlasterException *exception);
static ssize_t writenPlain(void *xa, const int fd, const char *ptr, const size_t nbytes);
static ssize_t writenCompressed(void *xa, const int fd, const char *ptr, const size_t nbytes);
static ssize_t readnPlain(void * userP, 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);

Dll_Export XmlBlasterAccessUnparsed *getXmlBlasterAccessUnparsed(int argc, const char* const* argv) {
   XmlBlasterAccessUnparsed * const xa = (XmlBlasterAccessUnparsed *)calloc(1, sizeof(XmlBlasterAccessUnparsed));
   if (xa == 0) return xa;
   xa->argc = argc;
   xa->argv = argv;
   xa->props = createProperties(xa->argc, xa->argv);
   if (xa->props == 0) {
      freeXmlBlasterAccessUnparsed(xa);
      return (XmlBlasterAccessUnparsed *)0;
   }
   xa->isInitialized = false;
   xa->isShutdown = false;
   xa->connectionP = 0;
   xa->callbackP = 0;
   xa->userObject = 0; /* A client can use this pointer to point to any client specific information */
   xa->userFp = 0;
   xa->connect = xmlBlasterConnect;
   xa->initialize = initialize;
   xa->disconnect = xmlBlasterDisconnect;
   xa->publish = xmlBlasterPublish;
   xa->publishArr = xmlBlasterPublishArr;
   xa->publishOneway = xmlBlasterPublishOneway;
   xa->subscribe = xmlBlasterSubscribe;
   xa->unSubscribe = xmlBlasterUnSubscribe;
   xa->erase = xmlBlasterErase;
   xa->get = xmlBlasterGet;
   xa->ping = xmlBlasterPing;
   xa->isConnected = isConnected;
   xa->logLevel = parseLogLevel(xa->props->getString(xa->props, "logLevel", "WARN"));
   xa->log = xmlBlasterDefaultLogging;
   xa->logUserP = 0;
   xa->clientsUpdateFp = 0;
   xa->callbackMultiThreaded = xa->props->getBool(xa->props, "plugin/socket/multiThreaded", true);
   xa->callbackMultiThreaded = xa->props->getBool(xa->props, "dispatch/callback/plugin/socket/multiThreaded", xa->callbackMultiThreaded);
   /*   xa->lowLevelAutoAck = xa->props->getBool(xa->props, "plugin/socket/lowLevelAutoAck", false); */
   /*   xa->lowLevelAutoAck = xa->props->getBool(xa->props, "dispatch/callback/plugin/socket/lowLevelAutoAck", xa->lowLevelAutoAck); */
   /* Currently forced to false: needs mutex and reference counter to not freeMsgUnitArr twice */
   xa->lowLevelAutoAck = false;

   /* We shouldn't do much logging here, as the caller had no chance to redirect it up to now */
   if (xa->callbackMultiThreaded == true) {
      if (xa->logLevel>=XMLBLASTER_LOG_DUMP) 
         xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_DUMP, __FILE__, "Multi threaded callback delivery is activated with -plugin/socket/multiThreaded true");
      /*xa->callbackMultiThreaded = false;*/
   }
   /* stdint.h: # define INT32_MAX              (2147483647) */
   xa->responseTimeout = xa->props->getLong(xa->props, "plugin/socket/responseTimeout", 2147483647L); /* Before xmlBlaster 1.1: One minute (given in millis) */
   xa->responseTimeout = xa->props->getLong(xa->props, "dispatch/connection/plugin/socket/responseTimeout", xa->responseTimeout);
   /* ERROR HANDLING ? xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_WARN, __FILE__, "Your configuration '-plugin/socket/responseTimeout %s' is invalid", argv[iarg]); */
   memset(&xa->callbackThreadId, 0, sizeof(pthread_t));
   xa->threadCounter = 0;

   if (xa->logLevel>=XMLBLASTER_LOG_DUMP) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_DUMP, __FILE__,
                                "Created handle: -logLevel=%s -plugin/socket/responseTimeout=%ld",
                                getLogLevelStr(xa->logLevel), xa->responseTimeout);

   /* See: http://www.llnl.gov/computing/tutorials/workshops/workshop/pthreads/MAIN.html */
   pthread_mutex_init(&xa->writenMutex, NULL); /* returns always 0 */
   pthread_mutex_init(&xa->readnMutex, NULL);
   return xa;
}

Dll_Export void freeXmlBlasterAccessUnparsed(XmlBlasterAccessUnparsed *xa)
{
   int rc;

   if (xa == 0) {
      char *stack = getStackTrace(10);
      printf("[%s:%d] Please provide a valid XmlBlasterAccessUnparsed pointer to freeXmlBlasterAccessUnparsed() %s",
                __FILE__, __LINE__, stack);
      free(stack);
      return;
   }

   if (xa->isShutdown) return; /* Avoid simultaneous multiple calls */
   xa->isShutdown = true;      /* Inhibit access to xa */

   if (xa->callbackP != 0) {
      xa->callbackP->shutdown(xa->callbackP);
   }
   if (xa->connectionP != 0) {
      xa->connectionP->shutdown(xa->connectionP);
   }

   if (xa->callbackP != 0) {
      /* Detach or join? On Linux both work fine. On Windows it blocks sometimes forever during join */
      const bool USE_DETACH_MODE = xa->props->getBool(xa->props, "plugin/socket/detachCbThread", true);
      int retVal;
      if (!xa->callbackP->isShutdown) {

         {  /* Wait for any pending update() dispatcher threads to die */
            int i;
            int num = 200;
            int interval = 10;
            for (i=0; i<num; i++) {
               if (xa->callbackP->isShutdown)
                  break;
               /*pthread_yield(0);*/
               sleepMillis(interval);
               if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
                   "freeXmlBlasterAccessUnparsed(): Sleeping %d millis for callback thread to join. %d/%d", interval, i, num);
            }
            if (i == num) {
               xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "Proper shutdown of callback thread failed, it seems to block on the socket");
            }
         }

         if (!USE_DETACH_MODE) {
            /* pthread_cancel() does not block. Who cleans up open resources? TODO: pthread_cleanup_push() */
            /* On Linux all works fine without pthread_cancel() but on Windows the later pthread_join() sometimes hangs without a pthread_cancel() */
            /*
            retVal = pthread_cancel(xa->callbackThreadId);
            if (retVal != 0) {
               xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "pthread_cancel problem return value is %d", retVal);
            }
            */
         }
      }

      if (USE_DETACH_MODE) {
         retVal = pthread_detach(xa->callbackThreadId); /* Frees resources (even if thread has died already), don't call multiple times on same thread! */
         if (retVal != 0) {
            xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "[%d] Detaching callback thread 0x%x failed with error number %d", __LINE__, get_pthread_id(xa->callbackThreadId), retVal);
         }
         else {
            if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
                                          "pthread_detach(id=%ld) succeeded for callback server thread", get_pthread_id(xa->callbackThreadId));
         }
      }
      else { /* JOIN mode */
         retVal = pthread_join(xa->callbackThreadId, 0);
         if (retVal != 0) {
            xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "pthread_join problem return value is %d", retVal);
         }
         else {
            if (xa->logLevel>=XMLBLASTER_LOG_INFO) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_INFO, __FILE__,
                                          "pthread_join(id=%ld) succeeded for callback server thread", get_pthread_id(xa->callbackThreadId));
         }
      }

      memset(&xa->callbackThreadId, 0, sizeof(pthread_t));
   }

   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "freeXmlBlasterAccessUnparsed() conP=0x%x cbP=0x%x", xa->connectionP, xa->callbackP);

   {  /* Wait for any pending update() dispatcher threads to die */
      int i;
      int num = 1000;
      int interval = 10;
      for (i=0; i<num; i++) {
         if ((int)xa->threadCounter < 1)
            break;
         sleepMillis(interval);
         if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
             "freeXmlBlasterAccessUnparsed(): Sleeping %d millis for update thread to join. %d/%d", interval, i, num);
      }
      if (i >= num) {
         if (xa->logLevel>=XMLBLASTER_LOG_ERROR) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__,
             "freeXmlBlasterAccessUnparsed(): There are active callback threads in user code which didn't return after sleeping for %ld millis, we continue now to shutdown ...", (long)interval*num);
      }
   }

   if (xa->connectionP != 0) {
      freeXmlBlasterConnectionUnparsed(&xa->connectionP);
   }

   if (xa->callbackP != 0) {
      freeCallbackServerUnparsed(&xa->callbackP);
   }

   freeProperties(xa->props);

   rc = pthread_mutex_destroy(&xa->writenMutex); /* On Linux this does nothing, but returns an error code EBUSY if the mutex was locked */
   if (rc != 0) /* EBUSY */
      xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_WARN, __FILE__, "pthread_mutex_destroy(writenMutex) returned %d, we ignore it", rc);

   rc = pthread_mutex_destroy(&xa->readnMutex);
   if (rc != 0) /* EBUSY */
      xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_WARN, __FILE__, "pthread_mutex_destroy(readnMutex) returned %d, we ignore it", rc);

   free(xa);
}

static bool initialize(XmlBlasterAccessUnparsed *xa, UpdateFp clientUpdateFp, XmlBlasterException *exception)
{
   int threadRet = 0;
   const char *compressType = 0;

   if (checkArgs(xa, "initialize", false, exception) == false) return false;

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

   if (clientUpdateFp == 0) {
      xa->clientsUpdateFp = 0;
      xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_INFO, "",
        "Your callback UpdateFp pointer is NULL, we use our default callback handler");
   }
   else {
      xa->clientsUpdateFp = clientUpdateFp;
   }

   if (xa->connectionP) {
      freeXmlBlasterConnectionUnparsed(&xa->connectionP);
   }
   xa->connectionP = getXmlBlasterConnectionUnparsed(xa->argc, xa->argv);
   if (xa->connectionP == 0) {
      strncpy0(exception->errorCode, "resource.outOfMemory", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
               "[%.100s:%d] Creating XmlBlasterConnectionUnparsed failed", __FILE__, __LINE__);
      if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
      return false;
   }
   xa->connectionP->log = xa->log;
   xa->connectionP->logUserP = xa->logUserP;
   xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "Created XmlBlasterConnectionUnparsed");


   /* Switch on compression? */
   compressType = xa->props->getString(xa->props, "plugin/socket/compress/type", "");
   compressType = xa->props->getString(xa->props, "dispatch/connection/plugin/socket/compress/type", compressType);
         
   if (!strcmp(compressType, "zlib:stream")) {
      xa->connectionP->writeToSocket.writeToSocketFuncP = writenCompressed;
      xa->connectionP->writeToSocket.userP = xa;
      xa->connectionP->readFromSocket.readFromSocketFuncP = readnCompressed;
      xa->connectionP->readFromSocket.userP = xa;
   }
   else {
      if (strcmp(compressType, "")) {
         xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "Unsupported compression type 'plugin/socket/compress/type=%s', falling back to plain mode", compressType);
      }
      xa->connectionP->writeToSocket.writeToSocketFuncP = writenPlain;
      xa->connectionP->writeToSocket.userP = xa;
      xa->connectionP->readFromSocket.readFromSocketFuncP = readnPlain;
      xa->connectionP->readFromSocket.userP = xa;
   }

   if (xa->connectionP->initConnection(xa->connectionP, exception) == false) /* Establish low level IP connection */
      return false;

   /* the fourth arg 'xa' is returned as 'void *userData' in update() method */
   if (xa->callbackP != 0) {
      freeCallbackServerUnparsed(&xa->callbackP);
   }
   xa->callbackP = getCallbackServerUnparsed(xa->argc, xa->argv, interceptUpdate, xa);
   if (xa->callbackP == 0) {
      strncpy0(exception->errorCode, "resource.outOfMemory", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
               "[%.100s:%d] Creating CallbackServerUnparsed failed", __FILE__, __LINE__);
      if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
      freeXmlBlasterConnectionUnparsed(&xa->connectionP);
      return false;
   }
   xa->callbackP->log = xa->log;
   xa->callbackP->logUserP = xa->logUserP;

   if (!strcmp(compressType, "zlib:stream")) {
      xa->callbackP->writeToSocket.writeToSocketFuncP = writenCompressed;
      xa->callbackP->writeToSocket.userP = xa;
      xa->callbackP->readFromSocket.readFromSocketFuncP = readnCompressed;
      xa->callbackP->readFromSocket.userP = xa;
   }
   else {
      xa->callbackP->writeToSocket.writeToSocketFuncP = writenPlain;
      xa->callbackP->writeToSocket.userP = xa;
      xa->callbackP->readFromSocket.readFromSocketFuncP = readnPlain;
      xa->callbackP->readFromSocket.userP = xa;
   }

   xa->callbackP->useThisSocket(xa->callbackP, xa->connectionP->socketToXmlBlaster, xa->connectionP->socketToXmlBlasterUdp);

   xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_INFO, __FILE__,
          "Created CallbackServerUnparsed instance, creating on a separate thread a listener on socket://%s:%d...",
          (xa->callbackP->hostCB == 0) ? "" : xa->callbackP->hostCB, xa->callbackP->portCB);

   /* Register our callback funtion which is called just before sending a message */
   xa->connectionP->preSendEvent = preSendEvent;
   xa->connectionP->preSendEvent_userP = xa;

   /* Register our callback funtion which is called just after sending a message */
   xa->connectionP->postSendEvent = postSendEvent;
   xa->connectionP->postSendEvent_userP = xa;

   /* thread blocks on socket listener */
   threadRet = pthread_create(&xa->callbackThreadId, (const pthread_attr_t *)0, (void * (*)(void *))xa->callbackP->runCallbackServer, (void *)xa->callbackP);
   if (threadRet != 0) {
      strncpy0(exception->errorCode, "resource.tooManyThreads", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
               "[%.100s:%d] Creating thread failed with error number %d",
               __FILE__, __LINE__, threadRet);
      if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
      freeCallbackServerUnparsed(&xa->callbackP);
      freeXmlBlasterConnectionUnparsed(&xa->connectionP);
      return false;
   }

   xa->isInitialized = true;
   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
                                "initialize() successful");
   return xa->isInitialized;
}

static bool isConnected(XmlBlasterAccessUnparsed *xa)
{
   if (xa == 0 || xa->isShutdown || xa->connectionP == 0) {
      return false;
   }
   return xa->connectionP->isConnected(xa->connectionP);
}

static MsgRequestInfo *preSendEvent(MsgRequestInfo *msgRequestInfoP, XmlBlasterException *exception)
{
   bool retVal;
   XmlBlasterAccessUnparsed * const xa = (XmlBlasterAccessUnparsed *)msgRequestInfoP->xa;

   /* if (!strcmp(XMLBLASTER_PUBLISH_ONEWAY, msgRequestInfoP->methodName)) */
   if (xbl_isOneway(MSG_TYPE_INVOKE, msgRequestInfoP->methodName))
      return msgRequestInfoP;

   /* ======== Initialize threading ====== */
   msgRequestInfoP->responseMutexIsLocked = false; /* Only to remember if the client thread holds the lock */

   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
                                "preSendEvent(%s) occurred", msgRequestInfoP->methodName);
   retVal = xa->callbackP->addResponseListener(xa->callbackP, msgRequestInfoP, responseEvent);
   if (retVal == false) {
      strncpy0(exception->errorCode, "user.internal", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
               "[%.100s:%d] Couldn't register as response listener", __FILE__, __LINE__);
      if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
      return (MsgRequestInfo *)0;
   }

   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
                  "preSendEvent(requestId=%s, msgRequestInfoP->responseBlob.dataLen=%d), entering lock",
                  msgRequestInfoP->requestIdStr, msgRequestInfoP->responseBlob.dataLen);
   pthread_mutex_init(&msgRequestInfoP->responseMutex, NULL); /* returns always 0 */
   if ((retVal = pthread_mutex_lock(&msgRequestInfoP->responseMutex)) != 0) {
      strncpy0(exception->errorCode, "user.internal", XMLBLASTEREXCEPTION_ERRORCODE_LEN);
      SNPRINTF(exception->message, XMLBLASTEREXCEPTION_MESSAGE_LEN,
               "[%.100s:%d] Error trying to lock responseMutex %d", __FILE__, __LINE__, retVal);
      if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, exception->message);
      return (MsgRequestInfo *)0;
   }
   msgRequestInfoP->responseMutexIsLocked = true; /* Only if the client thread holds the lock */

   return msgRequestInfoP;
}

static void responseEvent(MsgRequestInfo *msgRequestInfoP, void /*SocketDataHolder*/ *socketDataHolder) {
   int retVal;
   SocketDataHolder *s = (SocketDataHolder *)socketDataHolder;
   XmlBlasterAccessUnparsed *xa = (XmlBlasterAccessUnparsed *)msgRequestInfoP->xa;

   if ((retVal = pthread_mutex_lock(&msgRequestInfoP->responseMutex)) != 0) {
      xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "Trying to lock responseMutex in responseEvent() failed %d", retVal);
      /* return; */
   }
   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "responseEvent() responseMutex is LOCKED");

   blobcpyAlloc(&msgRequestInfoP->responseBlob, s->blob.data, s->blob.dataLen);
   msgRequestInfoP->responseType = s->type;

   if ((retVal = pthread_cond_signal(&msgRequestInfoP->responseCond)) != 0) {
      xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "Trying to signal waiting thread in responseEvent() fails %d", retVal);
      /* return; */
   }

   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__,
                                "responseEvent(requestId '%s', msgType=%c, dataLen=%d) occurred, wake up signal sent",
                                s->requestId, msgRequestInfoP->responseType, msgRequestInfoP->responseBlob.dataLen);

   if ((retVal = pthread_mutex_unlock(&msgRequestInfoP->responseMutex)) != 0) {
      xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_ERROR, __FILE__, "Trying to unlock responseMutex in responseEvent() failed %d", retVal);
      /* return; */
   }
   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "responseEvent() responseMutex is UNLOCKED");
}

static MsgRequestInfo *postSendEvent(MsgRequestInfo *msgRequestInfoP, XmlBlasterException *exception)
{
   XmlBlasterAccessUnparsed *xa = (XmlBlasterAccessUnparsed *)msgRequestInfoP->xa;
   struct timespec abstime;
   bool useTimeout = false;
   int retVal;

   if (msgRequestInfoP->rollback) {
      mutexUnlock(msgRequestInfoP, exception);
      return (MsgRequestInfo *)0;
   }

   if (xa->responseTimeout > 0 && getAbsoluteTime(xa->responseTimeout, &abstime) == true) {
      useTimeout = true;
   }

   if (xa->logLevel>=XMLBLASTER_LOG_TRACE) xa->log(xa->logUserP, xa->logLevel, XMLBLASTER_LOG_TRACE, __FILE__, "postSendEvent(requestId=%s) responseMutex is LOCKED, entering wait ...", msgRequestInfoP->requestIdStr);
   
   if ((retVal = pthread_cond_init(&msgRequestInfoP->responseCond, NULL)) != 0) {
      xa->