text: /*##############################################################################
  
      HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.
  
      Licensed under the Apache License, Version 2.0 (the "License");
      you may not use this file except in compliance with the License.
      You may obtain a copy of the License at
  
         http://www.apache.org/licenses/LICENSE-2.0
  
      Unless required by applicable law or agreed to in writing, software
      distributed under the License is distributed on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
      See the License for the specific language governing permissions and
      limitations under the License.
  ############################################################################## */
  
  #include "jliball.hpp"
  #include "jqueue.tpp"
  #include "jisem.hpp"
  
  #include "rtlformat.hpp"
  
  #include "thorxmlread.hpp"
  #include "thorxmlwrite.hpp"
  #include "thorcommon.ipp"
  #include "thorsoapcall.hpp"
  
  #include "securesocket.hpp"
  #include "eclrtl.hpp"
  #include "roxiemem.hpp"
  #ifdef _USE_ZLIB
  #include "zcrypt.hpp"
  #endif
  
  using roxiemem::OwnedRoxieString;
  
  #ifndef _WIN32
  #include <stdexcept>
  #endif
  
  #include <new>
  
  #define CONTENT_LENGTH "Content-Length: "
  #define CONTENT_ENCODING "Content-Encoding"
  #define ACCEPT_ENCODING "Accept-Encoding"
  
  unsigned soapTraceLevel = 1;
  
  #define WSCBUFFERSIZE 0x10000
  #define MAXWSCTHREADS 50    //Max Web Service Call Threads
  
  interface DECL_EXCEPTION IReceivedRoxieException : extends IException
  {
  public:
      virtual const void *errorRow() = 0;
  };
  
  #define EXCEPTION_PREFIX "ReceivedRoxieException:"
  
  class DECL_EXCEPTION ReceivedRoxieException: public IReceivedRoxieException, public CInterface
  {
  public:
      IMPLEMENT_IINTERFACE;
  
      ReceivedRoxieException(int code, const char *_msg, const void *_row=NULL) : errcode(code), msg(_msg), row(_row) { };
      int errorCode() const { return (errcode); };
      StringBuffer &  errorMessage(StringBuffer &str) const
      {
          if (strncmp(str.str(), EXCEPTION_PREFIX, strlen(EXCEPTION_PREFIX)) == 0)
              str.append(msg);
          else
              str.append(EXCEPTION_PREFIX).append(" (").append(msg).append(")");
  
          return str;
      };
      MessageAudience errorAudience() const { return (MSGAUD_user); };
      const void *errorRow() { return (row); };
  
  private:
      int errcode;
      StringAttr msg;
      const void *row;
  };
  
  //=================================================================================================
  
  class Url : public CInterface, implements IInterface
  {
  public:
      IMPLEMENT_IINTERFACE;
  
      StringAttr method;
      StringAttr host;
      unsigned port;
      StringBuffer path;
      StringBuffer userPasswordPair;
  
      StringBuffer &getUrlString(StringBuffer &url) const
      {
          return url.append(method).append("://").append(host).append(":").append(port).append(path);
      }
  
      IException *getUrlException(IException *e) const
      {
          StringBuffer url;
          StringBuffer text;
          e->errorMessage(text);
          rtlAddExceptionTag(text, "url", getUrlString(url).str());
          if (text.length() <= 1024)
              return MakeStringException(e->errorCode(), "%s", text.str());
          else
              return MakeStringExceptionDirect(e->errorCode(), text.str());
      }
  
      Url() : port(0)
      {
      }
  
  private:
      char translateHex(char hex) {
          if(hex >= 'A')
              return (hex & 0xdf) - 'A' + 10;
          else
              return hex - '0';
      }
  
      void userPassword_decode(const char* userpass, StringBuffer& result)
      {
          if(!userpass || !*userpass)
              return;
  
          const char *finger = userpass;
          while (*finger)
          {
              char c = *finger++;
              if (c == '%')
              {
                  if(*finger != '\0')
                  {
                      c = translateHex(*finger);
                      finger++;
                  }
                  if(*finger != '\0')
                  {
                      c = (char)(c*16 + translateHex(*finger));
                      finger++;
                  }
              }
              result.append(c);
          }
  
          return;
      }
  
  public:
      Url(char *urltext)
      {
          char *p;
  
          if ((p = strstr(urltext, "://")) != NULL)
          {
              *p = 0;
              p += 3; // skip past the colon-slash-slash
              method.set(urltext);
              urltext = p;
          }
          else
              throw MakeStringException(-1, "Malformed URL");
  
          if ((p = strchr(urltext, '@')) != NULL)
          {
              // extract username & password
              *p = 0;
              p++;
  
              userPassword_decode(urltext, userPasswordPair);
              urltext = p;
          }
  
          if ((p = strchr(urltext, ':')) != NULL)
          {
              // extract the port
              *p = 0;
              p++;
  
              port = atoi(p);
  
              host.set(urltext);
  
              if ((p = strchr(p, '/')) != NULL)
                  path.append(p);
              else
                  path.append("/");
          }
          else
          {
              // no port - look at method for port
              if (stricmp(method.str(), "https") == 0)
                  port = 443;
              else if (stricmp(method.str(), "http") == 0)
                  port = 80;
              else
                  throw MakeStringException(-1, "Unsupported access method");
  
              if ((p = strchr(urltext, '/')) != NULL)
              {
                  *p = 0;
                  p++;
  
                  host.set(urltext);
                  path.append("/").append(p);
              }
              else
              {
                  host.set(urltext);
                  path.append("/");
              }
          }
          // While the code below would make some sense, there is code that relies on being able to use invalid IP addresses and catching the
          // errors that result via ONFAIL.
  #if 0
          IpAddress ipaddr(host);
          if ( ipaddr.isNull())
              throw MakeStringException(-1, "Invalid IP address %s", host.str());
  #endif
      }
  };
  
  typedef IArrayOf<Url> UrlArray;
  
  //=================================================================================================
  
  //MORE: This whole class should really be replaced with a single function, avoiding at least one string clone.
  class UrlListParser
  {
  public:
      UrlListParser(const char * text)
      {
          fullText = text ? strdup(text) : NULL;
      }
  
      ~UrlListParser()
      {
          free(fullText);
      }
  
      unsigned getUrls(UrlArray &array)
      {
          if (fullText)
          {
              char *copyFullText = strdup(fullText);
  
              char *saveptr;
              char *url = strtok_r(copyFullText, "|", &saveptr);
              while (url != NULL)
              {
                  array.append(*new Url(url));
                  url = strtok_r(NULL, "|", &saveptr);
              }
  
              free(copyFullText);
          }
          return array.ordinality();
      }
  
  private:
      char *fullText;
  };
  
  //=================================================================================================
  
  #define BLACKLIST_RETRIES 10
  #define ROXIE_ABORT_EVENT  1407
  #define TIMELIMIT_EXCEEDED 1408
  
  class BlackLister : public CInterface, implements IThreadFactory
  {
      SocketEndpointArray list;
      Owned<IThreadPool> pool;
      CriticalSection crit;
  
  private:
      inline void checkRoxieAbortMonitor(IRoxieAbortMonitor * roxieAbortMonitor)
      {
          if (roxieAbortMonitor)
          {
              try 
              {
                  roxieAbortMonitor->checkForAbort();//throws
              }
              catch (IException *e)
              {
                  StringBuffer s;
                  e->errorMessage(s);
                  e->Release();
                  throw MakeStringException(ROXIE_ABORT_EVENT, "%s", s.str());
              }
          }
      }
  
  public:
      bool lookup(SocketEndpoint &ep, const IContextLogger &logctx)
      {
          CriticalBlock b(crit);
          if (list.find(ep)!=NotFound)
          {
              if (soapTraceLevel > 3)
              {
                  StringBuffer s;
                  logctx.CTXLOG("socket %s is blacklisted", ep.getUrlStr(s).str());
              }
              return true;
          }
          return false;
  
      }
      void blacklist(SocketEndpoint &ep, const IContextLogger &logctx)
      {
          CriticalBlock b(crit);
          if (list.find(ep)==NotFound)
          {
              if (soapTraceLevel > 0)
              {
                  StringBuffer s;
                  logctx.CTXLOG("Blacklisting socket %s", ep.getUrlStr(s).str());
              }
              list.append(ep);
              pool->start(&ep);
          }
      }
      void deblacklist(SocketEndpoint &ep)
      {
          CriticalBlock b(crit);
          unsigned idx = list.find(ep);
          if (idx!=NotFound)
          {
              if (soapTraceLevel > 0)
              {
                  StringBuffer s;
                  DBGLOG("De-blacklisting socket %s", ep.getUrlStr(s).str());
              }
              list.remove(idx);
          }
      }
  
  public:
      IMPLEMENT_IINTERFACE;
  
      BlackLister()
      {
          pool.setown(createThreadPool("SocketBlacklistPool", this, NULL, 0, 0));
      }
  
      ISocket* connect(SocketEndpoint &ep,
                       const IContextLogger &logctx,
                       unsigned retries,
                       unsigned timeoutMS,
                       IRoxieAbortMonitor * roxieAbortMonitor)
      {
          if (lookup(ep, logctx))
          {
              StringBuffer s;
              ep.getUrlStr(s);
              throw MakeStringException(-1, "blacklisted socket %s", s.str());
          }
          Owned<IException> exc;
          try
          {
              checkRoxieAbortMonitor(roxieAbortMonitor);
              Owned<ISocket> sock;
              Owned<ISocketConnectWait> scw = nonBlockingConnect(ep, timeoutMS == WAIT_FOREVER ? 60000 : timeoutMS*(retries+1));
              for (;;)
              {
                  sock.setown(scw->wait(1000));//throws if connect fails or timeoutMS
                  checkRoxieAbortMonitor(roxieAbortMonitor);
                  if (sock)
                      return sock.getLink();
              }
          }
  
          catch (IJSOCK_Exception *e)
          {
              EXCLOG(e,"BlackLister::connect");
              if (exc)
                  e->Release();
              else
                  exc.setown(e);
          }
  
          blacklist(ep, logctx);
          if (exc->errorCode()==JSOCKERR_connection_failed) {
              StringBuffer s;
              ep.getUrlStr(s);
              throw MakeStringException(JSOCKERR_connection_failed, "connection failed %s", s.str());
          }
          throw exc.getClear();
          return NULL;
      }
  
      bool blacklisted (unsigned short port, char const* host)
      {
          SocketEndpoint ep(host, port);
          return (list.find(ep)!=NotFound);
      }
  
      ISocket* connect(unsigned short port,
                       char const* host,
                       const IContextLogger &logctx,
                       unsigned retries,
                       unsigned timeoutMS,
                       IRoxieAbortMonitor * roxieAbortMonitor )
      {
          SocketEndpoint ep(host, port);
          return connect(ep, logctx, retries, timeoutMS, roxieAbortMonitor);
      }
  
      virtual IPooledThread *createNew()
      {
          class SocketDeblacklister : public CInterface, implements IPooledThread
          {
              SocketEndpoint ep;
              BlackLister &parent;
              Semaphore stopped;
          public:
              IMPLEMENT_IINTERFACE;
              SocketDeblacklister(BlackLister &_parent): parent(_parent)
              {
              }
              ~SocketDeblacklister()
              {
              }
  
              virtual void init(void *param) override
              {
                  ep.set(*(SocketEndpoint *) param);
              }
              virtual void threadmain() override
              {
                  unsigned delay = 5000;
                  for (unsigned i = 0; i < BLACKLIST_RETRIES; i++)
                  {
                      try
                      {
                          Owned<ISocket> s = ISocket::connect_timeout(ep, 10000);
                          s->close();
                          break;
                      }
                      catch (IJSOCK_Exception *E)
                      {
                          // EXCLOG(E, "While updating socket blacklist");  // MORE - may need to downgrade if this fires traps
                          E->Release();
                          if (stopped.wait(delay))
                              return;
                          delay += delay;
                      }
                  }
                  parent.deblacklist(ep);
              }
              virtual bool stop() override
              {
                  stopped.signal();
                  return true;
              }
              virtual bool canReuse() const override { return true; }
          };
          return new SocketDeblacklister(*this);
      }
  
      void stop()
      {
          pool->stopAll();
      }
  } *blacklist;
  
  MODULE_INIT(INIT_PRIORITY_STANDARD)
  {
      blacklist = new BlackLister;
  
      return true;
  }
  
  MODULE_EXIT()
  {
      blacklist->stop();
      delete blacklist;
  }
  
  //=================================================================================================
  
  class ColumnProvider : public IColumnProvider, public CInterface
  {
  public:
      ColumnProvider(unsigned _callLatencyMs) : callLatencyMs(_callLatencyMs), base(NULL) {}
      IMPLEMENT_IINTERFACE;
      virtual bool        getBool(const char * path) { return base->getBool(path); }
      virtual void        getData(size32_t len, void * text, const char * path) { base->getData(len, text, path); }
      virtual void        getDataX(size32_t & len, void * & text, const char * path) { base->getDataX(len, text, path); }
      virtual __int64     getInt(const char * path)
      {
          __int64 ret = base->getInt(path);
          if((ret==0) && path && *path=='_')
          {
              if (stricmp(path, "_call_latency_ms")==0)
                  ret = callLatencyMs;
              else if (stricmp(path, "_call_latency")==0)
                  ret = (callLatencyMs + 500) / 1000;
          }
          return ret;
      }
      virtual void        getQString(size32_t len, char * text, const char * path) { base->getQString(len, text, path); }
      virtual void        getString(size32_t len, char * text, const char * path) { base->getString(len, text, path); }
      virtual void        getStringX(size32_t & len, char * & text, const char * path) { base->getStringX(len, text, path); }
      virtual void        getUnicodeX(size32_t & len, UChar * & text, const char * path) { base->getUnicodeX(len, text, path); }
      virtual bool        getIsSetAll(const char * path) { return base->getIsSetAll(path); }
      virtual IColumnProviderIterator * getChildIterator(const char * path) { return base->getChildIterator(path); }
      virtual void        getUtf8X(size32_t & len, char * & text, const char * path) { base->getUtf8X(len, text, path); }
  
      virtual bool        readBool(const char * path, bool _default) { return base->readBool(path, _default); }
      virtual void        readData(size32_t len, void * text, const char * path, size32_t _lenDefault, const void * _default) { base->readData(len, text, path, _lenDefault, _default); }
      virtual void        readDataX(size32_t & len, void * & text, const char * path, size32_t _lenDefault, const void * _default) { base->readDataX(len, text, path, _lenDefault, _default); }
      virtual __int64     readInt(const char * path, __int64 _default)
      {
          if(path && *path=='_')
          {
              if (stricmp(path, "_call_latency_ms")==0)
                  return callLatencyMs;
              else if (stricmp(path, "_call_latency")==0)
                  return (callLatencyMs + 500) / 1000;
          }
          return base->readInt(path, _default);
      }
      virtual void        readQString(size32_t len, char * text, const char * path, size32_t _lenDefault, const char * _default) { base->readQString(len, text, path, _lenDefault, _default); }
      virtual void        readString(size32_t len, char * text, const char * path, size32_t _lenDefault, const char * _default) { base->readString(len, text, path, _lenDefault, _default); }
      virtual void        readStringX(size32_t & len, char * & text, const char * path, size32_t _lenDefault, const char * _default) { base->readStringX(len, text, path, _lenDefault, _default); }
      virtual void        readUnicodeX(size32_t & len, UChar * & text, const char * path, size32_t _lenDefault, const UChar * _default) { base->readUnicodeX(len, text, path, _lenDefault, _default); }
      virtual bool        readIsSetAll(const char * path, bool _default) { return base->readIsSetAll(path, _default); }
      virtual void        readUtf8X(size32_t & len, char * & text, const char * path, size32_t _lenDefault, const char * _default) { base->readUtf8X(len, text, path, _lenDefault, _default); }
  
      void setBase(IColumnProvider * _base) { base = _base; }
  
      virtual void        getDataRaw(size32_t len, void * text, const char * path) { base->getDataRaw(len, text, path); }
      virtual void        getDataRawX(size32_t & len, void * & text, const char * path) { base->getDataRawX(len, text, path); }
      virtual void        readDataRaw(size32_t len, void * text, const char * path, size32_t _lenDefault, const void * _default) { base->readDataRaw(len, text, path, _lenDefault, _default); }
      virtual void        readDataRawX(size32_t & len, void * & text, const char * path, size32_t _lenDefault, const void * _default) { base->readDataRawX(len, text, path, _lenDefault, _default); }
  
  protected:
      IColumnProvider * base;
  
  private:
      unsigned callLatencyMs;
  
  };
  
  //=================================================================================================
  //Same as ColumnProvider, except returns data as hex64Binary instead of 16 bit hex
  class ColumnProviderData64 : public ColumnProvider
  {
  public:
      ColumnProviderData64(unsigned _callLatencyMs) : ColumnProvider(_callLatencyMs) {}
      virtual void getData(size32_t len, void * text, const char * path)      { ColumnProvider::getDataRaw(len, text, path);  }
      virtual void getDataX(size32_t & len, void * & text, const char * path) { ColumnProvider::getDataRawX(len, text, path);     }
      virtual void readData(size32_t len, void * text, const char * path, size32_t _lenDefault, const void * _default) { ColumnProvider::readDataRaw(len, text, path, _lenDefault, _default); }
      virtual void readDataX(size32_t len, void * text, const char * path, size32_t _lenDefault, const void * _default) { ColumnProvider::readDataRawX(len, text, path, _lenDefault, _default); }
  };
  
  //=================================================================================================
  
  IColumnProvider * CreateColumnProvider(unsigned _callLatencyMs, bool _encoding)
  {
      if (!_encoding)
          return new ColumnProvider(_callLatencyMs);
      else
          return new ColumnProviderData64(_callLatencyMs);
  }
  
  //=================================================================================================
  
  
  enum WSCType{STsoap, SThttp} ;  //web service call type
  
  //Web Services Call Asynchronous For
  interface IWSCAsyncFor: public IInterface
  {
      virtual void processException(const Url &url, const void *row, IException *e) = 0;
      virtual void processException(const Url &url, ConstPointerArray &inputRows, IException *e) = 0;
      virtual void checkTimeLimitExceeded(unsigned * _remainingMS) = 0;
  
      virtual void createHttpRequest(Url &url, StringBuffer &request) = 0;
      virtual int readHttpResponse(StringBuffer &response, ISocket *socket) = 0;
      virtual void processResponse(Url &url, StringBuffer &response, ColumnProvider * meta) = 0;
  
      virtual StringBuffer getResponsePath() = 0;
      virtual ConstPointerArray & getInputRows() = 0;
      virtual IWSCHelper * getMaster() = 0;
      virtual IEngineRowAllocator * getOutputAllocator() = 0;
      
      virtual void For(unsigned num,unsigned maxatonce,bool abortFollowingException=false,bool shuffled=false) = 0;
      virtual void Do(unsigned idx=0)=0;
  };
  
  class CWSCHelper;
  IWSCAsyncFor * createWSCAsyncFor(CWSCHelper * _master, CommonXmlWriter &_xmlWriter, ConstPointerArray &_inputRows, PTreeReaderOptions _options);
  
  //=================================================================================================
  
  class CMatchCB : implements IXMLSelect, public CInterface
  {
      IWSCAsyncFor &parent;
      const Url &url;
      StringAttr tail;
      ColumnProvider * meta;
  public:
      IMPLEMENT_IINTERFACE;
  
      CMatchCB(IWSCAsyncFor &_parent, const Url &_url, const char *_tail, ColumnProvider * _meta) : parent(_parent), url(_url), tail(_tail), meta(_meta)
      {
      }
  
      virtual void match(IColumnProvider &entry, offset_t startOffset, offset_t endOffset)
      {
          Owned<IException> e;
          if (tail.length())
          {
              StringBuffer path(parent.getResponsePath());
              unsigned idx = (unsigned)entry.getInt(path.append("/@sequence").str());
              Owned<IColumnProviderIterator> excIter = entry.getChildIterator("Exception");
              IColumnProvider *excptEntry = excIter->first();
              if (excptEntry)
              {
                  int code = (int)excptEntry->getInt("Code");
                  size32_t len;
                  char *message;
                  excptEntry->getStringX(len, message, "Message");
                  StringBuffer mstr(len, message);
                  rtlFree(message);
                  DBGLOG("Roxie exception: %s", mstr.str());
                  e.setown(new ReceivedRoxieException(code, mstr.str()));
                  parent.processException(url, parent.getInputRows().item(idx), e.getLink());
              }
          }
          if (parent.getMaster()->getRowTransformer() && !e)
          {
              IEngineRowAllocator * outputAllocator = parent.getOutputAllocator();
              Owned<IColumnProviderIterator> iter = entry.getChildIterator(tail);
              IColumnProvider *rowEntry = iter->first();
              while (rowEntry)
              {
                  RtlDynamicRowBuilder rowBuilder(outputAllocator);
                  size32_t sizeGot;
                  if(meta)
                  {
                      meta->setBase(rowEntry);
                      sizeGot = parent.getMaster()->transformRow(rowBuilder, meta);
                  }
                  else
                  {
                      sizeGot = parent.getMaster()->transformRow(rowBuilder, rowEntry);
                  }
                  if (sizeGot > 0)
                      parent.getMaster()->putRow(rowBuilder.finalizeRowClear(sizeGot));
                  rowEntry = iter->next();
              }
          }
      }
  
  };
  
  //=================================================================================================
  
  bool isValidHttpValue(char const * in)
  {
      for (const byte * ptr = (byte*)in; *ptr; ++ptr)
          if (((*ptr <= '\x1F') && (*ptr != '\x09')) || (*ptr == '\x7F'))
              return false;
      return true;
  }
  
  //=================================================================================================
  //Web Service Call helper thread
  class CWSCHelperThread : public Thread
  {
  private:
      CWSCHelper * master;
      virtual void outputXmlRows(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows, const char *itemtag=NULL, bool encode_off=false, char const * itemns = NULL);
      virtual void createESPQuery(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows);
      virtual void createSOAPliteralOrEncodedQuery(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows);
      virtual void createXmlSoapQuery(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows);
      virtual void processQuery(ConstPointerArray &inputRows);
      
      //Thread
      virtual int run();
  public:
      CWSCHelperThread(CWSCHelper * _master) : Thread("CWSCHelperThread")
      {
          master = _master;
      }
  
      ~CWSCHelperThread()
      {
      }
  };
  
  //=================================================================================================
  
  class CWSCHelper : implements IWSCHelper, public CInterface
  {
  private:
      SimpleInterThreadQueueOf<const void, true> outputQ;
      SpinLock outputQLock;
      CriticalSection toXmlCrit, transformCrit, onfailCrit, timeoutCrit;
      unsigned done;
      Linked<ClientCertificate> clientCert;
  
      static CriticalSection secureContextCrit;
      static Owned<ISecureSocketContext> secureContext;
  
      CTimeMon timeLimitMon;
      bool complete, timeLimitExceeded;
      IRoxieAbortMonitor * roxieAbortMonitor;
  
  protected:
      CIArrayOf<CWSCHelperThread> threads;
      WSCType wscType;
  
  public:
      IMPLEMENT_IINTERFACE;
  
      CWSCHelper(IWSCRowProvider *_rowProvider, IEngineRowAllocator * _outputAllocator, const char *_authToken, WSCMode _wscMode, ClientCertificate *_clientCert, const IContextLogger &_logctx, IRoxieAbortMonitor *_roxieAbortMonitor, WSCType _wscType)
          : logctx(_logctx), outputAllocator(_outputAllocator), clientCert(_clientCert), roxieAbortMonitor(_roxieAbortMonitor)
      {
          wscMode = _wscMode;
          wscType = _wscType;
          done = 0;
          complete = aborted = timeLimitExceeded = false;
  
          rowProvider = _rowProvider;
          helper = rowProvider->queryActionHelper();
          callHelper = rowProvider->queryCallHelper();
          flags = helper->getFlags();
          OwnedRoxieString s;
  
          authToken.append(_authToken);
  
          if (helper->numRetries() < 0)
              maxRetries = 3; // default to 3 retries
          else
              maxRetries = helper->numRetries();
  
          //Allow all of these options to be specified separately.  Possibly useful, and the code is cleaner.
          logMin = (flags & SOAPFlogmin) != 0;
          logXML = (flags & SOAPFlog) != 0;
          logUserMsg = (flags & SOAPFlogusermsg) != 0;
  
          double dval = helper->getTimeout(); // In seconds, but may include fractions of a second...
          if (dval < 0.0) //not provided, or out of range
              timeoutMS = 300*1000; // 300 second default
          else if (dval == 0)
              timeoutMS = WAIT_FOREVER;
          else
              timeoutMS = (unsigned)(dval * 1000);
  
          dval = helper->getTimeLimit();
          if (dval <= 0.0)
              timeLimitMS = WAIT_FOREVER;
          else
              timeLimitMS = (unsigned)(dval * 1000);
  
          if (flags & SOAPFhttpheaders)
              httpHeaders.set(s.setown(helper->getHttpHeaders()));
  
          StringAttr proxyAddress;
          proxyAddress.set(s.setown(helper->getProxyAddress()));
          if (!proxyAddress.isEmpty())
          {
              UrlListParser proxyUrlListParser(proxyAddress);
              if (0 == proxyUrlListParser.getUrls(proxyUrlArray))
                  throw MakeStringException(0, "SOAPCALL PROXYADDRESS specified no URLs");
          }
  
          if (wscType == STsoap)
          {
              soapaction.set(s.setown(helper->getSoapAction()));
              if(soapaction.get() && !isValidHttpValue(soapaction.get()))
                  throw MakeStringException(-1, "SOAPAction value contained illegal characters: %s", soapaction.get());
  
              httpHeaderName.set(s.setown(helper->getHttpHeaderName()));
              if(httpHeaderName.get() && !isValidHttpValue(httpHeaderName.get()))
                  throw MakeStringException(-1, "HTTPHEADER name contained illegal characters: %s", httpHeaderName.get());
  
              httpHeaderValue.set(s.setown(helper->getHttpHeaderValue()));
              if(httpHeaderValue.get() && !isValidHttpValue(httpHeaderValue.get()))
                  throw MakeStringException(-1, "HTTPHEADER value contained illegal characters: %s", httpHeaderValue.get());
  
              if ((flags & SOAPFliteral) && (flags & SOAPFencoding))
                  throw MakeStringException(0, "SOAPCALL 'LITERAL' and 'ENCODING' options are mutually exclusive");
  
              header.set(s.setown(helper->getHeader()));
              footer.set(s.setown(helper->getFooter()));
              if(flags & SOAPFnamespace)
              {
                  OwnedRoxieString ns = helper->getNamespaceName();
                  if(ns && *ns)
                      xmlnamespace.set(ns);
              }
          }
  
          if (wscType == SThttp)
          {
              //Check for unsupported flags
              if ((flags & SOAPFliteral) || (flags & SOAPFencoding))
                  throw MakeStringException(0, "HTTPCALL 'LITERAL' and 'ENCODINGD' options not supported");
          }
  
          if (callHelper)
          {
              OwnedRoxieString iteratorPath(callHelper->getInputIteratorPath());
              char const * ipath = iteratorPath;
              if(ipath && (*ipath == '/'))
                  ++ipath;
              inputpath.set(ipath);
          }
          
          service.set(s.setown(helper->getService()));
          service.trim();
  
          if (wscType == SThttp)
          {
              service.toUpperCase();  //GET/PUT/POST
              if (strcmp(service.str(), "GET"))
                  throw MakeStringException(0, "HTTPCALL Only 'GET' http method currently supported");
              OwnedRoxieString acceptTypeSupplied(helper->getAcceptType()); // text/html, text/xml, etc
              acceptType.set(acceptTypeSupplied);
              acceptType.trim();
              acceptType.toLowerCase();
          }
  
          if (callHelper)
              rowTransformer = callHelper->queryInputTransformer();
          else
              rowTransformer = NULL;
  
          OwnedRoxieString hosts(helper->getHosts());
          UrlListParser urlListParser(hosts);
          if ((numUrls = urlListParser.getUrls(urlArray)) > 0)
          {
              if (wscMode == SCrow)
              {
                  numRowThreads = 1;
  
                  numUrlThreads = helper->numParallelThreads();
                  if (numUrlThreads == 0)
                      numUrlThreads = 1;
                  else if (numUrlThreads > MAXWSCTHREADS)
                      numUrlThreads = MAXWSCTHREADS;
  
                  numRecordsPerBatch = 1;
              }
              else
              {
                  unsigned totThreads = helper->numParallelThreads();
                  if (totThreads < 1)
                      totThreads = 2; // default to 2 threads
                  else if (totThreads > MAXWSCTHREADS)
                      totThreads = MAXWSCTHREADS;
  
                  numUrlThreads = (numUrls < totThreads)? numUrls: totThreads;
  
                  numRowThreads = totThreads / numUrlThreads;
                  if (numRowThreads < 1)
                      numRowThreads = 1;
                  else if (numRowThreads > MAXWSCTHREADS)
                      numRowThreads = MAXWSCTHREADS;
  
                  numRecordsPerBatch = helper->numRecordsPerBatch();
                  if (numRecordsPerBatch < 1)
                      numRecordsPerBatch = 1;
              }
          }
          else
              throw MakeStringException(0, "%sCALL specified no URLs",wscType == STsoap ? "SOAP" : "HTTP");
  
          for (unsigned i=0; i<numRowThreads; i++)
              threads.append(*new CWSCHelperThread(this));
      }
      ~CWSCHelper()
      {
          complete = true;
          waitUntilDone();
          threads.kill();
      }
      void waitUntilDone()
      {
          ForEachItemIn(i,threads)
              threads.item(i).join();
          for (;;)
          {
              const void *row = outputQ.dequeueNow();
              if (!row)
                  break;
              outputAllocator->releaseRow(row);
          }
          outputQ.reset();
      }
      void start()
      {
          if (timeLimitMS != WAIT_FOREVER)
              timeLimitMon.reset(timeLimitMS);
  
          done = 0;
          complete = aborted = timeLimitExceeded = false;
  
          ForEachItemIn(i,threads)
              threads.item(i).start();
      }
      void abort()
      {
          aborted = true;
          complete = true;
          outputQ.stop();
      }
      const void * getRow()
      {
          if (complete)
              return NULL;
          for (;;)
          {
              const void *row = outputQ.dequeue();
              if (aborted)
                  break;
              if (row)
                  return row;
              // should only be here if setDone() triggered
              complete = true;
              Owned<IException> e = getError();
              if (e)
                  throw e.getClear();
              break;
          }
          return NULL;
      }
      IException * getError()
      {
          SpinBlock sb(outputQLock);
          return error.getLink();
      }
      inline IEngineRowAllocator * queryOutputAllocator() const { return outputAllocator; }
  #ifdef _USE_OPENSSL
      ISecureSocket *createSecureSocket(ISocket *sock)
      {
          {
              CriticalBlock b(secureContextCrit);
              if (!secureContext)
              {
                  if (clientCert != NULL)
                      secureContext.setown(createSecureSocketContextEx(clientCert->certificate, clientCert->privateKey, clientCert->passphrase, ClientSocket));
                  else
                      secureContext.setown(createSecureSocketContext(ClientSocket));
              }
          }
          return secureContext->createSecureSocket(sock);
      }
  #endif
      bool isTimeLimitExceeded(unsigned *_remainingMS)
      {
          if (timeLimitMS != WAIT_FOREVER)
          {
              CriticalBlock block(timeoutCrit);
              if (timeLimitExceeded || timeLimitMon.timedout(_remainingMS))
              {
                  timeLimitExceeded = true;
                  return true;
              }
          }
          else
              *_remainingMS = (unsigned)-1;
          return false;
      }
  
      void addUserLogMsg(const byte * row)
      {
          if (logUserMsg)
          {
              size32_t lenText;
              rtlDataAttr text;
              helper->getLogText(lenText, text.refstr(), row);
              logctx.CTXLOG("%s: %.*s", wscCallTypeText(), lenText, text.getstr());
          }
      }
      inline IXmlToRowTransformer * getRowTransformer() { return rowTransformer; }
      inline const char * wscCallTypeText() const { return wscType == STsoap ? "SOAPCALL" : "HTTPCALL"; }
  
  protected:
      friend class CWSCHelperThread;
      friend class CWSCAsyncFor;
  
      void putRow(const void * row)
      {
          assertex(row);
          outputQ.enqueue(row);
      }
      void setDone()
      {
          bool doStop;
          {
              SpinBlock sb(outputQLock);
              done++;
              doStop = (done == numRowThreads);
          }
          if (doStop)
          {
              // Note - Don't stop the queue - that effectively discards what's already on there,
              // which is not what we want.
              // Instead, push a NULL to indicate the end of the output.
              outputQ.enqueue(NULL);
          }
      }
      void setErrorOwn(IException * e)
      {
          SpinBlock sb(outputQLock);
          if (error)
              ::Release(e);
          else
              error.setown(e);
      }
      void toXML(const byte * self, IXmlWriter & out) { CriticalBlock block(toXmlCrit); helper->toXML(self, out); }
      size32_t transformRow(ARowBuilder & rowBuilder, IColumnProvider * row) 
      { 
          CriticalBlock block(transformCrit); 
          NullDiskCallback callback;
          return rowTransformer->transform(rowBuilder, row, &callback); 
      }
      unsigned onFailTransform(ARowBuilder & rowBuilder, const void * left, IException * e) { CriticalBlock block(onfailCrit); return callHelper->onFailTransform(rowBuilder, left, e); }
  
      StringBuffer authToken;
      WSCMode wscMode;
      IWSCRowProvider * rowProvider;
      IHThorWebServiceCallActionArg * helper;
      IHThorWebServiceCallArg *   callHelper;
      Linked<IEngineRowAllocator> outputAllocator;
      Owned<IException> error;
      UrlArray urlArray;
      UrlArray proxyUrlArray;
      unsigned numRecordsPerBatch;
      unsigned numUrls;
      unsigned numRowThreads;
      unsigned numUrlThreads;
      unsigned maxRetries;
      unsigned timeoutMS;
      unsigned timeLimitMS;
      bool logXML;
      bool logMin;
      bool logUserMsg;
      bool aborted;
      const IContextLogger &logctx;
      unsigned flags;
      StringAttr soapaction;
      StringAttr httpHeaderName;
      StringAttr httpHeaderValue;
      StringAttr httpHeaders;
      StringAttr inputpath;
      StringBuffer service;
      StringBuffer acceptType;//for httpcall, text/plain, text/html, text/xml, etc
      StringAttr header;
      StringAttr footer;
      StringAttr xmlnamespace;
      IXmlToRowTransformer * rowTransformer;
  };
  
  CriticalSection CWSCHelper::secureContextCrit;
  Owned<ISecureSocketContext> CWSCHelper::secureContext; // created on first use
  
  //=================================================================================================
  
  void CWSCHelperThread::outputXmlRows(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows, const char *itemtag, bool encode_off, char const * itemns)
  {
      ForEachItemIn(idx, inputRows)
      {
          if (itemtag)                //TAG
          {
              xmlWriter.outputQuoted("<");
              xmlWriter.outputQuoted(itemtag);
              if(itemns)
              {
                  xmlWriter.outputQuoted(" xmlns=\"");
                  xmlWriter.outputQuoted(itemns);
                  xmlWriter.outputQuoted("\"");
              }
              xmlWriter.outputQuoted(">");
          }
  
          if (master->header.get())   //OPTIONAL HEADER (specified by "HEADING" option)
              xmlWriter.outputQuoted(master->header.get());
  
                                      //XML ROW CONTENT
          master->toXML((const byte *)inputRows.item(idx), xmlWriter);
  
          if (master->footer.get())   //OPTION FOOTER
              xmlWriter.outputQuoted(master->footer.get());
  
          if (encode_off)             //ENCODING
              xmlWriter.outputQuoted("<encode_>0</encode_>");
  
          if (itemtag)                //CLOSE TAG
          {
              xmlWriter.outputQuoted("</");
              xmlWriter.outputQuoted(itemtag);
              xmlWriter.outputQuoted(">");
          }
  
          master->addUserLogMsg((const byte *)inputRows.item(idx));
      }
  }
  
  void CWSCHelperThread::createESPQuery(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows)
  {
      StringBuffer method_tag;
      method_tag.append(master->service).append("Request");
      StringAttr method_ns;
  
      if (inputRows.ordinality() > 1)
      {
          xmlWriter.outputQuoted("<");
          xmlWriter.outputQuoted(method_tag.str());
          xmlWriter.outputQuoted("Array");
          if (master->xmlnamespace.get())
          {
              xmlWriter.outputQuoted(" xmlns=\"");
              xmlWriter.outputQuoted(master->xmlnamespace.get());
              xmlWriter.outputQuoted("\"");
          }
          xmlWriter.outputQuoted(">");
      }
      else
      {
          if(master->xmlnamespace.get())
              method_ns.set(master->xmlnamespace.get());
      }
  
      outputXmlRows(xmlWriter, inputRows, method_tag.str(), (inputRows.ordinality() == 1), method_ns.get());
  
      if (inputRows.ordinality() > 1)
      {
          xmlWriter.outputQuoted("<encode_>0</encode_>");
          xmlWriter.outputQuoted("</");
          xmlWriter.outputQuoted(method_tag.str());
          xmlWriter.outputQuoted("Array>");
      }
  }
  
  //Create servce xml request body, with binding usage of either Literal or Encoded
  //Note that Encoded usage requires type encoding for data fields
  void CWSCHelperThread::createSOAPliteralOrEncodedQuery(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows)
  {
      xmlWriter.outputQuoted("<");
      xmlWriter.outputQuoted(master->service);
  
      if (master->flags & SOAPFencoding)
          xmlWriter.outputQuoted(" soapenv:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\"");
  
      if (master->xmlnamespace.get())
      {
          xmlWriter.outputQuoted(" xmlns=\"");
          xmlWriter.outputQuoted(master->xmlnamespace.get());
          xmlWriter.outputQuoted("\"");
      }
  
      xmlWriter.outputQuoted(">");
  
      outputXmlRows(xmlWriter, inputRows);
  
      xmlWriter.outputQuoted("</");
      xmlWriter.outputQuoted(master->service);
      xmlWriter.outputQuoted(">");
  }
  
  //Create SOAP body of http request
  void CWSCHelperThread::createXmlSoapQuery(CommonXmlWriter &xmlWriter, ConstPointerArray &inputRows)
  {
      xmlWriter.outputQuoted("<?xml version=\"1.0\" encoding=\"UTF-8\"?>");
      xmlWriter.outputQuoted("<soap:Envelope");
  
      xmlWriter.outputQuoted(" xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\"");
      if (master->flags & SOAPFencoding)
      {   //SOAP RPC/encoded.  'Encoded' usage includes type encoding 
          xmlWriter.outputQuoted(" xmlns:xsd=\"http://www.w3.org/2001/XMLSchema\"");
          xmlWriter.outputQuoted(" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"");
      }
      xmlWriter.outputQuoted(">");
  
      xmlWriter.outputQuoted("<soap:Body>");
  
      if (master->flags & SOAPFliteral  ||  master->flags & SOAPFencoding)
          createSOAPliteralOrEncodedQuery(xmlWriter, inputRows);
      else
          createESPQuery(xmlWriter, inputRows);
  
      xmlWriter.outputQuoted("</soap:Body></soap:Envelope>");
  }
  
  void CWSCHelperThread::processQuery(ConstPointerArray &inputRows)
  {
      unsigned xmlWriteFlags = 0;
      unsigned xmlReadFlags = ptr_ignoreNameSpaces;
      if (master->flags & SOAPFtrim)
          xmlWriteFlags |= XWFtrim;
      if ((master->flags & SOAPFpreserveSpace) == 0)
          xmlReadFlags |= ptr_ignoreWhiteSpace;
          XMLWriterType xmlType = !(master->flags & SOAPFencoding) ? WTStandard : WTEncodingData64; 
      CommonXmlWriter *xmlWriter = CreateCommonXmlWriter(xmlWriteFlags, 0, NULL, xmlType);
      if (master->wscType == STsoap)
          createXmlSoapQuery(*xmlWriter, inputRows);
  
      Owned<IWSCAsyncFor> casyncfor = createWSCAsyncFor(master, *xmlWriter, inputRows, (PTreeReaderOptions) xmlReadFlags);
      casyncfor->For(master->numUrls, master->numUrlThreads,false,true); // shuffle URLS for poormans load balance
      delete xmlWriter;
  }
  
  int CWSCHelperThread::run()
  {
      ConstPointerArray inputRows;
  
      if (master->wscMode == SCrow)
      {
          inputRows.append(NULL);
          try
          {
              processQuery(inputRows);
          }
          catch (IException *e)
          {
              master->setErrorOwn(e);
          }
          inputRows.pop();
      }
      else
      {
          // following a bit odd but preserving previous semantics (except fixing abort leak)
          for (;;)
          {
              try
              {
                  while (!master->complete && !master->error.get())
                  {
                      if (master->aborted) {
                          while (inputRows.ordinality() > 0)
                              master->rowProvider->releaseRow(inputRows.popGet());
                          return 0;
                      }
                      const void *r = master->rowProvider->getNextRow();
                      if (!r)
                          break;
                      inputRows.append(r);
                      if (inputRows.ordinality() >= master->numRecordsPerBatch)
                          break;
                  }
                  if (inputRows.ordinality() == 0)
                      break;
                  processQuery(inputRows);
              }
              catch (IException *e)
              {
                  master->setErrorOwn(e); // going to exit next time round
              }
  
              while (inputRows.ordinality() > 0)
                  master->rowProvider->releaseRow(inputRows.popGet());
          }
      }
  
      master->setDone();
  
      return 0;
  }
  
  //=================================================================================================
  
  IWSCHelper * createSoapCallHelper(IWSCRowProvider *r, IEngineRowAllocator * outputAllocator, const char *authToken, WSCMode wscMode, ClientCertificate *clientCert, const IContextLogger &logctx, IRoxieAbortMonitor * roxieAbortMonitor)
  {
      return new CWSCHelper(r, outputAllocator, authToken, wscMode, clientCert, logctx, roxieAbortMonitor, STsoap);
  }
  
  IWSCHelper * createHttpCallHelper(IWSCRowProvider *r, IEngineRowAllocator * outputAllocator, const char *authToken, WSCMode wscMode, ClientCertificate *clientCert, const IContextLogger &logctx, IRoxieAbortMonitor * roxieAbortMonitor)
  {
      return new CWSCHelper(r, outputAllocator, authToken, wscMode, clientCert, logctx, roxieAbortMonitor, SThttp);
  }
  
  //=================================================================================================
  bool httpHeaderBlockContainsHeader(const char *httpheaders, const char *header)
  {
      if (!httpheaders || !*httpheaders)
          return false;
      VStringBuffer match("\n%s:", header);
      const char *matchStart = match.str()+1;
      if (!strncmp(httpheaders, matchStart, strlen(matchStart)))
          return true;
      if (strstr(httpheaders, match))
          return true;
      return false;
  }
  
  bool getHTTPHeader(const char *httpheaders, const char *header, StringBuffer& value)
  {
      if (!httpheaders || !*httpheaders || !header || !*header)
          return false;
  
      const char* pHeader = strstr(httpheaders, header);
      if (!pHeader)
          return false;
  
      pHeader += strlen(header);
      if (*pHeader != ':')
          return getHTTPHeader(pHeader, header, value);
  
      pHeader++;
      const char* ppHeader = strchr(pHeader, '\n');
      if (!ppHeader)
          value.append(pHeader);
      else
          value.append(pHeader, 0, ppHeader - pHeader);
      value.trim();
      return true;
  }
  
  class CWSCAsyncFor : implements IWSCAsyncFor, public CInterface, public CAsyncFor
  {
      class CSocketDataProvider : public CInterface
      {
          const char * buffer;
          size32_t currLen;
          size32_t curPosn;
          ISocket * socket;
          unsigned timeoutMS;
      public:
          CSocketDataProvider(const char * _buffer, size32_t _curPosn, size32_t _currLen, ISocket * _sock, unsigned _timeout )
              : buffer(_buffer), currLen(_currLen), curPosn(_curPosn), socket(_sock), timeoutMS(_timeout)
          {
          }
          size32_t getBytes(char * buf, size32_t len)
          {
              size32_t count;
              if ( len <= (currLen-curPosn))
              {   //its already in the buffer
                  memcpy(buf, (buffer + curPosn), len);
                  curPosn += len;
                  count = len;
              }
              else if (curPosn >= currLen)
              {   //nothing in buffer, read from socket
                  size32_t bytesRead=0;
                  count = 0;
                  do
                  {
                      socket->readtms(buf + count, 0, len - count, bytesRead, timeoutMS);
                      count += bytesRead;
                  } while (count != len);
                  currLen = curPosn = 0;
              }
              else
              {   //only some is in buffer, read rest from socket
                  size32_t bytesRead=0;
                  size32_t avail = currLen - curPosn;
                  memcpy(buf, (buffer + curPosn), avail);
                  count = avail;
                  do
                  {
                      size32_t read;
                      socket->readtms(buf+avail+bytesRead, 0, len-avail-bytesRead, read, timeoutMS);
                      bytesRead += read;
                  } while (len != (bytesRead + avail));
                  count += bytesRead;
                  currLen = curPosn = 0;
              }
              return count;
          }
      };
  
  private:
      CWSCHelper * master;
      ConstPointerArray &inputRows;
      CommonXmlWriter &xmlWriter;
      IEngineRowAllocator * outputAllocator;
      CriticalSection processExceptionCrit;
      StringBuffer responsePath;
      Owned<CSocketDataProvider> dataProvider;
      PTreeReaderOptions options;
      unsigned remainingMS;
  
      inline void checkRoxieAbortMonitor(IRoxieAbortMonitor * roxieAbortMonitor)
      {
          if (roxieAbortMonitor)
          {
              try 
              {
                  roxieAbortMonitor->checkForAbort();//throws
              }
              catch (IException *e)
              {
                  StringBuffer s;
                  throw MakeStringException(ROXIE_ABORT_EVENT, "%s", e->errorMessage(s).str());
              }
          }
      }
  
      bool checkContentEncodingSupported(const char* encoding)
      {
          if (strieq(encoding, "gzip"))
              return true;
          if (strieq(encoding, "deflate"))
              return true;
          if (strieq(encoding, "x-deflate"))
              return true;
          return false;
      }
  
      bool checkContentDecoding(const StringBuffer& headers, StringBuffer& content, StringBuffer& contentEncoding)
      {
          if ((headers.length() == 0) || (content.length() == 0))
              return false;
  
          getHTTPHeader(headers.str(), CONTENT_ENCODING, contentEncoding);
          if (contentEncoding.isEmpty())
              return false;
  
          if (!checkContentEncodingSupported(contentEncoding.str()))
              throw MakeStringException(-1, "Content-Encoding:%s not supported", contentEncoding.str());
          return true;
      }
  
      void decodeContent(const char* contentEncodingType, StringBuffer& content)
      {
  #ifdef _USE_ZLIB
          unsigned contentLength = content.length();
          StringBuffer contentDecoded;
  
          httpInflate((const byte*)content.str(), contentLength, contentDecoded, strieq(contentEncodingType, "gzip"));
          PROGLOG("Content decoded from %d bytes to %d bytes", contentLength, contentDecoded.length());
          content = contentDecoded;
  
          if (soapTraceLevel > 6 || master->logXML)
              master->logctx.CTXLOG("Content decoded. Original %s %d", CONTENT_LENGTH, contentLength);
  #else
              throw MakeStringException(-1, "_USE_ZLIB is required for Content-Encoding:%s", contentEncodingType);
  #endif
      }
  
      bool checkContentEncoding(const char* httpheaders, StringBuffer& contentEncodingType)
      {
          if (xmlWriter.length() == 0)
              return false;
  
          getHTTPHeader(httpheaders, CONTENT_ENCODING, contentEncodingType);
          if (contentEncodingType.isEmpty())
              return false;
  
          if (!checkContentEncodingSupported(contentEncodingType.str()))
              throw MakeStringException(-1, "Content-Encoding:%s not supported", contentEncodingType.str());
          return true;
      }
  
      ZlibCompressionType getEncodeFormat(const char *name)
      {
          if (strieq(name, "gzip"))
              return ZlibCompressionType::GZIP;
          if (strieq(name, "x-deflate"))
              return ZlibCompressionType::ZLIB_DEFLATE;
          if (strieq(name, "deflate"))
              return ZlibCompressionType::DEFLATE;
          return ZlibCompressionType::GZIP; //already checked above, shouldn't be here
      }
  
      void encodeContent(const char* contentEncodingType, MemoryBuffer& mb)
      {
  #ifdef _USE_ZLIB
          zlib_deflate(mb, xmlWriter.str(), xmlWriter.length(), GZ_BEST_SPEED, getEncodeFormat(contentEncodingType));
          PROGLOG("Content encoded from %d bytes to %d bytes", xmlWriter.length(), mb.length());
  #else
          throw MakeStringException(-1, "_USE_ZLIB is required for Content-Encoding:%s", contentEncodingType);
  #endif
      }
  
      void logRequest(bool contentEncoded, StringBuffer& request)
      {
          if (soapTraceLevel > 6 || master->logXML)
          {
              if (!contentEncoded)
                  master->logctx.CTXLOG("%s: request(%s)", master->wscCallTypeText(), request.str());
              else
                  master->logctx.CTXLOG("%s: request(%s), content encoded.", master->wscCallTypeText(), request.str());
          }
      }
  
      void createHttpRequest(Url &url, StringBuffer &request)
      {
          // Create the HTTP POST request
          if (master->wscType == STsoap)
              request.clear().append("POST ").append(url.path).append(" HTTP/1.1\r\n");
          else
              request.clear().append(master->service).append(" ").append(url.path).append(" HTTP/1.1\r\n");
  
          const char *httpheaders = master->httpHeaders.get();
          if (httpheaders && *httpheaders)
          {
              if (soapTraceLevel > 6 || master->logXML)
                  master->logctx.CTXLOG("%s: Adding HTTP Headers(%s)",  master->wscCallTypeText(), httpheaders);
              request.append(httpheaders);
          }
  
          if (!httpHeaderBlockContainsHeader(httpheaders, "Authorization"))
          {
              if (url.userPasswordPair.length() > 0)
              {
                  StringBuffer authToken;
                  JBASE64_Encode(url.userPasswordPair.str(), url.userPasswordPair.length(), authToken);
                  request.append("Authorization: Basic ").append(authToken).append("\r\n");
              }
              else if (master->authToken.length() > 0)
              {
                  request.append("Authorization: Basic ").append(master->authToken).append("\r\n");
              }
          }
  
  #ifdef _USE_ZLIB
          if (!httpHeaderBlockContainsHeader(httpheaders, ACCEPT_ENCODING))
              request.appendf("%s: gzip, deflate\r\n", ACCEPT_ENCODING);
  #endif
  
          if (master->wscType == STsoap)
          {
              if (master->soapaction.get())
                  request.append("SOAPAction: ").append(master->soapaction.get()).append("\r\n");
              if (master->httpHeaders.isEmpty() && master->httpHeaderName.get() && master->httpHeaderValue.get())
              {
                  //backward compatibility
                  StringBuffer hdr = master->httpHeaderName.get();
                  hdr.append(": ").append(master->httpHeaderValue);
                  if (soapTraceLevel > 6 || master->logXML)
                      master->logctx.CTXLOG("SOAPCALL: Adding HTTP Header(%s)", hdr.str());
                  request.append(hdr.append("\r\n"));
              }
              if (!httpHeaderBlockContainsHeader(httpheaders, "Content-Type"))
                  request.append("Content-Type: text/xml\r\n");
          }
          else if(master->wscType == SThttp)
              request.append("Accept: ").append(master->acceptType).append("\r\n");
          else
              assertex(false);
  
          if (master->wscType == STsoap)
          {
              request.append("Host: ").append(url.host).append(":").append(url.port).append("\r\n");//http 1.1
  
              StringBuffer contentEncodingType;
              if (!checkContentEncoding(httpheaders, contentEncodingType))
              {
                  request.append(CONTENT_LENGTH).append(xmlWriter.length()).append("\r\n\r\n");
                  request.append(xmlWriter.str());//add SOAP xml content
                  logRequest(false, request);
              }
              else
              {
                  logRequest(true, request);
                  MemoryBuffer encodedContentBuf;
                  encodeContent(contentEncodingType.str(), encodedContentBuf);
                  request.append(CONTENT_LENGTH).append(encodedContentBuf.length()).append("\r\n\r\n");
                  request.append(encodedContentBuf.length(), encodedContentBuf.toByteArray());
              }
          }
          else
          {
              request.append("Host: ").append(url.host);//http 1.1
              if (url.port != 80) //default port?
                  request.append(":").append(url.port);
              request.append("\r\n");//http 1.1
              request.append("\r\n");//httpcall
              logRequest(false, request);
          }
  
          if (master->logMin)
              master->logctx.CTXLOG("%s: request(%s:%u)", master->wscCallTypeText(), url.host.str(), url.port);
      }
  
      int readHttpResponse(StringBuffer &response, ISocket *socket)
      {
          // Read the POST reply
          // not doesn't *assume* is valid HTTP post format but if it is takes advantage of
          response.clear();
          unsigned bytesRead;
          MemoryAttr buf;
          char *buffer=(char *)buf.allocate(WSCBUFFERSIZE+1);
          int rval = 200;
  
          // first read header
          size32_t payloadofs = 0;
          size32_t payloadsize = 0;
          StringBuffer dbgheader, contentEncoding;
          bool chunked = false;
          size32_t read = 0;
          do {
              checkTimeLimitExceeded(&remainingMS);
              checkRoxieAbortMonitor(master->roxieAbortMonitor);
              socket->readtms(buffer+read, 0, WSCBUFFERSIZE-read, bytesRead, MIN(master->timeoutMS,remainingMS));
              checkTimeLimitExceeded(&remainingMS);
              checkRoxieAbortMonitor(master->roxieAbortMonitor);
  
              read += bytesRead;
              buffer[read] = 0;
              if (strncmp(buffer, "HTTP", 4) == 0) {
                  const char *s = strstr(buffer,"\r\n\r\n");
                  if (s) {
                      payloadofs = (size32_t)(s-buffer+4);
                      dbgheader.append(payloadofs,buffer);
                      s = strstr(buffer, " ");
                      if (s)
                          rval = atoi(s+1);
                      if (!strstr(buffer,"Transfer-Encoding: chunked"))
                      {
                          chunked = false;
                          s = strstr(buffer,CONTENT_LENGTH);
                          if (s) {
                              s += strlen(CONTENT_LENGTH);
                              if ((size32_t)(s-buffer) < payloadofs)
                                  payloadsize = atoi(s);
                          }
                      }
                      else
                      {
                          chunked = true;
                          size32_t chunkSize = 0;
                          size32_t dataLen = 0;
                          char ch;
  /*
                          //from http://www.w3.org/Protocols/rfc2616/rfc2616-sec19.html#sec19.4.4
                          "19.4.6 Introduction of Transfer-Encoding"
  
                          length := 0
                          read chunk-size, chunk-extension (if any) and CRLF
                          while (chunk-size > 0) 
                          {
                              read chunk-data and CRLF
                              append chunk-data to entity-body
                              length := length + chunk-size
                              read chunk-size and CRLF
                          }
  */
                          checkTimeLimitExceeded(&remainingMS);
                          dataProvider.setown(new CSocketDataProvider(buffer, payloadofs, read, socket, MIN(master->timeoutMS,remainingMS)));
                          dataProvider->getBytes(&ch, 1);
                          while (isalpha(ch) || isdigit(ch))
                          {   //get chunk-size
                              if (isdigit(ch))
                                  chunkSize = (chunkSize*16) + (ch - '0');
                              else
                                  chunkSize = (chunkSize*16) + 10 + (toupper(ch) - 'A');
                              dataProvider->getBytes(&ch, 1);
                          }
                          while (chunkSize && ch != '\n')//consume chunk-extension and CRLF
                              dataProvider->getBytes(&ch, 1);
                          while (chunkSize)
                          {
                              if (chunkSize > WSCBUFFERSIZE)
                                  DBGLOG("SOAPCALL chunk size %d", chunkSize);
  
                              //read chunk data directly into response
                              size32_t count = dataProvider->getBytes(response.reserve(dataLen + chunkSize), chunkSize);
                              assertex(count == chunkSize);
                              dataLen += count;
                              response.setLength(dataLen);
  
                              dataProvider->getBytes(&ch, 1);//consume CRLF at end of chunk
                              while (ch != '\n')
                                  dataProvider->getBytes(&ch, 1);
  
                              chunkSize = 0;
                              dataProvider->getBytes(&ch, 1);
                              while (isalpha(ch) || isdigit(ch))
                              {   //get next chunk size
                                  if (isdigit(ch))
                                      chunkSize = (chunkSize*16) + (ch - '0');
                                  else
                                      chunkSize = (chunkSize*16) + 10 + (toupper(ch) - 'A');
                                  dataProvider->getBytes(&ch, 1);
                              }
                              while(chunkSize && ch != '\n')//consume chunk-extension and CRLF
                                  dataProvider->getBytes(&ch, 1);
                          }
                      }
                      break;
                  }
              }
              if (bytesRead == 0)         // socket closed
                  break;
          } while (bytesRead&&(read<WSCBUFFERSIZE));
          if (!chunked)
          {
              if (payloadsize)
                  response.ensureCapacity(payloadsize);
              char *payload = buffer;
              if (payloadofs) {
                  read -= payloadofs;
                  payload += payloadofs;
                  if (payloadsize&&(read>payloadsize))
                      read = payloadsize;
              }
              if (read)
                  response.append(read,payload);
              if (payloadsize) {  // read directly into response
                  while (read<payloadsize) {
                      checkTimeLimitExceeded(&remainingMS);
                      checkRoxieAbortMonitor(master->roxieAbortMonitor);
                      socket->readtms(response.reserve(payloadsize-read), 0, payloadsize-read, bytesRead, MIN(master->timeoutMS,remainingMS));
                      checkTimeLimitExceeded(&remainingMS);
                      checkRoxieAbortMonitor(master->roxieAbortMonitor);
  
                      read += bytesRead;
                      response.setLength(read);
                      if (bytesRead==0) {
                          master->logctx.CTXLOG("%sCALL: Warning %sHTTP response terminated prematurely",master->wscType == STsoap ? "SOAP" : "HTTP",chunked?"CHUNKED ":NULL);
                          break; // oops  looks likesocket closed early
                      }
                  }
              }
              else {
                  for (;;) {
                      checkTimeLimitExceeded(&remainingMS);
                      checkRoxieAbortMonitor(master->roxieAbortMonitor);
                      socket->readtms(buffer, 0, WSCBUFFERSIZE, bytesRead, MIN(master->timeoutMS,remainingMS));
                      checkTimeLimitExceeded(&remainingMS);
                      checkRoxieAbortMonitor(master->roxieAbortMonitor);
  
                      if (bytesRead==0)
                          break;              // rely on socket closing to terminate message
                      response.append(bytesRead,buffer);
                  }
              }
          }
          if (checkContentDecoding(dbgheader, response, contentEncoding))
              decodeContent(contentEncoding.str(), response);
          if (soapTraceLevel > 6 || master->logXML)
              master->logctx.CTXLOG("%sCALL: LEN=%d %sresponse(%s%s)", master->wscType == STsoap ? "SOAP" : "HTTP",response.length(),chunked?"CHUNKED ":"", dbgheader.str(), response.str());
          else if (soapTraceLevel > 8)
              master->logctx.CTXLOG("%sCALL: LEN=%d %sresponse(%s)", master->wscType == STsoap ? "SOAP" : "HTTP",response.length(),chunked?"CHUNKED ":"", response.str()); // not sure this is that useful but...
          return rval;
      }
  
      void processEspResponse(Url &url, StringBuffer &response, ColumnProvider * meta)
      {
          StringBuffer path(responsePath);
          path.append("/Results/Result/");
          const char *tail;
          if (master->rowTransformer && master->inputpath.get())
          {
              StringBuffer ipath;
              ipath.append("/Envelope/Body/").append(master->inputpath.get());
              if((ipath.length() >= path.length()) && (0 == memcmp(ipath.str(), path.str(), path.length())))
              {
                  tail = ipath.str() + path.length();
              }
              else
              {
                  path.clear().append(ipath);
                  tail = NULL;
              }
          }
          else
              tail = "Dataset/Row";
  
          CMatchCB matchCB(*this, url, tail, meta);
          Owned<IXMLParse> xmlParser = createXMLParse((const void *)response.str(), (unsigned)response.length(), path.str(), matchCB, options, (master->flags&SOAPFusescontents)!=0);
          while (xmlParser->next());
      }
  
      void processLiteralResponse(Url &url, StringBuffer &response, ColumnProvider * meta)
      {
          StringBuffer path("/Envelope/Body/");
          if(master->rowTransformer && master->inputpath.get())
              path.append(master->inputpath.get());
          CMatchCB matchCB(*this, url, NULL, meta);
          Owned<IXMLParse> xmlParser = createXMLParse((const void *)response.str(), (unsigned)response.length(), path.str(), matchCB, options, (master->flags&SOAPFusescontents)!=0);
          while (xmlParser->next());
      }
  
      void processHttpResponse(Url &url, StringBuffer &response, ColumnProvider * meta)
      {
          StringBuffer path;
          if(master->rowTransformer && master->inputpath.get())
              path.append(master->inputpath.get());
          CMatchCB matchCB(*this, url, NULL, meta);
          Owned<IXMLParse> xmlParser;
          if (strieq(master->acceptType.str(), "application/json"))
              xmlParser.setown(createJSONParse((const void *)response.str(), (unsigned)response.length(), path.str(), matchCB, options, (master->flags&SOAPFusescontents)!=0, true));
          else
              xmlParser.setown(createXMLParse((const void *)response.str(), (unsigned)response.length(), path.str(), matchCB, options, (master->flags&SOAPFusescontents)!=0));
          while (xmlParser->next());
      }
  
      void processResponse(Url &url, StringBuffer &response, ColumnProvider * meta)
      {
          if (master->wscType == SThttp)
              processHttpResponse(url, response, meta);
          else if (master->flags & SOAPFliteral)
              processLiteralResponse(url, response, meta);
          else if (master->flags & SOAPFencoding)
              processLiteralResponse(url, response, meta);
          else
              processEspResponse(url, response, meta);
      }
  
      void processException(const Url &url, const void *row, IException *e)
      {
          CriticalBlock block(processExceptionCrit);
  
          Owned<IException> ne = url.getUrlException(e);
          e->Release();
  
          if ((master->flags & SOAPFonfail) && master->callHelper)
          {
              try
              {
                  RtlDynamicRowBuilder rowBuilder(outputAllocator);
                  size32_t sizeGot = master->onFailTransform(rowBuilder, row, ne);
                  if (sizeGot > 0)
                      master->putRow(rowBuilder.finalizeRowClear(sizeGot));
              }
              catch (IException *te)
              {
                  master->setErrorOwn(te);
              }
          }
          else
              master->setErrorOwn(ne.getClear());
      }
  
      void processException(const Url &url, ConstPointerArray &inputRows, IException *e)
      {
          Owned<IException> ne = url.getUrlException(e);
          e->Release();
  
          if ((master->flags & SOAPFonfail) && master->callHelper)
          {
              ForEachItemIn(idx, inputRows)
              {
                  try
                  {
                      RtlDynamicRowBuilder rowBuilder(outputAllocator);
                      size32_t sizeGot = master->onFailTransform(rowBuilder, inputRows.item(idx), ne);
                      if (sizeGot > 0)
                          master->putRow(rowBuilder.finalizeRowClear(sizeGot));
                  }
                  catch (IException *te)
                  {
                      master->setErrorOwn(te);
                      break;
                  }
              }
          }
          else
              master->setErrorOwn(ne.getClear());
      }
  
      inline void checkTimeLimitExceeded(unsigned * remainingMS)
      {
          if (master->isTimeLimitExceeded(remainingMS))
              throw MakeStringException(TIMELIMIT_EXCEEDED, "%sCALL TIMELIMIT(%ums) exceeded", master->wscType == STsoap ? "SOAP" : "HTTP", master->timeLimitMS);
      }
  
  public:
      CWSCAsyncFor(CWSCHelper * _master, CommonXmlWriter &_xmlWriter, ConstPointerArray &_inputRows, PTreeReaderOptions _options): xmlWriter(_xmlWriter), inputRows(_inputRows), options(_options)
      {
          master = _master;
          outputAllocator = master->queryOutputAllocator();
          responsePath.append("/Envelope/Body/");
          if (inputRows.ordinality() > 1)
          {
              // can we receive a roxie exceptions for the whole RequestArray?
              // if so, we need to handle them here
  
              responsePath.append(master->service).append("ResponseArray/");
          }
          responsePath.append(master->service).append("Response");
      }
  
      ~CWSCAsyncFor()
      {
      }
  
      IMPLEMENT_IINTERFACE;
  
      void For(unsigned num,unsigned maxatonce,bool abortFollowingException, bool shuffled)
      {
          CAsyncFor::For(num, maxatonce, abortFollowingException, shuffled);
      }
  
      void Do(unsigned idx)
      {
          StringBuffer request;
          StringBuffer response;
          unsigned numRetries = 0;
          unsigned retryInterval = 0;
  
          Url &url = master->urlArray.item(idx);
          createHttpRequest(url, request);
          unsigned startidx = idx;
          while (!master->aborted)
          {
              Owned<ISocket> socket;
              cycle_t startCycles, endCycles;
              startCycles = get_cycles_now();
              for (;;)
              {
                  try
                  {
                      checkTimeLimitExceeded(&remainingMS);
                      Url &connUrl = master->proxyUrlArray.empty() ? url : master->proxyUrlArray.item(0);
                      socket.setown(blacklist->connect(connUrl.port, connUrl.host, master->logctx, (unsigned)master->maxRetries, master->timeoutMS, master->roxieAbortMonitor));
                      if (stricmp(url.method, "https") == 0)
                      {
  #ifdef _USE_OPENSSL
                          Owned<ISecureSocket> ssock = master->createSecureSocket(socket.getClear());
                          if (ssock) 
                          {
                              checkTimeLimitExceeded(&remainingMS);
                              int status = ssock->secure_connect();
                              if (status < 0)
                              {
                                  StringBuffer err;
                                  err.append("Failure to establish secure connection to ");
                                  connUrl.getUrlString(err);
                                  err.append(": returned ").append(status);
                                  throw MakeStringExceptionDirect(0, err.str());
                              }
                              socket.setown(ssock.getLink());
                          }
  #else
                          StringBuffer err;
                          err.append("Failure to establish secure connection to ");
                          connUrl.getUrlString(err);
                          err.append(": OpenSSL disabled in build");
                          throw MakeStringExceptionDirect(0, err.str());
  #endif
                      }
                      break;
                  }
                  catch (IException *e)
                  {
                      if (master->timeLimitExceeded)
                      {
                          master->logctx.CTXLOG("%sCALL exiting: time limit (%ums) exceeded",master->wscType == STsoap ? "SOAP" : "HTTP", master->timeLimitMS);
                          processException(url, inputRows, e);
                          return;
                      }
  
                      if (e->errorCode() == ROXIE_ABORT_EVENT)
                      {
                          StringBuffer s;
                          master->logctx.CTXLOG("%sCALL exiting: Roxie Abort : %s",master->wscType == STsoap ? "SOAP" : "HTTP",e->errorMessage(s).str());
                          throw;
                      }
  
                      do 
                      {
                          idx++;  // try next socket not blacklisted
                          if (idx==master->urlArray.ordinality())
                              idx = 0;
                          if (idx==startidx)
                          {
                              StringBuffer s;
                              master->logctx.CTXLOG("Exception %s", e->errorMessage(s).str());
                              processException(url, inputRows, e);
                              return;
                          }
                      } while (blacklist->blacklisted(url.port, url.host));
                  }
              }
              try
              {
                  checkTimeLimitExceeded(&remainingMS);
                  checkRoxieAbortMonitor(master->roxieAbortMonitor);
                  socket->write(request.str(), request.length());
                  if (soapTraceLevel > 4)
                      master->logctx.CTXLOG("%sCALL: sent request (%s) to %s:%d", master->wscType == STsoap ? "SOAP" : "HTTP",master->service.str(), url.host.str(), url.port);
                  checkTimeLimitExceeded(&remainingMS);
                  checkRoxieAbortMonitor(master->roxieAbortMonitor);
  
                  int rval = readHttpResponse(response, socket);
  
                  if (soapTraceLevel > 4)
                      master->logctx.CTXLOG("%sCALL: received response (%s) from %s:%d", master->wscType == STsoap ? "SOAP" : "HTTP",master->service.str(), url.host.str(), url.port);
  
                  if (rval != 200)
                  {
                      if (rval == 503)
                          throw new ReceivedRoxieException(1001, "Server Too Busy");
  
                      StringBuffer text;
                      text.appendf("HTTP error (%d) in processQuery",rval);
                      rtlAddExceptionTag(text, "soapresponse", response.str());
                      throw MakeStringExceptionDirect(-1, text.str());
                  }
                  if (response.length() == 0)
                  {
                      throw MakeStringException(-1, "Zero length response in processQuery");
                  }
                  endCycles = get_cycles_now();
                  __int64 elapsedNs = cycle_to_nanosec(endCycles-startCycles);
                  master->logctx.noteStatistic(StTimeSoapcall, elapsedNs);
                  checkTimeLimitExceeded(&remainingMS);
                  ColumnProvider * meta = (ColumnProvider*)CreateColumnProvider((unsigned)nanoToMilli(elapsedNs), master->flags&SOAPFencoding?true:false);
                  processResponse(url, response, meta);
                  delete meta;
                  break;
              }
              catch (IReceivedRoxieException *e)
              {
                  // server busy ... Sleep and retry
                  if (e->errorCode() == 1001)
                  {
                      if (retryInterval)
                      {
                          int sleepTime = retryInterval + getRandom() % retryInterval;
                          master->logctx.CTXLOG("Server busy (1001), sleeping for %d milliseconds before retry", sleepTime);
                          Sleep(sleepTime);
                          retryInterval = (retryInterval*2 >= 10000)? 10000: retryInterval*2;
                      }
                      else
                      {
                          master->logctx.CTXLOG("Server busy (1001), retrying");
                          retryInterval = 10;
                      }
                      e->Release();
                  }
                  else
                  {
                      // other roxie exception ...
                      master->logctx.CTXLOG("Exiting: received Roxie exception");
                      if (e->errorRow())
                          processException(url, e->errorRow(), e);
                      else
                          processException(url, inputRows, e);
                      break;
                  }
              }
              catch (IException *e)
              {
                  if (master->timeLimitExceeded)
                  {
                      processException(url, inputRows, e);
                      master->logctx.CTXLOG("%sCALL exiting: time limit (%ums) exceeded", master->wscType == STsoap ? "SOAP" : "HTTP", master->timeLimitMS);
                      break;
                  }
  
                  if (e->errorCode() == ROXIE_ABORT_EVENT)
                  {
                      StringBuffer s;
                      master->logctx.CTXLOG("%sCALL exiting: Roxie Abort : %s",master->wscType == STsoap ? "SOAP" : "HTTP",e->errorMessage(s).str());
                      throw;
                  }
  
                  // other IException ... retry up to maxRetries
                  StringBuffer s;
                  master->logctx.CTXLOG("Exception %s", e->errorMessage(s).str());
  
                  if (numRetries >= master->maxRetries)
                  {
                      // error affects all inputRows
                      master->logctx.CTXLOG("Exiting: maxRetries %d exceeded", master->maxRetries);
                      processException(url, inputRows, e);
                      break;
                  }
                  numRetries++;
                  master->logctx.CTXLOG("Retrying: attempt %d of %d", numRetries, master->maxRetries);
                  e->Release();
              }
              catch (std::exception & es)
              {
                  if(dynamic_cast<std::bad_alloc *>(&es))
                      throw MakeStringException(-1, "std::exception: out of memory (std::bad_alloc) in CWSCAsyncFor processQuery");
                  throw MakeStringException(-1, "std::exception: standard library exception (%s) in CWSCAsyncFor processQuery",es.what());
              }
              catch (...)
              {
                  throw MakeStringException(-1, "Unknown exception in processQuery");
              }
          }
      }
      inline virtual StringBuffer getResponsePath() { return responsePath; }
      inline virtual ConstPointerArray & getInputRows() { return inputRows; }
      inline virtual CWSCHelper * getMaster() { return master; }
      inline virtual IEngineRowAllocator * getOutputAllocator() { return outputAllocator; }
  };
  
  IWSCAsyncFor * createWSCAsyncFor(CWSCHelper * _master, CommonXmlWriter &_xmlWriter, ConstPointerArray &_inputRows, PTreeReaderOptions _options)
  {
      return new CWSCAsyncFor(_master, _xmlWriter, _inputRows, _options);
  }
  