ampsplusplus.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2024 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//
#ifndef _AMPSPLUSPLUS_H_
#define _AMPSPLUSPLUS_H_
#include "amps/amps.h"
#include "amps/ampsver.h"
#include <string>
#include <map>
#include <sstream>
#include <iostream>
#include <memory>
#include <stdexcept>
#include <limits.h>
#include <list>
#include <memory>
#include <set>
#include <deque>
#include <vector>
#include <assert.h>
#ifndef _WIN32
  #include <inttypes.h>
#endif
#if defined(sun)
  #include <sys/atomic.h>
#endif
#include "amps/BookmarkStore.hpp"
#include "amps/MessageRouter.hpp"
#include "amps/util.hpp"
#include "amps/ampscrc.hpp"
#if __cplusplus >= 201100L || _MSC_VER >= 1900
  #include <atomic>
#endif

#ifndef AMPS_TESTING_SLOW_MESSAGE_STREAM
  #define AMPS_TESTING_SLOW_MESSAGE_STREAM
#endif





// For StoreBuffer implementations
#define AMPS_MEMORYBUFFER_DEFAULT_BUFFERS              10
#define AMPS_MEMORYBUFFER_DEFAULT_LENGTH            40960
#define AMPS_SUBSCRIPTION_MANAGER_DEFAULT_TIMEOUT       0
#define AMPS_HACLIENT_TIMEOUT_DEFAULT               10000
#define AMPS_HACLIENT_RECONNECT_DEFAULT               200
#define AMPS_DEFAULT_COMMAND_TIMEOUT                 5000
#define AMPS_DEFAULT_TOP_N                             -1
#define AMPS_DEFAULT_BATCH_SIZE                        10
#define AMPS_NUMBER_BUFFER_LEN                         20
#define AMPS_DEFAULT_QUEUE_ACK_TIMEOUT               1000

#if defined(_M_X64) || defined(__x86_64) || defined(_WIN64)
  #define AMPS_X64 1
#endif

static thread_local AMPS::Message publishStoreMessage;

namespace AMPS
{

  typedef std::map<std::string, std::string> ConnectionInfo;

  template<class Type>
  inline std::string asString(Type x_)
  {
    std::ostringstream os;
    os << x_;
    return os.str();
  }

  inline
  size_t convertToCharArray(char* buf_, amps_uint64_t seqNo_)
  {
    size_t pos = AMPS_NUMBER_BUFFER_LEN;
    for (int i = 0; i < AMPS_NUMBER_BUFFER_LEN; ++i)
    {
      if (seqNo_ > 0)
      {
        buf_[--pos] = (char)(seqNo_ % 10 + '0');
        seqNo_ /= 10;
      }
    }
    return pos;
  }

#ifdef _WIN32
  inline
  size_t convertToCharArray(char* buf_, unsigned long seqNo_)
  {
    size_t pos = AMPS_NUMBER_BUFFER_LEN;
    for (int i = 0; i < AMPS_NUMBER_BUFFER_LEN; ++i)
    {
      if (seqNo_ > 0)
      {
        buf_[--pos] = (char)(seqNo_ % 10 + '0');
        seqNo_ /= 10;
      }
    }
    return pos;
  }
#endif

  class Reason
  {
  public:
    static const char* duplicate()
    {
      return "duplicate";
    }
    static const char* badFilter()
    {
      return "bad filter";
    }
    static const char* badRegexTopic()
    {
      return "bad regex topic";
    }
    static const char* subscriptionAlreadyExists()
    {
      return "subscription already exists";
    }
    static const char* nameInUse()
    {
      return "name in use";
    }
    static const char* authFailure()
    {
      return "auth failure";
    }
    static const char* notEntitled()
    {
      return "not entitled";
    }
    static const char* authDisabled()
    {
      return "authentication disabled";
    }
    static const char* subidInUse()
    {
      return "subid in use";
    }
    static const char* noTopic()
    {
      return "no topic";
    }
  };

  class ExceptionListener
  {
  public:
    virtual ~ExceptionListener() {;}
    virtual void exceptionThrown(const std::exception&) const {;}
  };

  typedef ExceptionListener DefaultExceptionListener;


#define AMPS_CALL_EXCEPTION_WRAPPER(x) \
  try\
  {\
    x;\
  }\
  catch (std::exception& ex_)\
  {\
    try\
    {\
      _exceptionListener->exceptionThrown(ex_);\
    }\
    catch(...)\
    {\
      ;\
    }\
  }
  /*
   *  Note : we don't attempt to trap non std::exception exceptions
   *  here because doing so interferes with pthread_exit on some OSes.
  catch (...)\
  {\
    try\
    {\
      _exceptionListener->exceptionThrown(AMPS::AMPSException(\
          "An unhandled exception of unknown type was thrown by "\
          "the registered handler.", AMPS_E_USAGE));\
    }\
    catch(...)\
    {\
      ;\
    }\
  }
  */
#ifdef _WIN32
#define AMPS_CALL_EXCEPTION_WRAPPER_2(me,x) \
  try\
  {\
    while(me->_connected)\
    {\
      try\
      {\
        x;\
        break;\
      }\
      catch(MessageStreamFullException&)\
      {\
        try\
        {\
          me->checkAndSendHeartbeat(false);\
        }\
        catch (std::exception& ex_)\
        {\
          try\
          {\
            me->_exceptionListener->exceptionThrown(ex_);\
          }\
          catch(...)\
          {\
            ;\
          }\
          break;\
        }\
      }\
    }\
  }\
  catch (std::exception& ex_)\
  {\
    try\
    {\
      me->_exceptionListener->exceptionThrown(ex_);\
    }\
    catch(...)\
    {\
      ;\
    }\
  }
  /*
   *  Note : we don't attempt to trap non std::exception exceptions
   *  here because doing so interferes with pthread_exit on some OSes.
  catch (...)\
  {\
    try\
    {\
      me->_exceptionListener->exceptionThrown(AMPS::AMPSException(\
          "An unhandled exception of unknown type was thrown by "\
          "the registered handler.", AMPS_E_USAGE));\
    }\
    catch(...)\
    {\
      ;\
    }\
  }*/

#define AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me, x)\
  while(me->_connected)\
  {\
    try\
    {\
      x;\
      break;\
    }\
    catch(MessageStreamFullException&)\
    {\
      try\
      {\
        me->checkAndSendHeartbeat(false);\
      }\
      catch (std::exception& ex_)\
      {\
        try\
        {\
          me->_exceptionListener->exceptionThrown(ex_);\
        }\
        catch(...)\
        {\
          ;\
        }\
        break;\
      }\
    }\
  }
#else
#define AMPS_CALL_EXCEPTION_WRAPPER_2(me,x) \
  try\
  {\
    while(me->_connected)\
    {\
      try\
      {\
        x;\
        break;\
      }\
      catch(MessageStreamFullException& ex_)\
      {\
        try\
        {\
          me->checkAndSendHeartbeat(false);\
        }\
        catch (std::exception& ex_)\
        {\
          try\
          {\
            me->_exceptionListener->exceptionThrown(ex_);\
          }\
          catch(...)\
          {\
            ;\
          }\
          break;\
        }\
      }\
    }\
  }\
  catch (std::exception& ex_)\
  {\
    try\
    {\
      me->_exceptionListener->exceptionThrown(ex_);\
    }\
    catch(...)\
    {\
      ;\
    }\
  }
  /*
   *  Note : we don't attempt to trap non std::exception exceptions
   *  here because doing so interferes with pthread_exit on some OSes.
  catch (...)\
  {\
    try\
    {\
      me->_exceptionListener->exceptionThrown(AMPS::AMPSException(\
          "An unhandled exception of unknown type was thrown by "\
          "the registered handler.", AMPS_E_USAGE));\
    }\
    catch(...)\
    {\
      ;\
    }\
  }*/

#define AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me, x)\
  while(me->_connected)\
  {\
    try\
    {\
      x;\
      break;\
    }\
    catch(MessageStreamFullException& ex_)\
    {\
      try\
      {\
        me->checkAndSendHeartbeat(false);\
      }\
      catch (std::exception& ex_)\
      {\
        try\
        {\
          me->_exceptionListener->exceptionThrown(ex_);\
        }\
        catch(...)\
        {\
          ;\
        }\
        break;\
      }\
    }\
  }
#endif

#define AMPS_UNHANDLED_EXCEPTION(ex) \
  try\
  {\
    _exceptionListener->exceptionThrown(ex);\
  }\
  catch(...)\
  {;}

#define AMPS_UNHANDLED_EXCEPTION_2(me,ex) \
  try\
  {\
    me->_exceptionListener->exceptionThrown(ex);\
  }\
  catch(...)\
  {;}


  class Client;


  class Command
  {
    Message _message;
    unsigned _timeout;
    unsigned _batchSize;
    unsigned _flags;
    static const unsigned Subscribe           = 1;
    static const unsigned SOW                 = 2;
    static const unsigned NeedsSequenceNumber = 4;
    static const unsigned ProcessedAck        = 8;
    static const unsigned StatsAck            = 16;
    void init(Message::Command::Type command_)
    {
      _timeout = 0;
      _batchSize = 0;
      _flags = 0;
      _message.reset();
      _message.setCommandEnum(command_);
      _setIds();
    }
    void init(const std::string& command_)
    {
      _timeout = 0;
      _batchSize = 0;
      _flags = 0;
      _message.reset();
      _message.setCommand(command_);
      _setIds();
    }
    void init(const char* command_, size_t commandLen_)
    {
      _timeout = 0;
      _batchSize = 0;
      _flags = 0;
      _message.reset();
      _message.setCommand(command_, commandLen_);
      _setIds();
    }
    void _setIds(void)
    {
      Message::Command::Type command = _message.getCommandEnum();
      if (!(command & Message::Command::NoDataCommands))
      {
        _message.newCommandId();
        if (command == Message::Command::Subscribe ||
            command == Message::Command::SOWAndSubscribe ||
            command == Message::Command::DeltaSubscribe ||
            command == Message::Command::SOWAndDeltaSubscribe)
        {
          _message.setSubscriptionId(_message.getCommandId());
          _flags |= Subscribe;
        }
        if (command == Message::Command::SOW
            || command == Message::Command::SOWAndSubscribe
            || command == Message::Command::SOWAndDeltaSubscribe)
        {
          _message.setQueryID(_message.getCommandId());
          if (_batchSize == 0)
          {
            setBatchSize(AMPS_DEFAULT_BATCH_SIZE);
          }
          if (command == Message::Command::SOW)
          {
            _flags |= SOW;
          }
        }
        _flags |= ProcessedAck;
      }
      else if (command == Message::Command::SOWDelete)
      {
        _message.newCommandId();
        _flags |= ProcessedAck;
        _flags |= NeedsSequenceNumber;
      }
      else if (command == Message::Command::Publish
               || command == Message::Command::DeltaPublish)
      {
        _flags |= NeedsSequenceNumber;
      }
      else if (command == Message::Command::StopTimer)
      {
        _message.newCommandId();
      }
    }
  public:
    Command(const std::string& command_)
    {
      init(command_);
    }
    Command(const char* command_, size_t commandLen_)
    {
      init(command_, commandLen_);
    }
    Command(Message::Command::Type command_)
    {
      init(command_);
    }

    Command& reset(const std::string& command_)
    {
      init(command_);
      return *this;
    }
    Command& reset(const char* command_, size_t commandLen_)
    {
      init(command_, commandLen_);
      return *this;
    }
    Command& reset(Message::Command::Type command_)
    {
      init(command_);
      return *this;
    }
    Command& setSowKey(const std::string& sowKey_)
    {
      _message.setSowKey(sowKey_);
      return *this;
    }
    Command& setSowKey(const char* sowKey_, size_t sowKeyLen_)
    {
      _message.setSowKey(sowKey_, sowKeyLen_);
      return *this;
    }
    Command& setSowKeys(const std::string& sowKeys_)
    {
      _message.setSowKeys(sowKeys_);
      return *this;
    }
    Command& setSowKeys(const char* sowKeys_, size_t sowKeysLen_)
    {
      _message.setSowKeys(sowKeys_, sowKeysLen_);
      return *this;
    }
    Command& setCommandId(const std::string& cmdId_)
    {
      _message.setCommandId(cmdId_);
      return *this;
    }
    Command& setCommandId(const char* cmdId_, size_t cmdIdLen_)
    {
      _message.setCommandId(cmdId_, cmdIdLen_);
      return *this;
    }
    Command& setTopic(const std::string& topic_)
    {
      _message.setTopic(topic_);
      return *this;
    }
    Command& setTopic(const char* topic_, size_t topicLen_)
    {
      _message.setTopic(topic_, topicLen_);
      return *this;
    }
    Command& setFilter(const std::string& filter_)
    {
      _message.setFilter(filter_);
      return *this;
    }
    Command& setFilter(const char* filter_, size_t filterLen_)
    {
      _message.setFilter(filter_, filterLen_);
      return *this;
    }
    Command& setOrderBy(const std::string& orderBy_)
    {
      _message.setOrderBy(orderBy_);
      return *this;
    }
    Command& setOrderBy(const char* orderBy_, size_t orderByLen_)
    {
      _message.setOrderBy(orderBy_, orderByLen_);
      return *this;
    }
    Command& setSubId(const std::string& subId_)
    {
      _message.setSubscriptionId(subId_);
      return *this;
    }
    Command& setSubId(const char* subId_, size_t subIdLen_)
    {
      _message.setSubscriptionId(subId_, subIdLen_);
      return *this;
    }
    Command& setQueryId(const std::string& queryId_)
    {
      _message.setQueryId(queryId_);
      return *this;
    }
    Command& setQueryId(const char* queryId_, size_t queryIdLen_)
    {
      _message.setQueryId(queryId_, queryIdLen_);
      return *this;
    }
    Command& setBookmark(const std::string& bookmark_)
    {
      _message.setBookmark(bookmark_);
      return *this;
    }
    Command& setBookmark(const char* bookmark_, size_t bookmarkLen_)
    {
      _message.setBookmark(bookmark_, bookmarkLen_);
      return *this;
    }
    Command& setCorrelationId(const std::string& correlationId_)
    {
      _message.setCorrelationId(correlationId_);
      return *this;
    }
    Command& setCorrelationId(const char* correlationId_, size_t correlationIdLen_)
    {
      _message.setCorrelationId(correlationId_, correlationIdLen_);
      return *this;
    }
    Command& setOptions(const std::string& options_)
    {
      _message.setOptions(options_);
      return *this;
    }
    Command& setOptions(const char* options_, size_t optionsLen_)
    {
      _message.setOptions(options_, optionsLen_);
      return *this;
    }
    Command& setSequence(const std::string& seq_)
    {
      _message.setSequence(seq_);
      return *this;
    }
    Command& setSequence(const char* seq_, size_t seqLen_)
    {
      _message.setSequence(seq_, seqLen_);
      return *this;
    }
    Command& setSequence(const amps_uint64_t seq_)
    {
      std::ostringstream os;
      os << seq_;
      _message.setSequence(os.str());
      return *this;
    }
    amps_uint64_t getSequence() const
    {
      return amps_message_get_field_uint64(_message.getMessage(), AMPS_Sequence);
    }
    Command& setData(const std::string& data_)
    {
      _message.setData(data_);
      return *this;
    }
    Command& setData(const char* data_, size_t dataLen_)
    {
      _message.setData(data_, dataLen_);
      return *this;
    }
    Command& setTimeout(unsigned timeout_)
    {
      _timeout = timeout_;
      return *this;
    }
    Command& setTopN(unsigned topN_)
    {
      _message.setTopNRecordsReturned(topN_);
      return *this;
    }
    Command& setBatchSize(unsigned batchSize_)
    {
      _message.setBatchSize(batchSize_);
      _batchSize = batchSize_;
      return *this;
    }
    Command& setExpiration(unsigned expiration_)
    {
      _message.setExpiration(expiration_);
      return *this;
    }
    Command& addAckType(const std::string& ackType_)
    {
      _message.setAckType(_message.getAckType() + "," + ackType_);
      if (ackType_ == "processed")
      {
        _flags |= ProcessedAck;
      }
      else if (ackType_ == "stats")
      {
        _flags |= StatsAck;
      }
      return *this;
    }
    Command& setAckType(const std::string& ackType_)
    {
      _message.setAckType(ackType_);
      if (ackType_.find("processed") != std::string::npos)
      {
        _flags |= ProcessedAck;
      }
      else
      {
        _flags &= ~ProcessedAck;
      }
      if (ackType_.find("stats") != std::string::npos)
      {
        _flags |= StatsAck;
      }
      else
      {
        _flags &= ~StatsAck;
      }
      return *this;
    }
    Command& setAckType(unsigned ackType_)
    {
      _message.setAckTypeEnum(ackType_);
      if (ackType_ & Message::AckType::Processed)
      {
        _flags |= ProcessedAck;
      }
      else
      {
        _flags &= ~ProcessedAck;
      }
      if (ackType_ & Message::AckType::Stats)
      {
        _flags |= StatsAck;
      }
      else
      {
        _flags &= ~StatsAck;
      }
      return *this;
    }
    std::string getAckType() const
    {
      return (std::string)(_message.getAckType());
    }
    unsigned getAckTypeEnum() const
    {
      return _message.getAckTypeEnum();
    }

    Message& getMessage(void)
    {
      return _message;
    }
    unsigned getTimeout(void) const
    {
      return _timeout;
    }
    unsigned getBatchSize(void) const
    {
      return _batchSize;
    }
    bool isSubscribe(void) const
    {
      return _flags & Subscribe;
    }
    bool isSow(void) const
    {
      return (_flags & SOW) != 0;
    }
    bool hasProcessedAck(void) const
    {
      return (_flags & ProcessedAck) != 0;
    }
    bool hasStatsAck(void) const
    {
      return (_flags & StatsAck) != 0;
    }
    bool needsSequenceNumber(void) const
    {
      return (_flags & NeedsSequenceNumber) != 0;
    }
  };

  typedef void(*DisconnectHandlerFunc)(Client&, void* userData);

  class Message;
  typedef Handler<DisconnectHandlerFunc, Client&> DisconnectHandler;

  class Authenticator
  {
  public:
    virtual ~Authenticator() {;}

    virtual std::string authenticate(const std::string& userName_, const std::string& password_) = 0;
    virtual std::string retry(const std::string& userName_, const std::string& password_) = 0;
    virtual void completed(const std::string& userName_, const std::string& password_, const std::string& reason_) = 0;
  };

  class DefaultAuthenticator : public Authenticator
  {
  public:
    virtual ~DefaultAuthenticator() {;}
    std::string authenticate(const std::string& /*userName_*/, const std::string& password_)
    {
      return password_;
    }

    std::string retry(const std::string& /*userName_*/, const std::string& /*password_*/)
    {
      throw AuthenticationException("retry not implemented by DefaultAuthenticator.");
    }

    void completed(const std::string& /*userName_*/, const std::string& /* password_ */, const std::string& /* reason */) {;}

    static Authenticator& instance()
    {
      static DefaultAuthenticator d;
      return d;
    }
  };

  class StoreReplayer
  {
  public:

    virtual void execute(Message& message_) = 0;

    virtual ~StoreReplayer() {;}
  };

  class Store;

  typedef bool (*PublishStoreResizeHandler)(Store store_,
                                            size_t size_,
                                            void* userData_);

  class StoreImpl : public RefBody
  {
  public:
    StoreImpl(bool errorOnPublishGap_ = false)
      : _resizeHandler(NULL)
      , _resizeHandlerData(NULL)
      , _errorOnPublishGap(errorOnPublishGap_)
    {;}

    virtual amps_uint64_t store(const Message& message_) = 0;

    virtual void discardUpTo(amps_uint64_t index_) = 0;

    virtual void replay(StoreReplayer& replayer_) = 0;

    virtual bool replaySingle(StoreReplayer& replayer_, amps_uint64_t index_) = 0;

    virtual size_t unpersistedCount() const = 0;

    virtual ~StoreImpl() {;}

    virtual void flush(long timeout_) = 0;

    static inline size_t getUnsetPosition()
    {
      return AMPS_UNSET_INDEX;
    }

    static inline amps_uint64_t getUnsetSequence()
    {
      return AMPS_UNSET_SEQUENCE;
    }

    virtual amps_uint64_t getLowestUnpersisted() const = 0;

    virtual amps_uint64_t getLastPersisted() = 0;

    inline virtual void setResizeHandler(PublishStoreResizeHandler handler_,
                                         void* userData_)
    {
      _resizeHandler = handler_;
      _resizeHandlerData = userData_;
    }

    inline virtual PublishStoreResizeHandler getResizeHandler() const
    {
      return _resizeHandler;
    }

    bool callResizeHandler(size_t newSize_);

    inline virtual void setErrorOnPublishGap(bool errorOnPublishGap_)
    {
      _errorOnPublishGap = errorOnPublishGap_;
    }

    inline virtual bool getErrorOnPublishGap() const
    {
      return _errorOnPublishGap;
    }

  private:
    PublishStoreResizeHandler _resizeHandler;
    void*                     _resizeHandlerData;
    bool                      _errorOnPublishGap;
  };

  class Store
  {
    RefHandle<StoreImpl> _body;
  public:
    Store() {;}
    Store(StoreImpl* body_) : _body(body_) {;}
    Store(const Store& rhs) : _body(rhs._body) {;}
    Store& operator=(const Store& rhs)
    {
      _body = rhs._body;
      return *this;
    }

    amps_uint64_t store(const Message& message_)
    {
      return _body.get().store(message_);
    }

    void discardUpTo(amps_uint64_t index_)
    {
      _body.get().discardUpTo(index_);
    }

    void replay(StoreReplayer& replayer_)
    {
      _body.get().replay(replayer_);
    }

    bool replaySingle(StoreReplayer& replayer_, amps_uint64_t index_)
    {
      return _body.get().replaySingle(replayer_, index_);
    }

    size_t unpersistedCount() const
    {
      return _body.get().unpersistedCount();
    }

    bool isValid() const
    {
      return _body.isValid();
    }

    void flush(long timeout_ = 0)
    {
      return _body.get().flush(timeout_);
    }

    amps_uint64_t getLowestUnpersisted()
    {
      return _body.get().getLowestUnpersisted();
    }

    amps_uint64_t getLastPersisted()
    {
      return _body.get().getLastPersisted();
    }

    void setResizeHandler(PublishStoreResizeHandler handler_,
                          void* userData_)
    {
      _body.get().setResizeHandler(handler_, userData_);
    }

    PublishStoreResizeHandler getResizeHandler() const
    {
      return _body.get().getResizeHandler();
    }

    inline void setErrorOnPublishGap(bool errorOnPublishGap_)
    {
      _body.get().setErrorOnPublishGap(errorOnPublishGap_);
    }

    inline bool getErrorOnPublishGap() const
    {
      return _body.get().getErrorOnPublishGap();
    }

    StoreImpl* get()
    {
      if (_body.isValid())
      {
        return &_body.get();
      }
      else
      {
        return NULL;
      }
    }

  };

  class FailedWriteHandler
  {
  public:
    virtual ~FailedWriteHandler() {;}
    virtual void failedWrite(const Message& message_,
                             const char* reason_, size_t reasonLength_) = 0;
  };


  inline bool StoreImpl::callResizeHandler(size_t newSize_)
  {
    if (_resizeHandler)
    {
      return _resizeHandler(Store(this), newSize_, _resizeHandlerData);
    }
    return true;
  }

