ampsplusplus.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2020 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.h"
#include "ampsver.h"
#include <string>
#include <map>
#include <sstream>
#include <iostream>
#include <memory>
#include <stdexcept>
#include <limits.h>
#include <list>
#include <set>
#include <deque>
#include <vector>
#include <assert.h>
#ifndef _WIN32
#include <inttypes.h>
#endif
#if defined(sun)
#include <sys/atomic.h>
#endif
#include "MessageRouter.hpp"
#include "util.hpp"
#include "ampscrc.hpp"

#ifndef AMPS_TESTING_SLOW_MESSAGE_STREAM
#define AMPS_TESTING_SLOW_MESSAGE_STREAM
#endif





#define AMPS_INITIAL_LOG_SIZE  40960UL
#define AMPS_MEMORYBUFFER_DEFAULT_BUFFERS 10
#define AMPS_MEMORYBUFFER_DEFAULT_LENGTH 40960
#define AMPS_RING_POSITIONS 3
#define AMPS_RING_BYTES_SUBID 128
#define AMPS_RING_BYTES_BOOKMARK 64
#define AMPS_RING_ENTRIES 16384
#define AMPS_RING_ENTRY_SIZE  ( AMPS_RING_BYTES_SUBID + ( AMPS_RING_POSITIONS * AMPS_RING_BYTES_BOOKMARK ) )
#define AMPS_SUBSCRIPTION_MANAGER_DEFAULT_TIMEOUT 0

#define AMPS_BOOKMARK_RECENT "recent"

#define AMPS_BOOKMARK_EPOCH "0"

#define AMPS_BOOKMARK_NOW "0|1|"

/* @}  */

#define AMPS_INITIAL_MEMORY_BOOKMARK_SIZE   16384UL
#define AMPS_HACLIENT_TIMEOUT_DEFAULT       10000
#define AMPS_HACLIENT_RECONNECT_DEFAULT     200
#define AMPS_PERSISTED_BOOKMARK_MIN_VERSION 3080000
#define AMPS_MULTI_BOOKMARK_MIN_VERSION     4000000
#define AMPS_FLUSH_MIN_VERSION              4000000
#define AMPS_MIN_SOW_KEY_PUBLISH_VERSION    4030100
#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

#ifdef _WIN32
static __declspec ( thread ) AMPS::Message* publishStoreMessage = 0;
#else
static __thread AMPS::Message* publishStoreMessage = 0;
#endif

namespace AMPS
{

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

class PerThreadMessageTracker {
std::vector<AMPS::Message*> _messages;
public:
    PerThreadMessageTracker() {}
    ~PerThreadMessageTracker()
    {
        for (size_t i=0; i<_messages.size(); ++i)
        {
            delete _messages[i];
        }
    }
    void addMessage(AMPS::Message* message)
    {
        _messages.push_back(message);
    }
    static void addMessageToCleanupList(AMPS::Message* message)
    {
        static AMPS::Mutex _lock;
        AMPS::Lock<Mutex> l(_lock);
        _addMessageToCleanupList(message);
    }
    static void _addMessageToCleanupList(AMPS::Message* message)
    {
        static PerThreadMessageTracker tracker;
        tracker.addMessage(message);
    }
};

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_STREAM_FULL_2(me,x) \
    while(me->_connected)\
    {\
        try\
        {\
            x;\
            break;\
        }\
        catch(MessageStreamFullException&)\
        {\
            me->checkAndSendHeartbeat(false);\
        }\
    }
#else
#define AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me,x) \
    while(me->_connected)\
    {\
        try\
        {\
            x;\
            break;\
        }\
        catch(MessageStreamFullException& ex_)\
        {\
            me->checkAndSendHeartbeat(false);\
        }\
    }
#endif

#define AMPS_CALL_EXCEPTION_WRAPPER_2(me,x) \
  try\
  {\
      AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me,x) \
  }\
  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_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 _setIds(void)
    {
        Message::Command::Type command = _message.getCommandEnum();
        if (command != Message::Command::Publish && command != Message::Command::DeltaPublish)
        {
            _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::SOWDelete)
            {
                _flags |= NeedsSequenceNumber;
            }
            else
            {
                _message.setQueryID(_message.getCommandId());
                if (command == Message::Command::SOW)
                {
                    _flags |= SOW;
                    if (_batchSize == 0)
                    {
                        setBatchSize(AMPS_DEFAULT_BATCH_SIZE);
                    }
                }
                else if ((command == Message::Command::SOWAndSubscribe ||
                          command == Message::Command::SOWAndDeltaSubscribe)
                         && _batchSize == 0)
                {
                    setBatchSize(AMPS_DEFAULT_BATCH_SIZE);
                }
            }
            _flags |= ProcessedAck;
        }
        else
        {
            _flags |= NeedsSequenceNumber;
        }
    }
