MemorySubscriptionManager.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2020 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//

#ifndef _MEMORYSUBSCRIPTIONMANAGER_H_
#define _MEMORYSUBSCRIPTIONMANAGER_H_

#include <ampsplusplus.hpp>
#include <Field.hpp>
#include <algorithm>
#include <list>
#include <map>
#include <set>


namespace AMPS
{

class MemorySubscriptionManager : public SubscriptionManager
{
private:

    class SubscriptionInfo
    {
    public:
        SubscriptionInfo(MessageHandler messageHandler_,
                         const Message& message_,
                         unsigned requestedAckTypes_)
        : _handler(messageHandler_)
        , _requestedAckTypes(requestedAckTypes_)
        , _bookmark(!message_.getBookmark().empty())
        , _clearSubId(false)
        {
            _m = message_;
            _subId = _m.getSubscriptionId();
            std::string options = _m.getOptions();
            size_t replace = options.find("replace");
            // AMPS should be ok if options contains ,,
            static const size_t replaceLen = 7;
            if (replace != std::string::npos)
            {
                options.erase(replace, replaceLen);
                _m.setOptions(options);
            }
            _paused = (options.find("pause") != std::string::npos);
        }

        ~SubscriptionInfo()
        {
            if (_clearSubId)
            {
                _m.getSubscriptionId().clear();
            }
        }

        void resubscribe(Client& client_, int timeout_)
        {
            if(_bookmark)
            {
                if (_paused && !_recent.empty())
                    _m.setBookmark(_recent);
                else
                    _m.setBookmark(client_.getBookmarkStore().getMostRecent(_subId));
            }
            _m.newCommandId();
            _m.setAckTypeEnum(_requestedAckTypes);
            client_.send(_handler, _m, timeout_);
        }

        MessageHandler messageHandler() const
        {
            return _handler;
        }

        const Message::Field& subId() const
        {
            return _subId;
        }

        // Returns true if the last subId is removed, false otherwise
        bool removeSubId(const Message::Field& subId_)
        {
            size_t subIdLen = subId_.len();
            const char* subIdData = subId_.data();
            while (subIdLen && *subIdData == ',')
            {
                ++subIdData;
                --subIdLen;
            }
            while (subIdLen && subIdData[subIdLen-1] == ',')
            {
                --subIdLen;
            }
            if (subIdLen == 0 || subIdLen > _subId.len()) return _subId.empty();
            bool match = true;
            size_t matchStart = 0;
            size_t matchCount = 0;
            for (size_t i=0; i<_subId.len(); ++i)
            {
                if (_subId.data()[i] == ',')
                {
                    if (matchCount == subIdLen) break;
                    matchStart = i+1;
                    matchCount = 0;
                    match = true;
                }
                else if (match)
                {
                    if (_subId.data()[i] == subIdData[matchCount])
                        ++matchCount;
                    else
                    {
                        matchCount = 0;
                        match = false;
                    }
                }
            }
            if (match && matchCount == subIdLen)
            {
                size_t newLen = _subId.len() - matchCount;
                if (newLen > 1) // More than just ,
                {
                    while (matchStart + matchCount < _subId.len() &&
                           _subId.data()[matchStart+matchCount] == ',')
                    {
                        ++matchCount;
                        --newLen;
                    }
                    char* buffer = new char[newLen];
                    // Match is not first
                    if (matchStart > 0)
                    {
                        memcpy(buffer, _subId.data(), matchStart);
                    }
                    // Match is not last
                    if (matchStart + matchCount < _subId.len())
                    {
                        memcpy(buffer+matchStart,
                               _subId.data()+matchStart+matchCount,
                               _subId.len()-matchStart-matchCount);
                    }
                    if (_clearSubId)
                    {
                        _m.getSubscriptionId().clear();
                    }
                    else
                    {
                        _clearSubId = true;
                    }
                    _m.assignSubscriptionId(buffer, newLen);
                    _subId = _m.getSubscriptionId();
                    return false;
                }
                else
                {
                    if (_clearSubId)
                    {
                        _m.getSubscriptionId().clear();
                        _clearSubId = false;
                    }
                    else
                    {
                        _m.getSubscriptionId().assign(NULL, 0);
                    }
                    _subId = _m.getSubscriptionId();
                    return true;
                }
            }
            return _subId.empty();
        }