  inline bool DangerousFlushPublishStoreResizeHandler(Store store_, size_t /*size_*/,
                                                      void* data_)
  {
    long* timeoutp = (long*)data_;
    size_t count = store_.unpersistedCount();
    if (count == 0)
    {
      return false;
    }
    try
    {
      store_.flush(*timeoutp);
    }
#ifdef _WIN32
    catch (const TimedOutException&)
#else
    catch (const TimedOutException& e)
#endif
    {
      return true;
    }
    return (count == store_.unpersistedCount());
  }

  class FailedResubscribeHandler
  {
  public:
    virtual bool failure(const Message& message_, const MessageHandler& handler_,
                         unsigned requestedAckTypes_,
                         const AMPSException& exception_) = 0;
  };

  class SubscriptionManager
  {
  public:
    virtual ~SubscriptionManager() {;}
    virtual void subscribe(MessageHandler messageHandler_, const Message& message_,
                           unsigned requestedAckTypes_) = 0;
    virtual void unsubscribe(const Message::Field& subId_) = 0;
    virtual void clear() = 0;
    virtual void resubscribe(Client& client_) = 0;
    virtual void setFailedResubscribeHandler(std::shared_ptr<FailedResubscribeHandler> handler_)
    {
      _failedResubscribeHandler = handler_;
    }
  protected:
    std::shared_ptr<FailedResubscribeHandler> _failedResubscribeHandler;
  };


  class ConnectionStateListener
  {
  public:
    typedef enum { Disconnected = 0,
                   Shutdown = 1,
                   Connected = 2,
                   LoggedOn = 4,
                   PublishReplayed = 8,
                   HeartbeatInitiated = 16,
                   Resubscribed = 32,
                   UNKNOWN = 16384
                 } State;

    virtual void connectionStateChanged(State newState_) = 0;
    virtual ~ConnectionStateListener()  {;};
  };


  class MessageStreamImpl;
  class MessageStream;

  typedef void(*DeferredExecutionFunc)(void*);

  class ClientImpl : public RefBody // -V553
  {
    // Class to wrap turning of Nagle for things like flush and logon
    class NoDelay
    {
    private:
      AMPS_SOCKET _socket;
      int _noDelay;
      char* _valuePtr;
#ifdef _WIN32
      int _valueLen;
#else
      socklen_t _valueLen;
#endif
    public:
      NoDelay(amps_handle client_)
        : _socket(AMPS_INVALID_SOCKET), _noDelay(0), _valueLen(sizeof(int))
      {
        _valuePtr = (char*)&_noDelay;
        _socket = amps_client_get_socket(client_);
        if (_socket != AMPS_INVALID_SOCKET)
        {
          getsockopt(_socket, IPPROTO_TCP, TCP_NODELAY, _valuePtr, &_valueLen);
          if (!_noDelay)
          {
            _noDelay = 1;
            setsockopt(_socket, IPPROTO_TCP, TCP_NODELAY, _valuePtr, _valueLen);
          }
          else
          {
            _socket = AMPS_INVALID_SOCKET;
          }
        }
      }

      ~NoDelay()
      {
        if (_socket != AMPS_INVALID_SOCKET)
        {
          _noDelay = 0;
          setsockopt(_socket, IPPROTO_TCP, TCP_NODELAY, _valuePtr, _valueLen);
        }
      }
    };

    friend class Client;
  protected:
    amps_handle _client;
    DisconnectHandler _disconnectHandler;
    enum GlobalCommandTypeHandlers : size_t
    {
      Publish = 0,
      SOW = 1,
      GroupBegin = 2,
      GroupEnd = 3,
      Heartbeat = 4,
      OOF = 5,
      Ack = 6,
      LastChance = 7,
      DuplicateMessage = 8,
      COUNT = 9
    };
    std::vector<MessageHandler> _globalCommandTypeHandlers;
    Message _message, _readMessage, _publishMessage, _deltaMessage, _beatMessage;
    MessageRouter _routes;
    MessageRouter::RouteCache _routeCache;
    mutable Mutex _lock;
    std::string _name, _nameHash, _lastUri, _logonCorrelationData;
    amps_uint64_t _nameHashValue;
    BookmarkStore _bookmarkStore;
    Store _publishStore;
    bool _isRetryOnDisconnect;
    amps_unique_ptr<FailedWriteHandler> _failedWriteHandler;
#if __cplusplus >= 201100L || _MSC_VER >= 1900
    std::atomic<amps_uint64_t> _lastSentHaSequenceNumber;
#else
    volatile amps_uint64_t _lastSentHaSequenceNumber;
#endif
    AMPS_ATOMIC_TYPE_8 _logonInProgress;
    AMPS_ATOMIC_TYPE_8 _badTimeToHASubscribe;
    VersionInfo _serverVersion;
    Timer _heartbeatTimer;
    amps_unique_ptr<MessageStream> _pEmptyMessageStream;

    // queue data
    int           _queueAckTimeout;
    bool          _isAutoAckEnabled;
    unsigned      _ackBatchSize;
    unsigned      _queuedAckCount;
    unsigned      _defaultMaxDepth;
    struct QueueBookmarks
    {
      QueueBookmarks(const std::string& topic_)
        : _topic(topic_)
        , _oldestTime(0)
        , _bookmarkCount(0)
      {;}
      std::string   _topic;
      std::string   _data;
      amps_uint64_t _oldestTime;
      unsigned      _bookmarkCount;
    };
    typedef amps_uint64_t topic_hash;
    typedef std::map<topic_hash, QueueBookmarks> TopicHashMap;
    TopicHashMap _topicHashMap;

    class ClientStoreReplayer : public StoreReplayer
    {
      ClientImpl* _client;
    public:
      unsigned _version;
      amps_result _res;

      ClientStoreReplayer()
        : _client(NULL), _version(0), _res(AMPS_E_OK)
      {}

      ClientStoreReplayer(ClientImpl* client_)
        : _client(client_), _version(0), _res(AMPS_E_OK)
      {}

      void setClient(ClientImpl* client_)
      {
        _client = client_;
      }

      void execute(Message& message_)
      {
        if (!_client)
        {
          throw CommandException("Can't replay without a client.");
        }
        amps_uint64_t index = amps_message_get_field_uint64(message_.getMessage(),
                                                            AMPS_Sequence);
        if (index > _client->_lastSentHaSequenceNumber)
        {
          _client->_lastSentHaSequenceNumber = index;
        }

        _res = AMPS_E_OK;
        // Don't replay a queue cancel message after a reconnect.
        // Currently, the only messages that will have anything in options
        // are cancel messages.
        if (!message_.getCommand().empty() &&
            (!_client->_logonInProgress ||
             message_.getOptions().len() < 6))
        {
          _res = amps_client_send_with_version(_client->_client,
                                               message_.getMessage(),
                                               &_version);
          if (_res != AMPS_E_OK)
          {
            throw DisconnectedException("AMPS Server disconnected during replay");
          }
        }
      }

    };
    ClientStoreReplayer _replayer;

    class FailedWriteStoreReplayer : public StoreReplayer
    {
      ClientImpl* _parent;
      const char* _reason;
      size_t      _reasonLength;
      size_t      _replayCount;
    public:
      FailedWriteStoreReplayer(ClientImpl* parent, const char* reason_, size_t reasonLength_)
        : _parent(parent),
          _reason(reason_),
          _reasonLength(reasonLength_),
          _replayCount(0)
      {;}
      void execute(Message& message_)
      {
        if (_parent->_failedWriteHandler)
        {
          ++_replayCount;
          _parent->_failedWriteHandler->failedWrite(message_,
                                                    _reason, _reasonLength);
        }
      }
      size_t replayCount(void) const
      {
        return _replayCount;
      }
    };

    struct AckResponseImpl : public RefBody
    {
      std::string username, password, reason, status, bookmark, options;
      amps_uint64_t sequenceNo;
      amps_uint64_t nameHashValue;
      VersionInfo serverVersion;
#if __cplusplus >= 201100L || _MSC_VER >= 1900
      std::atomic<bool> responded;
      std::atomic<bool> abandoned;
#else
      volatile bool responded;
      volatile bool abandoned;
#endif
      unsigned connectionVersion;
      AckResponseImpl() :
        RefBody(),
        username(), password(), reason(), status(), bookmark(), options(),
        sequenceNo((amps_uint64_t)0),
        serverVersion(),
        responded(false),
        abandoned(false),
        connectionVersion(0)
      {
      }
    };

    class AckResponse
    {
      RefHandle<AckResponseImpl> _body;
    public:
      AckResponse() : _body(NULL) {;}
      AckResponse(const AckResponse& rhs) : _body(rhs._body) {;}
      static AckResponse create()
      {
        AckResponse r;
        r._body = new AckResponseImpl();
        return r;
      }

      const std::string& username()
      {
        return _body.get().username;
      }
      void setUsername(const char* data_, size_t len_)
      {
        if (data_)
        {
          _body.get().username.assign(data_, len_);
        }
        else
        {
          _body.get().username.clear();
        }
      }
      const std::string& password()
      {
        return _body.get().password;
      }
      void setPassword(const char* data_, size_t len_)
      {
        if (data_)
        {
          _body.get().password.assign(data_, len_);
        }
        else
        {
          _body.get().password.clear();
        }
      }
      const std::string& reason()
      {
        return _body.get().reason;
      }
      void setReason(const char* data_, size_t len_)
      {
        if (data_)
        {
          _body.get().reason.assign(data_, len_);
        }
        else
        {
          _body.get().reason.clear();
        }
      }
      const std::string& status()
      {
        return _body.get().status;
      }
      void setStatus(const char* data_, size_t len_)
      {
        if (data_)
        {
          _body.get().status.assign(data_, len_);
        }
        else
        {
          _body.get().status.clear();
        }
      }
      const std::string& bookmark()
      {
        return _body.get().bookmark;
      }
      void setBookmark(const Field& bookmark_)
      {
        if (!bookmark_.empty())
        {
          _body.get().bookmark.assign(bookmark_.data(), bookmark_.len());
          Field::parseBookmark(bookmark_, _body.get().nameHashValue,
                               _body.get().sequenceNo);
        }
        else
        {
          _body.get().bookmark.clear();
          _body.get().sequenceNo = (amps_uint64_t)0;
          _body.get().nameHashValue = (amps_uint64_t)0;
        }
      }
      amps_uint64_t sequenceNo() const
      {
        return _body.get().sequenceNo;
      }
      amps_uint64_t nameHashValue() const
      {
        return _body.get().nameHashValue;
      }
      void setSequenceNo(const char* data_, size_t len_)
      {
        amps_uint64_t result = (amps_uint64_t)0;
        if (data_)
        {
          for (size_t i = 0; i < len_; ++i)
          {
            result *= (amps_uint64_t)10;
            result += (amps_uint64_t)(data_[i] - '0');
          }
        }
        _body.get().sequenceNo = result;
      }
      VersionInfo serverVersion() const
      {
        return _body.get().serverVersion;
      }
      void setServerVersion(const char* data_, size_t len_)
      {
        if (data_)
        {
          _body.get().serverVersion.setVersion(std::string(data_, len_));
        }
      }
      bool responded()
      {
        return _body.get().responded;
      }
      void setResponded()
      {
        _body.get().responded = true;
      }
      bool abandoned()
      {
        return _body.get().abandoned;
      }
      void setAbandoned()
      {
        if (_body.isValid())
        {
          _body.get().abandoned = true;
        }
      }

      void setConnectionVersion(unsigned connectionVersion)
      {
        _body.get().connectionVersion = connectionVersion;
      }

      unsigned getConnectionVersion()
      {
        return _body.get().connectionVersion;
      }
      void setOptions(const char* data_, size_t len_)
      {
        if (data_)
        {
          _body.get().options.assign(data_, len_);
        }
        else
        {
          _body.get().options.clear();
        }
      }

      const std::string& options()
      {
        return _body.get().options;
      }

      AckResponse& operator=(const AckResponse& rhs)
      {
        _body = rhs._body;
        return *this;
      }
    };


    typedef std::map<std::string, AckResponse> AckMap;
    AckMap _ackMap;
    Mutex  _ackMapLock;
    DefaultExceptionListener _defaultExceptionListener;
  protected:

    struct DeferredExecutionRequest
    {
      DeferredExecutionRequest(DeferredExecutionFunc func_,
                               void*                 userData_)
        : _func(func_),
          _userData(userData_)
      {;}

      DeferredExecutionFunc _func;
      void*                 _userData;
    };
    const ExceptionListener* _exceptionListener;
    std::shared_ptr<const ExceptionListener> _pExceptionListener;
    amps_unique_ptr<SubscriptionManager> _subscriptionManager;
    volatile bool _connected;
    std::string _username;
    typedef std::set<ConnectionStateListener*> ConnectionStateListeners;
    ConnectionStateListeners _connectionStateListeners;
    typedef std::vector<DeferredExecutionRequest> DeferredExecutionList;
    Mutex _deferredExecutionLock;
    DeferredExecutionList _deferredExecutionList;
    unsigned _heartbeatInterval;
    unsigned _readTimeout;

    void broadcastConnectionStateChanged(ConnectionStateListener::State newState_)
    {
      // If we disconnected before we got to notification, don't notify.
      // This should only be able to happen for Resubscribed, since the lock
      // is released to let the subscription manager run resubscribe so a
      // disconnect could be called before the change is broadcast.
      if (!_connected && newState_ > ConnectionStateListener::Connected)
      {
        return;
      }
      for (ConnectionStateListeners::iterator it = _connectionStateListeners.begin(); it != _connectionStateListeners.end(); ++it)
      {
        AMPS_CALL_EXCEPTION_WRAPPER(
          (*it)->connectionStateChanged(newState_));
      }
    }
    unsigned processedAck(Message& message);
    unsigned persistedAck(Message& meesage);
    void lastChance(Message& message);
    void checkAndSendHeartbeat(bool force = false);
    virtual ConnectionInfo getConnectionInfo() const;
    static amps_result
    ClientImplMessageHandler(amps_handle message, void* userData);
    static void
    ClientImplPreDisconnectHandler(amps_handle client, unsigned failedConnectionVersion, void* userData);
    static amps_result
    ClientImplDisconnectHandler(amps_handle client, void* userData);

    void unsubscribeInternal(const std::string& id)
    {
      if (id.empty())
      {
        return;
      }
      // remove the handler first to avoid any more message delivery
      Message::Field subId;
      subId.assign(id.data(), id.length());
      _routes.removeRoute(subId);
      // Lock is already acquired
      if (_subscriptionManager)
      {
        // Have to unlock before calling into sub manager to avoid deadlock
        Unlock<Mutex> unlock(_lock);
        _subscriptionManager->unsubscribe(subId);
      }
      _message.reset();
      _message.setCommandEnum(Message::Command::Unsubscribe);
      _message.newCommandId();
      _message.setSubscriptionId(id);
      _sendWithoutRetry(_message);
      deferredExecution(&amps_noOpFn, NULL);
    }

    AckResponse syncAckProcessing(long timeout_, Message& message_,
                                  bool isHASubscribe_)
    {
      return syncAckProcessing(timeout_, message_,
                               (amps_uint64_t)0, isHASubscribe_);
    }

    AckResponse syncAckProcessing(long timeout_, Message& message_,
                                  amps_uint64_t haSeq = (amps_uint64_t)0,
                                  bool isHASubscribe_ = false)
    {
      // inv: we already have _lock locked up.
      AckResponse ack = AckResponse::create();
      if (1)
      {
        Lock<Mutex> guard(_ackMapLock);
        _ackMap[message_.getCommandId()] = ack;
      }
      ack.setConnectionVersion((unsigned)_send(message_, haSeq, isHASubscribe_));
      if (ack.getConnectionVersion() == 0)
      {
        // Send failed
        throw DisconnectedException("Connection closed while waiting for response.");
      }
      bool timedOut = false;
      AMPS_START_TIMER(timeout_)
      while (!timedOut && !ack.responded() && !ack.abandoned() && _connected)
      {
        if (timeout_)
        {
          timedOut = !_lock.wait(timeout_);
          // May have woken up early, check real time
          if (timedOut)
          {
            AMPS_RESET_TIMER(timedOut, timeout_);
          }
        }
        else
        {
          // Using a timeout version to ensure python can interrupt
          _lock.wait(1000);
          Unlock<Mutex> unlck(_lock);
          amps_invoke_waiting_function();
        }
      }
      if (ack.responded())
      {
        if (ack.status() != "failure")
        {
          if (message_.getCommand() == "logon")
          {
            amps_uint64_t ackSequence = ack.sequenceNo();
            if (_lastSentHaSequenceNumber < ackSequence)
            {
              _lastSentHaSequenceNumber = ackSequence;
            }
            if (_publishStore.isValid())
            {
              // If this throws, logon will fail and eitehr be
              // handled in HAClient/ServerChooser or by the caller
              // of logon.
              _publishStore.discardUpTo(ackSequence);
              if (_lastSentHaSequenceNumber < _publishStore.getLastPersisted())
              {
                _lastSentHaSequenceNumber = _publishStore.getLastPersisted();
              }
            }
            _nameHash = ack.bookmark().substr(0, ack.bookmark().find('|'));
            _nameHashValue = ack.nameHashValue();
            _serverVersion = ack.serverVersion();
            if (_bookmarkStore.isValid())
            {
              _bookmarkStore.setServerVersion(_serverVersion);
            }
          }
          if (_ackBatchSize)
          {
            const std::string& options = ack.options();
            size_t index = options.find_first_of("max_backlog=");
            if (index != std::string::npos)
            {
              unsigned data = 0;
              const char* c = options.c_str() + index + 12;
              while (*c && *c != ',')
              {
                data = (data * 10) + (unsigned)(*c++ -48);
              }
              if (_ackBatchSize > data)
              {
                _ackBatchSize = data;
              }
            }
          }
          return ack;
        }
        const size_t NotEntitled = 12;
        std::string ackReason = ack.reason();
        if (ackReason.length() == 0)
        {
          return ack;  // none
        }
        if (ackReason.length() == NotEntitled &&
            ackReason[0] == 'n' &&
            message_.getUserId().len() == 0)
        {
          message_.assignUserId(_username);
        }
        message_.throwFor(_client, ackReason);
      }
      else     // !ack.responded()
      {
        if (!ack.abandoned())
        {
          throw TimedOutException("timed out waiting for operation.");
        }
        else
        {
          throw DisconnectedException("Connection closed while waiting for response.");
        }
      }
      return ack;
    }

    void _cleanup(void)
    {
      if (!_client)
      {
        return;
      }
      amps_client_set_predisconnect_handler(_client, NULL, 0L);
      amps_client_set_disconnect_handler(_client, NULL, 0L);
      AMPS_CALL_EXCEPTION_WRAPPER(ClientImpl::disconnect());
      _pEmptyMessageStream.reset(NULL);
      amps_client_destroy(_client);
      _client = NULL;
    }

  public:

    ClientImpl(const std::string& clientName)
      : _client(NULL), _name(clientName)
      , _isRetryOnDisconnect(true)
      , _lastSentHaSequenceNumber((amps_uint64_t)0), _logonInProgress(0)
      , _badTimeToHASubscribe(0), _serverVersion()
      , _queueAckTimeout(AMPS_DEFAULT_QUEUE_ACK_TIMEOUT)
      , _isAutoAckEnabled(false)
      , _ackBatchSize(0)
      , _queuedAckCount(0)
      , _defaultMaxDepth(0)
      , _connected(false)
      , _heartbeatInterval(0)
      , _readTimeout(0)
    {
      _replayer.setClient(this);
      _client = amps_client_create(clientName.c_str());
      amps_client_set_message_handler(_client,
                                      (amps_handler)ClientImpl::ClientImplMessageHandler,
                                      this);
      amps_client_set_predisconnect_handler(_client,
                                            (amps_predisconnect_handler)ClientImpl::ClientImplPreDisconnectHandler,
                                            this);
      amps_client_set_disconnect_handler(_client,
                                         (amps_handler)ClientImpl::ClientImplDisconnectHandler,
                                         this);
      _exceptionListener = &_defaultExceptionListener;
      for (size_t i = 0; i < GlobalCommandTypeHandlers::COUNT; ++i)
      {
#ifdef AMPS_USE_EMPLACE
        _globalCommandTypeHandlers.emplace_back(MessageHandler());
#else
        _globalCommandTypeHandlers.push_back(MessageHandler());
#endif
      }
    }

    virtual ~ClientImpl()
    {
      _cleanup();
    }

    const std::string& getName() const
    {
      return _name;
    }

    const std::string& getNameHash() const
    {
      return _nameHash;
    }

    const amps_uint64_t getNameHashValue() const
    {
      return _nameHashValue;
    }

    void setName(const std::string& name)
    {
      // This operation will fail if the client's
      // name is already set.
      amps_result result = amps_client_set_name(_client, name.c_str());
      if (result != AMPS_E_OK)
      {
        AMPSException::throwFor(_client, result);
      }
      _name = name;
    }

    const std::string& getLogonCorrelationData() const
    {
      return _logonCorrelationData;
    }

    void setLogonCorrelationData(const std::string& logonCorrelationData_)
    {
      _logonCorrelationData = logonCorrelationData_;
    }

    size_t getServerVersion() const
    {
      return _serverVersion.getOldStyleVersion();
    }

    VersionInfo getServerVersionInfo() const
    {
      return _serverVersion;
    }

    const std::string& getURI() const
    {
      return _lastUri;
    }

    virtual void connect(const std::string& uri)
    {
      Lock<Mutex> l(_lock);
      _connect(uri);
    }

    virtual void _connect(const std::string& uri)
    {
      _lastUri = uri;
      amps_result result = amps_client_connect(_client, uri.c_str());
      if (result != AMPS_E_OK)
      {
        AMPSException::throwFor(_client, result);
      }
      _message.reset();
      _deltaMessage.setCommandEnum(Message::Command::DeltaPublish);
      _publishMessage.setCommandEnum(Message::Command::Publish);
      _beatMessage.setCommandEnum(Message::Command::Heartbeat);
      _beatMessage.setOptions("beat");
      _readMessage.setClientImpl(this);
      if (_queueAckTimeout)
      {
        result = amps_client_set_idle_time(_client, _queueAckTimeout);
        if (result != AMPS_E_OK)
        {
          AMPSException::throwFor(_client, result);
        }
      }
      _connected = true;
      broadcastConnectionStateChanged(ConnectionStateListener::Connected);
    }

    void setDisconnected()
    {
      {
        Lock<Mutex> l(_lock);
        if (_connected)
        {
          AMPS_CALL_EXCEPTION_WRAPPER(broadcastConnectionStateChanged(ConnectionStateListener::Disconnected));
        }
        _connected = false;
        _heartbeatTimer.setTimeout(0.0);
      }
      clearAcks(INT_MAX);
      amps_client_disconnect(_client);
      _routes.clear();
    }

    virtual void disconnect()
    {
      AMPS_CALL_EXCEPTION_WRAPPER(flushAcks());
      setDisconnected();
      AMPS_CALL_EXCEPTION_WRAPPER(processDeferredExecutions());
      Lock<Mutex> l(_lock);
      broadcastConnectionStateChanged(ConnectionStateListener::Shutdown);
    }