public:
    Command(const std::string& command_)
    {
        init(command_);
    }
    Command(Message::Command::Type command_)
    {
        init(command_);
    }

    Command& reset(const std::string& command_)
    {
        init(command_);
        return *this;
    }
    Command& reset(Message::Command::Type command_)
    {
        init(command_);
        return *this;
    }
    Command& setSowKey(const std::string& sowKey_) { _message.setSowKey(sowKey_); return *this; }
    Command& setSowKeys(const std::string& sowKeys_) { _message.setSowKeys(sowKeys_); return *this; }
    Command& setCommandId(const std::string& v_)     { _message.setCommandId(v_);     return *this; }
    Command& setTopic(const std::string& v_)         { _message.setTopic(v_);         return *this; }
    Command& setFilter(const std::string& v_)        { _message.setFilter(v_);        return *this; }
    Command& setOrderBy(const std::string& v_)       { _message.setOrderBy(v_);       return *this; }
    Command& setSubId(const std::string& v_)         { _message.setSubscriptionId(v_); return *this; }
    Command& setBookmark(const std::string& v_)      { _message.setBookmark(v_);      return *this; }
    Command& setCorrelationId(const std::string& v_) { _message.setCorrelationId(v_); return *this; }
    Command& setOptions(const std::string& v_)       { _message.setOptions(v_);       return *this; }
    Command& setData(const std::string& v_)          { _message.setData(v_);          return *this; }
    Command& setData(const char* v_, size_t length_) { _message.setData(v_, length_); return *this; }
    Command& setTimeout(unsigned v_)                 { _timeout = v_;                 return *this; }
    Command& setTopN(unsigned v_)                    { _message.setTopNRecordsReturned(v_); return *this; }
    Command& setBatchSize(unsigned v_)               { _message.setBatchSize(v_); _batchSize = v_;    return *this; }
    Command& setExpiration(unsigned v_)              { _message.setExpiration(v_);    return *this; }
    Command& addAckType(const std::string& v_)
    {
        _message.setAckType(_message.getAckType() + "," + v_);
        if (v_ == "processed") _flags |= ProcessedAck;
        else if (v_ == "stats") _flags |= StatsAck;
        return *this;
    }

    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 BookmarkStore;

typedef bool (*BookmarkStoreResizeHandler)(BookmarkStore store_,
                                           const Message::Field& subId_,
                                           size_t size_,
                                           void* userData_);

class BookmarkStoreImpl : public RefBody
{
public:
    BookmarkStoreImpl()
    : _resizeHandler(NULL)
    , _resizeHandlerData(NULL)
    , _maxSubIdLength(AMPS_MAX_SUBID_LEN)
    {;}

    virtual ~BookmarkStoreImpl() {;}

    virtual size_t log(Message& message_) = 0;

    virtual void discard(const Message::Field& subId_,
                         size_t bookmarkSeqNo_) = 0;

    virtual void discard(const Message& message_) = 0;

    virtual Message::Field getMostRecent(const Message::Field& subId_) = 0;

    virtual bool isDiscarded(Message& message_) = 0;

    virtual void purge() = 0;

    virtual void purge(const Message::Field& subId_) = 0;

    virtual size_t getOldestBookmarkSeq(const Message::Field& subId_) = 0;

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

    virtual void persisted(const Message::Field& subId_,
                           const Message::Field& bookmark_) = 0;

    virtual Message::Field persisted(const Message::Field& subId_, size_t bookmark_) = 0;

    virtual void noPersistedAcks(const Message::Field& subId_) = 0;

    virtual void setServerVersion(size_t version_) = 0;

    virtual void setServerVersion(const VersionInfo& version_) = 0;

    bool callResizeHandler(const Message::Field& subId_, size_t newSize_);

    inline void prune(std::string tmpFileName_ = std::string())
    {
        _prune(tmpFileName_);
    }

    virtual void _prune(std::string) { return; }

    size_t getMaxSubIdLength() const
    {
        return _maxSubIdLength;
    }

    void setMaxSubIdLength(size_t maxSubIdLength_)
    {
        _maxSubIdLength = maxSubIdLength_;
    }

private:
    BookmarkStoreResizeHandler  _resizeHandler;
    void*                       _resizeHandlerData;
    size_t                      _maxSubIdLength;
};

class BookmarkStore
{
    RefHandle<BookmarkStoreImpl> _body;
public:
    BookmarkStore() {;}

    BookmarkStore(BookmarkStoreImpl* impl_) : _body(impl_) {;}

    BookmarkStore(const BookmarkStore& rhs) : _body(rhs._body) {;}

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

    ~BookmarkStore() {;}

    void setImplementation(BookmarkStoreImpl* impl_)
    {
        _body = impl_;
    }

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

    size_t log(Message& message_)
    {
        if (_body.isValid())
            return _body.get().log(message_);
        return Message::BOOKMARK_NONE;
    }

    void discard(const Message::Field& subId_, size_t bookmarkSeqNo_)
    {
        if (_body.isValid())
            _body.get().discard(subId_, bookmarkSeqNo_);
    }

    void discard(const Message& message_)
    {
        if (_body.isValid())
            _body.get().discard(message_);
    }

    Message::Field getMostRecent(const Message::Field& subId_)
    {
        if (_body.isValid())
            return _body.get().getMostRecent(subId_);
        return Message::Field(AMPS_BOOKMARK_EPOCH);
    }