        bool paused() const
        {
            return _paused;
        }

        void pause()
        {
            if (_paused) return;
            std::string opts(Message::Options::Pause());
            opts.append(_m.getOptions());
            _m.setOptions(opts);
            _paused = true;
        }

        std::string getMostRecent(Client& client_)
        {
            if (!_recent.empty()) return _recent;
            std::map<amps_uint64_t, amps_uint64_t> publishers;
            const char* start = _subId.data();
            const char* end = _subId.data() + _subId.len();
            while (start < end)
            {
                const char* comma = (const char*)memchr(start, ',',
                                                        (size_t)(end-start));
                // No more commas found, just use start->end
                if(!comma) comma = end;
                if(comma == start)
                {
                    start = comma + 1;
                    continue;
                }
                Message::Field sid(start, (size_t)(comma-start));
                Message::Field sidRecent = client_.getBookmarkStore().getMostRecent(sid);
                const char* sidRecentStart = sidRecent.data();
                const char* sidRecentEnd = sidRecent.data() + sidRecent.len();
                while (sidRecentStart < sidRecentEnd)
                {
                    const char* sidRecentComma = (const char*)
                                                  memchr(sidRecentStart, ',',
                                                   (size_t)(sidRecentEnd-sidRecentStart));
                    // No more commas found, just use start->end
                    if(!sidRecentComma) sidRecentComma = sidRecentEnd;
                    if(sidRecentComma == sidRecentStart)
                    {
                        sidRecentStart = sidRecentComma + 1;
                        continue;
                    }
                    Message::Field bookmark(sidRecentStart,
                                            (size_t)(sidRecentComma-sidRecentStart));
                    amps_uint64_t publisher = (amps_uint64_t)0;
                    amps_uint64_t seq = (amps_uint64_t)0;
                    Field::parseBookmark(bookmark, publisher, seq);
                    if (publishers.count(publisher) == 0
                        || publishers[publisher] > seq)
                    {
                        publishers[publisher] = seq;
                    }
                    // Move past comma
                    sidRecentStart = sidRecentComma + 1;
                }
                // Move past comma
                start = comma + 1;
            }
            std::ostringstream os;
            for(std::map<amps_uint64_t, amps_uint64_t>::iterator i = publishers.begin();
                                                                 i != publishers.end();
                                                                 ++i)
            {
                if (i->first == 0 && i->second == 0) continue;
                if (os.tellp() > 0) os << ',';
                os << i->first << '|' << i->second << "|";
            }
            _recent = os.str();
            return _recent;
        }

        void setMostRecent(const std::string& recent_)
        {
            _recent = recent_;
        }

    private:
        std::string    _recent;
        MessageHandler _handler;
        Message        _m;
        Message::Field _subId;
        unsigned       _requestedAckTypes;
        bool           _bookmark;
        bool           _paused;
        bool           _clearSubId;

    };//class SubscriptionInfo

    class Resubscriber
    {
            Client& _client;
            int _timeout;
        public:
            Resubscriber(Client& client_, int timeout_)
            : _client(client_), _timeout(timeout_) {}
            void operator()(std::pair<Message::Field, SubscriptionInfo*> iter_)
            {
                iter_.second->resubscribe(_client, _timeout);
            }
            void operator()(SubscriptionInfo* iter_)
            {
                iter_->resubscribe(_client, _timeout);
            }
    };

    class Deleter
    {
            bool _clearSubId;
        public:
            Deleter(bool clearSubId_ = false) : _clearSubId(clearSubId_) {}
            void operator()(std::pair<Message::Field, SubscriptionInfo*> iter_)
            {
                if (_clearSubId) iter_.first.clear();
                else delete iter_.second;
            }
            void operator()(SubscriptionInfo* iter_)
            {
                delete iter_;
            }
    };

public:
    typedef std::map<Message::Field, SubscriptionInfo*, Message::Field::FieldHash> SubscriptionMap;