    void clearAcks(unsigned failedVersion)
    {
      // Have to lock to prevent race conditions
      Lock<Mutex> guard(_ackMapLock);
      {
        // Go ahead and signal any waiters if they are around...
        std::vector<std::string> worklist;
        for (AckMap::iterator i = _ackMap.begin(), e = _ackMap.end(); i != e; ++i)
        {
          if (i->second.getConnectionVersion() <= failedVersion)
          {
            i->second.setAbandoned();
            worklist.push_back(i->first);
          }
        }

        for (std::vector<std::string>::iterator j = worklist.begin(), e = worklist.end(); j != e; ++j)
        {
          _ackMap.erase(*j);
        }
      }

      _lock.signalAll();
    }

    int send(const Message& message)
    {
      Lock<Mutex> l(_lock);
      return _send(message);
    }

    void sendWithoutRetry(const Message& message_)
    {
      Lock<Mutex> l(_lock);
      // If we got here while logon was in progress, then we tried to send
      // while we were disconnected so throw DisconnectedException
      if (_logonInProgress)
      {
        throw DisconnectedException("The client has been disconnected.");
      }
      _sendWithoutRetry(message_);
    }

    void _sendWithoutRetry(const Message& message_)
    {
      amps_result result = amps_client_send(_client, message_.getMessage());
      if (result != AMPS_E_OK)
      {
        AMPSException::throwFor(_client, result);
      }
    }

    int _send(const Message& message, amps_uint64_t haSeq = (amps_uint64_t)0,
              bool isHASubscribe_ = false)
    {
      // Lock is already acquired
      amps_result result = AMPS_E_RETRY;

      // Create a local reference to this message, as we'll need to hold on
      // to a reference to it in case reconnect occurs.
      Message localMessage = message;
      unsigned version = 0;

      while (result == AMPS_E_RETRY)
      {
        if (haSeq && _logonInProgress)
        {
          // If retrySend is disabled, do not wait for the reconnect
          // to finish, just throw.
          if (!_isRetryOnDisconnect)
          {
            AMPSException::throwFor(_client, AMPS_E_RETRY);
          }
          if (!_lock.wait(1000))
          {
            amps_invoke_waiting_function();
          }
        }
        else
        {
          if ((haSeq && haSeq <= _lastSentHaSequenceNumber) ||
              (isHASubscribe_ && _badTimeToHASubscribe))
          {
            return (int)version;
          }
          // It's possible to get here out of order, but this way we'll
          // always send in order.
          if (haSeq > _lastSentHaSequenceNumber)
          {
            while (haSeq > _lastSentHaSequenceNumber + 1)
            {
              try
              {
                // Replayer updates _lastSentHaSsequenceNumber
                if (!_publishStore.replaySingle(_replayer,
                                                _lastSentHaSequenceNumber + 1))
                {
                  //++_lastSentHaSequenceNumber;
                  continue;
                }
                result = AMPS_E_OK;
                version = _replayer._version;
              }
#ifdef _WIN32
              catch (const DisconnectedException&)
#else
              catch (const DisconnectedException& e)
#endif
              {
                result = _replayer._res;
                break;
              }
            }
            result = amps_client_send_with_version(_client,
                                                   localMessage.getMessage(),
                                                   &version);
            ++_lastSentHaSequenceNumber;
          }
          else
          {
            if (_logonInProgress && localMessage.getCommand().data()[0] != 'l')
            {
              while (_logonInProgress)
              {
                if (!_lock.wait(1000))
                {
                  amps_invoke_waiting_function();
                }
              }
            }
            result = amps_client_send_with_version(_client,
                                                   localMessage.getMessage(),
                                                   &version);
          }
          if (result != AMPS_E_OK)
          {
            if (!isHASubscribe_ && !haSeq &&
                localMessage.getMessage() == message.getMessage())
            {
              localMessage = message.deepCopy();
            }
            if (_isRetryOnDisconnect)
            {
              Unlock<Mutex> u(_lock);
              result = amps_client_attempt_reconnect(_client, version);
              // If this is an HA publish or subscribe command, it was
              // stored first and will have already been replayed by the
              // store or sub manager after reconnect, so just return.
              if ((isHASubscribe_ || haSeq) &&
                  result == AMPS_E_RETRY)
              {
                return (int)version;
              }
            }
            else
            {
              // retrySend is disabled so throw the error
              // from the send as an exception, do not retry.
              AMPSException::throwFor(_client, result);
            }
          }
        }
        if (result == AMPS_E_RETRY)
        {
          amps_invoke_waiting_function();
        }
      }

      if (result != AMPS_E_OK)
      {
        AMPSException::throwFor(_client, result);
      }
      return (int)version;
    }

    void addMessageHandler(const Field& commandId_,
                           const AMPS::MessageHandler& messageHandler_,
                           unsigned requestedAcks_, Message::Command::Type commandType_)
    {
      Lock<Mutex> lock(_lock);
      _routes.addRoute(commandId_, messageHandler_, requestedAcks_,
                       0, commandType_);
    }

    bool removeMessageHandler(const Field& commandId_)
    {
      Lock<Mutex> lock(_lock);
      return _routes.removeRoute(commandId_);
    }

    std::string send(const MessageHandler& messageHandler_, Message& message_, int timeout_ = 0)
    {
      Field id = message_.getCommandId();
      Field subId = message_.getSubscriptionId();
      Field qid = message_.getQueryId();
      bool isSubscribeOnly = false;
      bool replace = false;
      unsigned requestedAcks = message_.getAckTypeEnum();
      unsigned systemAddedAcks = Message::AckType::None;
      Message::Command::Type commandType = message_.getCommandEnum();

      switch (commandType)
      {
      case Message::Command::Subscribe:
      case Message::Command::DeltaSubscribe:
        replace = message_.getOptions().operator std::string().find(AMPS_OPTIONS_REPLACE, 0, strlen(AMPS_OPTIONS_REPLACE) - 1) != std::string::npos;
        isSubscribeOnly = true;
      // fall through
      case Message::Command::SOWAndSubscribe:
      case Message::Command::SOWAndDeltaSubscribe:
        if (id.empty())
        {
          id = message_.newCommandId().getCommandId();
        }
        else
        {
          while (!replace && id != subId && _routes.hasRoute(id))
          {
            id = message_.newCommandId().getCommandId();
          }
        }
        if (subId.empty())
        {
          message_.setSubscriptionId(id);
          subId = id;
        }
        if (!message_.getBookmark().empty() && _bookmarkStore.isValid())
        {
          systemAddedAcks |= Message::AckType::Persisted;
        }
      // fall through
      case Message::Command::SOW:
        if (id.empty())
        {
          id = message_.newCommandId().getCommandId();
        }
        else
        {
          while (!replace && id != subId && _routes.hasRoute(id))
          {
            message_.newCommandId();
            if (qid == id)
            {
              qid = message_.getCommandId();
              message_.setQueryId(qid);
            }
            id = message_.getCommandId();
          }
        }
        if (!isSubscribeOnly)
        {
          if (qid.empty())
          {
            message_.setQueryID(id);
            qid = id;
          }
          else
          {
            while (!replace && qid != subId && qid != id
                   && _routes.hasRoute(qid))
            {
              qid = message_.newQueryId().getQueryId();
            }
          }
        }
        systemAddedAcks |= Message::AckType::Processed;
        message_.setAckTypeEnum(requestedAcks | systemAddedAcks);
        {
          int routesAdded = 0;
          Lock<Mutex> l(_lock);
          if (!subId.empty() && messageHandler_.isValid())
          {
            if (!_routes.hasRoute(subId))
            {
              ++routesAdded;
            }
            // This can replace a non-subscribe with a matching id
            // with a subscription but not another subscription.
            _routes.addRoute(subId, messageHandler_, requestedAcks,
                             systemAddedAcks, commandType);
          }
          if (!isSubscribeOnly && !qid.empty()
              && messageHandler_.isValid() && qid != subId)
          {
            if (routesAdded == 0)
            {
              _routes.addRoute(qid, messageHandler_,
                               requestedAcks, systemAddedAcks, commandType);
            }
            else
            {
              void* data = NULL;
              {
                Unlock<Mutex> u(_lock);
                data = amps_invoke_copy_route_function(
                         messageHandler_.userData());
              }
              if (!data)
              {
                _routes.addRoute(qid, messageHandler_, requestedAcks,
                                 systemAddedAcks, commandType);
              }
              else
              {
                _routes.addRoute(qid,
                                 MessageHandler(messageHandler_.function(),
                                                data),
                                 requestedAcks, systemAddedAcks, commandType);
              }
            }
            ++routesAdded;
          }
          if (!id.empty() && messageHandler_.isValid()
              && requestedAcks & ~Message::AckType::Persisted
              && id != subId && id != qid)
          {
            if (routesAdded == 0)
            {
              _routes.addRoute(id, messageHandler_, requestedAcks,
                               systemAddedAcks, commandType);
            }
            else
            {
              void* data = NULL;
              {
                Unlock<Mutex> u(_lock);
                data = amps_invoke_copy_route_function(
                         messageHandler_.userData());
              }
              if (!data)
              {
                _routes.addRoute(id, messageHandler_, requestedAcks,
                                 systemAddedAcks, commandType);
              }
              else
              {
                _routes.addRoute(id,
                                 MessageHandler(messageHandler_.function(),
                                                data),
                                 requestedAcks,
                                 systemAddedAcks, commandType);
              }
            }
            ++routesAdded;
          }
          try
          {
            // We aren't adding to subscription manager, so this isn't
            // an HA subscribe.
            syncAckProcessing(timeout_, message_, 0, false);
            message_.setAckTypeEnum(requestedAcks);
          }
          catch (...)
          {
            _routes.removeRoute(message_.getQueryID());
            _routes.removeRoute(message_.getSubscriptionId());
            _routes.removeRoute(id);
            message_.setAckTypeEnum(requestedAcks);
            throw;
          }
        }
        break;
      // These are valid commands that are used as-is
      case Message::Command::Unsubscribe:
      case Message::Command::Heartbeat:
      case Message::Command::Logon:
      case Message::Command::StartTimer:
      case Message::Command::StopTimer:
      case Message::Command::SOWDelete:
      {
        Lock<Mutex> l(_lock);
        // if an ack is requested, it'll need a command ID.
        if (message_.getAckTypeEnum() != Message::AckType::None)
        {
          if (id.empty())
          {
            message_.newCommandId();
            id = message_.getCommandId();
          }
          if (messageHandler_.isValid())
          {
            _routes.addRoute(id, messageHandler_, requestedAcks,
                             Message::AckType::None, commandType);
          }
        }
        _send(message_);
      }
      break;
      case Message::Command::DeltaPublish:
      case Message::Command::Publish:
      {
        bool useSync = message_.getFilter().len() > 0;
        Lock<Mutex> l(_lock);
        // if an ack is requested, it'll need a command ID.
        unsigned ackType = message_.getAckTypeEnum();
        if (ackType != Message::AckType::None
            || useSync)
        {
          if (id.empty())
          {
            message_.newCommandId();
            id = message_.getCommandId();
          }
          if (messageHandler_.isValid())
          {
            _routes.addRoute(id, messageHandler_, requestedAcks,
                             Message::AckType::None, commandType);
          }
        }
        if (useSync)
        {
          message_.setAckTypeEnum(ackType | Message::AckType::Processed);
          syncAckProcessing(timeout_, message_, 0, false);
        }
        else
        {
          _send(message_);
        }
      }
      break;
      // These are things that shouldn't be sent (not meaningful)
      case Message::Command::GroupBegin:
      case Message::Command::GroupEnd:
      case Message::Command::OOF:
      case Message::Command::Ack:
      case Message::Command::Unknown:
      default:
        throw CommandException("Command type " + message_.getCommand() + " can not be sent directly to AMPS");
      }
      message_.setAckTypeEnum(requestedAcks);
      return id;
    }

    void setDisconnectHandler(const DisconnectHandler& disconnectHandler)
    {
      Lock<Mutex> l(_lock);
      _disconnectHandler = disconnectHandler;
    }

    void setGlobalCommandTypeMessageHandler(const std::string& command_, const MessageHandler& handler_)
    {
      switch (command_[0])
      {
#if 0 // Not currently implemented to avoid an extra branch in delivery
      case 'p':
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Publish] = handler_;
        break;
      case 's':
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::SOW] = handler_;
        break;
#endif
      case 'h':
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Heartbeat] = handler_;
        break;
#if 0 // Not currently implemented to avoid an extra branch in delivery
      case 'g':
        if (command_[6] == 'b')
        {
          _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupBegin] = handler_;
        }
        else if (command_[6] == 'e')
        {
          _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupEnd] = handler_;
        }
        else
        {
          std::ostringstream os;
          os << "Invalid command '" << command_ << "' passed to setGlobalCommandTypeHandler";
          throw CommandException(os.str());
        }
        break;
      case 'o':
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::OOF] = handler_;
        break;
#endif
      case 'a':
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Ack] = handler_;
        break;
      case 'l':
      case 'L':
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::LastChance] = handler_;
        break;
      case 'd':
      case 'D':
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage] = handler_;
        break;
      default:
        std::ostringstream os;
        os << "Invalid command '" << command_ << "' passed to setGlobalCommandTypeHandler";
        throw CommandException(os.str());
        break;
      }
    }

    void setGlobalCommandTypeMessageHandler(const Message::Command::Type command_, const MessageHandler& handler_)
    {
      switch (command_)
      {
#if 0 // Not currently implemented to avoid an extra branch in delivery
      case Message::Command::Publish:
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Publish] = handler_;
        break;
      case Message::Command::SOW:
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::SOW] = handler_;
        break;
#endif
      case Message::Command::Heartbeat:
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Heartbeat] = handler_;
        break;
#if 0 // Not currently implemented to avoid an extra branch in delivery
      case Message::Command::GroupBegin:
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupBegin] = handler_;
        break;
      case Message::Command::GroupEnd:
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::GroupEnd] = handler_;
        break;
      case Message::Command::OOF:
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::OOF] = handler_;
        break;