    bool isDiscarded(Message& message_)
    {
        if (_body.isValid())
            return _body.get().isDiscarded(message_);
        return false;
    }

    void purge()
    {
        if (_body.isValid())
            _body.get().purge();
    }

    void purge(const Message::Field& subId_)
    {
        if (_body.isValid())
            _body.get().purge(subId_);
    }

    void setResizeHandler(BookmarkStoreResizeHandler handler_, void* userData_)
    {
        if (_body.isValid())
            _body.get().setResizeHandler(handler_, userData_);
    }

    size_t getOldestBookmarkSeq(const std::string& subId_)
    {
        if (_body.isValid())
            return _body.get().getOldestBookmarkSeq(Message::Field(subId_.c_str(),
                                                                   subId_.length()));
        return AMPS_UNSET_INDEX;
    }

    size_t getOldestBookmarkSeq(const Message::Field& subId_)
    {
        if (_body.isValid())
            return _body.get().getOldestBookmarkSeq(subId_);
        return AMPS_UNSET_INDEX;
    }

    void persisted(const Message::Field& subId_, const Message::Field& bookmark_)
    {
        if (_body.isValid())
            _body.get().persisted(subId_, bookmark_);
    }

    void persisted(const Message::Field& subId_, size_t bookmark_)
    {
        if (_body.isValid())
            _body.get().persisted(subId_, bookmark_);
    }

    void noPersistedAcks(const Message::Field& subId_)
    {
        if (_body.isValid())
            _body.get().noPersistedAcks(subId_);
    }

    void setServerVersion(size_t version_)
    {
        if (_body.isValid())
            _body.get().setServerVersion(version_);
    }

    void setServerVersion(const VersionInfo& version_)
    {
        if (_body.isValid())
            _body.get().setServerVersion(version_);
    }

    void prune(std::string tmpFileName_ = "")
    {
        if (_body.isValid())
            _body.get().prune(tmpFileName_);
    }

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

    size_t getMaxSubIdLength() const
    {
        if (_body.isValid())
            return _body.get().getMaxSubIdLength();
        else
            return 0;
    }

    void setMaxSubIdLength(size_t maxSubIdLength_)
    {
        if (_body.isValid())
            _body.get().setMaxSubIdLength(maxSubIdLength_);
    }

};

inline bool BookmarkStoreImpl::callResizeHandler(const Message::Field& subId_,
                                                 size_t newSize_)
{
    if (_resizeHandler)
        return _resizeHandler(BookmarkStore(this), subId_, newSize_, _resizeHandlerData);
    return true;
}

inline bool ThrowawayBookmarkResizeHandler(BookmarkStore store_,
                                           const Message::Field& subId_,
                                           size_t newSize_, void* data_)
{
    size_t* maxSizep = (size_t*)data_;
    if (newSize_ > *maxSizep)
    {
        size_t discardSeq = store_.getOldestBookmarkSeq(subId_);
        store_.discard(subId_, discardSeq);
        store_.persisted(subId_, discardSeq);
        return false;
    }
    return true;
}

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() : _resizeHandler(NULL), _resizeHandlerData(NULL) {;}

    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_);

private:
    PublishStoreResizeHandler _resizeHandler;
    void*                     _resizeHandlerData;
};

class Store
{
    RefHandle<StoreImpl> _body;
public:
    Store() {;}
    Store(StoreImpl* body_) : _body(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()
    {
        return _body.get().getResizeHandler();
    }

    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 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;
};


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
{
    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;
    MessageHandler _lastChanceMessageHandler, _duplicateMessageHandler;
    Message _message, _readMessage, _publishMessage, _deltaMessage, _beatMessage;
    MessageRouter _routes;
    MessageRouter::RouteCache _routeCache;
    mutable Mutex _lock;
    std::string _name, _lastUri, _logonCorrelationData;
    BookmarkStore _bookmarkStore;
    Store _publishStore;
    bool _isRetryOnDisconnect;
    amps_unique_ptr<FailedWriteHandler> _failedWriteHandler;
    volatile amps_uint64_t _lastSentHaSequenceNumber;
    ATOMIC_TYPE_8 _badTimeToHAPublish;
    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->_badTimeToHAPublish ||
                 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;
        VersionInfo serverVersion;
        volatile bool responded, abandoned;
        unsigned connectionVersion;
        AckResponseImpl() :
            RefBody(),
            sequenceNo((amps_uint64_t)0),
            serverVersion(),
            responded(false),
            abandoned(false),
            connectionVersion(0)
        {
        }
    };

    class AckResponse
    {
        RefHandle<AckResponseImpl> _body;
    public:
        AckResponse() : _body(NULL) {;}
        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 char* data_, size_t len_)
        {
            if (data_) _body.get().bookmark.assign(data_, len_);
            else _body.get().bookmark.clear();
        }
        amps_uint64_t sequenceNo() const
        {
            return _body.get().sequenceNo;
        }
        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(bool responded_)
        {
            _body.get().responded = responded_;
        }
        bool abandoned()
        {
            return _body.get().abandoned;
        }
        void setAbandoned(bool abandoned_)
        {
            if (_body.isValid())
                _body.get().abandoned = abandoned_;
        }

        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 _acks;
    DefaultExceptionListener _defaultExceptionListener;
protected:

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

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