    MemorySubscriptionManager()
    : _resubscribing(0)
    , _resubscriptionTimeout(getDefaultResubscriptionTimeout())
    { ; }

    ~MemorySubscriptionManager()
    {
        clear();
    }

    void subscribe(MessageHandler messageHandler_,
                   const Message& message_, unsigned requestedAckTypes_)
    {
        const Message::Field& subId = message_.getSubscriptionId();
        if(!subId.empty())
        {
            Lock<Mutex> l(_lock);
            while(_resubscribing != 0) _lock.wait(10);
            std::string options = message_.getOptions();
            if (options.find("resume") != std::string::npos)
            {
                // For a resume, we store each sub id with a single Subscription
                SubscriptionInfo* subInfo = new SubscriptionInfo(MessageHandler(),
                                                                 message_,
                                                                 requestedAckTypes_);
                bool saved = false;
                const char* start = subInfo->subId().data();
                const char* end = subInfo->subId().data() + subInfo->subId().len();
                while (start < end)
                {
                    const char* comma = (const char*)memchr(start, ',',
                                                      (size_t)(end-start));
                    // No more commas found, just use start->end
                    if(!comma) comma = end;
                    if(comma == start)
                    {
                        start = comma + 1;
                        continue;
                    }
                    Message::Field sid = Message::Field(start,
                                                        (size_t)(comma-start));
                    // Calling resume on something already resumed is ignored,
                    // so don't update anything that exists.
                    if (_resumed.find(sid) == _resumed.end())
                    {
                        _resumed[sid.deepCopy()] = subInfo;
                        saved = true;
                    }
                    // Move past comma
                    start = comma + 1;
                }
                if (saved) _resumedSet.insert(subInfo);
                else delete subInfo;
            }
            else if (options.find("pause") != std::string::npos)
            {
                const char* start = subId.data();
                const char* end = subId.data() + subId.len();
                while (start < end)
                {
                    MessageHandler messageHandler = messageHandler_;
                    const char* comma = (const char*)memchr(start, ',',
                                                          (size_t)(end-start));
                    // No more commas found, just use start->end
                    if(!comma) comma = end;
                    if(comma == start)
                    {
                        start = comma + 1;
                        continue;
                    }
                    Message::Field sid = Message::Field(start,
                                                        (size_t)(comma-start));
                    SubscriptionMap::iterator resume = _resumed.find(sid);
                    if (resume != _resumed.end())
                    {
                        SubscriptionInfo* subPtr = resume->second;
                        Message::Field subField(resume->first);
                        _resumed.erase(resume); // Remove mapping for sid
                        subField.clear();
                        // If last subId, remove completely
                        if (subPtr->removeSubId(sid))
                        {
                            _resumedSet.erase(subPtr);
                            delete subPtr;
                        }
                    }
                    // Move past comma
                    start = comma + 1;
                    SubscriptionMap::iterator item = _active.find(sid);
                    if(item != _active.end())
                    {
                        if (options.find("replace") != std::string::npos)
                        {
                            messageHandler = item->second->messageHandler();
                            delete item->second;
                            _active.erase(item);
                        }
                        else
                        {
                            item->second->pause();
                            continue; // Leave current one
                        }
                    }
                    Message m = message_.deepCopy();
                    m.setSubscriptionId(sid.data(), sid.len());
                    SubscriptionInfo* s = new SubscriptionInfo(messageHandler, m,
                                                               requestedAckTypes_);
                    // Insert using the subId from s, which is deep copy of original
                    _active[s->subId()] = s;
                }
            }
            else // Not a pause or resume
            {
                MessageHandler messageHandler = messageHandler_;
                SubscriptionMap::iterator item = _active.find(subId);
                if(item != _active.end())
                {
                    messageHandler = item->second->messageHandler();
                    delete item->second;
                    _active.erase(item);
                }
                SubscriptionInfo* s = new SubscriptionInfo(messageHandler,
                                                           message_,
                                                           requestedAckTypes_);
                // Insert using the subId from s, which is deep copy of original
                _active[s->subId()] = s;
            }
        }
    }