#endif
      case Message::Command::Ack:
        _globalCommandTypeHandlers[GlobalCommandTypeHandlers::Ack] = handler_;
        break;
      default:
        unsigned bits = 0;
        unsigned command = command_;
        while (command > 0)
        {
          ++bits;
          command >>= 1;
        }
        char errBuf[128];
        AMPS_snprintf(errBuf, sizeof(errBuf),
                      "Invalid command '%.*s' passed to setGlobalCommandTypeHandler",
                      CommandConstants<0>::Lengths[bits],
                      CommandConstants<0>::Values[bits]);
        throw CommandException(errBuf);
        break;
      }
    }

    void setGlobalCommandTypeMessageHandler(const GlobalCommandTypeHandlers handlerType_, const MessageHandler& handler_)
    {
      _globalCommandTypeHandlers[handlerType_] = handler_;
    }

    void setFailedWriteHandler(FailedWriteHandler* handler_)
    {
      Lock<Mutex> l(_lock);
      _failedWriteHandler.reset(handler_);
    }

    void setPublishStore(const Store& publishStore_)
    {
      Lock<Mutex> l(_lock);
      if (_connected)
      {
        throw AlreadyConnectedException("Setting a publish store on a connected client is undefined behavior");
      }
      _publishStore = publishStore_;
    }

    void setBookmarkStore(const BookmarkStore& bookmarkStore_)
    {
      Lock<Mutex> l(_lock);
      if (_connected)
      {
        throw AlreadyConnectedException("Setting a bookmark store on a connected client is undefined behavior");
      }
      _bookmarkStore = bookmarkStore_;
    }

    void setSubscriptionManager(SubscriptionManager* subscriptionManager_)
    {
      Lock<Mutex> l(_lock);
      _subscriptionManager.reset(subscriptionManager_);
    }

    SubscriptionManager* getSubscriptionManager() const
    {
      return const_cast<SubscriptionManager*>(_subscriptionManager.get());
    }

    DisconnectHandler getDisconnectHandler() const
    {
      return _disconnectHandler;
    }

    MessageHandler getDuplicateMessageHandler() const
    {
      return _globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage];
    }

    FailedWriteHandler* getFailedWriteHandler() const
    {
      return const_cast<FailedWriteHandler*>(_failedWriteHandler.get());
    }

    Store getPublishStore() const
    {
      return _publishStore;
    }

    BookmarkStore getBookmarkStore() const
    {
      return _bookmarkStore;
    }

    amps_uint64_t publish(const char* topic_, size_t topicLen_, const char* data_, size_t dataLen_)
    {
      if (!_publishStore.isValid())
      {
        Lock<Mutex> l(_lock);
        _publishMessage.assignTopic(topic_, topicLen_);
        _publishMessage.assignData(data_, dataLen_);
        _send(_publishMessage);
        return 0;
      }
      else
      {
        publishStoreMessage.reset();
        publishStoreMessage.setCommandEnum(Message::Command::Publish);
        return _publish(topic_, topicLen_, data_, dataLen_);
      }
    }

    amps_uint64_t publish(const char* topic_, size_t topicLen_, const char* data_,
                          size_t dataLen_, unsigned long expiration_)
    {
      if (!_publishStore.isValid())
      {
        Lock<Mutex> l(_lock);
        _publishMessage.assignTopic(topic_, topicLen_);
        _publishMessage.assignData(data_, dataLen_);
        char exprBuf[AMPS_NUMBER_BUFFER_LEN];
        size_t pos = convertToCharArray(exprBuf, expiration_);
        _publishMessage.assignExpiration(exprBuf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
        _send(_publishMessage);
        _publishMessage.assignExpiration(NULL, 0);
        return 0;
      }
      else
      {
        publishStoreMessage.reset();
        char exprBuf[AMPS_NUMBER_BUFFER_LEN];
        size_t exprPos = convertToCharArray(exprBuf, expiration_);
        publishStoreMessage.setCommandEnum(Message::Command::Publish)
        .assignExpiration(exprBuf + exprPos,
                          AMPS_NUMBER_BUFFER_LEN - exprPos);
        return _publish(topic_, topicLen_, data_, dataLen_);
      }
    }

    class FlushAckHandler : ConnectionStateListener
    {
    private:
      ClientImpl* _pClient;
      Field _cmdId;
#if __cplusplus >= 201100L || _MSC_VER >= 1900
      std::atomic<bool> _acked;
      std::atomic<bool> _disconnected;
#else
      volatile bool _acked;
      volatile bool _disconnected;
#endif
    public:
      FlushAckHandler(ClientImpl* pClient_)
        : _pClient(pClient_), _cmdId(), _acked(false), _disconnected(false)
      {
        pClient_->addConnectionStateListener(this);
      }
      ~FlushAckHandler()
      {
        _pClient->removeConnectionStateListener(this);
        _pClient->removeMessageHandler(_cmdId);
        _cmdId.clear();
      }
      void setCommandId(const Field& cmdId_)
      {
        _cmdId.deepCopy(cmdId_);
      }
      void invoke(const Message&)
      {
        _acked = true;
      }
      void connectionStateChanged(State state_)
      {
        if (state_ <= Shutdown)
        {
          _disconnected = true;
        }
      }
      bool acked()
      {
        return _acked;
      }
      bool done()
      {
        return _acked || _disconnected;
      }
    };

    void publishFlush(long timeout_, unsigned ackType_)
    {
      static const char* processed = "processed";
      static const size_t processedLen = strlen(processed);
      static const char* persisted = "persisted";
      static const size_t persistedLen = strlen(persisted);
      static const char* flush = "flush";
      static const size_t flushLen = strlen(flush);
      static VersionInfo minPersisted("5.3.3.0");
      static VersionInfo minFlush("4");
      if (ackType_ != Message::AckType::Processed
          && ackType_ != Message::AckType::Persisted)
      {
        throw CommandException("Flush can only be used with processed or persisted acks.");
      }
      FlushAckHandler flushHandler(this);
      if (_serverVersion >= minFlush)
      {
        Lock<Mutex> l(_lock);
        if (!_connected)
        {
          throw DisconnectedException("Not connected trying to flush");
        }
        _message.reset();
        _message.newCommandId();
        _message.assignCommand(flush, flushLen);
        if (_serverVersion < minPersisted
            || ackType_ == Message::AckType::Processed)
        {
          _message.assignAckType(processed, processedLen);
        }
        else
        {
          _message.assignAckType(persisted, persistedLen);
        }
        flushHandler.setCommandId(_message.getCommandId());
        addMessageHandler(_message.getCommandId(),
                          std::bind(&FlushAckHandler::invoke,
                                    std::ref(flushHandler),
                                    std::placeholders::_1),
                          ackType_, _message.getCommandEnum());
        NoDelay noDelay(_client);
        if (_send(_message) == -1)
        {
          throw DisconnectedException("Disconnected trying to flush");
        }
      }
      if (_publishStore.isValid())
      {
        try
        {
          _publishStore.flush(timeout_);
        }
        catch (const AMPSException& ex)
        {
          AMPS_UNHANDLED_EXCEPTION(ex);
          throw;
        }
      }
      else if (_serverVersion < minFlush)
      {
        if (timeout_ > 0)
        {
          AMPS_USLEEP(timeout_ * 1000);
        }
        else
        {
          AMPS_USLEEP(1000 * 1000);
        }
        return;
      }
      if (timeout_)
      {
        Timer timer((double)timeout_);
        timer.start();
        while (!timer.check() && !flushHandler.done())
        {
          AMPS_USLEEP(10000);
          amps_invoke_waiting_function();
        }
      }
      else
      {
        while (!flushHandler.done())
        {
          AMPS_USLEEP(10000);
          amps_invoke_waiting_function();
        }
      }
      // No response or disconnect in timeout interval
      if (!flushHandler.done())
      {
        throw TimedOutException("Timed out waiting for flush");
      }
      // We got disconnected and there is no publish store
      if (!flushHandler.acked() && !_publishStore.isValid())
      {
        throw DisconnectedException("Disconnected waiting for flush");
      }
    }

    amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
                               const char* data_, size_t dataLength_)
    {
      if (!_publishStore.isValid())
      {
        Lock<Mutex> l(_lock);
        _deltaMessage.assignTopic(topic_, topicLength_);
        _deltaMessage.assignData(data_, dataLength_);
        _send(_deltaMessage);
        return 0;
      }
      else
      {
        publishStoreMessage.reset();
        publishStoreMessage.setCommandEnum(Message::Command::DeltaPublish);
        return _publish(topic_, topicLength_, data_, dataLength_);
      }
    }

    amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
                               const char* data_, size_t dataLength_,
                               unsigned long expiration_)
    {
      if (!_publishStore.isValid())
      {
        Lock<Mutex> l(_lock);
        _deltaMessage.assignTopic(topic_, topicLength_);
        _deltaMessage.assignData(data_, dataLength_);
        char exprBuf[AMPS_NUMBER_BUFFER_LEN];
        size_t pos = convertToCharArray(exprBuf, expiration_);
        _deltaMessage.assignExpiration(exprBuf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
        _send(_deltaMessage);
        _deltaMessage.assignExpiration(NULL, 0);
        return 0;
      }
      else
      {
        publishStoreMessage.reset();
        char exprBuf[AMPS_NUMBER_BUFFER_LEN];
        size_t exprPos = convertToCharArray(exprBuf, expiration_);
        publishStoreMessage.setCommandEnum(Message::Command::DeltaPublish)
        .assignExpiration(exprBuf + exprPos,
                          AMPS_NUMBER_BUFFER_LEN - exprPos);
        return _publish(topic_, topicLength_, data_, dataLength_);
      }
    }

    amps_uint64_t _publish(const char* topic_, size_t topicLength_,
                           const char* data_, size_t dataLength_)
    {
      publishStoreMessage.assignTopic(topic_, topicLength_)
      .setAckTypeEnum(Message::AckType::Persisted)
      .assignData(data_, dataLength_);
      amps_uint64_t haSequenceNumber = _publishStore.store(publishStoreMessage);
      char buf[AMPS_NUMBER_BUFFER_LEN];
      size_t pos = convertToCharArray(buf, haSequenceNumber);
      publishStoreMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
      {
        Lock<Mutex> l(_lock);
        _send(publishStoreMessage, haSequenceNumber);
      }
      return haSequenceNumber;
    }

    virtual std::string logon(long timeout_, Authenticator& authenticator_,
                              const char* options_ = NULL)
    {
      Lock<Mutex> l(_lock);
      return _logon(timeout_, authenticator_, options_);
    }

    virtual std::string _logon(long timeout_, Authenticator& authenticator_,
                               const char* options_ = NULL)
    {
      _message.reset();
      _message.newCommandId();
      std::string newCommandId = _message.getCommandId();
      _message.setCommandEnum(Message::Command::Logon);
      _message.setClientName(_name);
#ifdef AMPS_CLIENT_VERSION_WITH_LANGUAGE
      _message.assignVersion(AMPS_CLIENT_VERSION_WITH_LANGUAGE,
                             strlen(AMPS_CLIENT_VERSION_WITH_LANGUAGE));
#endif
      URI uri(_lastUri);
      if (uri.user().size())
      {
        _message.setUserId(uri.user());
      }
      if (uri.password().size())
      {
        _message.setPassword(uri.password());
      }
      if (uri.protocol() == "amps" && uri.messageType().size())
      {
        _message.setMessageType(uri.messageType());
      }
      if (uri.isTrue("pretty"))
      {
        _message.setOptions("pretty");
      }

      _message.setPassword(authenticator_.authenticate(_message.getUserId(), _message.getPassword()));
      if (!_logonCorrelationData.empty())
      {
        _message.assignCorrelationId(_logonCorrelationData);
      }
      if (options_)
      {
        _message.setOptions(options_);
      }
      _username = _message.getUserId();
      try
      {
        AtomicFlagFlip pubFlip(&_logonInProgress);
        NoDelay noDelay(_client);
        while (true)
        {
          _message.setAckTypeEnum(Message::AckType::Processed);
          AckResponse ack = syncAckProcessing(timeout_, _message);
          if (ack.status() == "retry")
          {
            _message.setPassword(authenticator_.retry(ack.username(), ack.password()));
            _username = ack.username();
            _message.setUserId(_username);
          }
          else
          {
            authenticator_.completed(ack.username(), ack.password(), ack.reason());
            break;
          }
        }
        broadcastConnectionStateChanged(ConnectionStateListener::LoggedOn);

        // Now re-send the heartbeat command if configured
        _sendHeartbeat();
        // Signal any threads waiting for _logonInProgress
        _lock.signalAll();
      }
      catch (const AMPSException& ex)
      {
        _lock.signalAll();
        AMPS_UNHANDLED_EXCEPTION(ex);
        throw;
      }
      catch (...)
      {
        _lock.signalAll();
        throw;
      }

      if (_publishStore.isValid())
      {
        try
        {
          _publishStore.replay(_replayer);
          broadcastConnectionStateChanged(ConnectionStateListener::PublishReplayed);
        }
        catch (const PublishStoreGapException& ex)
        {
          _lock.signalAll();
          AMPS_UNHANDLED_EXCEPTION(ex);
          throw;
        }
        catch (const StoreException& ex)
        {
          _lock.signalAll();
          std::ostringstream os;
          os << "A local store exception occurred while logging on."
             << ex.toString();
          throw ConnectionException(os.str());
        }
        catch (const AMPSException& ex)
        {
          _lock.signalAll();
          AMPS_UNHANDLED_EXCEPTION(ex);
          throw;
        }
        catch (const std::exception& ex)
        {
          _lock.signalAll();
          AMPS_UNHANDLED_EXCEPTION(ex);
          throw;
        }
        catch (...)
        {
          _lock.signalAll();
          throw;
        }
      }
      _lock.signalAll();
      return newCommandId;
    }

    std::string subscribe(const MessageHandler& messageHandler_,
                          const std::string& topic_,
                          long timeout_,
                          const std::string& filter_,
                          const std::string& bookmark_,
                          const std::string& options_,
                          const std::string& subId_,
                          bool isHASubscribe_ = true)
    {
      isHASubscribe_ &= (bool)_subscriptionManager;
      Lock<Mutex> l(_lock);
      _message.reset();
      _message.setCommandEnum(Message::Command::Subscribe);
      _message.newCommandId();
      std::string subId(subId_);
      if (subId.empty())
      {
        if (options_.find(AMPS_OPTIONS_REPLACE, 0, strlen(AMPS_OPTIONS_REPLACE) - 1) != std::string::npos)
        {
          throw ConnectionException("Cannot issue a replacement subscription; a valid subscription id is required.");
        }

        subId = _message.getCommandId();
      }
      _message.setSubscriptionId(subId);
      // we need to deep copy this before sending the message; while we are
      // waiting for a response, the fields in _message may get blown away for
      // other operations.
      AMPS::Message::Field subIdField(subId);
      unsigned ackTypes = Message::AckType::Processed;

      if (!bookmark_.empty() && _bookmarkStore.isValid())
      {
        ackTypes |= Message::AckType::Persisted;
      }
      _message.setTopic(topic_);

      if (filter_.length())
      {
        _message.setFilter(filter_);
      }
      if (bookmark_.length())
      {
        if (bookmark_ == AMPS_BOOKMARK_RECENT)
        {
          Message::Field mostRecent = _bookmarkStore.getMostRecent(subIdField);
          _message.setBookmark(mostRecent);
        }
        else
        {
          _message.setBookmark(bookmark_);
          if (_bookmarkStore.isValid())
          {
            if (bookmark_ != AMPS_BOOKMARK_NOW &&
                bookmark_ != AMPS_BOOKMARK_EPOCH)
            {
              _bookmarkStore.log(_message);
              _bookmarkStore.discard(_message);
              _bookmarkStore.persisted(subIdField, _message.getBookmark());
            }
          }
        }
      }
      if (options_.length())
      {
        _message.setOptions(options_);
      }

      Message message = _message;
      if (isHASubscribe_)
      {
        message = _message.deepCopy();
        Unlock<Mutex> u(_lock);
        _subscriptionManager->subscribe(messageHandler_, message,
                                        Message::AckType::None);
        if (_badTimeToHASubscribe)
        {
          return subId;
        }
      }
      if (!_routes.hasRoute(_message.getSubscriptionId()))
      {
        _routes.addRoute(_message.getSubscriptionId(), messageHandler_,
                         Message::AckType::None, ackTypes, _message.getCommandEnum());
      }
      message.setAckTypeEnum(ackTypes);
      if (!options_.empty())
      {
        message.setOptions(options_);
      }
      try
      {
        syncAckProcessing(timeout_, message, isHASubscribe_);
      }
      catch (const DisconnectedException&)
      {
        if (!isHASubscribe_)
        {
          _routes.removeRoute(subIdField);
          throw;
        }
        else
        {
          AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subIdField));
          throw;
        }
      }
      catch (const TimedOutException&)
      {
        AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subIdField));
        throw;
      }
      catch (...)
      {
        if (isHASubscribe_)
        {
          // Have to unlock before calling into sub manager to avoid deadlock
          Unlock<Mutex> unlock(_lock);
          _subscriptionManager->unsubscribe(subIdField);
        }
        _routes.removeRoute(subIdField);
        throw;
      }

      return subId;
    }
    std::string deltaSubscribe(const MessageHandler& messageHandler_,
                               const std::string& topic_,
                               long timeout_,
                               const std::string& filter_,
                               const std::string& bookmark_,
                               const std::string& options_,
                               const std::string& subId_ = "",
                               bool isHASubscribe_ = true)
    {
      isHASubscribe_ &= (bool)_subscriptionManager;
      Lock<Mutex> l(_lock);
      _message.reset();
      _message.setCommandEnum(Message::Command::DeltaSubscribe);
      _message.newCommandId();
      std::string subId(subId_);
      if (subId.empty())
      {
        subId = _message.getCommandId();
      }
      _message.setSubscriptionId(subId);
      // we need to deep copy this before sending the message; while we are
      // waiting for a response, the fields in _message may get blown away for
      // other operations.
      AMPS::Message::Field subIdField(subId);
      unsigned ackTypes = Message::AckType::Processed;

      if (!bookmark_.empty() && _bookmarkStore.isValid())
      {
        ackTypes |= Message::AckType::Persisted;
      }
      _message.setTopic(topic_);
      if (filter_.length())
      {
        _message.setFilter(filter_);
      }
      if (bookmark_.length())
      {
        if (bookmark_ == AMPS_BOOKMARK_RECENT)
        {
          Message::Field mostRecent = _bookmarkStore.getMostRecent(subIdField);
          _message.setBookmark(mostRecent);
        }
        else
        {
          _message.setBookmark(bookmark_);
          if (_bookmarkStore.isValid())
          {
            if (bookmark_ != AMPS_BOOKMARK_NOW &&
                bookmark_ != AMPS_BOOKMARK_EPOCH)
            {
              _bookmarkStore.log(_message);
              _bookmarkStore.discard(_message);
              _bookmarkStore.persisted(subIdField, _message.getBookmark());
            }
          }
        }
      }
      if (options_.length())
      {
        _message.setOptions(options_);
      }
      Message message = _message;
      if (isHASubscribe_)
      {
        message = _message.deepCopy();
        Unlock<Mutex> u(_lock);
        _subscriptionManager->subscribe(messageHandler_, message,
                                        Message::AckType::None);
        if (_badTimeToHASubscribe)
        {
          return subId;
        }
      }
      if (!_routes.hasRoute(_message.getSubscriptionId()))
      {
        _routes.addRoute(_message.getSubscriptionId(), messageHandler_,
                         Message::AckType::None, ackTypes, _message.getCommandEnum());
      }
      message.setAckTypeEnum(ackTypes);
      if (!options_.empty())
      {
        message.setOptions(options_);
      }
      try
      {
        syncAckProcessing(timeout_, message, isHASubscribe_);
      }
      catch (const DisconnectedException&)
      {
        if (!isHASubscribe_)
        {
          _routes.removeRoute(subIdField);
          throw;
        }
      }
      catch (const TimedOutException&)
      {
        AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subIdField));
        throw;
      }
      catch (...)
      {
        if (isHASubscribe_)
        {
          // Have to unlock before calling into sub manager to avoid deadlock
          Unlock<Mutex> unlock(_lock);
          _subscriptionManager->unsubscribe(subIdField);
        }
        _routes.removeRoute(subIdField);
        throw;
      }
      return subId;
    }

    void unsubscribe(const std::string& id)
    {
      Lock<Mutex> l(_lock);
      unsubscribeInternal(id);
    }

    void unsubscribe(void)
    {
      if (_subscriptionManager)
      {
        _subscriptionManager->clear();
      }
      {
        _routes.unsubscribeAll();
        Lock<Mutex> l(_lock);
        _message.reset();
        _message.setCommandEnum(Message::Command::Unsubscribe);
        _message.newCommandId();
        _message.setSubscriptionId("all");
        _sendWithoutRetry(_message);
      }
      deferredExecution(&amps_noOpFn, NULL);
    }

    std::string sow(const MessageHandler& messageHandler_,
                    const std::string& topic_,
                    const std::string& filter_ = "",
                    const std::string& orderBy_ = "",
                    const std::string& bookmark_ = "",
                    int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
                    int topN_ = AMPS_DEFAULT_TOP_N,
                    const std::string& options_ = "",
                    long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT)
    {
      Lock<Mutex> l(_lock);
      _message.reset();
      _message.setCommandEnum(Message::Command::SOW);
      _message.newCommandId();
      // need to keep our own copy of the command ID.
      std::string commandId = _message.getCommandId();
      _message.setQueryID(_message.getCommandId());
      unsigned ackTypes = Message::AckType::Processed | Message::AckType::Completed;
      _message.setAckTypeEnum(ackTypes);
      _message.setTopic(topic_);
      if (filter_.length())
      {
        _message.setFilter(filter_);
      }
      if (orderBy_.length())
      {
        _message.setOrderBy(orderBy_);
      }
      if (bookmark_.length())
      {
        _message.setBookmark(bookmark_);
      }
      _message.setBatchSize(AMPS::asString(batchSize_));
      if (topN_ != AMPS_DEFAULT_TOP_N)
      {
        _message.setTopNRecordsReturned(AMPS::asString(topN_));
      }
      if (options_.length())
      {
        _message.setOptions(options_);
      }

      _routes.addRoute(_message.getQueryID(), messageHandler_,
                       Message::AckType::None, ackTypes, _message.getCommandEnum());

      try
      {
        syncAckProcessing(timeout_, _message);
      }
      catch (...)
      {
        AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId));
        throw;
      }

      return commandId;
    }

    std::string sow(const MessageHandler& messageHandler_,
                    const std::string& topic_,
                    long timeout_,
                    const std::string& filter_ = "",
                    int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
                    int topN_ = AMPS_DEFAULT_TOP_N)
    {
      std::string notSet;
      return sow(messageHandler_,
                 topic_,
                 filter_,
                 notSet,   // orderBy
                 notSet,   // bookmark
                 batchSize_,
                 topN_,
                 notSet,
                 timeout_);
    }

    std::string sowAndSubscribe(const MessageHandler& messageHandler_,
                                const std::string& topic_,
                                const std::string& filter_ = "",
                                const std::string& orderBy_ = "",
                                const std::string& bookmark_ = "",
                                int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
                                int topN_ = AMPS_DEFAULT_TOP_N,
                                const std::string& options_ = "",
                                long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT,
                                bool isHASubscribe_ = true)
    {
      isHASubscribe_ &= (bool)_subscriptionManager;
      unsigned ackTypes = Message::AckType::Processed;
      Lock<Mutex> l(_lock);
      _message.reset();
      _message.setCommandEnum(Message::Command::SOWAndSubscribe);
      _message.newCommandId();
      Field cid = _message.getCommandId();
      std::string subId = cid;
      _message.setQueryID(cid).setSubscriptionId(cid).setTopic(topic_);
      if (filter_.length())
      {
        _message.setFilter(filter_);
      }
      if (orderBy_.length())
      {
        _message.setOrderBy(orderBy_);
      }
      if (bookmark_.length())
      {
        _message.setBookmark(bookmark_);
        Message::Field bookmark = _message.getBookmark();
        if (_bookmarkStore.isValid())
        {
          ackTypes |= Message::AckType::Persisted;
          if (bookmark == AMPS_BOOKMARK_RECENT)
          {
            _message.setBookmark(_bookmarkStore.getMostRecent(_message.getSubscriptionId()));
          }
          else if (bookmark != AMPS_BOOKMARK_NOW &&
                   bookmark != AMPS_BOOKMARK_EPOCH)
          {
            _bookmarkStore.log(_message);
            if (!BookmarkRange::isRange(bookmark))
            {
              _bookmarkStore.discard(_message);
              _bookmarkStore.persisted(_message.getSubscriptionId(),
                                       bookmark);
            }
          }
        }
        else if (bookmark == AMPS_BOOKMARK_RECENT)
        {
          _message.setBookmark(AMPS_BOOKMARK_EPOCH);
        }
      }
      _message.setBatchSize(AMPS::asString(batchSize_));
      if (topN_ != AMPS_DEFAULT_TOP_N)
      {
        _message.setTopNRecordsReturned(AMPS::asString(topN_));
      }
      if (options_.length())
      {
        _message.setOptions(options_);
      }

      Message message = _message;
      if (isHASubscribe_)
      {
        message = _message.deepCopy();
        Unlock<Mutex> u(_lock);
        _subscriptionManager->subscribe(messageHandler_, message,
                                        Message::AckType::None);
        if (_badTimeToHASubscribe)
        {
          return subId;
        }
      }
      _routes.addRoute(cid, messageHandler_,
                       Message::AckType::None, ackTypes, _message.getCommandEnum());
      message.setAckTypeEnum(ackTypes);
      if (!options_.empty())
      {
        message.setOptions(options_);
      }
      try
      {
        syncAckProcessing(timeout_, message, isHASubscribe_);
      }
      catch (const DisconnectedException&)
      {
        if (!isHASubscribe_)
        {
          _routes.removeRoute(subId);
          throw;
        }
      }
      catch (const TimedOutException&)
      {
        AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subId));
        throw;
      }
      catch (...)
      {
        if (isHASubscribe_)
        {
          // Have to unlock before calling into sub manager to avoid deadlock
          Unlock<Mutex> unlock(_lock);
          _subscriptionManager->unsubscribe(cid);
        }
        _routes.removeRoute(subId);
        throw;
      }
      return subId;
    }

    std::string sowAndSubscribe(const MessageHandler& messageHandler_,
                                const std::string& topic_,
                                long timeout_,
                                const std::string& filter_ = "",
                                int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
                                bool oofEnabled_ = false,
                                int topN_ = AMPS_DEFAULT_TOP_N,
                                bool isHASubscribe_ = true)
    {
      std::string notSet;
      return sowAndSubscribe(messageHandler_,
                             topic_,
                             filter_,
                             notSet,   // orderBy
                             notSet,   // bookmark
                             batchSize_,
                             topN_,
                             (oofEnabled_ ? "oof" : ""),
                             timeout_,
                             isHASubscribe_);
    }

    std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
                                     const std::string& topic_,
                                     const std::string& filter_ = "",
                                     const std::string& orderBy_ = "",
                                     int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
                                     int topN_ = AMPS_DEFAULT_TOP_N,
                                     const std::string& options_ = "",
                                     long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT,
                                     bool isHASubscribe_ = true)
    {
      isHASubscribe_ &= (bool)_subscriptionManager;
      Lock<Mutex> l(_lock);
      _message.reset();
      _message.setCommandEnum(Message::Command::SOWAndDeltaSubscribe);
      _message.newCommandId();
      _message.setQueryID(_message.getCommandId());
      _message.setSubscriptionId(_message.getCommandId());
      std::string subId = _message.getSubscriptionId();
      _message.setTopic(topic_);
      if (filter_.length())
      {
        _message.setFilter(filter_);
      }
      if (orderBy_.length())
      {
        _message.setOrderBy(orderBy_);
      }
      _message.setBatchSize(AMPS::asString(batchSize_));
      if (topN_ != AMPS_DEFAULT_TOP_N)
      {
        _message.setTopNRecordsReturned(AMPS::asString(topN_));
      }
      if (options_.length())
      {
        _message.setOptions(options_);
      }
      Message message = _message;
      if (isHASubscribe_)
      {
        message = _message.deepCopy();
        Unlock<Mutex> u(_lock);
        _subscriptionManager->subscribe(messageHandler_, message,
                                        Message::AckType::None);
        if (_badTimeToHASubscribe)
        {
          return subId;
        }
      }
      _routes.addRoute(message.getQueryID(), messageHandler_,
                       Message::AckType::None, Message::AckType::Processed, message.getCommandEnum());
      message.setAckTypeEnum(Message::AckType::Processed);
      if (!options_.empty())
      {
        message.setOptions(options_);
      }
      try
      {
        syncAckProcessing(timeout_, message, isHASubscribe_);
      }
      catch (const DisconnectedException&)
      {
        if (!isHASubscribe_)
        {
          _routes.removeRoute(subId);
          throw;
        }
      }
      catch (const TimedOutException&)
      {
        AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subId));
        throw;
      }
      catch (...)
      {
        if (isHASubscribe_)
        {
          // Have to unlock before calling into sub manager to avoid deadlock
          Unlock<Mutex> unlock(_lock);
          _subscriptionManager->unsubscribe(Field(subId));
        }
        _routes.removeRoute(subId);
        throw;
      }
      return subId;
    }

    std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
                                     const std::string& topic_,
                                     long timeout_,
                                     const std::string& filter_ = "",
                                     int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
                                     bool oofEnabled_ = false,
                                     bool sendEmpties_ = false,
                                     int topN_ = AMPS_DEFAULT_TOP_N,
                                     bool isHASubscribe_ = true)
    {
      std::string notSet;
      Message::Options options;
      if (oofEnabled_)
      {
        options.setOOF();
      }
      if (sendEmpties_ == false)
      {
        options.setNoEmpties();
      }
      return sowAndDeltaSubscribe(messageHandler_,
                                  topic_,
                                  filter_,
                                  notSet,   // orderBy
                                  batchSize_,
                                  topN_,
                                  options,
                                  timeout_,
                                  isHASubscribe_);
    }

    std::string sowDelete(const MessageHandler& messageHandler_,
                          const std::string& topic_,
                          const std::string& filter_,
                          long timeout_,
                          Message::Field commandId_ = Message::Field())
    {
      if (_publishStore.isValid())
      {
        unsigned ackType = Message::AckType::Processed |
                           Message::AckType::Stats |
                           Message::AckType::Persisted;
        publishStoreMessage.reset();
        if (commandId_.empty())
        {
          publishStoreMessage.newCommandId();
          commandId_ = publishStoreMessage.getCommandId();
        }
        else
        {
          publishStoreMessage.setCommandId(commandId_.data(), commandId_.len());
        }
        publishStoreMessage.setCommandEnum(Message::Command::SOWDelete)
        .assignSubscriptionId(commandId_.data(), commandId_.len())
        .assignQueryID(commandId_.data(), commandId_.len())
        .setAckTypeEnum(ackType)
        .assignTopic(topic_.c_str(), topic_.length())
        .assignFilter(filter_.c_str(), filter_.length());
        amps_uint64_t haSequenceNumber = _publishStore.store(publishStoreMessage);
        char buf[AMPS_NUMBER_BUFFER_LEN];
        size_t pos = convertToCharArray(buf, haSequenceNumber);
        publishStoreMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
        {
          try
          {
            Lock<Mutex> l(_lock);
            _routes.addRoute(commandId_, messageHandler_,
                             Message::AckType::Stats,
                             Message::AckType::Processed | Message::AckType::Persisted,
                             publishStoreMessage.getCommandEnum());
            syncAckProcessing(timeout_, publishStoreMessage,
                              haSequenceNumber);
          }
          catch (const DisconnectedException&)
          {
            // -V565
            // Pass - it will get replayed upon reconnect
          }
          catch (...)
          {
            AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
            throw;
          }
        }
        return (std::string)commandId_;
      }
      else
      {
        Lock<Mutex> l(_lock);
        _message.reset();
        if (commandId_.empty())
        {
          _message.newCommandId();
          commandId_ = _message.getCommandId();
        }
        else
        {
          _message.setCommandId(commandId_.data(), commandId_.len());
        }
        _message.setCommandEnum(Message::Command::SOWDelete)
        .assignSubscriptionId(commandId_.data(), commandId_.len())
        .assignQueryID(commandId_.data(), commandId_.len())
        .setAckTypeEnum(Message::AckType::Processed |
                        Message::AckType::Stats)
        .assignTopic(topic_.c_str(), topic_.length())
        .assignFilter(filter_.c_str(), filter_.length());
        _routes.addRoute(commandId_, messageHandler_,
                         Message::AckType::Stats,
                         Message::AckType::Processed,
                         _message.getCommandEnum());
        try
        {
          syncAckProcessing(timeout_, _message);
        }
        catch (...)
        {
          AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
          throw;
        }
        return (std::string)commandId_;
      }
    }

    std::string sowDeleteByData(const MessageHandler& messageHandler_,
                                const std::string& topic_,
                                const std::string& data_,
                                long timeout_,
                                Message::Field commandId_ = Message::Field())
    {
      if (_publishStore.isValid())
      {
        unsigned ackType = Message::AckType::Processed |
                           Message::AckType::Stats |
                           Message::AckType::Persisted;
        publishStoreMessage.reset();
        if (commandId_.empty())
        {
          publishStoreMessage.newCommandId();
          commandId_ = publishStoreMessage.getCommandId();
        }
        else
        {
          publishStoreMessage.setCommandId(commandId_.data(), commandId_.len());
        }
        publishStoreMessage.setCommandEnum(Message::Command::SOWDelete)
        .assignSubscriptionId(commandId_.data(), commandId_.len())
        .assignQueryID(commandId_.data(), commandId_.len())
        .setAckTypeEnum(ackType)
        .assignTopic(topic_.c_str(), topic_.length())
        .assignData(data_.c_str(), data_.length());
        amps_uint64_t haSequenceNumber = _publishStore.store(publishStoreMessage);
        char buf[AMPS_NUMBER_BUFFER_LEN];
        size_t pos = convertToCharArray(buf, haSequenceNumber);
        publishStoreMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
        {
          try
          {
            Lock<Mutex> l(_lock);
            _routes.addRoute(commandId_, messageHandler_,
                             Message::AckType::Stats,
                             Message::AckType::Processed | Message::AckType::Persisted,
                             publishStoreMessage.getCommandEnum());
            syncAckProcessing(timeout_, publishStoreMessage,
                              haSequenceNumber);
          }
          catch (const DisconnectedException&)
          {
            // -V565
            // Pass - it will get replayed upon reconnect
          }
          catch (...)
          {
            AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
            throw;
          }
        }
        return (std::string)commandId_;
      }
      else
      {
        Lock<Mutex> l(_lock);
        _message.reset();
        if (commandId_.empty())
        {
          _message.newCommandId();
          commandId_ = _message.getCommandId();
        }
        else
        {
          _message.setCommandId(commandId_.data(), commandId_.len());
        }
        _message.setCommandEnum(Message::Command::SOWDelete)
        .assignSubscriptionId(commandId_.data(), commandId_.len())
        .assignQueryID(commandId_.data(), commandId_.len())
        .setAckTypeEnum(Message::AckType::Processed |
                        Message::AckType::Stats)
        .assignTopic(topic_.c_str(), topic_.length())
        .assignData(data_.c_str(), data_.length());
        _routes.addRoute(commandId_, messageHandler_,
                         Message::AckType::Stats,
                         Message::AckType::Processed,
                         _message.getCommandEnum());
        try
        {
          syncAckProcessing(timeout_, _message);
        }
        catch (...)
        {
          AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
          throw;
        }
        return (std::string)commandId_;
      }
    }

    std::string sowDeleteByKeys(const MessageHandler& messageHandler_,
                                const std::string& topic_,
                                const std::string& keys_,
                                long timeout_,
                                Message::Field commandId_ = Message::Field())
    {
      if (_publishStore.isValid())
      {
        unsigned ackType = Message::AckType::Processed |
                           Message::AckType::Stats |
                           Message::AckType::Persisted;
        publishStoreMessage.reset();
        if (commandId_.empty())
        {
          publishStoreMessage.newCommandId();
          commandId_ = publishStoreMessage.getCommandId();
        }
        else
        {
          publishStoreMessage.setCommandId(commandId_.data(), commandId_.len());
        }
        publishStoreMessage.setCommandEnum(Message::Command::SOWDelete)
        .assignSubscriptionId(commandId_.data(), commandId_.len())
        .assignQueryID(commandId_.data(), commandId_.len())
        .setAckTypeEnum(ackType)
        .assignTopic(topic_.c_str(), topic_.length())
        .assignSowKeys(keys_.c_str(), keys_.length());
        amps_uint64_t haSequenceNumber = _publishStore.store(publishStoreMessage);
        char buf[AMPS_NUMBER_BUFFER_LEN];
        size_t pos = convertToCharArray(buf, haSequenceNumber);
        publishStoreMessage.assignSequence(buf + pos, AMPS_NUMBER_BUFFER_LEN - pos);
        {
          try
          {
            Lock<Mutex> l(_lock);
            _routes.addRoute(commandId_, messageHandler_,
                             Message::AckType::Stats,
                             Message::AckType::Processed | Message::AckType::Persisted,
                             publishStoreMessage.getCommandEnum());
            syncAckProcessing(timeout_, publishStoreMessage,
                              haSequenceNumber);
          }
          catch (const DisconnectedException&)
          {
            // -V565
            // Pass - it will get replayed upon reconnect
          }
          catch (...)
          {
            AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
            throw;
          }
        }
        return (std::string)commandId_;
      }
      else
      {
        Lock<Mutex> l(_lock);
        _message.reset();
        if (commandId_.empty())
        {
          _message.newCommandId();
          commandId_ = _message.getCommandId();
        }
        else
        {
          _message.setCommandId(commandId_.data(), commandId_.len());
        }
        _message.setCommandEnum(Message::Command::SOWDelete)
        .assignSubscriptionId(commandId_.data(), commandId_.len())
        .assignQueryID(commandId_.data(), commandId_.len())
        .setAckTypeEnum(Message::AckType::Processed |
                        Message::AckType::Stats)
        .assignTopic(topic_.c_str(), topic_.length())
        .assignSowKeys(keys_.c_str(), keys_.length());
        _routes.addRoute(commandId_, messageHandler_,
                         Message::AckType::Stats,
                         Message::AckType::Processed,
                         _message.getCommandEnum());
        try
        {
          syncAckProcessing(timeout_, _message);
        }
        catch (...)
        {
          AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
          throw;
        }
        return (std::string)commandId_;
      }
    }

    void startTimer(void)
    {
      if (_serverVersion >= "5.3.2.0")
      {
        throw CommandException("The start_timer command is deprecated.");
      }
      Lock<Mutex> l(_lock);
      _message.reset();
      _message.setCommandEnum(Message::Command::StartTimer);

      _send(_message);
    }

    std::string stopTimer(MessageHandler messageHandler_)
    {
      if (_serverVersion >= "5.3.2.0")
      {
        throw CommandException("The stop_timer command is deprecated.");
      }
      return executeAsync(Command("stop_timer").addAckType("completed"), messageHandler_);
    }

    amps_handle getHandle(void)
    {
      return _client;
    }

    void setExceptionListener(const std::shared_ptr<const ExceptionListener>& pListener_)
    {
      _pExceptionListener = pListener_;
      _exceptionListener = _pExceptionListener.get();
    }

    void setExceptionListener(const ExceptionListener& listener_)
    {
      _exceptionListener = &listener_;
    }

    const ExceptionListener& getExceptionListener(void) const
    {
      return *_exceptionListener;
    }

    void setHeartbeat(unsigned heartbeatInterval_, unsigned readTimeout_)
    {
      if (readTimeout_ < heartbeatInterval_)
      {
        throw UsageException("The socket read timeout must be >= the heartbeat interval.");
      }
      Lock<Mutex> l(_lock);
      if (_heartbeatInterval != heartbeatInterval_ ||
          _readTimeout       != readTimeout_)
      {
        _heartbeatInterval = heartbeatInterval_;
        _readTimeout = readTimeout_;
        _sendHeartbeat();
      }
    }

    void _sendHeartbeat(void)
    {
      if (_connected && _heartbeatInterval != 0)
      {
        std::ostringstream options;
        options << "start," << _heartbeatInterval;
        _beatMessage.setOptions(options.str());

        _heartbeatTimer.setTimeout(_heartbeatInterval * 1000.0);
        _heartbeatTimer.start();
        try
        {
          _sendWithoutRetry(_beatMessage);
          broadcastConnectionStateChanged(ConnectionStateListener::HeartbeatInitiated);
        }
        catch (ConnectionException& ex_)
        {
          // If we are disconnected when we attempt to send, that's OK;
          //   we'll send this message after we re-connect (if we do).
          AMPS_UNHANDLED_EXCEPTION(ex_);
        }
        _beatMessage.setOptions("beat");
      }
      amps_result result = AMPS_E_OK;
      if (_readTimeout && _connected)
      {
        result = amps_client_set_read_timeout(_client, (int)_readTimeout);
        if (result != AMPS_E_OK && result != AMPS_E_DISCONNECTED)
        {
          AMPSException::throwFor(_client, result);
        }
        if (!_queueAckTimeout)
        {
          result = amps_client_set_idle_time(_client,
                                             (int)(_heartbeatInterval * 1000));
          if (result != AMPS_E_OK && result != AMPS_E_DISCONNECTED)
          {
            AMPSException::throwFor(_client, result);
          }
        }
      }
    }

    void addConnectionStateListener(ConnectionStateListener* listener_)
    {
      Lock<Mutex> lock(_lock);
      _connectionStateListeners.insert(listener_);
    }

    void removeConnectionStateListener(ConnectionStateListener* listener_)
    {
      Lock<Mutex> lock(_lock);
      _connectionStateListeners.erase(listener_);
    }

    void clearConnectionStateListeners()
    {
      Lock<Mutex> lock(_lock);
      _connectionStateListeners.clear();
    }

    void _registerHandler(Command& command_, Message::Field& cid_,
                          MessageHandler& handler_, unsigned requestedAcks_,
                          unsigned systemAddedAcks_, Message::Command::Type commandType_)
    {
      Message message = command_.getMessage();
      Message::Command::Type commandType = message.getCommandEnum();
      Message::Field subid = message.getSubscriptionId();
      Message::Field qid = message.getQueryID();
      // If we have an id, we're good, even if it's an existing route
      bool added = qid.len() || subid.len() || cid_.len();
      bool cidIsQid = cid_ == qid;
      bool cidUnique = !cidIsQid && cid_.len() > 0 && cid_ != subid;
      int addedCount = 0;
      if (subid.len() > 0)
      {
        // This can replace a non-subscribe with a matching id
        // with a subscription but not another subscription.
        addedCount += _routes.addRoute(subid, handler_, requestedAcks_,
                                       systemAddedAcks_, commandType_);
        if (!cidUnique
            && (commandType == Message::Command::Subscribe
                || commandType == Message::Command::DeltaSubscribe))
        {
          // We don't need to do anything else
          cid_ = subid;
          return;
        }
      }
      if (qid.len() > 0 && qid != subid
          && (commandType == Message::Command::SOW
              || commandType == Message::Command::SOWDelete
              || commandType == Message::Command::SOWAndSubscribe
              || commandType == Message::Command::SOWAndDeltaSubscribe))
      {
        while (_routes.hasRoute(qid))
        {
          message.newQueryId();
          if (cidIsQid)
          {
            cid_ = message.getQueryId();
          }
          qid = message.getQueryId();
        }
        if (addedCount == 0)
        {
          _routes.addRoute(qid, handler_, requestedAcks_,
                           systemAddedAcks_, commandType_);
        }
        else
        {
          void* data = NULL;
          {
            Unlock<Mutex> u(_lock);
            data = amps_invoke_copy_route_function(handler_.userData());
          }
          if (!data)
          {
            _routes.addRoute(qid, handler_, requestedAcks_,
                             systemAddedAcks_, commandType_);
          }
          else
          {
            _routes.addRoute(qid,
                             MessageHandler(handler_.function(),
                                            data),
                             requestedAcks_,
                             systemAddedAcks_, commandType_);
          }
        }
        ++addedCount;
      }
      if (cidUnique && requestedAcks_ & ~Message::AckType::Persisted)
      {
        while (_routes.hasRoute(cid_))
        {
          cid_ = message.newCommandId().getCommandId();
        }
        if (addedCount == 0)
        {
          _routes.addRoute(cid_, handler_, requestedAcks_,
                           systemAddedAcks_, commandType_);
        }
        else
        {
          void* data = NULL;
          {
            Unlock<Mutex> u(_lock);
            data = amps_invoke_copy_route_function(handler_.userData());
          }
          if (!data)
          {
            _routes.addRoute(cid_, handler_, requestedAcks_,
                             systemAddedAcks_, commandType_);
          }
          else
          {
            _routes.addRoute(cid_,
                             MessageHandler(handler_.function(),
                                            data),
                             requestedAcks_,
                             systemAddedAcks_, commandType_);
          }
        }
      }
      else if ((commandType == Message::Command::Publish ||
                commandType == Message::Command::DeltaPublish)
               && requestedAcks_ & ~Message::AckType::Persisted)
      {
        cid_ = command_.getMessage().newCommandId().getCommandId();
        _routes.addRoute(cid_, handler_, requestedAcks_,
                         systemAddedAcks_, commandType_);
        added = true;
      }
      if (!added)
      {
        throw UsageException("To use a messagehandler, you must also supply a command or subscription ID.");
      }
    }

    std::string executeAsyncNoLock(Command& command_, MessageHandler& handler_,
                                   bool isHASubscribe_ = true)
    {
      isHASubscribe_ &= (bool)_subscriptionManager;
      Message& message = command_.getMessage();
      unsigned systemAddedAcks = (handler_.isValid() || command_.hasProcessedAck()) ?
                                 Message::AckType::Processed : Message::AckType::None;
      unsigned requestedAcks = message.getAckTypeEnum();
      bool isPublishStore = _publishStore.isValid() && command_.needsSequenceNumber();
      Message::Command::Type commandType = message.getCommandEnum();
      if (commandType == Message::Command::StopTimer)
      {
        systemAddedAcks |= Message::AckType::Completed;
      }
      Message::Field cid = message.getCommandId();
      if (handler_.isValid() && cid.empty())
      {
        cid = message.newCommandId().getCommandId();
      }
      if (message.getBookmark().len() > 0)
      {
        if (command_.isSubscribe())
        {
          Message::Field bookmark = message.getBookmark();
          if (_bookmarkStore.isValid())
          {
            systemAddedAcks |= Message::AckType::Persisted;
            if (bookmark == AMPS_BOOKMARK_RECENT)
            {
              message.setBookmark(_bookmarkStore.getMostRecent(message.getSubscriptionId()));
            }
            else if (bookmark != AMPS_BOOKMARK_NOW &&
                     bookmark != AMPS_BOOKMARK_EPOCH)
            {
              _bookmarkStore.log(message);
              if (!BookmarkRange::isRange(bookmark))
              {
                _bookmarkStore.discard(message);
                _bookmarkStore.persisted(message.getSubscriptionId(),
                                         bookmark);
              }
            }
          }
          else if (bookmark == AMPS_BOOKMARK_RECENT)
          {
            message.setBookmark(AMPS_BOOKMARK_EPOCH);
          }
        }
      }
      if (isPublishStore)
      {
        systemAddedAcks |= Message::AckType::Persisted;
      }
      bool isSubscribe = command_.isSubscribe();
      if (handler_.isValid() && !isSubscribe)
      {
        _registerHandler(command_, cid, handler_,
                         requestedAcks, systemAddedAcks, commandType);
      }
      if (isPublishStore)
      {
        bool useSyncSend = cid.len() > 0 && command_.hasProcessedAck();
        amps_uint64_t haSequenceNumber = (amps_uint64_t)0;
        message.setAckTypeEnum(requestedAcks | systemAddedAcks);
        {
          Unlock<Mutex> u(_lock);
          haSequenceNumber = _publishStore.store(message);
        }
        message.setSequence(haSequenceNumber);
        try
        {
          if (useSyncSend)
          {
            syncAckProcessing((long)command_.getTimeout(), message,
                              haSequenceNumber);
          }
          else
          {
            _send(message, haSequenceNumber);
          }
        }
        catch (const DisconnectedException&)
        {
          // -V565
          // Pass - message will get replayed when reconnected
        }
        catch (...)
        {
          AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
          throw;
        }
      }
      else
      {
        if (isSubscribe)
        {
          const Message::Field& subId = message.getSubscriptionId();
          if (isHASubscribe_)
          {
            Unlock<Mutex> u(_lock);
            _subscriptionManager->subscribe(handler_,
                                            message.deepCopy(),
                                            requestedAcks);
            if (_badTimeToHASubscribe)
            {
              message.setAckTypeEnum(requestedAcks);
              return std::string(subId.data(), subId.len());
            }
          }
          if (handler_.isValid())
          {
            _registerHandler(command_, cid, handler_,
                             requestedAcks, systemAddedAcks, commandType);
          }
          message.setAckTypeEnum(requestedAcks | systemAddedAcks);
          try
          {
            syncAckProcessing((long)command_.getTimeout(), message,
                              isHASubscribe_);
          }
          catch (const DisconnectedException&)
          {
            if (!isHASubscribe_)
            {
              AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
              AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(subId));
              AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
              message.setAckTypeEnum(requestedAcks);
              throw;
            }
          }
          catch (const TimedOutException&)
          {
            AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(cid));
            AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subId));
            AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(message.getQueryId()));
            message.setAckTypeEnum(requestedAcks);
            throw;
          }
          catch (...)
          {
            if (isHASubscribe_)
            {
              // Have to unlock before calling into sub manager to avoid deadlock
              Unlock<Mutex> unlock(_lock);
              _subscriptionManager->unsubscribe(subId);
            }
            if (message.getQueryID().len() > 0)
            {
              _routes.removeRoute(message.getQueryID());
            }
            _routes.removeRoute(cid);
            _routes.removeRoute(subId);
            message.setAckTypeEnum(requestedAcks);
            throw;
          }
          if (subId.len() > 0)
          {
            message.setAckTypeEnum(requestedAcks);
            return std::string(subId.data(), subId.len());
          }
        }
        else
        {
          // SOW, Flush, etc. should always be sync. Publish/delete may not be.
          bool useSyncSend = commandType & ~Message::Command::NoDataCommands
                             || (cid.len() > 0 && command_.hasProcessedAck());
          message.setAckTypeEnum(requestedAcks | systemAddedAcks);
          try
          {
            if (useSyncSend)
            {
              syncAckProcessing((long)(command_.getTimeout()), message);
            }
            else
            {
              _send(message);
            }
          }
          catch (const TimedOutException&)
          {
            AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(cid));
            AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(message.getQueryId()));
            message.setAckTypeEnum(requestedAcks);
            throw;
          }
          catch (const DisconnectedException&)
          {
            AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
            AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
            message.setAckTypeEnum(requestedAcks);
            throw;
          }
          catch (...)
          {
            AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
            AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
            message.setAckTypeEnum(requestedAcks);
            throw;
          }
        }
      }
      message.setAckTypeEnum(requestedAcks);
      return cid;
    }

    MessageStream getEmptyMessageStream(void);

    std::string executeAsync(Command& command_, MessageHandler& handler_,
                             bool isHASubscribe_ = true)
    {
      Lock<Mutex> lock(_lock);
      return executeAsyncNoLock(command_, handler_, isHASubscribe_);
    }

    // Queue Methods //
    void setAutoAck(bool isAutoAckEnabled_)
    {
      _isAutoAckEnabled = isAutoAckEnabled_;
    }
    bool getAutoAck(void) const
    {
      return _isAutoAckEnabled;
    }
    void setAckBatchSize(const unsigned batchSize_)
    {
      _ackBatchSize = batchSize_;
      if (!_queueAckTimeout)
      {
        _queueAckTimeout = AMPS_DEFAULT_QUEUE_ACK_TIMEOUT;
        amps_client_set_idle_time(_client, _queueAckTimeout);
      }
    }
    unsigned getAckBatchSize(void) const
    {
      return _ackBatchSize;
    }
    int getAckTimeout(void) const
    {
      return _queueAckTimeout;
    }
    void setAckTimeout(const int ackTimeout_)
    {
      amps_client_set_idle_time(_client, ackTimeout_);
      _queueAckTimeout = ackTimeout_;
    }
    size_t _ack(QueueBookmarks& queueBookmarks_)
    {
      if (queueBookmarks_._bookmarkCount)
      {
        publishStoreMessage.reset();
        publishStoreMessage.setCommandEnum(Message::Command::SOWDelete)
        .setTopic(queueBookmarks_._topic)
        .setBookmark(queueBookmarks_._data)
        .setCommandId("AMPS-queue-ack");
        amps_uint64_t haSequenceNumber = 0;
        if (_publishStore.isValid())
        {
          haSequenceNumber = _publishStore.store(publishStoreMessage);
          publishStoreMessage.setAckType("persisted")
          .setSequence(haSequenceNumber);
          queueBookmarks_._data.erase();
          queueBookmarks_._bookmarkCount = 0;
        }
        _send(publishStoreMessage, haSequenceNumber);
        if (!_publishStore.isValid())
        {
          queueBookmarks_._data.erase();
          queueBookmarks_._bookmarkCount = 0;
        }
        return 1;
      }
      return 0;
    }
    void ack(const Field& topic_, const Field& bookmark_, const char* options_ = NULL)
    {
      if (_isAutoAckEnabled)
      {
        return;
      }
      _ack(topic_, bookmark_, options_);
    }
    void _ack(const Field& topic_, const Field& bookmark_, const char* options_ = NULL)
    {
      if (bookmark_.len() == 0)
      {
        return;
      }
      Lock<Mutex> lock(_lock);
      if (_ackBatchSize < 2 || options_ != NULL)
      {
        publishStoreMessage.reset();
        publishStoreMessage.setCommandEnum(Message::Command::SOWDelete)
        .setCommandId("AMPS-queue-ack")
        .setTopic(topic_).setBookmark(bookmark_);
        if (options_)
        {
          publishStoreMessage.setOptions(options_);
        }
        amps_uint64_t haSequenceNumber = 0;
        if (_publishStore.isValid())
        {
          haSequenceNumber = _publishStore.store(publishStoreMessage);
          publishStoreMessage.setAckType("persisted")
          .setSequence(haSequenceNumber);
        }
        _send(publishStoreMessage, haSequenceNumber);
        return;
      }
      // have we acked anything for this hash
      topic_hash hash = CRC<0>::crcNoSSE(topic_.data(), topic_.len());
      TopicHashMap::iterator it = _topicHashMap.find(hash);
      if (it == _topicHashMap.end())
      {
        // add a new one to the map
#ifdef AMPS_USE_EMPLACE
        it = _topicHashMap.emplace(TopicHashMap::value_type(hash, QueueBookmarks(topic_))).first;
#else
        it = _topicHashMap.insert(TopicHashMap::value_type(hash, QueueBookmarks(topic_))).first;
#endif
      }
      QueueBookmarks& queueBookmarks = it->second;
      if (queueBookmarks._data.length())
      {
        queueBookmarks._data.append(",");
      }
      else
      {
        queueBookmarks._oldestTime = amps_now();
      }
      queueBookmarks._data.append(bookmark_);
      if (++queueBookmarks._bookmarkCount >= _ackBatchSize)
      {
        _ack(queueBookmarks);
      }
    }
    void flushAcks(void)
    {
      size_t sendCount = 0;
      if (!_connected)
      {
        return;
      }
      else
      {
        Lock<Mutex> lock(_lock);
        typedef TopicHashMap::iterator iterator;
        for (iterator it = _topicHashMap.begin(), end = _topicHashMap.end(); it != end; ++it)
        {
          QueueBookmarks& queueBookmarks = it->second;
          sendCount += _ack(queueBookmarks);
        }
      }
      if (sendCount && _connected)
      {
        publishFlush(0, Message::AckType::Processed);
      }
    }
    // called when there's idle time, to see if we need to flush out any "acks"
    void checkQueueAcks(void)
    {
      if (!_topicHashMap.size())
      {
        return;
      }
      Lock<Mutex> lock(_lock);
      try
      {
        amps_uint64_t threshold = amps_now()
                                  - (amps_uint64_t)_queueAckTimeout;
        typedef TopicHashMap::iterator iterator;
        for (iterator it = _topicHashMap.begin(), end = _topicHashMap.end(); it != end; ++it)
        {
          QueueBookmarks& queueBookmarks = it->second;
          if (queueBookmarks._bookmarkCount && queueBookmarks._oldestTime < threshold)
          {
            _ack(queueBookmarks);
          }
        }
      }
      catch (std::exception& ex)
      {
        AMPS_UNHANDLED_EXCEPTION(ex);
      }
    }

    void deferredExecution(DeferredExecutionFunc func_, void* userData_)
    {
      Lock<Mutex> lock(_deferredExecutionLock);
#ifdef AMPS_USE_EMPLACE
      _deferredExecutionList.emplace_back(
        DeferredExecutionRequest(func_, userData_));
#else
      _deferredExecutionList.push_back(
        DeferredExecutionRequest(func_, userData_));
#endif
    }

    inline void processDeferredExecutions(void)
    {
      if (_deferredExecutionList.size())
      {
        Lock<Mutex> lock(_deferredExecutionLock);
        DeferredExecutionList::iterator it = _deferredExecutionList.begin();
        DeferredExecutionList::iterator end = _deferredExecutionList.end();
        for (; it != end; ++it)
        {
          try
          {
            it->_func(it->_userData);
          }
          catch (...)
          {
            // -V565
            // Intentionally ignore errors
          }
        }
        _deferredExecutionList.clear();
        _routes.invalidateCache();
        _routeCache.invalidateCache();
      }
    }

    bool getRetryOnDisconnect(void) const
    {
      return _isRetryOnDisconnect;
    }

    void setRetryOnDisconnect(bool isRetryOnDisconnect_)
    {
      _isRetryOnDisconnect = isRetryOnDisconnect_;
    }

    void setDefaultMaxDepth(unsigned maxDepth_)
    {
      _defaultMaxDepth = maxDepth_;
    }

    unsigned getDefaultMaxDepth(void) const
    {
      return _defaultMaxDepth;
    }

    void setTransportFilterFunction(amps_transport_filter_function filter_,
                                    void* userData_)
    {
      amps_client_set_transport_filter_function(_client, filter_, userData_);
    }

    void setThreadCreatedCallback(amps_thread_created_callback callback_,
                                  void* userData_)
    {
      amps_client_set_thread_created_callback(_client, callback_, userData_);
    }
  }; // class ClientImpl

  class MessageStream
  {
    RefHandle<MessageStreamImpl> _body;
  public:
    class iterator
    {
      MessageStream* _pStream;
      Message _current;
      inline void advance(void);

    public:
      iterator() // end
        : _pStream(NULL)
      {;}
      iterator(MessageStream* pStream_)
        : _pStream(pStream_)
      {
        advance();
      }

      bool operator==(const iterator& rhs) const
      {
        return _pStream == rhs._pStream;
      }
      bool operator!=(const iterator& rhs) const
      {
        return _pStream != rhs._pStream;
      }
      void operator++(void)
      {
        advance();
      }
      Message operator*(void)
      {
        return _current;
      }
      Message* operator->(void)
      {
        return &_current;
      }
    };
    bool isValid() const
    {
      return _body.isValid();
    }

    iterator begin(void)
    {
      if (!_body.isValid())
      {
        throw UsageException("This MessageStream is not valid and cannot be iterated.");
      }
      return iterator(this);
    }
    //           For non-SOW queries, the end is never reached.
    iterator end(void)
    {
      return iterator();
    }
    inline MessageStream(void);

    MessageStream timeout(unsigned timeout_);

    MessageStream conflate(void);
    MessageStream maxDepth(unsigned maxDepth_);
    unsigned getMaxDepth(void) const;
    unsigned getDepth(void) const;

  private:
    inline MessageStream(const Client& client_);
    inline void setSOWOnly(const std::string& commandId_,
                           const std::string& queryId_ = "");
    inline void setSubscription(const std::string& subId_,
                                const std::string& commandId_ = "",
                                const std::string& queryId_ = "");
    inline void setStatsOnly(const std::string& commandId_,
                             const std::string& queryId_ = "");
    inline void setAcksOnly(const std::string& commandId_, unsigned acks_);

    inline operator MessageHandler(void);

    inline static MessageStream fromExistingHandler(const MessageHandler& handler);

    friend class Client;

  };

  class Client // -V553
  {
  protected:
    BorrowRefHandle<ClientImpl> _body;
  public:
    static const int DEFAULT_COMMAND_TIMEOUT = AMPS_DEFAULT_COMMAND_TIMEOUT;
    static const int DEFAULT_BATCH_SIZE      = AMPS_DEFAULT_BATCH_SIZE;
    static const int DEFAULT_TOP_N           = AMPS_DEFAULT_TOP_N;

    Client(const std::string& clientName = "")
      : _body(new ClientImpl(clientName), true)
    {;}

    Client(ClientImpl* existingClient)
      : _body(existingClient, true)
    {;}

    Client(ClientImpl* existingClient, bool isRef)
      : _body(existingClient, isRef)
    {;}

    Client(const Client& rhs) : _body(rhs._body) {;}
    virtual ~Client(void) {;}

    Client& operator=(const Client& rhs)
    {
      _body = rhs._body;
      return *this;
    }

    bool isValid()
    {
      return _body.isValid();
    }

    void setName(const std::string& name)
    {
      _body.get().setName(name);
    }

    const std::string& getName() const
    {
      return _body.get().getName();
    }

    const std::string& getNameHash() const
    {
      return _body.get().getNameHash();
    }

    const amps_uint64_t getNameHashValue() const
    {
      return _body.get().getNameHashValue();
    }

    void setLogonCorrelationData(const std::string& logonCorrelationData_)
    {
      _body.get().setLogonCorrelationData(logonCorrelationData_);
    }

    const std::string& getLogonCorrelationData() const
    {
      return _body.get().getLogonCorrelationData();
    }

    size_t getServerVersion() const
    {
      return _body.get().getServerVersion();
    }

    VersionInfo getServerVersionInfo() const
    {
      return _body.get().getServerVersionInfo();
    }

    static size_t convertVersionToNumber(const std::string& version_)
    {
      return AMPS::convertVersionToNumber(version_.c_str(), version_.length());
    }

    static size_t convertVersionToNumber(const char* data_, size_t len_)
    {
      return AMPS::convertVersionToNumber(data_, len_);
    }

    const std::string& getURI() const
    {
      return _body.get().getURI();
    }



    void connect(const std::string& uri)
    {
      _body.get().connect(uri);
    }

    void disconnect()
    {
      _body.get().disconnect();
    }

    void send(const Message& message)
    {
      _body.get().send(message);
    }

    void addMessageHandler(const Field& commandId_,
                           const AMPS::MessageHandler& messageHandler_,
                           unsigned requestedAcks_, bool isSubscribe_)
    {
      Message::Command::Type commandType = isSubscribe_ ? Message::Command::Subscribe : Message::Command::SOW;
      _body.get().addMessageHandler(commandId_, messageHandler_,
                                    requestedAcks_, commandType);
    }

    void addMessageHandler(const Field& commandId_,
                           const AMPS::MessageHandler& messageHandler_,
                           unsigned requestedAcks_, Message::Command::Type commandType_)
    {
      _body.get().addMessageHandler(commandId_, messageHandler_,
                                    requestedAcks_, commandType_);
    }

    bool removeMessageHandler(const Field& commandId_)
    {
      return _body.get().removeMessageHandler(commandId_);
    }

    std::string send(const MessageHandler& messageHandler, Message& message, int timeout = 0)
    {
      return _body.get().send(messageHandler, message, timeout);
    }

    void setDisconnectHandler(const DisconnectHandler& disconnectHandler)
    {
      _body.get().setDisconnectHandler(disconnectHandler);
    }

    DisconnectHandler getDisconnectHandler(void) const
    {
      return _body.get().getDisconnectHandler();
    }

    virtual ConnectionInfo getConnectionInfo() const
    {
      return _body.get().getConnectionInfo();
    }

    void setBookmarkStore(const BookmarkStore& bookmarkStore_)
    {
      _body.get().setBookmarkStore(bookmarkStore_);
    }

    BookmarkStore getBookmarkStore()
    {
      return _body.get().getBookmarkStore();
    }

    SubscriptionManager* getSubscriptionManager()
    {
      return _body.get().getSubscriptionManager();
    }

    void setSubscriptionManager(SubscriptionManager* subscriptionManager_)
    {
      _body.get().setSubscriptionManager(subscriptionManager_);
    }

    void setPublishStore(const Store& publishStore_)
    {
      _body.get().setPublishStore(publishStore_);
    }

    Store getPublishStore()
    {
      return _body.get().getPublishStore();
    }

    void setDuplicateMessageHandler(const MessageHandler& duplicateMessageHandler_)
    {
      _body.get().setGlobalCommandTypeMessageHandler(ClientImpl::GlobalCommandTypeHandlers::DuplicateMessage,
                                                     duplicateMessageHandler_);
    }

    MessageHandler getDuplicateMessageHandler(void)
    {
      return _body.get().getDuplicateMessageHandler();
    }

    void setFailedWriteHandler(FailedWriteHandler* handler_)
    {
      _body.get().setFailedWriteHandler(handler_);
    }

    FailedWriteHandler* getFailedWriteHandler()
    {
      return _body.get().getFailedWriteHandler();
    }


    amps_uint64_t publish(const std::string& topic_, const std::string& data_)
    {
      return _body.get().publish(topic_.c_str(), topic_.length(),
                                 data_.c_str(), data_.length());
    }

    amps_uint64_t publish(const char* topic_, size_t topicLength_,
                          const char* data_, size_t dataLength_)
    {
      return _body.get().publish(topic_, topicLength_, data_, dataLength_);
    }

    amps_uint64_t publish(const std::string& topic_, const std::string& data_,
                          unsigned long expiration_)
    {
      return _body.get().publish(topic_.c_str(), topic_.length(),
                                 data_.c_str(), data_.length(), expiration_);
    }

    amps_uint64_t publish(const char* topic_, size_t topicLength_,
                          const char* data_, size_t dataLength_,
                          unsigned long expiration_)
    {
      return _body.get().publish(topic_, topicLength_,
                                 data_, dataLength_, expiration_);
    }

    void publishFlush(long timeout_ = 0, unsigned ackType_ = Message::AckType::Processed)
    {
      _body.get().publishFlush(timeout_, ackType_);
    }


    amps_uint64_t deltaPublish(const std::string& topic_, const std::string& data_)
    {
      return _body.get().deltaPublish(topic_.c_str(), topic_.length(),
                                      data_.c_str(), data_.length());
    }

    amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
                               const char* data_, size_t dataLength_)
    {
      return _body.get().deltaPublish(topic_, topicLength_,
                                      data_, dataLength_);
    }

    amps_uint64_t deltaPublish(const std::string& topic_, const std::string& data_,
                               unsigned long expiration_)
    {
      return _body.get().deltaPublish(topic_.c_str(), topic_.length(),
                                      data_.c_str(), data_.length(),
                                      expiration_);
    }

    amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
                               const char* data_, size_t dataLength_,
                               unsigned long expiration_)
    {
      return _body.get().deltaPublish(topic_, topicLength_,
                                      data_, dataLength_, expiration_);
    }

    std::string logon(int timeout_ = 0,
                      Authenticator& authenticator_ = DefaultAuthenticator::instance(),
                      const char* options_ = NULL)
    {
      return _body.get().logon(timeout_, authenticator_, options_);
    }
    std::string logon(const char* options_, int timeout_ = 0)
    {
      return _body.get().logon(timeout_, DefaultAuthenticator::instance(),
                               options_);
    }

    std::string logon(const std::string& options_, int timeout_ = 0)
    {
      return _body.get().logon(timeout_, DefaultAuthenticator::instance(),
                               options_.c_str());
    }

    std::string subscribe(const MessageHandler& messageHandler_,
                          const std::string& topic_,
                          long timeout_ = 0,
                          const std::string& filter_ = "",
                          const std::string& options_ = "",
                          const std::string& subId_ = "")
    {
      return _body.get().subscribe(messageHandler_, topic_, timeout_,
                                   filter_, "", options_, subId_);
    }

    MessageStream subscribe(const std::string& topic_,
                            long timeout_ = 0, const std::string& filter_ = "",
                            const std::string& options_ = "",
                            const std::string& subId_ = "")
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().subscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_, "",
                               options_, subId_, false));
      return result;
    }

    MessageStream subscribe(const char* topic_,
                            long timeout_ = 0, const std::string& filter_ = "",
                            const std::string& options_ = "",
                            const std::string& subId_ = "")
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().subscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_, "",
                               options_, subId_, false));
      return result;
    }

    std::string deltaSubscribe(const MessageHandler& messageHandler_,
                               const std::string& topic_,
                               long timeout_,
                               const std::string& filter_ = "",
                               const std::string& options_ = "",
                               const std::string& subId_ = "")
    {
      return _body.get().deltaSubscribe(messageHandler_, topic_, timeout_,
                                        filter_, "", options_, subId_);
    }
    MessageStream deltaSubscribe(const std::string& topic_,
                                 long timeout_, const std::string& filter_ = "",
                                 const std::string& options_ = "",
                                 const std::string& subId_ = "")
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().deltaSubscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_, "",
                               options_, subId_, false));
      return result;
    }

    MessageStream deltaSubscribe(const char* topic_,
                                 long timeout_, const std::string& filter_ = "",
                                 const std::string& options_ = "",
                                 const std::string& subId_ = "")
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().deltaSubscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_, "",
                               options_, subId_, false));
      return result;
    }

    std::string bookmarkSubscribe(const MessageHandler& messageHandler_,
                                  const std::string& topic_,
                                  long timeout_,
                                  const std::string& bookmark_,
                                  const std::string& filter_ = "",
                                  const std::string& options_ = "",
                                  const std::string& subId_ = "")
    {
      return _body.get().subscribe(messageHandler_, topic_, timeout_,
                                   filter_, bookmark_, options_, subId_);
    }
    MessageStream bookmarkSubscribe(const std::string& topic_,
                                    long timeout_,
                                    const std::string& bookmark_,
                                    const std::string& filter_ = "",
                                    const std::string& options_ = "",
                                    const std::string& subId_ = "")
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().subscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_,
                               bookmark_, options_,
                               subId_, false));
      return result;
    }

    MessageStream bookmarkSubscribe(const char* topic_,
                                    long timeout_,
                                    const std::string& bookmark_,
                                    const std::string& filter_ = "",
                                    const std::string& options_ = "",
                                    const std::string& subId_ = "")
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().subscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_,
                               bookmark_, options_,
                               subId_, false));
      return result;
    }

    void unsubscribe(const std::string& commandId)
    {
      return _body.get().unsubscribe(commandId);
    }

    void unsubscribe()
    {
      return _body.get().unsubscribe();
    }


    std::string sow(const MessageHandler& messageHandler_,
                    const std::string& topic_,
                    const std::string& filter_ = "",
                    const std::string& orderBy_ = "",
                    const std::string& bookmark_ = "",
                    int batchSize_ = DEFAULT_BATCH_SIZE,
                    int topN_ = DEFAULT_TOP_N,
                    const std::string& options_ = "",
                    long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      return _body.get().sow(messageHandler_, topic_, filter_, orderBy_,
                             bookmark_, batchSize_, topN_, options_,
                             timeout_);
    }
    MessageStream sow(const std::string& topic_,
                      const std::string& filter_ = "",
                      const std::string& orderBy_ = "",
                      const std::string& bookmark_ = "",
                      int batchSize_ = DEFAULT_BATCH_SIZE,
                      int topN_ = DEFAULT_TOP_N,
                      const std::string& options_ = "",
                      long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSOWOnly(_body.get().sow(result.operator MessageHandler(),
                                        topic_, filter_, orderBy_, bookmark_,
                                        batchSize_, topN_, options_, timeout_));
      return result;
    }

    MessageStream sow(const char* topic_,
                      const std::string& filter_ = "",
                      const std::string& orderBy_ = "",
                      const std::string& bookmark_ = "",
                      int batchSize_ = DEFAULT_BATCH_SIZE,
                      int topN_ = DEFAULT_TOP_N,
                      const std::string& options_ = "",
                      long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSOWOnly(_body.get().sow(result.operator MessageHandler(),
                                        topic_, filter_, orderBy_, bookmark_,
                                        batchSize_, topN_, options_, timeout_));
      return result;
    }
    std::string sow(const MessageHandler& messageHandler_,
                    const std::string& topic_,
                    long timeout_,
                    const std::string& filter_ = "",
                    int batchSize_ = DEFAULT_BATCH_SIZE,
                    int topN_ = DEFAULT_TOP_N)
    {
      return _body.get().sow(messageHandler_, topic_, timeout_, filter_,
                             batchSize_, topN_);
    }
    std::string sowAndSubscribe(const MessageHandler& messageHandler_,
                                const std::string& topic_,
                                long timeout_,
                                const std::string& filter_ = "",
                                int batchSize_ = DEFAULT_BATCH_SIZE,
                                bool oofEnabled_ = false,
                                int topN_ = DEFAULT_TOP_N)
    {
      return _body.get().sowAndSubscribe(messageHandler_, topic_, timeout_,
                                         filter_, batchSize_, oofEnabled_,
                                         topN_);
    }

    MessageStream sowAndSubscribe(const std::string& topic_,
                                  long timeout_,
                                  const std::string& filter_ = "",
                                  int batchSize_ = DEFAULT_BATCH_SIZE,
                                  bool oofEnabled_ = false,
                                  int topN_ = DEFAULT_TOP_N)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndSubscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_,
                               batchSize_, oofEnabled_,
                               topN_, false));
      return result;
    }
    MessageStream sowAndSubscribe(const char* topic_,
                                  long timeout_,
                                  const std::string& filter_ = "",
                                  int batchSize_ = DEFAULT_BATCH_SIZE,
                                  bool oofEnabled_ = false,
                                  int topN_ = DEFAULT_TOP_N)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndSubscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_,
                               batchSize_, oofEnabled_,
                               topN_, false));
      return result;
    }


    std::string sowAndSubscribe(const MessageHandler& messageHandler_,
                                const std::string& topic_,
                                const std::string& filter_ = "",
                                const std::string& orderBy_ = "",
                                const std::string& bookmark_ = "",
                                int batchSize_ = DEFAULT_BATCH_SIZE,
                                int topN_ = DEFAULT_TOP_N,
                                const std::string& options_ = "",
                                long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      return _body.get().sowAndSubscribe(messageHandler_, topic_, filter_,
                                         orderBy_, bookmark_, batchSize_,
                                         topN_, options_, timeout_);
    }

    MessageStream sowAndSubscribe(const std::string& topic_,
                                  const std::string& filter_ = "",
                                  const std::string& orderBy_ = "",
                                  const std::string& bookmark_ = "",
                                  int batchSize_ = DEFAULT_BATCH_SIZE,
                                  int topN_ = DEFAULT_TOP_N,
                                  const std::string& options_ = "",
                                  long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndSubscribe(
                               result.operator MessageHandler(),
                               topic_, filter_, orderBy_,
                               bookmark_, batchSize_, topN_,
                               options_, timeout_, false));
      return result;
    }

    MessageStream sowAndSubscribe(const char* topic_,
                                  const std::string& filter_ = "",
                                  const std::string& orderBy_ = "",
                                  const std::string& bookmark_ = "",
                                  int batchSize_ = DEFAULT_BATCH_SIZE,
                                  int topN_ = DEFAULT_TOP_N,
                                  const std::string& options_ = "",
                                  long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndSubscribe(
                               result.operator MessageHandler(),
                               topic_, filter_, orderBy_,
                               bookmark_, batchSize_, topN_,
                               options_, timeout_, false));
      return result;
    }

    std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
                                     const std::string& topic_,
                                     const std::string& filter_ = "",
                                     const std::string& orderBy_ = "",
                                     int batchSize_ = DEFAULT_BATCH_SIZE,
                                     int topN_ = DEFAULT_TOP_N,
                                     const std::string& options_ = "",
                                     long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      return _body.get().sowAndDeltaSubscribe(messageHandler_, topic_,
                                              filter_, orderBy_, batchSize_,
                                              topN_, options_, timeout_);
    }
    MessageStream sowAndDeltaSubscribe(const std::string& topic_,
                                       const std::string& filter_ = "",
                                       const std::string& orderBy_ = "",
                                       int batchSize_ = DEFAULT_BATCH_SIZE,
                                       int topN_ = DEFAULT_TOP_N,
                                       const std::string& options_ = "",
                                       long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndDeltaSubscribe(
                               result.operator MessageHandler(),
                               topic_, filter_, orderBy_,
                               batchSize_, topN_, options_,
                               timeout_, false));
      return result;
    }

    MessageStream sowAndDeltaSubscribe(const char* topic_,
                                       const std::string& filter_ = "",
                                       const std::string& orderBy_ = "",
                                       int batchSize_ = DEFAULT_BATCH_SIZE,
                                       int topN_ = DEFAULT_TOP_N,
                                       const std::string& options_ = "",
                                       long timeout_ = DEFAULT_COMMAND_TIMEOUT)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndDeltaSubscribe(
                               result.operator MessageHandler(),
                               topic_, filter_, orderBy_,
                               batchSize_, topN_, options_,
                               timeout_, false));
      return result;
    }

    std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
                                     const std::string& topic_,
                                     long timeout_,
                                     const std::string& filter_ = "",
                                     int batchSize_ = DEFAULT_BATCH_SIZE,
                                     bool oofEnabled_ = false,
                                     bool sendEmpties_ = false,
                                     int topN_ = DEFAULT_TOP_N)
    {
      return _body.get().sowAndDeltaSubscribe(messageHandler_, topic_,
                                              timeout_, filter_, batchSize_,
                                              oofEnabled_, sendEmpties_,
                                              topN_);
    }

    MessageStream sowAndDeltaSubscribe(const std::string& topic_,
                                       long timeout_,
                                       const std::string& filter_ = "",
                                       int batchSize_ = DEFAULT_BATCH_SIZE,
                                       bool oofEnabled_ = false,
                                       bool sendEmpties_ = false,
                                       int topN_ = DEFAULT_TOP_N)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndDeltaSubscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_,
                               batchSize_, oofEnabled_,
                               sendEmpties_, topN_, false));
      return result;
    }
    MessageStream sowAndDeltaSubscribe(const char* topic_,
                                       long timeout_,
                                       const std::string& filter_ = "",
                                       int batchSize_ = DEFAULT_BATCH_SIZE,
                                       bool oofEnabled_ = false,
                                       bool sendEmpties_ = false,
                                       int topN_ = DEFAULT_TOP_N)
    {
      MessageStream result(*this);
      if (_body.get().getDefaultMaxDepth())
      {
        result.maxDepth(_body.get().getDefaultMaxDepth());
      }
      result.setSubscription(_body.get().sowAndDeltaSubscribe(
                               result.operator MessageHandler(),
                               topic_, timeout_, filter_,
                               batchSize_, oofEnabled_,
                               sendEmpties_, topN_, false));
      return result;
    }
    std::string sowDelete(const MessageHandler& messageHandler,
                          const std::string& topic,
                          const std::string& filter,
                          long timeout)
    {
      return _body.get().sowDelete(messageHandler, topic, filter, timeout);
    }
    Message sowDelete(const std::string& topic, const std::string& filter,
                      long timeout = 0)
    {
      MessageStream stream(*this);
      char buf[Message::IdentifierLength + 1];
      buf[Message::IdentifierLength] = 0;
      AMPS_snprintf(buf, Message::IdentifierLength + 1, "%lx", MessageImpl::newId());
      Field cid(buf);
      try
      {
        stream.setStatsOnly(cid);
        _body.get().sowDelete(stream.operator MessageHandler(), topic, filter, timeout, cid);
        return *(stream.begin());
      }
      catch (const DisconnectedException&)
      {
        removeMessageHandler(cid);
        throw;
      }
    }

    void startTimer()
    {
      _body.get().startTimer();
    }

    std::string stopTimer(const MessageHandler& messageHandler)
    {
      return _body.get().stopTimer(messageHandler);
    }

    std::string sowDeleteByKeys(const MessageHandler& messageHandler_,
                                const std::string& topic_,
                                const std::string& keys_,
                                long timeout_ = 0)
    {
      return _body.get().sowDeleteByKeys(messageHandler_, topic_, keys_, timeout_);
    }
    Message sowDeleteByKeys(const std::string& topic_, const std::string& keys_,
                            long timeout_ = 0)
    {
      MessageStream stream(*this);
      char buf[Message::IdentifierLength + 1];
      buf[Message::IdentifierLength] = 0;
      AMPS_snprintf(buf, Message::IdentifierLength + 1, "%lx", MessageImpl::newId());
      Field cid(buf);
      try
      {
        stream.setStatsOnly(cid);
        _body.get().sowDeleteByKeys(stream.operator MessageHandler(), topic_, keys_, timeout_, cid);
        return *(stream.begin());
      }
      catch (const DisconnectedException&)
      {
        removeMessageHandler(cid);
        throw;
      }
    }

    std::string sowDeleteByData(const MessageHandler& messageHandler_,
                                const std::string& topic_, const std::string& data_,
                                long timeout_ = 0)
    {
      return _body.get().sowDeleteByData(messageHandler_, topic_, data_, timeout_);
    }

    Message sowDeleteByData(const std::string& topic_, const std::string& data_,
                            long timeout_ = 0)
    {
      MessageStream stream(*this);
      char buf[Message::IdentifierLength + 1];
      buf[Message::IdentifierLength] = 0;
      AMPS_snprintf(buf, Message::IdentifierLength + 1, "%lx", MessageImpl::newId());
      Field cid(buf);
      try
      {
        stream.setStatsOnly(cid);
        _body.get().sowDeleteByData(stream.operator MessageHandler(), topic_, data_, timeout_, cid);
        return *(stream.begin());
      }
      catch (const DisconnectedException&)
      {
        removeMessageHandler(cid);
        throw;
      }
    }

    amps_handle getHandle()
    {
      return _body.get().getHandle();
    }

    void setExceptionListener(const std::shared_ptr<const ExceptionListener>& pListener_)
    {
      _body.get().setExceptionListener(pListener_);
    }

    void setExceptionListener(const ExceptionListener& listener_)
    {
      _body.get().setExceptionListener(listener_);
    }

    const ExceptionListener& getExceptionListener(void) const
    {
      return _body.get().getExceptionListener();
    }

    //  type of message) from the server for the specified interval (plus a grace period),
    void setHeartbeat(unsigned heartbeatTime_, unsigned readTimeout_)
    {
      _body.get().setHeartbeat(heartbeatTime_, readTimeout_);
    }

    //  type of message) from the server for the specified interval (plus a grace period),
    void setHeartbeat(unsigned heartbeatTime_)
    {
      _body.get().setHeartbeat(heartbeatTime_, 2 * heartbeatTime_);
    }

    void setUnhandledMessageHandler(const AMPS::MessageHandler& messageHandler)
    {
      setLastChanceMessageHandler(messageHandler);
    }

    void setLastChanceMessageHandler(const AMPS::MessageHandler& messageHandler)
    {
      _body.get().setGlobalCommandTypeMessageHandler(ClientImpl::GlobalCommandTypeHandlers::LastChance,
                                                     messageHandler);
    }

    void setGlobalCommandTypeMessageHandler(const std::string& command_, const MessageHandler& handler_)
    {
      _body.get().setGlobalCommandTypeMessageHandler(command_, handler_);
    }

    void setGlobalCommandTypeMessageHandler(const Message::Command::Type command_, const MessageHandler& handler_)
    {
      _body.get().setGlobalCommandTypeMessageHandler(command_, handler_);
    }

    static const char* BOOKMARK_NOW()
    {
      return AMPS_BOOKMARK_NOW;
    }
    static const char* NOW()
    {
      return AMPS_BOOKMARK_NOW;
    }

    static const char* BOOKMARK_EPOCH()
    {
      return AMPS_BOOKMARK_EPOCH;
    }

    static const char* EPOCH()
    {
      return AMPS_BOOKMARK_EPOCH;
    }

    static const char* BOOKMARK_MOST_RECENT()
    {
      return AMPS_BOOKMARK_RECENT;
    }

    static const char* MOST_RECENT()
    {
      return AMPS_BOOKMARK_RECENT;
    }

    static const char* BOOKMARK_RECENT()
    {
      return AMPS_BOOKMARK_RECENT;
    }


    void addConnectionStateListener(ConnectionStateListener* listener)
    {
      _body.get().addConnectionStateListener(listener);
    }

    void removeConnectionStateListener(ConnectionStateListener* listener)
    {
      _body.get().removeConnectionStateListener(listener);
    }

    void clearConnectionStateListeners()
    {
      _body.get().clearConnectionStateListeners();
    }

    std::string executeAsync(Command& command_, MessageHandler handler_)
    {
      return _body.get().executeAsync(command_, handler_);
    }

    std::string executeAsyncNoResubscribe(Command& command_,
                                          MessageHandler handler_)
    {
      std::string id;
      try
      {
        if (command_.isSubscribe())
        {
          Message& message = command_.getMessage();
          Field subId = message.getSubscriptionId();
          bool useExistingHandler = !subId.empty() && !message.getOptions().empty() && message.getOptions().contains("replace", 7);
          if (useExistingHandler)
          {
            MessageHandler existingHandler;
            if (_body.get()._routes.getRoute(subId, existingHandler))
            {
              // we found an existing handler.
              _body.get().executeAsync(command_, existingHandler, false);
              return id; // empty string indicates existing
            }
          }
        }
        id = _body.get().executeAsync(command_, handler_, false);
      }
      catch (const DisconnectedException&)
      {
        removeMessageHandler(command_.getMessage().getCommandId());
        if (command_.isSubscribe())
        {
          removeMessageHandler(command_.getMessage().getSubscriptionId());
        }
        if (command_.isSow())
        {
          removeMessageHandler(command_.getMessage().getQueryID());
        }
        throw;
      }
      return id;
    }

    MessageStream execute(Command& command_);

    void ack(Field& topic_, Field& bookmark_, const char* options_ = NULL)
    {
      _body.get().ack(topic_, bookmark_, options_);
    }

    void ack(Message& message_, const char* options_ = NULL)
    {
      _body.get().ack(message_.getTopic(), message_.getBookmark(), options_);
    }
    void ack(const std::string& topic_, const std::string& bookmark_,
             const char* options_ = NULL)
    {
      _body.get().ack(Field(topic_.data(), topic_.length()), Field(bookmark_.data(), bookmark_.length()), options_);
    }

    void ackDeferredAutoAck(Field& topic_, Field& bookmark_, const char* options_ = NULL)
    {
      _body.get()._ack(topic_, bookmark_, options_);
    }
    void flushAcks(void)
    {
      _body.get().flushAcks();
    }

    bool getAutoAck(void) const
    {
      return _body.get().getAutoAck();
    }
    void setAutoAck(bool isAutoAckEnabled_)
    {
      _body.get().setAutoAck(isAutoAckEnabled_);
    }
    unsigned getAckBatchSize(void) const
    {
      return _body.get().getAckBatchSize();
    }
    void setAckBatchSize(const unsigned ackBatchSize_)
    {
      _body.get().setAckBatchSize(ackBatchSize_);
    }

    int getAckTimeout(void) const
    {
      return _body.get().getAckTimeout();
    }
    void setAckTimeout(const int ackTimeout_)
    {
      if (!ackTimeout_ && _body.get().getAckBatchSize() > 1)
      {
        throw UsageException("Ack timeout must be > 0 when ack batch size > 1");
      }
      _body.get().setAckTimeout(ackTimeout_);
    }


    void setRetryOnDisconnect(bool isRetryOnDisconnect_)
    {
      _body.get().setRetryOnDisconnect(isRetryOnDisconnect_);
    }

    bool getRetryOnDisconnect(void) const
    {
      return _body.get().getRetryOnDisconnect();
    }

    void setDefaultMaxDepth(unsigned maxDepth_)
    {
      _body.get().setDefaultMaxDepth(maxDepth_);
    }

    unsigned getDefaultMaxDepth(void) const
    {
      return _body.get().getDefaultMaxDepth();
    }

    void setTransportFilterFunction(amps_transport_filter_function filter_,
                                    void* userData_)
    {
      return _body.get().setTransportFilterFunction(filter_, userData_);
    }

    void setThreadCreatedCallback(amps_thread_created_callback callback_,
                                  void* userData_)
    {
      return _body.get().setThreadCreatedCallback(callback_, userData_);
    }

    void deferredExecution(DeferredExecutionFunc func_, void* userData_)
    {
      _body.get().deferredExecution(func_, userData_);
    }
  };

  inline void
  ClientImpl::lastChance(AMPS::Message& message)
  {
    AMPS_CALL_EXCEPTION_WRAPPER(_globalCommandTypeHandlers[GlobalCommandTypeHandlers::LastChance].invoke(message));
  }

  inline unsigned
  ClientImpl::persistedAck(AMPS::Message& message)
  {
    unsigned deliveries = 0;
    try
    {
      /*
      * Best Practice:  If you don't care about the dupe acks that
      * occur during failover or rapid disconnect/reconnect, then just
      * ignore them. We could discard each duplicate from the
      * persisted store, but the storage costs of doing 1 record
      * discards is heavy.  In most scenarios we'll just quickly blow
      * through the duplicates and get back to processing the
      * non-dupes.
      */
      const char* data = NULL;
      size_t len = 0;
      const char* status = NULL;
      size_t statusLen = 0;
      amps_handle messageHandle = message.getMessage();
      const size_t NotEntitled = 12, Duplicate = 9, Failure = 7;
      amps_message_get_field_value(messageHandle, AMPS_Reason, &data, &len);
      amps_message_get_field_value(messageHandle, AMPS_Status, &status, &statusLen);
      if (len == NotEntitled || len == Duplicate ||
          (statusLen == Failure && status[0] == 'f'))
      {
        if (_failedWriteHandler)
        {
          if (_publishStore.isValid())
          {
            amps_uint64_t sequence =
              amps_message_get_field_uint64(messageHandle, AMPS_Sequence);
            FailedWriteStoreReplayer replayer(this, data, len);
            AMPS_CALL_EXCEPTION_WRAPPER(_publishStore.replaySingle(
                                          replayer, sequence));
          }
          else // Call the handler with what little we have
          {
            static Message emptyMessage;
            emptyMessage.setSequence(message.getSequence());
            AMPS_CALL_EXCEPTION_WRAPPER(
              _failedWriteHandler->failedWrite(emptyMessage,
                                               data, len));
          }
          ++deliveries;
        }
      }
      if (_publishStore.isValid())
      {
        // Ack for publisher will have sequence while
        // ack for bookmark subscribe won't
        amps_uint64_t seq = amps_message_get_field_uint64(messageHandle,
                                                          AMPS_Sequence);
        if (seq > 0)
        {
          ++deliveries;
          AMPS_CALL_EXCEPTION_WRAPPER(_publishStore.discardUpTo(seq));
        }
      }

      if (!deliveries && _bookmarkStore.isValid())
      {
        amps_message_get_field_value(messageHandle, AMPS_SubscriptionId,
                                     &data, &len);
        if (len > 0)
        {
          Message::Field subId(data, len);
          const char* bookmarkData = NULL;
          size_t bookmarkLen = 0;
          amps_message_get_field_value(messageHandle,
                                       AMPS_Bookmark,
                                       &bookmarkData,
                                       &bookmarkLen);
          // Everything is there and not unsubscribed AC-912
          if (bookmarkLen > 0 && _routes.hasRoute(subId))
          {
            ++deliveries;
            _bookmarkStore.persisted(subId, Message::Field(bookmarkData, bookmarkLen));
          }
        }
      }
    }
    catch (std::exception& ex)
    {
      AMPS_UNHANDLED_EXCEPTION(ex);
    }
    return deliveries;
  }

  inline unsigned
  ClientImpl::processedAck(Message& message)
  {
    unsigned deliveries = 0;
    AckResponse ack;
    const char* data = NULL;
    size_t len = 0;
    amps_handle messageHandle = message.getMessage();
    amps_message_get_field_value(messageHandle, AMPS_CommandId, &data, &len);
    Lock<Mutex> l(_lock);
    if (data && len)
    {
      Lock<Mutex> guard(_ackMapLock);
      AckMap::iterator i = _ackMap.find(std::string(data, len));
      if (i != _ackMap.end())
      {
        ++deliveries;
        ack = i->second;
        _ackMap.erase(i);
      }
    }
    if (deliveries)
    {
      amps_message_get_field_value(messageHandle, AMPS_Status, &data, &len);
      ack.setStatus(data, len);
      amps_message_get_field_value(messageHandle, AMPS_Reason, &data, &len);
      ack.setReason(data, len);
      amps_message_get_field_value(messageHandle, AMPS_UserId, &data, &len);
      ack.setUsername(data, len);
      amps_message_get_field_value(messageHandle, AMPS_Password, &data, &len);
      ack.setPassword(data, len);
      amps_message_get_field_value(messageHandle, AMPS_Version, &data, &len);
      ack.setServerVersion(data, len);
      amps_message_get_field_value(messageHandle, AMPS_Options, &data, &len);
      ack.setOptions(data, len);
      // This sets bookmark, nameHashValue, and sequenceNo
      ack.setBookmark(message.getBookmark());
      ack.setResponded();
      _lock.signalAll();
    }
    return deliveries;
  }

  inline void
  ClientImpl::checkAndSendHeartbeat(bool force)
  {
    if (force || _heartbeatTimer.check())
    {
      _heartbeatTimer.start();
      try
      {
        sendWithoutRetry(_beatMessage);
      }
      catch (const AMPSException&)
      {
        ;
      }
    }
  }

  inline ConnectionInfo ClientImpl::getConnectionInfo() const
  {
    ConnectionInfo info;
    std::ostringstream writer;

    info["client.uri"] = _lastUri;
    info["client.name"] = _name;
    info["client.username"] = _username;
    if (_publishStore.isValid())
    {
      writer << _publishStore.unpersistedCount();
      info["publishStore.unpersistedCount"] = writer.str();
      writer.clear();
      writer.str("");
    }

    return info;
  }

  inline amps_result
  ClientImpl::ClientImplMessageHandler(amps_handle messageHandle_, void* userData_)
  {
    const unsigned SOWMask = Message::Command::SOW | Message::Command::GroupBegin | Message::Command::GroupEnd;
    const unsigned PublishMask = Message::Command::OOF | Message::Command::Publish | Message::Command::DeltaPublish;
    ClientImpl* me = (ClientImpl*) userData_;
    AMPS_CALL_EXCEPTION_WRAPPER_2(me, me->processDeferredExecutions());
    if (!messageHandle_)
    {
      if (me->_queueAckTimeout)
      {
        me->checkQueueAcks();
      }
      me->checkAndSendHeartbeat();
      return AMPS_E_OK;
    }

    me->_readMessage.replace(messageHandle_);
    Message& message = me->_readMessage;
    Message::Command::Type commandType = message.getCommandEnum();
    if (commandType & SOWMask)
    {
#if 0 // Not currently implemented, to avoid an extra branch in delivery
      // A small cheat here to get the right handler, using knowledge of the
      // Command values of SOW (8), GroupBegin (8192), and GroupEnd (16384)
      // and their GlobalCommandTypeHandlers values 1, 2, 3.
      AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                    me->_globalCommandTypeHandlers[1 + (commandType / 8192)].invoke(message));
#endif
      AMPS_CALL_EXCEPTION_WRAPPER_2(me, me->_routes.deliverData(message,
                                                                message.getQueryID()));
    }
    else if (commandType & PublishMask)
    {
#if 0 // Not currently implemented, to avoid an extra branch in delivery
      AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                    me->_globalCommandTypeHandlers[(commandType == Message::Command::Publish ?
                                                                    GlobalCommandTypeHandlers::Publish :
                                                                    GlobalCommandTypeHandlers::OOF)].invoke(message));
