ampsplusplus.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2021 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 <memory>
#include <set>
#include <deque>
#include <vector>
#include <assert.h>
#ifndef _WIN32
#include <inttypes.h>
#endif
#if defined(sun)
#include <sys/atomic.h>
#endif
#include "BookmarkStore.hpp"
#include "MessageRouter.hpp"
#include "util.hpp"
#include "ampscrc.hpp"

#ifndef AMPS_TESTING_SLOW_MESSAGE_STREAM
#define AMPS_TESTING_SLOW_MESSAGE_STREAM
#endif





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

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

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

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

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

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

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


class Client;


class Command
{
 Message _message;
 unsigned _timeout;
 unsigned _batchSize;
 unsigned _flags;
 static const unsigned Subscribe = 1;
 static const unsigned SOW = 2;
 static const unsigned NeedsSequenceNumber = 4;
 static const unsigned ProcessedAck = 8;
 static const unsigned StatsAck = 16;
 void init(Message::Command::Type command_)
 {
 _timeout = 0;
 _batchSize = 0;
 _flags = 0;
 _message.reset();
 _message.setCommandEnum(command_);
 _setIds();
 }
 void init(const std::string& command_)
 {
 _timeout = 0;
 _batchSize = 0;
 _flags = 0;
 _message.reset();
 _message.setCommand(command_);
 _setIds();
 }
 void _setIds(void)
 {
 Message::Command::Type command = _message.getCommandEnum();
 if (!(command & Message::Command::NoDataCommands))
 {
 _message.newCommandId();
 if (command == Message::Command::Subscribe ||
 command == Message::Command::SOWAndSubscribe ||
 command == Message::Command::DeltaSubscribe ||
 command == Message::Command::SOWAndDeltaSubscribe)
 {
 _message.setSubscriptionId(_message.getCommandId());
 _flags |= Subscribe;
 }
 if (command == Message::Command::SOW
 || command == Message::Command::SOWAndSubscribe
 || command == Message::Command::SOWAndDeltaSubscribe)
 {
 _message.setQueryID(_message.getCommandId());
 if (_batchSize == 0)
 {
 setBatchSize(AMPS_DEFAULT_BATCH_SIZE);
 }
 if (command == Message::Command::SOW)
 {
 _flags |= SOW;
 }
 }
 _flags |= ProcessedAck;
 }
 else if (command == Message::Command::SOWDelete)
 {
 _message.newCommandId();
 _flags |= ProcessedAck;
 _flags |= NeedsSequenceNumber;
 }
 else if (command == Message::Command::Publish
 || command == Message::Command::DeltaPublish)
 {
 _flags |= NeedsSequenceNumber;
 }
 else if (command == Message::Command::StopTimer)
 {
 _message.newCommandId();
 }
 }
public:
 Command(const std::string& command_)
 {
 init(command_);
 }
 Command(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& setQueryId(const std::string& v_) { _message.setQueryId(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& setSequence(const std::string& v_) { _message.setSequence(v_); return *this; }
 Command& setSequence(const amps_uint64_t v_)
 {
 std::ostringstream os;
 os << v_;
 _message.setSequence(os.str());
 return *this;
 }
 amps_uint64_t getSequence() const { return amps_message_get_field_uint64(_message.getMessage(),AMPS_Sequence); }
 Command& setData(const std::string& 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;
 }
 Command& setAckType(const std::string& v_)
 {
 _message.setAckType(v_);
 if (v_.find("processed") != std::string::npos) _flags |= ProcessedAck;
 else _flags &= ~ProcessedAck;
 if (v_.find("stats") != std::string::npos) _flags |= StatsAck;
 else _flags &= ~StatsAck;
 return *this;
 }
 Command& setAckType(unsigned v_)
 {
 _message.setAckTypeEnum(v_);
 if (v_ & Message::AckType::Processed) _flags |= ProcessedAck;
 else _flags &= ~ProcessedAck;
 if (v_ & Message::AckType::Stats) _flags |= StatsAck;
 else _flags &= ~StatsAck;
 return *this;
 }
 std::string getAckType() const
 {
 return (std::string)(_message.getAckType());
 }
 unsigned getAckTypeEnum() const
 {
 return _message.getAckTypeEnum();
 }

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

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

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

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

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

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

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

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

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

class StoreReplayer
{
public:

 virtual void execute(Message& message_) = 0;

 virtual ~StoreReplayer() {;}
};

class Store;

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

class StoreImpl : public RefBody
{
public:
 StoreImpl() : _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(const Store& rhs) : _body(rhs._body) {;}
 Store& operator=(const Store& rhs)
 {
 _body = rhs._body;
 return *this;
 }

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

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

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

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

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

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

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

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

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

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

 PublishStoreResizeHandler getResizeHandler()
 {
 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 // -V553
{
 friend class Client;
protected:
 amps_handle _client;
 DisconnectHandler _disconnectHandler;
 enum GlobalCommandTypeHandlers : size_t
 {
 Publish = 0,
 SOW = 1,
 GroupBegin = 2,
 GroupEnd = 3,
 Heartbeat = 4,
 OOF = 5,
 Ack = 6,
 LastChance = 7,
 DuplicateMessage = 8,
 COUNT = 9
 };
 std::vector<MessageHandler> _globalCommandTypeHandlers;
 Message _message, _readMessage, _publishMessage, _deltaMessage, _beatMessage;
 MessageRouter _routes;
 MessageRouter::RouteCache _routeCache;
 mutable Mutex _lock;
 std::string _name, _nameHash, _lastUri, _logonCorrelationData;
 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) {;}
 AckResponse(const AckResponse& rhs) : _body(rhs._body) {;}
 static AckResponse create()
 {
 AckResponse r;
 r._body = new AckResponseImpl();
 return r;
 }

 const std::string& username()
 {
 return _body.get().username;
 }
 void setUsername(const char* data_, size_t len_)
 {
 if (data_) _body.get().username.assign(data_, len_);
 else _body.get().username.clear();
 }
 const std::string& password()
 {
 return _body.get().password;
 }
 void setPassword(const char* data_, size_t len_)
 {
 if (data_) _body.get().password.assign(data_, len_);
 else _body.get().password.clear();
 }
 const std::string& reason()
 {
 return _body.get().reason;
 }
 void setReason(const char* data_, size_t len_)
 {
 if (data_) _body.get().reason.assign(data_, len_);
 else _body.get().reason.clear();
 }
 const std::string& status()
 {
 return _body.get().status;
 }
 void setStatus(const char* data_, size_t len_)
 {
 if (data_) _body.get().status.assign(data_, len_);
 else _body.get().status.clear();
 }
 const std::string& bookmark()
 {
 return _body.get().bookmark;
 }
 void setBookmark(const 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 _ackMap;
 Mutex _ackMapLock;
 DefaultExceptionListener _defaultExceptionListener;
protected:

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

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

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

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

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

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

public:

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

 virtual ~ClientImpl()
 {
 _cleanup();
 }

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

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

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

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

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

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

 VersionInfo getServerVersionInfo() const
 {
 return _serverVersion;
 }

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

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

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

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

 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_CALL_EXCEPTION_WRAPPER(processDeferredExecutions());
 Lock<Mutex> l(_lock);
 broadcastConnectionStateChanged(ConnectionStateListener::Shutdown);
 }

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

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

 _lock.signalAll();
 }

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

 void sendWithoutRetry(const Message& message_)
 {
 Lock<Mutex> l(_lock);
 _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(const MessageHandler& messageHandler_, Message& message_, int timeout_ = 0)
 {
 Field id = message_.getCommandId();
 Field subId = message_.getSubscriptionId();
 Field qid = message_.getQueryId();
 bool isSubscribe = false;
 bool isSubscribeOnly = false;
 bool replace = false;
 unsigned requestedAcks = message_.getAckTypeEnum();
 unsigned systemAddedAcks = Message::AckType::None;

 switch(message_.getCommandEnum())
 {
 case Message::Command::Subscribe:
 case Message::Command::DeltaSubscribe:
 replace = message_.getOptions().operator std::string().find(AMPS_OPTIONS_REPLACE, 0, strlen(AMPS_OPTIONS_REPLACE)-1) != std::string::npos;
 if (!message_.getBookmark().empty() && _bookmarkStore.isValid())
 {
 systemAddedAcks |= Message::AckType::Persisted;
 }
 isSubscribeOnly = true;
 // fall through
 case Message::Command::SOWAndSubscribe:
 case Message::Command::SOWAndDeltaSubscribe:
 if (id.empty())
 {
 id = message_.newCommandId().getCommandId();
 }
 else
 {
 while (!replace && id != subId && _routes.hasRoute(id))
 {
 id = message_.newCommandId().getCommandId();
 }
 }
 if (subId.empty())
 {
 message_.setSubscriptionId(id);
 subId = id;
 }
 isSubscribe = true;
 // fall through
 case Message::Command::SOW:
 if(id.empty())
 {
 id = message_.newCommandId().getCommandId();
 }
 else
 {
 while (!replace && id != subId && _routes.hasRoute(id))
 {
 message_.newCommandId();
 if (qid == id)
 {
 qid = message_.getCommandId();
 message_.setQueryId(qid);
 }
 id = message_.getCommandId();
 }
 }
 if (!isSubscribeOnly)
 {
 if (qid.empty())
 {
 message_.setQueryID(id);
 qid = id;
 }
 else
 {
 while (!replace && qid != subId && qid != id
 && _routes.hasRoute(qid))
 {
 qid = message_.newQueryId().getQueryId();
 }
 }
 }
 systemAddedAcks |= Message::AckType::Processed;
 // for SOW only, we get a completed ack so we know when to remove the handler.
 if (!isSubscribeOnly) systemAddedAcks |= Message::AckType::Completed;
 message_.setAckTypeEnum(requestedAcks | systemAddedAcks);
 {
 int routesAdded = 0;
 Lock<Mutex> l(_lock);
 if (!subId.empty() && messageHandler_.isValid())
 {
 if (!_routes.hasRoute(subId))
 {
 ++routesAdded;
 }
 // This can replace a non-subscribe with a matching id
 // with a subscription but not another subscription.
 _routes.addRoute(subId, messageHandler_, requestedAcks,
 systemAddedAcks, isSubscribe);
 }
 if (!isSubscribeOnly && !qid.empty()
 && messageHandler_.isValid() && qid != subId)
 {
 if (routesAdded == 0)
 {
 _routes.addRoute(qid, messageHandler_,
 requestedAcks, systemAddedAcks, false);
 }
 else
 {
 void* data = NULL;
 {
 Unlock<Mutex> u(_lock);
 data = amps_invoke_copy_route_function(
 messageHandler_.userData());
 }
 if (!data)
 {
 _routes.addRoute(qid, messageHandler_, requestedAcks,
 systemAddedAcks, false);
 }
 else
 {
 _routes.addRoute(qid,
 MessageHandler(messageHandler_.function(),
 data),
 requestedAcks, systemAddedAcks, false);
 }
 }
 ++routesAdded;
 }
 if (!id.empty() && messageHandler_.isValid()
 && requestedAcks & ~Message::AckType::Persisted
 && id != subId && id != qid)
 {
 if (routesAdded == 0)
 {
 _routes.addRoute(id, messageHandler_, requestedAcks,
 systemAddedAcks, false);
 }
 else
 {
 void* data = NULL;
 {
 Unlock<Mutex> u(_lock);
 data = amps_invoke_copy_route_function(
 messageHandler_.userData());
 }
 if (!data)
 {
 _routes.addRoute(id, messageHandler_, requestedAcks,
 systemAddedAcks, false);
 }
 else
 {
 _routes.addRoute(id,
 MessageHandler(messageHandler_.function(),
 data),
 requestedAcks,
 systemAddedAcks, false);
 }
 }
 ++routesAdded;
 }
 try
 {
 // We aren't adding to subscription manager, so this isn't
 // an HA subscribe.
 syncAckProcessing(timeout_, message_, 0, false);
 message_.setAckTypeEnum(requestedAcks);
 }
 catch (...)
 {
 _routes.removeRoute(message_.getQueryID());
 _routes.removeRoute(message_.getSubscriptionId());
 _routes.removeRoute(id);
 message_.setAckTypeEnum(requestedAcks);
 throw;
 }
 }
 break;
 // These are valid commands that are used as-is
 case Message::Command::Unsubscribe:
 case Message::Command::Heartbeat:
 case Message::Command::Logon:
 case Message::Command::StartTimer:
 case Message::Command::StopTimer:
 case Message::Command::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(const DisconnectHandler& disconnectHandler)
 {
 Lock<Mutex> l(_lock);
 _disconnectHandler = disconnectHandler;
 }

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

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

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

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

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

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

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

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

 DisconnectHandler getDisconnectHandler() const
 {
 return _disconnectHandler;
 }

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

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

 Store getPublishStore() const
 {
 return _publishStore;
 }

 BookmarkStore getBookmarkStore() const
 {
 return _bookmarkStore;
 }

 amps_uint64_t publish(const char* topic_, size_t topicLen_, const char* data_,size_t dataLen_)
 {
 if (!_publishStore.isValid())
 {
 Lock<Mutex> l(_lock);
 _publishMessage.assignTopic(topic_, topicLen_);
 _publishMessage.assignData(data_, dataLen_);
 _send(_publishMessage);
 return 0;
 }
 else
 {
 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_);
 }
 }

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

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

 amps_uint64_t deltaPublish(const char* topic_, size_t topicLength_,
 const char* data_, size_t dataLength_)
 {
 if (!_publishStore.isValid())
 {
 Lock<Mutex> l(_lock);
 _deltaMessage.assignTopic(topic_, topicLength_);
 _deltaMessage.assignData(data_, dataLength_);
 _send(_deltaMessage);
 return 0;
 }
 else
 {
 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);
 return _logon(timeout_, authenticator_, options_);
 }

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

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

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

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

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

 if (!bookmark_.empty() && _bookmarkStore.isValid())
 {
 ackTypes |= Message::AckType::Persisted;
 }
 _message.setTopic(topic_);
 if (filter_.length()) _message.setFilter(filter_);
 if (bookmark_.length())
 {
 if (bookmark_ == AMPS_BOOKMARK_RECENT)
 {
 Message::Field mostRecent = _bookmarkStore.getMostRecent(subIdField);
 _message.setBookmark(mostRecent);
 }
 else
 {
 _message.setBookmark(bookmark_);
 if (_bookmarkStore.isValid())
 {
 if (bookmark_ != AMPS_BOOKMARK_NOW &&
 bookmark_ != AMPS_BOOKMARK_EPOCH)
 {
 _bookmarkStore.log(_message);
 _bookmarkStore.discard(_message);
 _bookmarkStore.persisted(subIdField, _message.getBookmark());
 }
 }
 }
 }
 if (options_.length()) _message.setOptions(options_);
 Message message = _message;
 if (isHASubscribe_)
 {
 message = _message.deepCopy();
 Unlock<Mutex> u(_lock);
 _subscriptionManager->subscribe(messageHandler_, message,
 Message::AckType::None);
 if (_badTimeToHASubscribe) return subId;
 }
 if (!_routes.hasRoute(_message.getSubscriptionId()))
 {
 _routes.addRoute(_message.getSubscriptionId(), messageHandler_,
 Message::AckType::None, ackTypes, 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);
 }
 deferredExecution(&amps_noOpFn, NULL);
 }

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

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

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

 return commandId;
 }

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

 std::string sowAndSubscribe(const MessageHandler& messageHandler_,
 const std::string& topic_,
 const std::string& filter_ = "",
 const std::string& orderBy_ = "",
 const std::string& bookmark_ = "",
 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
 int topN_ = AMPS_DEFAULT_TOP_N,
 const std::string& options_ = "",
 long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT,
 bool isHASubscribe_ = true)
 {
 isHASubscribe_ &= (bool)_subscriptionManager;
 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;
 }
 _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(const MessageHandler& messageHandler_,
 const std::string& topic_,
 long timeout_,
 const std::string& filter_ = "",
 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
 bool oofEnabled_ = false,
 int topN_ = AMPS_DEFAULT_TOP_N,
 bool isHASubscribe_ = true)
 {
 std::string notSet;
 return sowAndSubscribe(messageHandler_,
 topic_,
 filter_,
 notSet, // orderBy
 notSet, // bookmark
 batchSize_,
 topN_,
 (oofEnabled_ ? "oof" : ""),
 timeout_,
 isHASubscribe_);
 }

 std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
 const std::string& topic_,
 const std::string& filter_ = "",
 const std::string& orderBy_ = "",
 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
 int topN_ = AMPS_DEFAULT_TOP_N,
 const std::string& options_ = "",
 long timeout_ = AMPS_DEFAULT_COMMAND_TIMEOUT,
 bool isHASubscribe_ = true)
 {
 isHASubscribe_ &= (bool)_subscriptionManager;
 Lock<Mutex> l(_lock);
 _message.reset();
 _message.setCommandEnum(Message::Command::SOWAndDeltaSubscribe);
 _message.newCommandId();
 _message.setQueryID(_message.getCommandId());
 _message.setSubscriptionId(_message.getCommandId());
 std::string subId = _message.getSubscriptionId();
 _message.setTopic(topic_);
 if (filter_.length()) _message.setFilter(filter_);
 if (orderBy_.length()) _message.setOrderBy(orderBy_);
 _message.setBatchSize(AMPS::asString(batchSize_));
 if (topN_ != AMPS_DEFAULT_TOP_N) _message.setTopNRecordsReturned(AMPS::asString(topN_));
 if (options_.length()) _message.setOptions(options_);
 Message message = _message;
 if (isHASubscribe_)
 {
 message = _message.deepCopy();
 Unlock<Mutex> u(_lock);
 _subscriptionManager->subscribe(messageHandler_, message,
 Message::AckType::None);
 if (_badTimeToHASubscribe) return subId;
 }
 _routes.addRoute(message.getQueryID(), messageHandler_,
 Message::AckType::None, Message::AckType::Processed, 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(const MessageHandler& messageHandler_,
 const std::string& topic_,
 long timeout_,
 const std::string& filter_ = "",
 int batchSize_ = AMPS_DEFAULT_BATCH_SIZE,
 bool oofEnabled_ = false,
 bool sendEmpties_ = false,
 int topN_ = AMPS_DEFAULT_TOP_N,
 bool isHASubscribe_ = true)
 {
 std::string notSet;
 Message::Options options;
 if (oofEnabled_) options.setOOF();
 if (sendEmpties_ == false) options.setNoEmpties();
 return sowAndDeltaSubscribe(messageHandler_,
 topic_,
 filter_,
 notSet, // orderBy
 batchSize_,
 topN_,
 options,
 timeout_,
 isHASubscribe_);
 }

 std::string sowDelete(const MessageHandler& messageHandler_,
 const std::string& topic_,
 const std::string& filter_,
 long timeout_,
 Message::Field commandId_ = Message::Field())
 {
 if (_publishStore.isValid())
 {
 unsigned ackType = Message::AckType::Processed |
 Message::AckType::Stats |
 Message::AckType::Persisted;
 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&)
 { // -V565
 // Pass - it will get replayed upon reconnect
 }
 catch (...)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
 throw;
 }
 }
 return (std::string)commandId_;
 }
 else
 {
 Lock<Mutex> l(_lock);
 _message.reset();
 if (commandId_.empty())
 {
 _message.newCommandId();
 commandId_ = _message.getCommandId();
 }
 else
 {
 _message.setCommandId(commandId_.data(), commandId_.len());
 }
 _message.setCommandEnum(Message::Command::SOWDelete)
 .assignSubscriptionId(commandId_.data(), commandId_.len())
 .assignQueryID(commandId_.data(), commandId_.len())
 .setAckTypeEnum(Message::AckType::Processed |
 Message::AckType::Stats)
 .assignTopic(topic_.c_str(), topic_.length())
 .assignFilter(filter_.c_str(), filter_.length());
 _routes.addRoute(commandId_, messageHandler_,
 Message::AckType::Stats,
 Message::AckType::Processed,
 false);
 try
 {
 syncAckProcessing(timeout_, _message);
 }
 catch (...)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
 throw;
 }
 return (std::string)commandId_;
 }
 }

 std::string sowDeleteByData(const MessageHandler& messageHandler_,
 const std::string& topic_,
 const std::string& data_,
 long timeout_,
 Message::Field commandId_ = Message::Field())
 {
 if (_publishStore.isValid())
 {
 unsigned ackType = Message::AckType::Processed |
 Message::AckType::Stats |
 Message::AckType::Persisted;
 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&)
 { // -V565
 // Pass - it will get replayed upon reconnect
 }
 catch (...)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
 throw;
 }
 }
 return (std::string)commandId_;
 }
 else
 {
 Lock<Mutex> l(_lock);
 _message.reset();
 if (commandId_.empty())
 {
 _message.newCommandId();
 commandId_ = _message.getCommandId();
 }
 else
 {
 _message.setCommandId(commandId_.data(), commandId_.len());
 }
 _message.setCommandEnum(Message::Command::SOWDelete)
 .assignSubscriptionId(commandId_.data(), commandId_.len())
 .assignQueryID(commandId_.data(), commandId_.len())
 .setAckTypeEnum(Message::AckType::Processed |
 Message::AckType::Stats)
 .assignTopic(topic_.c_str(), topic_.length())
 .assignData(data_.c_str(), data_.length());
 _routes.addRoute(commandId_, messageHandler_,
 Message::AckType::Stats,
 Message::AckType::Processed,
 false);
 try
 {
 syncAckProcessing(timeout_, _message);
 }
 catch (...)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
 throw;
 }
 return (std::string)commandId_;
 }
 }

 std::string sowDeleteByKeys(const MessageHandler& messageHandler_,
 const std::string& topic_,
 const std::string& keys_,
 long timeout_,
 Message::Field commandId_ = Message::Field())
 {
 if (_publishStore.isValid())
 {
 unsigned ackType = Message::AckType::Processed |
 Message::AckType::Stats |
 Message::AckType::Persisted;
 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&)
 { // -V565
 // Pass - it will get replayed upon reconnect
 }
 catch (...)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
 throw;
 }
 }
 return (std::string)commandId_;
 }
 else
 {
 Lock<Mutex> l(_lock);
 _message.reset();
 if (commandId_.empty())
 {
 _message.newCommandId();
 commandId_ = _message.getCommandId();
 }
 else
 {
 _message.setCommandId(commandId_.data(), commandId_.len());
 }
 _message.setCommandEnum(Message::Command::SOWDelete)
 .assignSubscriptionId(commandId_.data(), commandId_.len())
 .assignQueryID(commandId_.data(), commandId_.len())
 .setAckTypeEnum(Message::AckType::Processed |
 Message::AckType::Stats)
 .assignTopic(topic_.c_str(), topic_.length())
 .assignSowKeys(keys_.c_str(), keys_.length());
 _routes.addRoute(commandId_, messageHandler_,
 Message::AckType::Stats,
 Message::AckType::Processed,
 false);
 try
 {
 syncAckProcessing(timeout_, _message);
 }
 catch (...)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(commandId_));
 throw;
 }
 return (std::string)commandId_;
 }
 }

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

 _send(_message);
 }

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

 amps_handle getHandle(void)
 {
 return _client;
 }

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

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

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

 void setHeartbeat(unsigned heartbeatInterval_, unsigned readTimeout_)
 {
 if (readTimeout_ && 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();
 int addedCount = 0;
 if (subid.len() > 0)
 {
 // This can replace a non-subscribe with a matching id
 // with a subscription but not another subscription.
 addedCount += _routes.addRoute(subid, handler_, requestedAcks_,
 systemAddedAcks_, isSubscribe_);
 }
 if (qid.len() > 0 && qid != subid)
 {
 while (_routes.hasRoute(qid))
 {
 message.newQueryId();
 if (cid_ == qid)
 cid_ = message.getQueryId();
 qid = message.getQueryId();
 }
 if (addedCount == 0)
 {
 _routes.addRoute(qid, handler_, requestedAcks_,
 systemAddedAcks_, isSubscribe_);
 }
 else
 {
 void* data = NULL;
 {
 Unlock<Mutex> u(_lock);
 data = amps_invoke_copy_route_function(handler_.userData());
 }
 if (!data)
 {
 _routes.addRoute(qid, handler_, requestedAcks_,
 systemAddedAcks_, false);
 }
 else
 {
 _routes.addRoute(qid,
 MessageHandler(handler_.function(),
 data),
 requestedAcks_,
 systemAddedAcks_, false);
 }
 }
 ++addedCount;
 }
 if (cid_.len() > 0 && cid_ != qid && cid_ != subid
 && requestedAcks_ & ~Message::AckType::Persisted)
 {
 while (_routes.hasRoute(cid_))
 {
 cid_ = message.newCommandId().getCommandId();
 }
 if (addedCount == 0)
 {
 _routes.addRoute(cid_, handler_, requestedAcks_,
 systemAddedAcks_, false);
 }
 else
 {
 void* data = NULL;
 {
 Unlock<Mutex> u(_lock);
 data = amps_invoke_copy_route_function(handler_.userData());
 }
 if (!data)
 {
 _routes.addRoute(cid_, handler_, requestedAcks_,
 systemAddedAcks_, false);
 }
 else
 {
 _routes.addRoute(cid_,
 MessageHandler(handler_.function(),
 data),
 requestedAcks_,
 systemAddedAcks_, false);
 }
 }
 }
 else if (commandType == Message::Command::Publish ||
 commandType == Message::Command::DeltaPublish)
 {
 cid_ = command_.getMessage().newCommandId().getCommandId();
 _routes.addRoute(cid_, handler_, requestedAcks_,
 systemAddedAcks_, false);
 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::SOWAndSubscribe
 || commandType == Message::Command::SOWAndDeltaSubscribe
 || commandType == Message::Command::StopTimer)
 systemAddedAcks |= Message::AckType::Completed;
 Message::Field cid = message.getCommandId();
 if (handler_.isValid() && cid.empty())
 {
 cid = message.newCommandId().getCommandId();
 }
 if (message.getBookmark().len() > 0)
 {
 if (command_.isSubscribe())
 {
 Message::Field bookmark = message.getBookmark();
 if (_bookmarkStore.isValid())
 {
 systemAddedAcks |= Message::AckType::Persisted;
 if (bookmark == AMPS_BOOKMARK_RECENT)
 {
 message.setBookmark(_bookmarkStore.getMostRecent(message.getSubscriptionId()));
 }
 else if (bookmark != AMPS_BOOKMARK_NOW &&
 bookmark != AMPS_BOOKMARK_EPOCH)
 {
 _bookmarkStore.log(message);
 if (!BookmarkRange::isRange(bookmark))
 {
 _bookmarkStore.discard(message);
 _bookmarkStore.persisted(message.getSubscriptionId(),
 bookmark);
 }
 }
 }
 else if (bookmark == AMPS_BOOKMARK_RECENT)
 {
 message.setBookmark(AMPS_BOOKMARK_EPOCH);
 }
 }
 }
 if (isPublishStore)
 {
 systemAddedAcks |= Message::AckType::Persisted;
 }
 bool isSubscribe = command_.isSubscribe();
 if (handler_.isValid() && !isSubscribe)
 {
 _registerHandler(command_, cid, handler_,
 requestedAcks, systemAddedAcks, isSubscribe);
 }
 bool useSyncSend = cid.len() > 0 && command_.hasProcessedAck();
 if (isPublishStore)
 {
 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&)
 { // -V565
 // 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_)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(subId));
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
 message.setAckTypeEnum(requestedAcks);
 throw;
 }
 }
 catch (const TimedOutException&)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(cid));
 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(subId));
 AMPS_CALL_EXCEPTION_WRAPPER(unsubscribeInternal(message.getQueryId()));
 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&)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
 message.setAckTypeEnum(requestedAcks);
 throw;
 }
 catch (...)
 {
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(cid));
 AMPS_CALL_EXCEPTION_WRAPPER(_routes.removeRoute(message.getQueryId()));
 message.setAckTypeEnum(requestedAcks);
 throw;
 }
 }
 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(!_connected)
 {
 return;
 }
 else
 {
 Lock<Mutex> lock(_lock);
 typedef TopicHashMap::iterator iterator;
 for(iterator it = _topicHashMap.begin(), end = _topicHashMap.end(); it!=end; ++it)
 {
 QueueBookmarks& queueBookmarks = it->second;
 sendCount += _ack(queueBookmarks);
 }
 }
 if(sendCount && _connected) publishFlush(0, Message::AckType::Processed);
 }
 // called when there's idle time, to see if we need to flush out any "acks"
 void checkQueueAcks(void)
 {
 if(!_topicHashMap.size()) return;
 Lock<Mutex> lock(_lock);
 try
 {
 amps_uint64_t threshold = amps_now() - (amps_uint64_t)_queueAckTimeout;
 typedef TopicHashMap::iterator iterator;
 for(iterator it = _topicHashMap.begin(), end = _topicHashMap.end(); it!=end; ++it)
 {
 QueueBookmarks& queueBookmarks = it->second;
 if(queueBookmarks._bookmarkCount && queueBookmarks._oldestTime < threshold) _ack(queueBookmarks);
 }
 }
 catch(std::exception& ex)
 {
 AMPS_UNHANDLED_EXCEPTION(ex);
 }
 }

 void deferredExecution(DeferredExecutionFunc func_, void* userData_)
 {
 Lock<Mutex> lock(_deferredExecutionLock);
 _deferredExecutionList.push_back(
 DeferredExecutionRequest(func_,userData_));
 }

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

 bool getRetryOnDisconnect(void) const
 {
 return _isRetryOnDisconnect;
 }

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

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

 unsigned getDefaultMaxDepth(void) const
 {
 return _defaultMaxDepth;
 }

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

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

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

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

 bool operator==(const iterator& rhs)
 {
 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_,
 const std::string& queryId_ = "");
 inline void setSubscription(const std::string& subId_,
 const std::string& commandId_ = "",
 const std::string& queryId_ = "");
 inline void setStatsOnly(const std::string& commandId_,
 const std::string& queryId_ = "");
 inline void setAcksOnly(const std::string& commandId_, unsigned acks_);

 inline operator MessageHandler(void);

 inline static MessageStream fromExistingHandler(const MessageHandler& handler);

 friend class Client;

};

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

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

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

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

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

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

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

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

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

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

 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(const MessageHandler& messageHandler, Message& message, int timeout = 0)
 {
 return _body.get().send(messageHandler, message, timeout);
 }

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

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

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

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


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

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

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

 std::string sowAndDeltaSubscribe(const MessageHandler& messageHandler_,
 const std::string& topic_,
 const std::string& filter_ = "",
 const std::string& orderBy_ = "",
 int batchSize_ = DEFAULT_BATCH_SIZE,
 int topN_ = DEFAULT_TOP_N,
 const std::string& options_ = "",
 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
 {
 return _body.get().sowAndDeltaSubscribe(messageHandler_, topic_,
 filter_, orderBy_, batchSize_,
 topN_, options_, timeout_);
 }
 MessageStream sowAndDeltaSubscribe(const std::string& topic_,
 const std::string& filter_ = "",
 const std::string& orderBy_ = "",
 int batchSize_ = DEFAULT_BATCH_SIZE,
 int topN_ = DEFAULT_TOP_N,
 const std::string& options_ = "",
 long timeout_ = DEFAULT_COMMAND_TIMEOUT)
 {
 MessageStream result(*this);
 if (_body.get().getDefaultMaxDepth())
 result.maxDepth(_body.get().getDefaultMaxDepth());
 result.setSubscription(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(const MessageHandler& messageHandler_,
 const std::string& topic_,
 long timeout_,
 const std::string& filter_ = "",
 int batchSize_ = DEFAULT_BATCH_SIZE,
 bool oofEnabled_ = false,
 bool sendEmpties_ = false,
 int topN_ = DEFAULT_TOP_N)
 {
 return _body.get().sowAndDeltaSubscribe(messageHandler_, topic_,
 timeout_, filter_, batchSize_,
 oofEnabled_, sendEmpties_,
 topN_);
 }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 static const char* BOOKMARK_NOW() { return AMPS_BOOKMARK_NOW; }
 static const char* 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
 {
 if (command_.isSubscribe())
 {
 Message& message = command_.getMessage();
 Field subId = message.getSubscriptionId();
 bool useExistingHandler = !subId.empty() && !message.getOptions().empty() && message.getOptions().contains("replace",7);
 if(useExistingHandler)
 {
 MessageHandler existingHandler;
 if (_body.get()._routes.getRoute(subId, existingHandler))
 {
 // we found an existing handler.
 _body.get().executeAsync(command_, existingHandler, false);
 return id; // empty string indicates existing
 }
 }
 }
 id = _body.get().executeAsync(command_, handler_, false);
 }
 catch (const DisconnectedException&)
 {
 removeMessageHandler(command_.getMessage().getCommandId());
 if (command_.isSubscribe())
 {
 removeMessageHandler(command_.getMessage().getSubscriptionId());
 }
 if (command_.isSow())
 {
 removeMessageHandler(command_.getMessage().getQueryID());
 }
 throw;
 }
 return id;
 }

 MessageStream execute(Command& command_);

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

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

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

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

 int getAckTimeout(void) const
 {
 return _body.get().getAckTimeout();
 }
 void setAckTimeout(const int ackTimeout_)
 {
 _body.get().setAckTimeout(ackTimeout_);
 }


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

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

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

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

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

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

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

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

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

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

inline unsigned
ClientImpl::processedAck(Message &message)
{
 unsigned deliveries = 0;
 AckResponse ack;
 const char* data = NULL;
 size_t len = 0;
 amps_handle messageHandle = message.getMessage();
 amps_message_get_field_value(messageHandle, AMPS_CommandId, &data, &len);
 Lock<Mutex> l(_lock);
 if (data && len)
 {
 Lock<Mutex> guard(_ackMapLock);
 AckMap::iterator i = _ackMap.find(std::string(data, len));
 if (i != _ackMap.end())
 {
 ++deliveries;
 ack = i->second;
 _ackMap.erase(i);
 }
 }
 if (deliveries)
 {
 amps_message_get_field_value(messageHandle, AMPS_Status, &data,
 &len);
 ack.setStatus(data, len);
 amps_message_get_field_value(messageHandle, AMPS_Reason, &data,
 &len);
 ack.setReason(data, len);
 amps_message_get_field_value(messageHandle, AMPS_UserId, &data,
 &len);
 ack.setUsername(data, len);
 amps_message_get_field_value(messageHandle, AMPS_Password, &data,
 &len);
 ack.setPassword(data, len);
 amps_message_get_field_value(messageHandle, AMPS_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);
 amps_message_get_field_value(messageHandle, AMPS_Bookmark, &data, &len);
 ack.setBookmark(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_;
 AMPS_CALL_EXCEPTION_WRAPPER_2(me,me->processDeferredExecutions());
 if(!messageHandle_)
 {
 if(me->_queueAckTimeout) me->checkQueueAcks();
 return AMPS_E_OK;
 }

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

 if (!isMessageQueue && !bookmark.empty() &&
 me->_bookmarkStore.isValid())
 {
 if (me->_bookmarkStore.isDiscarded(me->_readMessage))
 {
 //Call duplicate message handler in handlers map
 if (me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage].isValid())
 {
 AMPS_CALL_EXCEPTION_WRAPPER_2(me, me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::DuplicateMessage].invoke(message));
 }
 }
 else
 {
 me->_bookmarkStore.log(me->_readMessage);
 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
 handler.invoke(message));
 }
 }
 else
 {
 if(isMessageQueue && isAutoAck)
 {
 try
 {
 AMPS_CALL_EXCEPTION_WRAPPER_STREAM_FULL_2(me, handler.invoke(message));
 if (!message.getIgnoreAutoAck())
 {
 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
 me->_ack(message.getTopic(),message.getBookmark()));
 }
 }
 catch(std::exception& ex)
 {
 if (!message.getIgnoreAutoAck())
 {
 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
 me->_ack(message.getTopic(),message.getBookmark(),"cancel"));
 }
 AMPS_UNHANDLED_EXCEPTION_2(me,ex);
 }
 }
 else
 {
 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
 handler.invoke(message));
 }
 }
 }
 else me->lastChance(message);
 } // for (subidsEnd)
 }
 else if (commandType == Message::Command::Ack)
 {
 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
 me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::Ack].invoke(message));
 unsigned ackType = message.getAckTypeEnum();
 unsigned deliveries = 0U;
 switch (ackType)
 {
 case Message::AckType::Persisted:
 deliveries += me->persistedAck(message);
 break;
 case Message::AckType::Processed: // processed
 deliveries += me->processedAck(message);
 break;
 }
 AMPS_CALL_EXCEPTION_WRAPPER_2(me, deliveries += me->_routes.deliverAck(message, ackType));
 if (deliveries == 0)
 {
 me->lastChance(message);
 }
 }
 else if (commandType == Message::Command::Heartbeat)
 {
 AMPS_CALL_EXCEPTION_WRAPPER_2(me,
 me->_globalCommandTypeHandlers[GlobalCommandTypeHandlers::Heartbeat].invoke(message));
 if(me->_heartbeatTimer.getTimeout() != 0.0) // -V550
 {
 me->checkAndSendHeartbeat(true);
 }
 else
 {
 me->lastChance(message);
 }
 return AMPS_E_OK;
 }
 else if (!message.getCommandId().empty())
 {
 unsigned deliveries = 0U;
 try
 {
 while(me->_connected) // Keep sending heartbeats when stream is full
 {
 try
 {
 deliveries = me->_routes.deliverData(message, message.getCommandId());
 break;
 }
#ifdef _WIN32
 catch(MessageStreamFullException&)
#else
 catch(MessageStreamFullException& ex_)
#endif
 {
 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;
 std::string _subId;
 std::string _queryId;
 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 (...) {/*Ignore exception listener exceptions*/} // -V565
 }
 if (_client.isValid())
 {
 _client.removeConnectionStateListener(this);
 Client c = _client;
 _client = Client((ClientImpl*)NULL);
 c.deferredExecution(MessageStreamImpl::destroyer, this);
 }
 else
 {
 delete this;
 }
 }

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

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

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

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

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

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

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

 bool next(Message& current_)
 {
 Lock<Mutex> lock(_lock);
 if (!_previousTopic.empty() && !_previousBookmark.empty())
 {
 try
 {
 if (_client.isValid())
 {
 _client.ackDeferredAutoAck(_previousTopic, _previousBookmark);
 }
 }
#ifdef _WIN32
 catch (AMPSException&)
#else
 catch (AMPSException& e)
#endif
 {
 current_.invalidate();
 _previousTopic.clear();
 _previousBookmark.clear();
 return false;
 }
 _previousTopic.clear();
 _previousBookmark.clear();
 }
 double minWaitTime = (double)((_timeout && _timeout > 1000)
 ? _timeout : 1000);
 Timer timer(minWaitTime);
 timer.start();
 while(_q.empty() && _state & Running)
 {
 // Using timeout so python can interrupt
 _lock.wait((long)minWaitTime);
 {
 Unlock<Mutex> unlck(_lock);
 amps_invoke_waiting_function();
 }
 if (_timeout)
 {
 // In case we woke up early, see how much longer to wait
 if(timer.checkAndGetRemaining(&minWaitTime))
 {
 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 (_client.isValid())
 {
 if (_state == SOWOnly || _state == Subscribe) //not delete
 {
 if (!_commandId.empty()) _client.unsubscribe(_commandId);
 if (!_subId.empty()) _client.unsubscribe(_subId);
 if (!_queryId.empty()) _client.unsubscribe(_queryId);
 }
 else
 {
 if (!_commandId.empty()) _client.removeMessageHandler(_commandId);
 if (!_subId.empty()) _client.removeMessageHandler(_subId);
 if (!_queryId.empty()) _client.removeMessageHandler(_queryId);
 }
 }
 if(_state==SOWOnly || _state==Subscribe || _state==Unset)
 {
 _state = Closed;
 }
 }
 static void _messageHandler(const Message& message_, MessageStreamImpl* this_)
 {
 Lock<Mutex> lock(this_->_lock);
 if(this_->_state != Conflate)
 {
 AMPS_TESTING_SLOW_MESSAGE_STREAM
 if(this_->_q.size() >= this_->_maxDepth)
 {
 // We throw here so that heartbeats can be sent. The exception
 // will be handled internally only, and the same Message will
 // come back to try again. Make sure to signal.
 this_->_lock.signalAll();
 throw MessageStreamFullException("Stream is currently full.");
 }
 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) // -V712
 {
 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_,
 const std::string& queryId_)
{
 _body->setSOWOnly(commandId_, queryId_);
}
inline void MessageStream::setSubscription(const std::string& subId_,
 const std::string& commandId_,
 const std::string& queryId_)
{
 _body->setSubscription(subId_, commandId_, queryId_);
}
inline void MessageStream::setStatsOnly(const std::string& commandId_,
 const std::string& queryId_)
{
 _body->setStatsOnly(commandId_, queryId_);
}
inline void MessageStream::setAcksOnly(const std::string& commandId_,
 unsigned acks_)
{
 _body->setAcksOnly(commandId_, acks_);
}
inline MessageStream MessageStream::timeout(unsigned timeout_)
{
 _body->timeout(timeout_);
 return *this;
}
inline MessageStream MessageStream::conflate(void)
{
 _body->conflate();
 return *this;
}
inline MessageStream MessageStream::maxDepth(unsigned maxDepth_)
{
 _body->maxDepth(maxDepth_);
 return *this;
}
inline unsigned MessageStream::getMaxDepth(void) const
{
 return _body->getMaxDepth();
}
inline unsigned MessageStream::getDepth(void) const
{
 return _body->getDepth();
}

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

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

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