    void broadcastConnectionStateChanged(ConnectionStateListener::State newState_)
    {
        // If we disconnectd 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)
    {
        // Lock is already acquired
        if (_subscriptionManager)
        {
            Message::Field subId;
            subId.assign(id.data(), id.length());
            // 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);
        // remove the handler before sending the message, in case send fails.
        _routes.removeRoute(_message.getSubscriptionId());
        _sendWithoutRetry(_message);
    }

    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();
        _acks[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())
        {
            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);
                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())
                    {
                        _publishStore.discardUpTo(ackSequence);
                        if (_lastSentHaSequenceNumber < _publishStore.getLastPersisted())
                        {
                            _lastSentHaSequenceNumber = _publishStore.getLastPersisted();
                        }
                    }
                    _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_disconnect_handler(_client,
                                           NULL,
                                           0L);
        AMPS_CALL_EXCEPTION_WRAPPER(ClientImpl::disconnect());
        _pEmptyMessageStream.reset(NULL);
        amps_client_destroy(_client);
        processDeferredExecutions();
        _client = NULL;
    }

public:

    ClientImpl(const std::string& clientName)
        : _client(NULL), _name(clientName)
        , _isRetryOnDisconnect(true)
        , _lastSentHaSequenceNumber((amps_uint64_t)0), _badTimeToHAPublish(0)
        , _badTimeToHASubscribe(0), _serverVersion()
        , _queueAckTimeout(0)
        , _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)
        {
            _globalCommandTypeHandlers.push_back(MessageHandler());
        }
    }


    virtual ~ClientImpl()
    {
        _cleanup();
    }


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

    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
    {
        Lock<Mutex> l(_lock);
        return _lastUri;
    }

    virtual void connect(const std::string& uri)
    {
        Lock<Mutex> l(_lock);
        _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)
        {
            amps_client_set_idle_time(_client,_queueAckTimeout);
        }
        _connected = true;
        broadcastConnectionStateChanged(ConnectionStateListener::Connected);
    }

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

    virtual void disconnect()
    {
        {
            Lock<Mutex> l(_lock);
            _message.reset();
            _message.setCommandEnum(Message::Command::Unsubscribe);
            _message.newCommandId();
            _message.setSubscriptionId("all");
            AMPS_CALL_EXCEPTION_WRAPPER(_sendWithoutRetry(_message));
        }
        AMPS_CALL_EXCEPTION_WRAPPER(flushAcks());
        setDisconnected();
        amps_client_disconnect(_client);
        Lock<Mutex> l(_lock);
        broadcastConnectionStateChanged(ConnectionStateListener::Shutdown);
    }

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

            for(std::list<std::string>::iterator j = worklist.begin(); j != worklist.end(); j++)
            {
                _acks.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);
        _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 != (amps_uint64_t)0 && _badTimeToHAPublish > 0)
            {
                // If retrySend is disabled, do not wait for the reconnect
                // to finish, just throw.
                if(!_isRetryOnDisconnect)
                {
                  AMPSException::throwFor(_client,AMPS_E_RETRY);
                }
                Unlock<Mutex> l(_lock);
#ifdef _WIN32
                Sleep(0);
#elif defined(sun)
                sched_yield();
#else
                pthread_yield();
#endif
            }
            else
            {
                if ((haSeq && haSeq <= _lastSentHaSequenceNumber) ||
                    (isHASubscribe_ && _badTimeToHASubscribe != 0))
                {
                    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
                    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 subscrbie 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_, bool isSubscribe_)
    {
        Lock<Mutex> lock(_lock);
        _routes.addRoute(commandId_, messageHandler_, requestedAcks_,
                         0, isSubscribe_);
    }

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

    std::string send(MessageHandler messageHandler_, Message& message_, int timeout_ = 0)
    {
        Field id = message_.getCommandId();
        bool isSubscribe = false;
        unsigned requestedAcks = message_.getAckTypeEnum();
        unsigned systemAddedAcks = Message::AckType::None;
        switch(message_.getCommandEnum())
        {
        case Message::Command::Subscribe:
        case Message::Command::DeltaSubscribe:
            if (!message_.getBookmark().empty() && _bookmarkStore.isValid())
            {
                systemAddedAcks |= Message::AckType::Persisted;
            }
            // fall through
        case Message::Command::SOWAndSubscribe:
        case Message::Command::SOWAndDeltaSubscribe:
            if (id.empty())
            {
                message_.newCommandId();
                id = message_.getCommandId();
            }
            if (message_.getSubscriptionId().empty())
            {
                message_.setSubscriptionId(id);
            }
            isSubscribe = true;
            // fall through
        case Message::Command::SOW:
            if(id.empty())
            {
                message_.newCommandId();
                id = message_.getCommandId();
            }
            if (message_.getQueryID().empty())
            {
                message_.setQueryID(id);
            }
            systemAddedAcks |= Message::AckType::Processed;
            // for SOW only, we get a completed ack so we know when to remove the handler.
            if (!isSubscribe) systemAddedAcks |= Message::AckType::Completed;
            message_.setAckTypeEnum(requestedAcks | systemAddedAcks);
            {
                Lock<Mutex> l(_lock);
                _routes.addRoute(message_.getQueryID(), messageHandler_, requestedAcks,
                                 systemAddedAcks, isSubscribe);
                if (!message_.getSubscriptionId().empty() &&
                    message_.getQueryID() != message_.getSubscriptionId() &&
                    messageHandler_.isValid() &&
                    !_routes.hasRoute(message_.getSubscriptionId()))
                {
                    _routes.addRoute(message_.getSubscriptionId(), messageHandler_,
                                     requestedAcks, systemAddedAcks, true);
                }
                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::DeltaPublish:
        case Message::Command::Publish:
        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, false);
                    }
                }
                _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(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()
    {
        Lock<Mutex> l(_lock);
        return _subscriptionManager.get();
    }

    DisconnectHandler getDisconnectHandler()
    {
        Lock<Mutex> l(_lock);
        return _disconnectHandler;
    }

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

    FailedWriteHandler* getFailedWriteHandler()
    {
        Lock<Mutex> l(_lock);
        return _failedWriteHandler.get();
    }

    Store getPublishStore()
    {
        Lock<Mutex> l(_lock);
        return _publishStore;
    }

    BookmarkStore getBookmarkStore()
    {
        Lock<Mutex> l(_lock);
        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
        {
            if (!publishStoreMessage)
            {
                publishStoreMessage = new Message();
                PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
            }
            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
        {
            if (!publishStoreMessage)
            {
                publishStoreMessage = new Message();
                PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
            }
            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_);
        }
    }

    void publishFlush(long timeout_)
    {
        static const char* processed = "processed";
        static const size_t processedLen = strlen(processed);
        if (!_publishStore.isValid())
        {
            if (_serverVersion.getOldStyleVersion() >= AMPS_FLUSH_MIN_VERSION)
            {
                Lock<Mutex> l(_lock);
                _message.reset();
                _message.newCommandId();
                _message.assignAckType(processed, processedLen);
                static const char* flush = "flush";
                static const size_t flushLen = strlen(flush);
                _message.assignCommand(flush, flushLen);
                try
                {
                    syncAckProcessing(timeout_, _message);
                }
                catch(const AMPSException& ex)
                {
                    AMPS_UNHANDLED_EXCEPTION(ex);
                    throw;
                }
            }
            else
            {
                if (timeout_ > 0) { AMPS_USLEEP(timeout_ * 1000); }
                else { AMPS_USLEEP(1000 * 1000); }
                return;
            }
        }
        else
        {
            try
            {
                _publishStore.flush(timeout_);
            }
            catch (const AMPSException& ex)
            {
                AMPS_UNHANDLED_EXCEPTION(ex);
                throw;
            }
        }
    }

    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
        {
            if (!publishStoreMessage)
            {
                publishStoreMessage = new Message();
                PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
            }
            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
        {
            if (!publishStoreMessage)
            {
                publishStoreMessage = new Message();
                PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
            }
            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);
        AtomicFlagFlip pubFlip(&_badTimeToHAPublish);
        _message.reset();
        _message.setCommandEnum(Message::Command::Logon);
        _message.newCommandId();
        std::string newCommandId = _message.getCommandId();
        _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
        {
            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();
        }
        catch(const AMPSException& ex)
        {
            AMPS_UNHANDLED_EXCEPTION(ex);
            throw;
        }
        catch(...)
        {
            throw;
        }

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

    std::string subscribe(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, true);
        }
        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(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, true);
        }
        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);
        }
    }

    std::string sow(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, false);

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

        return commandId;
    }

    std::string sow(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(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;
        Lock<Mutex> l(_lock);
        _message.reset();
        _message.setCommandEnum(Message::Command::SOWAndSubscribe);
        _message.newCommandId();
        Field cid = _message.getCommandId();
        _message.setQueryID(cid);
        _message.setSubscriptionId(cid);
        std::string subId = cid;
        _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_);

        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(cid))
        {
            _routes.addRoute(cid, messageHandler_,
                             Message::AckType::None, Message::AckType::Processed, true);
        }
        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(cid);
            }
            _routes.removeRoute(subId);
            throw;
        }
        return subId;
    }

    std::string sowAndSubscribe(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(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;
        }
        if (!_routes.hasRoute(message.getSubscriptionId()))
        {
            _routes.addRoute(message.getQueryID(), messageHandler_,
                             Message::AckType::None, Message::AckType::Processed, true);
        }
        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(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(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;
            if (!publishStoreMessage)
            {
                publishStoreMessage = new Message();
                PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
            }
            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,
                                     false);
                    syncAckProcessing(timeout_, *publishStoreMessage,
                                      haSequenceNumber);
                }
                catch (const DisconnectedException&)
                {
                    // 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,
                             false);
            try
            {
                syncAckProcessing(timeout_, _message);
            }
            catch (...)
            {
                AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
                throw;
            }
            return (std::string)commandId_;
        }
    }

    std::string sowDeleteByData(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;
            if (!publishStoreMessage)
            {
                publishStoreMessage = new Message();
                PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
            }
            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,
                                     false);
                    syncAckProcessing(timeout_, *publishStoreMessage,
                                      haSequenceNumber);
                }
                catch (const DisconnectedException&)
                {
                    // 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,
                             false);
            try
            {
                syncAckProcessing(timeout_, _message);
            }
            catch (...)
            {
                AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
                throw;
            }
            return (std::string)commandId_;
        }
    }

    std::string sowDeleteByKeys(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;
            if (!publishStoreMessage)
            {
                publishStoreMessage = new Message();
                PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
            }
            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,
                                     false);
                    syncAckProcessing(timeout_, *publishStoreMessage,
                                      haSequenceNumber);
                }
                catch (const DisconnectedException&)
                {
                    // 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,
                             false);
            try
            {
                syncAckProcessing(timeout_, _message);
            }
            catch (...)
            {
                AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
                throw;
            }
            return (std::string)commandId_;
        }
    }

    void startTimer(void)
    {
        Lock<Mutex> l(_lock);
        _message.reset();
        _message.setCommandEnum(Message::Command::StartTimer);

        _send(_message);
    }

    std::string stopTimer(MessageHandler messageHandler_)
    {
        return executeAsync(Command("stop_timer").addAckType("completed"), messageHandler_);
    }

    amps_handle getHandle(void)
    {
        return _client;
    }

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

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

    void setHeartbeat(unsigned heartbeatInterval_, unsigned readTimeout_)
    {
        if (readTimeout_ && 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;
            Message startMessage = Message()
              .setCommandEnum(Message::Command::Heartbeat)
              .setOptions(options.str());

            _heartbeatTimer.setTimeout(_heartbeatInterval * 1000.0);
            _heartbeatTimer.start();
            try
            {
                _sendWithoutRetry(startMessage);
                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_);
            }
        }
        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);
        }
    }

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

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

    void _registerHandler(Command& command_, Message::Field& cid_,
                          MessageHandler& handler_, unsigned requestedAcks_,
                          unsigned systemAddedAcks_, bool isSubscribe_)
    {
        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();
        if (qid.len() > 0)
        {
            if (!_routes.hasRoute(qid))
            {
                _routes.addRoute(qid, handler_, requestedAcks_,
                                 systemAddedAcks_, isSubscribe_);
            }
        }
        if (subid.len() > 0)
        {
            if (!_routes.hasRoute(subid))
            {
                _routes.addRoute(subid, handler_, requestedAcks_,
                                 systemAddedAcks_, isSubscribe_);
            }
        }
        if (cid_.len() > 0)
        {
            if (!_routes.hasRoute(cid_))
            {
                _routes.addRoute(cid_, handler_, requestedAcks_,
                                 systemAddedAcks_, isSubscribe_);
            }
        }
        else if (commandType == Message::Command::Publish ||
                commandType == Message::Command::DeltaPublish)
        {
            cid_ = command_.getMessage().newCommandId().getCommandId();
            _routes.addRoute(cid_, handler_, requestedAcks_,
                             systemAddedAcks_, isSubscribe_);
            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::SOW ||
                commandType == Message::Command::StopTimer)
            systemAddedAcks |= Message::AckType::Completed;
        Message::Field cid = message.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);
                        _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, isSubscribe);
        }
        bool useSyncSend = cid.len() > 0 && command_.hasProcessedAck();
        if (_publishStore.isValid() && command_.needsSequenceNumber())
        {
            amps_uint64_t haSequenceNumber = (amps_uint64_t)0;
            message.setAckTypeEnum(requestedAcks|systemAddedAcks);
            {
                Unlock<Mutex> u(_lock);
                haSequenceNumber = _publishStore.store(message);
            }
            message.setSequence(haSequenceNumber);
            if (useSyncSend)
            {
                try
                {
                    syncAckProcessing((long)command_.getTimeout(), message,
                                      haSequenceNumber);
                }
                catch (const DisconnectedException&)
                {
                    // Pass - message will get replayed when reconnected
                }
                catch (...)
                {
                    AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
                    throw;
                }
            }
            else _send(message, haSequenceNumber);
        }
        else
        {
            if(command_.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, isSubscribe);
                }
                message.setAckTypeEnum(requestedAcks|systemAddedAcks);
                if (useSyncSend)
                {
                    try
                    {
                        syncAckProcessing((long)command_.getTimeout(), message,
                                          isHASubscribe_);
                    }
                    catch (const DisconnectedException&)
                    {
                        if (!isHASubscribe_)
                        {
                            message.setAckTypeEnum(requestedAcks);
                            throw;
                        }
                    }
                    catch (const TimedOutException&)
                    {
                        AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(cid));
                        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(subId);
                        }
                        if (message.getQueryID().len() > 0)
                            _routes.removeRoute(message.getQueryID());
                        _routes.removeRoute(cid);
                        _routes.removeRoute(subId);
                        throw;
                    }
                }
                else _send(message);
                if (subId.len() > 0)
                {
                    message.setAckTypeEnum(requestedAcks);
                    return std::string(subId.data(), subId.len());
                }
            }
            else
            {
                message.setAckTypeEnum(requestedAcks|systemAddedAcks);
                if (useSyncSend)
                {
                    try
                    {
                        syncAckProcessing((long)(command_.getTimeout()), message);
                    }
                    catch (const DisconnectedException&)
                    {
                        message.setAckTypeEnum(requestedAcks);
                        throw;
                    }
                    catch (...)
                    {
                        AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
                        message.setAckTypeEnum(requestedAcks);
                        throw;
                    }
                }
                else _send(message);
            }
        }
        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)
      {
        if (!publishStoreMessage)
        {
            publishStoreMessage = new Message();
            PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
        }
        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)
      {
        if (!publishStoreMessage)
        {
            publishStoreMessage = new Message();
            PerThreadMessageTracker::addMessageToCleanupList(publishStoreMessage);
        }
        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
        it = _topicHashMap.insert(TopicHashMap::value_type(hash,QueueBookmarks(topic_))).first;
      }
      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(1)
      {
        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) publishFlush(0);
    }
    // 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(_lock);
        _deferredExecutionList.push_back(
            DeferredExecutionRequest(func_,userData_));
    }

    inline void processDeferredExecutions(void)
    {
        if(_deferredExecutionList.size())
        {
            Lock<Mutex> lock(_lock);
            DeferredExecutionList::iterator it = _deferredExecutionList.begin();
            DeferredExecutionList::iterator end = _deferredExecutionList.end();
            for(; it != end; ++it)
            {
                it->_func(it->_userData);
            }
            _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_result result = amps_client_set_transport_filter_function(_client, filter_, userData_);
        if (result != AMPS_E_OK)
        {
            AMPSException::throwFor(_client, result);
        }
    }
}; // 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)
      {
        return _pStream == rhs._pStream;
      }
      bool operator!=(const iterator& rhs)
      {
        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_);
  inline void setSubscription(const std::string& commandId_);
  inline void setStatsOnly(const std::string& commandId_);
  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
{
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();
    }

    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_)
    {
        _body.get().addMessageHandler(commandId_, messageHandler_,
                                      requestedAcks_, isSubscribe_);
    }

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

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

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

    DisconnectHandler getDisconnectHandler(void)
    {
        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)
    {
        _body.get().publishFlush(timeout_);
    }


    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(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(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(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(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(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(sow(result.operator MessageHandler(), topic_, filter_, orderBy_, bookmark_, batchSize_, topN_, options_, timeout_));
        return result;
    }
    std::string sow(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(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(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(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(sowAndDeltaSubscribe(result.operator MessageHandler(), topic_, filter_, orderBy_, batchSize_, topN_, options_, timeout_));
        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(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(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(MessageHandler messageHandler)
    {
        return _body.get().stopTimer(messageHandler);
    }

    std::string sowDeleteByKeys(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(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 ExceptionListener& listener_)
    {
        _body.get().setExceptionListener(listener_);
    }

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

    void setHeartbeat(unsigned heartbeatTime_, unsigned readTimeout_)
    {
        _body.get().setHeartbeat(heartbeatTime_, readTimeout_);
    }

    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* BOOKMARK_EPOCH() { return AMPS_BOOKMARK_EPOCH; }
    static const char* BOOKMARK_RECENT() { return AMPS_BOOKMARK_RECENT; }
    static const char* BOOKMARK_MOST_RECENT() { return AMPS_BOOKMARK_RECENT; }
    static const char* NOW() { return AMPS_BOOKMARK_NOW; }
    static const char* EPOCH() { return AMPS_BOOKMARK_EPOCH; }
    static const char* MOST_RECENT() { return AMPS_BOOKMARK_RECENT; }


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

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

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

    std::string executeAsyncNoResubscribe(Command& command_,
                                          MessageHandler handler_)
    {
        std::string id;
        try
        {
            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_)
    {
      _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 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)
            {
                amps_uint64_t sequence =
                                amps_message_get_field_uint64(messageHandle,
                                                              AMPS_Sequence);
                if (_publishStore.isValid())
                {
                    FailedWriteStoreReplayer replayer(this, data, len);
                    _publishStore.replaySingle(replayer, sequence);
                }
                else // Call the handler with what little we have
                {
                    static Message emptyMessage;
                    _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;
                _publishStore.discardUpTo(seq);
            }
        }

        if (!deliveries && _bookmarkStore.isValid())
        {
            amps_message_get_field_value(messageHandle, AMPS_SubscriptionId,
                &data, &len);
            if (len > 0)
            {
                const char* bookmarkData = NULL;
                size_t bookmarkLen = 0;
                amps_message_get_field_value(messageHandle, AMPS_Bookmark,
                    &bookmarkData, &bookmarkLen);
                if (bookmarkLen > 0)
                {
                    ++deliveries;
                    Message::Field subId(data, len);
                    _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)
    {
        AckMap::iterator i = _acks.find(std::string(data, len));
        if (i != _acks.end())
        {
            ++deliveries;
            ack = i->second;
            _acks.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_Sequence, &data,
            &len);
        ack.setSequenceNo(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);
        ack.setResponded(true);
        _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_;
    if(!messageHandle_)
    {
      if(me->_queueAckTimeout) me->checkQueueAcks();
      return AMPS_E_OK;
    }
    me->processDeferredExecutions();

    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())
                        {
                            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)
        {
            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
                {
                    me->checkAndSendHeartbeat(false);
                }
            }
        }
        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
        {
            AtomicFlagFlip pubFlip(&me->_badTimeToHAPublish);
            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);
        }

        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;
    }
};

class MessageStreamImpl : public AMPS::RefBody, AMPS::ConnectionStateListener
{
    Mutex                _lock;
    std::deque<Message>  _q;
    std::string          _commandId;
    Client               _client;
    unsigned             _timeout;
    unsigned             _maxDepth;
    unsigned             _requestedAcks;
    Message::Field       _previousTopic;
    Message::Field       _previousBookmark;
    volatile enum { Unset=0x0, Running=0x10, Subscribe=0x11, SOWOnly=0x12, AcksOnly=0x13, Conflate=0x14, Closed=0x1, Disconnected=0x2 } _state;
    typedef std::map<std::string, Message*> SOWKeyMap;
    SOWKeyMap _sowKeyMap;
    public:
    MessageStreamImpl(const Client& client_)
        : _client(client_),
          _timeout(0),
          _maxDepth((unsigned)~0),
          _requestedAcks(0),
          _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 (...) {}
        }
        if (_client.isValid())
        {
            _client.removeConnectionStateListener(this);
            _client.deferredExecution(MessageStreamImpl::destroyer, this);
            _client = Client((ClientImpl*)NULL);
        }
        else
        {
            delete this;
        }
    }

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

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

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

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

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

    void connectionStateChanged(ConnectionStateListener::State state_)
    {
        Lock<Mutex> lock(_lock);
        if(state_ == AMPS::ConnectionStateListener::Disconnected)
        {
            _state = Disconnected;
        }
        _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();
        }
        unsigned minWaitTime = _timeout ? _timeout : 1000;
        size_t waited = 0;
        while(_q.empty() && _state & Running)
        {
            // Using timeout so python can interrupt
            _lock.wait((long)minWaitTime);
            amps_invoke_waiting_function();
            if (_timeout)
            {
                waited += minWaitTime;
                if(waited >= _timeout) break;
            }
        }
        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_.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((_state==SOWOnly || _state==Subscribe) && _client.isValid()) //not delete
        {
            _client.unsubscribe(_commandId);
        }
        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.");
            }
            this_->_q.push_back(message_.deepCopy());
            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) = message_.deepCopy();
                }
                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.");
                    }
                    this_->_q.push_back(message_.deepCopy());
                    this_->_sowKeyMap[sowKey] = &(this_->_q.back());
                }
            }
            else
            {
                while(this_->_q.size() >= this_->_maxDepth)
                {
                    this_->_lock.wait(1);
                }
                this_->_q.push_back(message_.deepCopy());
            }
        }
        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_)
{
    _body->setSOWOnly(commandId_);
}
inline void MessageStream::setSubscription(const std::string& commandId_)
{
    _body->setSubscription(commandId_);
}
inline void MessageStream::setStatsOnly(const std::string& commandId_)
{
    _body->setStatsOnly(commandId_);
}
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.
    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.get()._routes.getRoute(subId, existingHandler))
            {
                // we found an existing handler. It might not be a message stream, but that's okay.
                _body.get().executeAsync(command_, existingHandler, false);
                return MessageStream::fromExistingHandler(existingHandler);
            }
        }
        // fall through; we'll a new handler altogether.
    }
    Message::Command::Type command = command_.getMessage().getCommandEnum();
    if ((command == Message::Command::Publish ||
         command == Message::Command::DeltaPublish) &&
        (ackTypes == Message::AckType::Persisted ||
        ackTypes == Message::AckType::None))
    {
        executeAsync(command_, MessageHandler());
        if (!_body.get()._pEmptyMessageStream)
        {
            _body.get()._pEmptyMessageStream.reset(new MessageStream((ClientImpl*)0));
            _body.get()._pEmptyMessageStream.get()->_body->close();
        }
        return _body.get().getEmptyMessageStream();
    }
    MessageStream stream(*this);
    if (_body.get().getDefaultMaxDepth())
        stream.maxDepth(_body.get().getDefaultMaxDepth());
    MessageHandler handler = stream.operator MessageHandler();
    std::string commandID = _body.get().executeAsync(command_, handler, false);
    if (command_.hasStatsAck())
    {
        stream.setStatsOnly(commandID);
    }
    else if (command_.isSow())
    {
        stream.setSOWOnly(commandID);
    }
    else if (command_.isSubscribe())
    {
        stream.setSubscription(commandID);
    }
    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