#endif
      const char* subIds = NULL;
      size_t subIdsLen = 0;
      // Publish command, send to subscriptions
      amps_message_get_field_value(messageHandle_, AMPS_SubscriptionIds,
                                   &subIds, &subIdsLen);
      size_t subIdCount = me->_routes.parseRoutes(AMPS::Field(subIds, subIdsLen), me->_routeCache);
      for (size_t i = 0; i < subIdCount; ++i)
      {
        MessageRouter::RouteCache::value_type& lookupResult = me->_routeCache[i];
        MessageHandler& handler = lookupResult.handler;
        if (handler.isValid())
        {
          amps_message_set_field_value(messageHandle_,
                                       AMPS_SubscriptionId,
                                       subIds + lookupResult.idOffset,
                                       lookupResult.idLength);
          Message::Field bookmark = message.getBookmark();
          bool isMessageQueue = message.getLeasePeriod().len() != 0;
          bool isAutoAck = me->_isAutoAckEnabled;

          if (!isMessageQueue && !bookmark.empty() &&
              me->_bookmarkStore.isValid())
          {
            if (me->_bookmarkStore.isDiscarded(me->_readMessage))
            {
              //Call duplicate message handler in handlers map
              if (me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage].isValid())
              {
                AMPS_CALL_EXCEPTION_WRAPPER_2(me, me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage].invoke(message));
              }
            }
            else
            {
              me->_bookmarkStore.log(me->_readMessage);
              AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                            handler.invoke(message));
            }
          }
          else
          {
            if (isMessageQueue && isAutoAck)
            {
              try
              {
                AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me, handler.invoke(message));
                if (!message.getIgnoreAutoAck())
                {
                  AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                                me->_ack(message.getTopic(), message.getBookmark()));
                }
              }
              catch (std::exception& ex)
              {
                if (!message.getIgnoreAutoAck())
                {
                  AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                                me->_ack(message.getTopic(), message.getBookmark(), "cancel"));
                }
                AMPS_UNHANDLED_EXCEPTION_2(me, ex);
              }
            }
            else
            {
              AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                            handler.invoke(message));
            }
          }
        }
        else
        {
          me->lastChance(message);
        }
      } // for (subidsEnd)
    }
    else if (commandType == Message::Command::Ack)
    {
      AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                    me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::Ack].invoke(message));
      unsigned ackType = message.getAckTypeEnum();
      unsigned deliveries = 0U;
      switch (ackType)
      {
      case Message::AckType::Persisted:
        deliveries += me->persistedAck(message);
        break;
      case Message::AckType::Processed: // processed
        deliveries += me->processedAck(message);
        break;
      }
      AMPS_CALL_EXCEPTION_WRAPPER_2(me, deliveries += me->_routes.deliverAck(message, ackType));
      if (deliveries == 0)
      {
        me->lastChance(message);
      }
    }
    else if (commandType == Message::Command::Heartbeat)
    {
      AMPS_CALL_EXCEPTION_WRAPPER_2(me,
                                    me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::Heartbeat].invoke(message));
      if (me->_heartbeatTimer.getTimeout() != 0.0) // -V550
      {
        me->checkAndSendHeartbeat(true);
      }
      else
      {
        me->lastChance(message);
      }
      return AMPS_E_OK;
    }
    else if (!message.getCommandId().empty())
    {
      unsigned deliveries = 0U;
      try
      {
        while (me->_connected) // Keep sending heartbeats when stream is full
        {
          try
          {
            deliveries = me->_routes.deliverData(message, message.getCommandId());
            break;
          }
#ifdef _WIN32
          catch (MessageStreamFullException&)
#else
          catch (MessageStreamFullException& ex_)
#endif
          {
            try
            {
              me->checkAndSendHeartbeat(false);
            }
#ifdef _WIN32
            catch (std::exception&)
#else
            catch (std::exception& ex_)
#endif
            {
              ;
            }
          }
        }
      }
      catch (std::exception& ex_)
      {
        try
        {
          me->_exceptionListener->exceptionThrown(ex_);
        }
        catch (...)
        {
          ;
        }
      }
      if (deliveries == 0)
      {
        me->lastChance(message);
      }
    }
    me->checkAndSendHeartbeat();
    return AMPS_E_OK;
  }

  inline void
  ClientImpl::ClientImplPreDisconnectHandler(amps_handle /*client*/, unsigned failedConnectionVersion, void* userData)
  {
    ClientImpl* me = (ClientImpl*) userData;
    //Client wrapper(me);
    // Go ahead and signal any waiters if they are around...
    me->clearAcks(failedConnectionVersion);
  }

  inline amps_result
  ClientImpl::ClientImplDisconnectHandler(amps_handle /*client*/, void* userData)
  {
    ClientImpl* me = (ClientImpl*) userData;
    Lock<Mutex> l(me->_lock);
    Client wrapper(me, false);
    if (me->_connected)
    {
      me->broadcastConnectionStateChanged(ConnectionStateListener::Disconnected);
    }
    while (true)
    {
      AtomicFlagFlip subFlip(&me->_badTimeToHASubscribe);
      try
      {
        me->_connected = false;
        {
          // Have to release the lock here or receive thread can't
          // invoke the message handler.
          Unlock<Mutex> unlock(me->_lock);
          me->_disconnectHandler.invoke(wrapper);
        }
      }
      catch (const std::exception& ex)
      {
        AMPS_UNHANDLED_EXCEPTION_2(me, ex);
      }
      me->_lock.signalAll();

      if (!me->_connected)
      {
        me->broadcastConnectionStateChanged(ConnectionStateListener::Shutdown);
        AMPS_UNHANDLED_EXCEPTION_2(me, DisconnectedException("Reconnect failed."));
        return AMPS_E_DISCONNECTED;
      }
      try
      {
        // Resubscribe
        if (me->_subscriptionManager)
        {
          {
            // Have to release the lock here or receive thread can't
            // invoke the message handler.
            Unlock<Mutex> unlock(me->_lock);
            me->_subscriptionManager->resubscribe(wrapper);
          }
          me->broadcastConnectionStateChanged(ConnectionStateListener::Resubscribed);
        }
        return AMPS_E_OK;
      }
      catch (const AMPSException& subEx)
      {
        AMPS_UNHANDLED_EXCEPTION_2(me, subEx);
      }
      catch (const std::exception& subEx)
      {
        AMPS_UNHANDLED_EXCEPTION_2(me, subEx);
        return AMPS_E_RETRY;
      }
      catch (...)
      {
        return AMPS_E_RETRY;
      }
    }
    return AMPS_E_RETRY;
  }

  class FIX
  {
    const char* _data;
    size_t _len;
    char _fieldSep;
  public:
    class iterator
    {
      const char* _data;
      size_t _len;
      size_t _pos;
      char _fieldSep;
      iterator(const char* data_, size_t len_, size_t pos_, char fieldSep_)
        : _data(data_), _len(len_), _pos(pos_), _fieldSep(fieldSep_)
      {
        while (_pos != _len && _data[_pos] == _fieldSep)
        {
          ++_pos;
        }
      }
    public:
      typedef void* difference_type;
      typedef std::forward_iterator_tag iterator_category;
      typedef std::pair<Message::Field, Message::Field> value_type;
      typedef value_type* pointer;
      typedef value_type& reference;
      bool operator==(const iterator& rhs) const
      {
        return _pos == rhs._pos;
      }
      bool operator!=(const iterator& rhs) const
      {
        return _pos != rhs._pos;
      }
      iterator& operator++()
      {
        // Skip through the data
        while (_pos != _len && _data[_pos] != _fieldSep)
        {
          ++_pos;
        }
        // Skip through any field separators
        while (_pos != _len && _data[_pos] == _fieldSep)
        {
          ++_pos;
        }
        return *this;
      }

      value_type operator*() const
      {
        value_type result;
        size_t i = _pos, keyLength = 0, valueStart = 0, valueLength = 0;
        for (; i < _len && _data[i] != '='; ++i)
        {
          ++keyLength;
        }

        result.first.assign(_data + _pos, keyLength);

        if (i < _len && _data[i] == '=')
        {
          ++i;
          valueStart = i;
          for (; i < _len && _data[i] != _fieldSep; ++i)
          {
            valueLength++;
          }
        }
        result.second.assign(_data + valueStart, valueLength);
        return result;
      }

      friend class FIX;
    };
    class reverse_iterator
    {
      const char* _data;
      size_t _len;
      const char* _pos;
      char _fieldSep;
    public:
      typedef std::pair<Message::Field, Message::Field> value_type;
      reverse_iterator(const char* data, size_t len, const char* pos, char fieldsep)
        : _data(data), _len(len), _pos(pos), _fieldSep(fieldsep)
      {
        if (_pos)
        {
          // skip past meaningless trailing fieldseps
          while (_pos >= _data && *_pos == _fieldSep)
          {
            --_pos;
          }
          while (_pos > _data && *_pos != _fieldSep)
          {
            --_pos;
          }
          // if we stopped before the 0th character, it's because
          // it's a field sep.  advance one to point to the first character
          // of a key.
          if (_pos > _data || (_pos == _data && *_pos == _fieldSep))
          {
            ++_pos;
          }
          if (_pos < _data)
          {
            _pos = 0;
          }
        }
      }
      bool operator==(const reverse_iterator& rhs) const
      {
        return _pos == rhs._pos;
      }
      bool operator!=(const reverse_iterator& rhs) const
      {
        return _pos != rhs._pos;
      }
      reverse_iterator& operator++()
      {
        if (_pos == _data)
        {
          _pos = 0;
        }
        else
        {
          // back up 1 to a field separator
          --_pos;
          // keep backing up through field separators
          while (_pos >= _data && *_pos == _fieldSep)
          {
            --_pos;
          }
          // now back up to the beginning of this field
          while (_pos > _data && *_pos != _fieldSep)
          {
            --_pos;
          }
          if (_pos > _data || (_pos == _data && *_pos == _fieldSep))
          {
            ++_pos;
          }
          if (_pos < _data)
          {
            _pos = 0;
          }
        }
        return *this;
      }
      value_type operator*() const
      {
        value_type result;
        size_t keyLength = 0, valueStart = 0, valueLength = 0;
        size_t i = (size_t)(_pos - _data);
        for (; i < _len && _data[i] != '='; ++i)
        {
          ++keyLength;
        }
        result.first.assign(_pos, keyLength);
        if (i < _len && _data[i] == '=')
        {
          ++i;
          valueStart = i;
          for (; i < _len && _data[i] != _fieldSep; ++i)
          {
            valueLength++;
          }
        }
        result.second.assign(_data + valueStart, valueLength);
        return result;
      }
    };
    FIX(const Message::Field& data, char fieldSeparator = 1)
      : _data(data.data()), _len(data.len()),
        _fieldSep(fieldSeparator)
    {
    }

    FIX(const char* data, size_t len, char fieldSeparator = 1)
      : _data(data), _len(len), _fieldSep(fieldSeparator)
    {
    }

    iterator begin() const
    {
      return iterator(_data, _len, 0, _fieldSep);
    }
    iterator end() const
    {
      return iterator(_data, _len, _len, _fieldSep);
    }


    reverse_iterator rbegin() const
    {
      return reverse_iterator(_data, _len, _data + (_len - 1), _fieldSep);
    }

    reverse_iterator rend() const
    {
      return reverse_iterator(_data, _len, 0, _fieldSep);
    }
  };



  template <class T>
  class _FIXBuilder
  {
    std::stringstream _data;
    char _fs;
  public:
    _FIXBuilder(char fieldSep_ = (char)1) : _fs(fieldSep_) {;}

    void append(const T& tag, const char* value, size_t offset, size_t length)
    {
      _data << tag << '=';
      _data.write(value + offset, (std::streamsize)length);
      _data << _fs;
    }
    void append(const T& tag, const std::string& value)
    {
      _data << tag << '=' << value << _fs;
    }

    std::string getString() const
    {
      return _data.str();
    }
    operator std::string() const
    {
      return _data.str();
    }

    void reset()
    {
      _data.str(std::string());
    }
  };


  typedef _FIXBuilder<int> FIXBuilder;


  typedef _FIXBuilder<std::string> NVFIXBuilder;



  class FIXShredder
  {
    char _fs;
  public:
    FIXShredder(char fieldSep_ = (char)1) : _fs(fieldSep_) {;}

    typedef std::map<Message::Field, Message::Field> map_type;

    map_type toMap(const Message::Field& data)
    {
      FIX fix(data, _fs);
      map_type retval;
      for (FIX::iterator a = fix.begin(); a != fix.end(); ++a)
      {
        retval.insert(*a);
      }

      return retval;
    }
  };