    void unsubscribe(const Message::Field& subId_)
    {
        Lock<Mutex> l(_lock);
        SubscriptionMap::iterator item = _active.find(subId_);
        if(item != _active.end())
        {
            SubscriptionInfo* subPtr = item->second;
            _active.erase(item);
            while(_resubscribing != 0) _lock.wait(10);
            delete subPtr;
        }
        item = _resumed.find(subId_);
        if(item != _resumed.end())
        {
            SubscriptionInfo* subPtr = item->second;
            Message::Field subField(item->first);
            _resumed.erase(item);
            subField.clear();
            // If last subId, remove completely
            if (subPtr->removeSubId(subId_))
            {
                _resumedSet.erase(subPtr);
                while(_resubscribing != 0) _lock.wait(10);
                delete subPtr;
            }
        }
    }

    void clear()
    {
        Lock<Mutex> l(_lock);
        while(_resubscribing != 0) _lock.wait(10);
        std::for_each(_active.begin(), _active.end(), Deleter());
        std::for_each(_resumedSet.begin(), _resumedSet.end(), Deleter());
        std::for_each(_resumed.begin(), _resumed.end(), Deleter(true));
        _active.clear();
        _resumed.clear();
        _resumedSet.clear();
    }

    void resubscribe(Client& client_)
    {
        std::list<SubscriptionInfo*> subscriptions;
        try
        {
            AtomicFlagFlip resubFlip(&_resubscribing);
            {
                Lock<Mutex> l(_lock);
                subscriptions.assign(_resumedSet.begin(), _resumedSet.end());
                for (SubscriptionMap::iterator iter = _active.begin();
                     iter != _active.end(); ++iter)
                {
                    SubscriptionInfo* sub = iter->second;
                    if (sub->paused())
                    {
                        SubscriptionMap::iterator resIter = _resumed.find(sub->subId());
                        // All pause subs resuming together should be sent with
                        // bookmark as list of the resumes' most recents
                        if (resIter != _resumed.end())
                        {
                            sub->setMostRecent(resIter->second->getMostRecent(client_));
                        }
                    }
                    subscriptions.push_front(iter->second);
                }
            }
            Resubscriber resubscriber(client_, _resubscriptionTimeout);
            std::for_each(subscriptions.begin(), subscriptions.end(),
                          resubscriber);
        }
        catch(const AMPSException& e)
        {
            throw e;
        }
        catch(const std::exception& e)
        {
            throw e;
        }
    }

    void setResubscriptionTimeout(int timeout_)
    {
        if (timeout_ >= 0)
        {
            _resubscriptionTimeout = timeout_;
        }
    }

    int getResubscriptionTimeout(void)
    {
        return _resubscriptionTimeout;
    }

    static int setDefaultResubscriptionTimeout(int timeout_)
    {
        static int _defaultResubscriptionTimeout =
                                    AMPS_SUBSCRIPTION_MANAGER_DEFAULT_TIMEOUT;
        if (timeout_ >= 0)
        {
            _defaultResubscriptionTimeout = timeout_;
        }
        return _defaultResubscriptionTimeout;
    }

    static int getDefaultResubscriptionTimeout(void)
    {
        return setDefaultResubscriptionTimeout(-1);
    }

private:

    SubscriptionMap _active;
    SubscriptionMap _resumed;
    std::set<SubscriptionInfo*> _resumedSet;
    Mutex _lock;
    ATOMIC_TYPE_8 _resubscribing;
    int _resubscriptionTimeout;
}; //class MemorySubscriptionManager

} // namespace AMPS

#endif //_MEMORYSUBSCRIPTIONMANAGER_H_