#define AMPS_MESSAGE_STREAM_CACHE_MAX 128
  class MessageStreamImpl : public AMPS::RefBody, AMPS::ConnectionStateListener
  {
    Mutex                _lock;
    std::deque<Message>  _q;
    std::deque<Message>  _cache;
    std::string          _commandId;
    std::string          _subId;
    std::string          _queryId;
    Client               _client;
    unsigned             _timeout;
    unsigned             _maxDepth;
    unsigned             _requestedAcks;
    size_t               _cacheMax;
    Message::Field       _previousTopic;
    Message::Field       _previousBookmark;
    typedef enum : unsigned int { Unset = 0x0, Running = 0x10, Subscribe = 0x11, SOWOnly = 0x12, AcksOnly = 0x13, Conflate = 0x14, Closed = 0x1, Disconnected = 0x2 } State;
#if __cplusplus >= 201100L || _MSC_VER >= 1900
    std::atomic<State> _state;
#else
    volatile State _state;
#endif
    typedef std::map<std::string, Message*> SOWKeyMap;
    SOWKeyMap _sowKeyMap;
  public:
    MessageStreamImpl(const Client& client_)
      : _client(client_),
        _timeout(0),
        _maxDepth((unsigned)~0),
        _requestedAcks(0),
        _cacheMax(AMPS_MESSAGE_STREAM_CACHE_MAX),
        _state(Unset)
    {
      if (_client.isValid())
      {
        _client.addConnectionStateListener(this);
      }
    }

    MessageStreamImpl(ClientImpl* client_)
      : _client(client_),
        _timeout(0),
        _maxDepth((unsigned)~0),
        _requestedAcks(0),
        _state(Unset)
    {
      if (_client.isValid())
      {
        _client.addConnectionStateListener(this);
      }
    }

    ~MessageStreamImpl()
    {
    }

    virtual void destroy()
    {
      try
      {
        close();
      }
      catch (std::exception& e)
      {
        try
        {
          if (_client.isValid())
          {
            _client.getExceptionListener().exceptionThrown(e);
          }
        }
        catch (...) {/*Ignore exception listener exceptions*/}   // -V565
      }
      if (_client.isValid())
      {
        _client.removeConnectionStateListener(this);
        Client c = _client;
        _client = Client((ClientImpl*)NULL);
        c.deferredExecution(MessageStreamImpl::destroyer, this);
      }
      else
      {
        delete this;
      }
    }

    static void destroyer(void* vpMessageStreamImpl_)
    {
      delete ((MessageStreamImpl*)vpMessageStreamImpl_);
    }

    void setSubscription(const std::string& subId_,
                         const std::string& commandId_ = "",
                         const std::string& queryId_ = "")
    {
      Lock<Mutex> lock(_lock);
      _subId = subId_;
      if (!commandId_.empty() && commandId_ != subId_)
      {
        _commandId = commandId_;
      }
      if (!queryId_.empty() && queryId_ != subId_ && queryId_ != commandId_)
      {
        _queryId = queryId_;
      }
      // It's possible to disconnect between creation/registration and here.
      if (Disconnected == _state)
      {
        return;
      }
      assert(Unset == _state);
      _state = Subscribe;
    }

    void setSOWOnly(const std::string& commandId_,
                    const std::string& queryId_ = "")
    {
      Lock<Mutex> lock(_lock);
      _commandId = commandId_;
      if (!queryId_.empty() && queryId_ != commandId_)
      {
        _queryId = queryId_;
      }
      // It's possible to disconnect between creation/registration and here.
      if (Disconnected == _state)
      {
        return;
      }
      assert(Unset == _state);
      _state = SOWOnly;
    }

    void setStatsOnly(const std::string& commandId_,
                      const std::string& queryId_ = "")
    {
      Lock<Mutex> lock(_lock);
      _commandId = commandId_;
      if (!queryId_.empty() && queryId_ != commandId_)
      {
        _queryId = queryId_;
      }
      // It's possible to disconnect between creation/registration and here.
      if (Disconnected == _state)
      {
        return;
      }
      assert(Unset == _state);
      _state = AcksOnly;
      _requestedAcks = Message::AckType::Stats;
    }

    void setAcksOnly(const std::string& commandId_, unsigned acks_)
    {
      Lock<Mutex> lock(_lock);
      _commandId = commandId_;
      // It's possible to disconnect between creation/registration and here.
      if (Disconnected == _state)
      {
        return;
      }
      assert(Unset == _state);
      _state = AcksOnly;
      _requestedAcks = acks_;
    }

    void connectionStateChanged(ConnectionStateListener::State state_)
    {
      Lock<Mutex> lock(_lock);
      if (state_ == AMPS::ConnectionStateListener::Disconnected)
      {
        _state = Disconnected;
        close();
      }
      _lock.signalAll();
    }

    void timeout(unsigned timeout_)
    {
      _timeout = timeout_;
    }
    void conflate(void)
    {
      if (_state == Subscribe)
      {
        _state = Conflate;
      }
    }
    void maxDepth(unsigned maxDepth_)
    {
      if (maxDepth_)
      {
        _maxDepth = maxDepth_;
      }
      else
      {
        _maxDepth = (unsigned)~0;
      }
    }
    unsigned getMaxDepth(void) const
    {
      return _maxDepth;
    }
    unsigned getDepth(void) const
    {
      return (unsigned)(_q.size());
    }

    bool next(Message& current_)
    {
      Lock<Mutex> lock(_lock);
      if (!_previousTopic.empty() && !_previousBookmark.empty())
      {
        try
        {
          if (_client.isValid())
          {
            _client.ackDeferredAutoAck(_previousTopic, _previousBookmark);
          }
        }
#ifdef _WIN32
        catch (AMPSException&)
#else
        catch (AMPSException& e)
#endif
        {
          current_.invalidate();
          _previousTopic.clear();
          _previousBookmark.clear();
          return false;
        }
        _previousTopic.clear();
        _previousBookmark.clear();
      }
      // Don't wait to wait more than 1s at a time
      long minWaitTime = (_timeout && _timeout < 1000) ? _timeout : 1000;
      Timer timer((double)_timeout);
      timer.start();
      while (_q.empty() && _state & Running)
      {
        // Using timeout so python can interrupt
        _lock.wait(minWaitTime);
        {
          Unlock<Mutex> unlck(_lock);
          amps_invoke_waiting_function();
        }
        if (_timeout)
        {
          // In case we woke up early, see how much longer to wait
          if (timer.checkAndGetRemaining(&minWaitTime))
          {
            // No time left
            break;
          }
          // Adjust next wait time
          minWaitTime = (minWaitTime < 1000) ? minWaitTime : 1000;
        }
      }
      if (current_.isValid() && _cache.size() < _cacheMax)
      {
        current_.reset();
        _cache.push_back(current_);
      }
      if (!_q.empty())
      {
        current_ = _q.front();
        if (_q.size() == _maxDepth)
        {
          _lock.signalAll();
        }
        _q.pop_front();
        if (_state == Conflate)
        {
          std::string sowKey = current_.getSowKey();
          if (sowKey.length())
          {
            _sowKeyMap.erase(sowKey);
          }
        }
        else if (_state == AcksOnly)
        {
          _requestedAcks &= ~(current_.getAckTypeEnum());
        }
        if ((_state == AcksOnly && _requestedAcks == 0) ||
            (_state == SOWOnly && current_.getCommand() == "group_end"))
        {
          _state = Closed;
        }
        else if (current_.isValid()
                 && current_.getCommandEnum() == Message::Command::Publish
                 && _client.isValid() && _client.getAutoAck()
                 && !current_.getLeasePeriod().empty()
                 && !current_.getBookmark().empty())
        {
          _previousTopic = current_.getTopic().deepCopy();
          _previousBookmark = current_.getBookmark().deepCopy();
        }
        return true;
      }
      if (_state == Disconnected)
      {
        throw DisconnectedException("Connection closed.");
      }
      current_.invalidate();
      if (_state == Closed)
      {
        return false;
      }
      return _timeout != 0;
    }
    void close(void)
    {
      if (_client.isValid())
      {
        if (_state == SOWOnly || _state == Subscribe) //not delete
        {
          if (!_commandId.empty())
          {
            _client.unsubscribe(_commandId);
          }
          if (!_subId.empty())
          {
            _client.unsubscribe(_subId);
          }
          if (!_queryId.empty())
          {
            _client.unsubscribe(_queryId);
          }
        }
        else
        {
          if (!_commandId.empty())
          {
            _client.removeMessageHandler(_commandId);
          }
          if (!_subId.empty())
          {
            _client.removeMessageHandler(_subId);
          }
          if (!_queryId.empty())
          {
            _client.removeMessageHandler(_queryId);
          }
        }
      }
      if (_state == SOWOnly || _state == Subscribe || _state == Unset)
      {
        _state = Closed;
      }
    }
    static void _messageHandler(const Message& message_, MessageStreamImpl* this_)
    {
      Lock<Mutex> lock(this_->_lock);
      if (this_->_state != Conflate)
      {
        AMPS_TESTING_SLOW_MESSAGE_STREAM
        if (this_->_q.size() >= this_->_maxDepth)
        {
          // We throw here so that heartbeats can be sent. The exception
          // will be handled internally only, and the same Message will
          // come back to try again. Make sure to signal.
          this_->_lock.signalAll();
          throw MessageStreamFullException("Stream is currently full.");
        }
        if (!this_->_cache.empty())
        {
          this_->_cache.front().deepCopy(message_);
          this_->_q.push_back(this_->_cache.front());
          this_->_cache.pop_front();
        }
        else
        {
#ifdef AMPS_USE_EMPLACE
          this_->_q.emplace_back(message_.deepCopy());
#else
          this_->_q.push_back(message_.deepCopy());
#endif
        }
        if (message_.getCommandEnum() == Message::Command::Publish &&
            this_->_client.isValid() && this_->_client.getAutoAck() &&
            !message_.getLeasePeriod().empty() &&
            !message_.getBookmark().empty())
        {
          message_.setIgnoreAutoAck();
        }
      }
      else
      {
        std::string sowKey = message_.getSowKey();
        if (sowKey.length())
        {
          SOWKeyMap::iterator it = this_->_sowKeyMap.find(sowKey);
          if (it != this_->_sowKeyMap.end())
          {
            it->second->deepCopy(message_);
          }
          else
          {
            if (this_->_q.size() >= this_->_maxDepth)
            {
              // We throw here so that heartbeats can be sent. The
              // exception will be handled internally only, and the
              // same Message will come back to try again. Make sure
              // to signal.
              this_->_lock.signalAll();
              throw MessageStreamFullException("Stream is currently full.");
            }
            if (!this_->_cache.empty())
            {
              this_->_cache.front().deepCopy(message_);
              this_->_q.push_back(this_->_cache.front());
              this_->_cache.pop_front();
            }
            else
            {
#ifdef AMPS_USE_EMPLACE
              this_->_q.emplace_back(message_.deepCopy());
#else
              this_->_q.push_back(message_.deepCopy());
#endif
            }
            this_->_sowKeyMap[sowKey] = &(this_->_q.back());
          }
        }
        else
        {
          if (this_->_q.size() >= this_->_maxDepth)
          {
            // We throw here so that heartbeats can be sent. The exception
            // will be handled internally only, and the same Message will
            // come back to try again. Make sure to signal.
            this_->_lock.signalAll();
            throw MessageStreamFullException("Stream is currently full.");
          }
          if (!this_->_cache.empty())
          {
            this_->_cache.front().deepCopy(message_);
            this_->_q.push_back(this_->_cache.front());
            this_->_cache.pop_front();
          }
          else
          {
#ifdef   AMPS_USE_EMPLACE
            this_->_q.emplace_back(message_.deepCopy());
#else
            this_->_q.push_back(message_.deepCopy());
#endif
          }
          if (message_.getCommandEnum() == Message::Command::Publish &&
              this_->_client.isValid() && this_->_client.getAutoAck() &&
              !message_.getLeasePeriod().empty() &&
              !message_.getBookmark().empty())
          {
            message_.setIgnoreAutoAck();
          }
        }
      }
      this_->_lock.signalAll();
    }
  };
  inline MessageStream::MessageStream(void)
  {
  }
  inline MessageStream::MessageStream(const Client& client_)
    : _body(new MessageStreamImpl(client_))
  {
  }
  inline void MessageStream::iterator::advance(void)
  {
    _pStream = _pStream->_body->next(_current) ? _pStream : NULL;
  }
  inline MessageStream::operator MessageHandler(void)
  {
    return MessageHandler((void(*)(const Message&, void*))MessageStreamImpl::_messageHandler, &_body.get());
  }
  inline MessageStream MessageStream::fromExistingHandler(const MessageHandler& handler_)
  {
    MessageStream result;
    if (handler_._func == (MessageHandler::FunctionType)MessageStreamImpl::_messageHandler)
    {
      result._body = (MessageStreamImpl*)(handler_._userData);
    }
    return result;
  }

  inline void MessageStream::setSOWOnly(const std::string& commandId_,
                                        const std::string& queryId_)
  {
    _body->setSOWOnly(commandId_, queryId_);
  }
  inline void MessageStream::setSubscription(const std::string& subId_,
                                             const std::string& commandId_,
                                             const std::string& queryId_)
  {
    _body->setSubscription(subId_, commandId_, queryId_);
  }
  inline void MessageStream::setStatsOnly(const std::string& commandId_,
                                          const std::string& queryId_)
  {
    _body->setStatsOnly(commandId_, queryId_);
  }
  inline void MessageStream::setAcksOnly(const std::string& commandId_,
                                         unsigned acks_)
  {
    _body->setAcksOnly(commandId_, acks_);
  }
  inline MessageStream MessageStream::timeout(unsigned timeout_)
  {
    _body->timeout(timeout_);
    return *this;
  }
  inline MessageStream MessageStream::conflate(void)
  {
    _body->conflate();
    return *this;
  }
  inline MessageStream MessageStream::maxDepth(unsigned maxDepth_)
  {
    _body->maxDepth(maxDepth_);
    return *this;
  }
  inline unsigned MessageStream::getMaxDepth(void) const
  {
    return _body->getMaxDepth();
  }
  inline unsigned MessageStream::getDepth(void) const
  {
    return _body->getDepth();
  }

  inline MessageStream ClientImpl::getEmptyMessageStream(void)
  {
    return *(_pEmptyMessageStream.get());
  }

  inline MessageStream Client::execute(Command& command_)
  {
    // If the command is sow and has a sub_id, OR
    // if the command has a replace option, return the existing
    // messagestream, don't create a new one.
    ClientImpl& body = _body.get();
    Message& message = command_.getMessage();
    Field subId = message.getSubscriptionId();
    unsigned ackTypes = message.getAckTypeEnum();
    bool useExistingHandler = !subId.empty() && ((!message.getOptions().empty() && message.getOptions().contains("replace", 7)) || message.getCommandEnum() == Message::Command::SOW);
    if (useExistingHandler)
    {
      // Try to find the existing message handler.
      if (!subId.empty())
      {
        MessageHandler existingHandler;
        if (body._routes.getRoute(subId, existingHandler))
        {
          // we found an existing handler. It might not be a message stream, but that's okay.
          body.executeAsync(command_, existingHandler, false);
          return MessageStream::fromExistingHandler(existingHandler);
        }
      }
      // fall through; we'll a new handler altogether.
    }
    // Make sure something will be returned to the stream or use the empty one
    // Check that: it's a command that doesn't normally return data, and there
    // are no acks requested for the cmd id
    Message::Command::Type command = message.getCommandEnum();
    if ((command & Message::Command::NoDataCommands)
        && (ackTypes == Message::AckType::Persisted
            || ackTypes == Message::AckType::None))
    {
      executeAsync(command_, MessageHandler());
      if (!body._pEmptyMessageStream)
      {
        body._pEmptyMessageStream.reset(new MessageStream((ClientImpl*)0));
        body._pEmptyMessageStream.get()->_body->close();
      }
      return body.getEmptyMessageStream();
    }
    MessageStream stream(*this);
    if (body.getDefaultMaxDepth())
    {
      stream.maxDepth(body.getDefaultMaxDepth());
    }
    MessageHandler handler = stream.operator MessageHandler();
    std::string commandID = body.executeAsync(command_, handler, false);
    if (command_.hasStatsAck())
    {
      stream.setStatsOnly(commandID, command_.getMessage().getQueryId());
    }
    else if (command_.isSow())
    {
      if (command_.getAckTypeEnum() & Message::AckType::Completed)
      {
        stream.setAcksOnly(commandID,
                           ackTypes);
      }
      else
      {
        stream.setSOWOnly(commandID, command_.getMessage().getQueryId());
      }
    }
    else if (command_.isSubscribe())
    {
      stream.setSubscription(commandID,
                             command_.getMessage().getCommandId(),
                             command_.getMessage().getQueryId());
    }
    else
    {
      // Persisted acks for writes don't come back with command id
      if (command == Message::Command::Publish ||
          command == Message::Command::DeltaPublish ||
          command == Message::Command::SOWDelete)
      {
        stream.setAcksOnly(commandID,
                           ackTypes & (unsigned)~Message::AckType::Persisted);
      }
      else
      {
        stream.setAcksOnly(commandID, ackTypes);
      }
    }
    return stream;
  }

// This is here because it uses api from Client.
  inline void Message::ack(const char* options_) const
  {
    ClientImpl* pClient = _body.get().clientImpl();
    Message::Field bookmark = getBookmark();
    if (pClient && bookmark.len() &&
        !pClient->getAutoAck())
      //(!pClient->getAutoAck() || getIgnoreAutoAck()))
    {
      pClient->ack(getTopic(), bookmark, options_);
    }
  }
}// end namespace AMPS
#endif