MessageRouter.hpp

//
// 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 _MESSAGEROUTER_HPP_
#define _MESSAGEROUTER_HPP_
#include <map>
#include "util.hpp"
#include "Message.hpp"

namespace AMPS
{
template <typename Func, typename Object>
class Handler
{
protected:
    friend class MessageStream;
    Func _func;
    void* _userData;
#ifdef AMPS_USE_FUNCTIONAL
    std::function<void(Object)> _callable;
#endif
    bool _isValid;

public:
    // No op function for handlers
    static void noOpHandler(Object) {;}

    typedef Func FunctionType;
    Handler() : _func(NULL), _userData(NULL)
#ifdef AMPS_USE_FUNCTIONAL
        , _callable(Handler<Func,Object>::noOpHandler)
#endif
        , _isValid(false)
    {
    }

    Handler(Func func_, void* userData_)
        : _func(func_), _userData(userData_)
#ifdef AMPS_USE_FUNCTIONAL
        , _callable(noOpHandler)
#endif
        , _isValid(true)
    {
    }

    Handler(const Handler& orig_)
        : _func(orig_._func), _userData(orig_._userData)
#ifdef AMPS_USE_FUNCTIONAL
        , _callable(orig_._callable)
#endif
        , _isValid(true)
    {
    }
#ifdef AMPS_USE_FUNCTIONAL
    template <typename T>
    Handler(const T& callback_)
        : _func(NULL), _userData(NULL), _callable(callback_), _isValid(true)
    {
    }
#endif
    void invoke(Object message)
    {
        if (_func)
            _func(message, _userData);
#ifdef AMPS_USE_FUNCTIONAL
        else
            _callable(message);
#endif
    }

    bool  isValid(void) const { return _isValid; }
    Func  function(void) const { return _func; }
    void* userData(void) const { return _userData; }
};
class Message;
typedef void(*MessageHandlerFunc)(const Message&, void* userData);

typedef Handler<MessageHandlerFunc, const Message&> MessageHandler;

class MessageRouter
{
private:
  MessageHandler _emptyMessageHandler;

  class MessageRoute
  {
    MessageHandler _messageHandler;
    unsigned _requestedAcks, _systemAcks, _terminationAck;
  public:
    MessageRoute() : _requestedAcks(0), _systemAcks(0), _terminationAck(0) {;}
    MessageRoute(const MessageRoute& rhs_) :
      _messageHandler(rhs_._messageHandler),
      _requestedAcks (rhs_._requestedAcks),
      _systemAcks    (rhs_._systemAcks),
      _terminationAck(rhs_._terminationAck)
    {;}
    const MessageRoute& operator=(const MessageRoute& rhs_)
    {
      _messageHandler = rhs_._messageHandler;
      _requestedAcks  = rhs_._requestedAcks;
      _systemAcks     = rhs_._systemAcks;
      _terminationAck = rhs_._terminationAck;
      return *this;
    }
    MessageRoute(MessageHandler messageHandler_, unsigned requestedAcks_,
                 unsigned systemAcks_, bool isSubscribe_) :
        _messageHandler(messageHandler_),
        _requestedAcks(requestedAcks_),
        _systemAcks(systemAcks_),
        _terminationAck(0)
    {
      if(!isSubscribe_)
      {
        // The ack to terminate the route on is whatever the highest
        // bit set in requestedAcks is.
        unsigned bitCounter = (requestedAcks_|systemAcks_)>>1;
        _terminationAck = 1;
        while(bitCounter > 0) { bitCounter>>=1; _terminationAck<<=1; }
      }
    }

    // Deliver an ack to registered handler if the ack type was requested
    unsigned deliverAck(const Message& message_, unsigned ackType_)
    {
      if ( (_requestedAcks & ackType_) == 0) return 0;
      try {
        _messageHandler.invoke(message_);
      } catch (std::exception &ex)
      {
        std::cerr << ex.what() << std::endl;
      }
      return 1;
    }
    bool isTerminationAck(unsigned ackType_) const
    {
      return ackType_ == _terminationAck;
    }
    unsigned deliverData(const Message& message_)
    {
      _messageHandler.invoke(message_);
      return 1;
    }
    const MessageHandler& getMessageHandler() const
    {
      return _messageHandler;
    }
    MessageHandler& getMessageHandler()
    {
      return _messageHandler;
    }
  };

public:
  MessageRouter()
  : _previousCommandId(0),
    _lookupGenerationCount(0),
    _generationCount(0)
  {;}

  void addRoute(const Field& commandId_, const AMPS::MessageHandler& messageHandler_,
                unsigned requestedAcks_, unsigned systemAcks_, bool isSubscribe_)
  {
    Lock<Mutex> lock(_lock);
    RouteMap::iterator i = _routes.find(commandId_);
    if (i == _routes.end()) {
        _routes[commandId_.deepCopy()] = MessageRoute(messageHandler_, requestedAcks_, systemAcks_, isSubscribe_);
    }
    else {
        i->second = MessageRoute(messageHandler_, requestedAcks_, systemAcks_, isSubscribe_);
    }
  }

  // returns true if a route was removed.
  bool removeRoute(const Field& commandId_)
  {
    Lock<Mutex> lock(_lock);
    RouteMap::iterator i = _routes.find(commandId_);
    if (i == _routes.end()) return false;
    return _removeRoute(i);
  }

  void clear()
  {
    AMPS_FETCH_ADD(&_generationCount,1);
    Lock<Mutex> lock(_lock);
    for (RouteMap::iterator i = _routes.begin(); i != _routes.end(); ++i)
    {
        // Make a non-const copy of Field and clear it, which will clear i as well
        // but won't actually affect the map, which is unaware that the key's shared
        // pointer has been deleted.
        Field f = i->first;
        f.clear();
    }
    _routes.clear();
  }

  // Returns true if a route exists for a single id.
  bool hasRoute(const Field& commandId_) const
  {
    Lock<Mutex> lock(_lock);
    RouteMap::const_iterator it = _routes.find(commandId_);
    return it != _routes.end();
  }

  // Find a single route and return true if here, setting result_ to the handler.
  bool getRoute(const Field& commandId_, MessageHandler& result_) const
  {
    Lock<Mutex> lock(_lock);
    RouteMap::const_iterator it = _routes.find(commandId_);
    if (it != _routes.end())
    {
      result_ = it->second.getMessageHandler();
      return true;
    }
    else
    {
      result_ = _emptyMessageHandler;
      return false;
    }
  }

  // RouteCache is the result type for a parseRoutes(); we do extra work
  //   to avoid hitting the map or its lock when the subids field on
  //   publish messages does not change.
  struct RouteLookup
  {
    size_t         idOffset;
    size_t         idLength;
    MessageHandler handler;
  };
  class RouteCache : public std::vector<RouteLookup>
  {
    RouteCache(const RouteCache&);
    void operator=(const RouteCache&);
    public:
      RouteCache(void)
        : _generationCount(0),
          _hashVal(0)
      {;}

      void invalidateCache(void)
      {
        _generationCount = 0;
        _hashVal = 0;
        clear();
      }
      void invalidateCache(ATOMIC_TYPE generationCount_, size_t hashVal_)
      {
        _generationCount = generationCount_;
        _hashVal = hashVal_;
        clear();
      }

      bool isCacheHit(ATOMIC_TYPE generationCount_, size_t hashVal_) const
      {
        return _generationCount == generationCount_ && _hashVal == hashVal_;
      }

    private:
      ATOMIC_TYPE _generationCount;
      size_t _hashVal;
  };

  // Parses the command id list into the route lookup vector and assigns
  // the found handlers into the list. Only intended to be called by the
  // message handler thread. Returns the number of command/sub IDs parsed.
  size_t parseRoutes(const Field& commandIdList_, RouteCache& result_)
  {
    // Super shortcut: if the whole subID list is the same as the previous one,
    // then assume the result_ contains all the right handlers already, and that
    // the offsets and lengths of subIds are unchanged.
    size_t listHash = commandIdList_.hash_function();
    if (result_.isCacheHit(_generationCount, listHash))
    {
      return result_.size();
    }
    result_.invalidateCache(_generationCount, listHash);

    // Lock required now that we'll be using the route map.
    Lock<Mutex> lockGuard(_lock);
    size_t resultCount = 0;
    const char* pStart = commandIdList_.data();
    for(const char* p = pStart, *e = commandIdList_.len() + pStart; p < e;
                    ++p, ++resultCount)
    {
      const char* delimiter = p;
      while(delimiter != e && *delimiter != ',')
      {
        ++delimiter;
      }
      AMPS::Field subId(p, (size_t)(delimiter-p));
      result_.push_back(RouteLookup());
      // Push back and then copy over fields; would emplace_back if available on
      // all supported compilers.
      RouteLookup& result = result_[resultCount];
      result.idOffset = (size_t)(p - pStart);
      result.idLength = (size_t)(delimiter - p);

      RouteMap::const_iterator it = _routes.find(subId);
      if (it != _routes.end())
      {
        result.handler = it->second.getMessageHandler();
      }
      else
      {
        result.handler = _emptyMessageHandler;
      }
      p = delimiter;
    }
    return resultCount;
  }
  unsigned deliverAck(const Message& ackMessage_, unsigned ackType_)
  {
    assert(ackMessage_.getCommand() == "ack");
    unsigned messagesDelivered = 0;
    Field key;

    // Call _deliverAck, which will deliver to any waiting handlers
    // AND remove the route if it's a termination ack
    if(key = ackMessage_.getCommandId(), !key.empty())
    {
      messagesDelivered += _deliverAck(ackMessage_, ackType_, key);
    }
    if(key = ackMessage_.getQueryID(),
       !key.empty() && messagesDelivered == 0)
    {
        messagesDelivered += _deliverAck(ackMessage_, ackType_, key);
    }
    if(key = ackMessage_.getSubscriptionId(),
       !key.empty() && messagesDelivered == 0)
    {
        messagesDelivered += _deliverAck(ackMessage_, ackType_, key);
    }
    return messagesDelivered;
  }

  // deliverData may only be called by the message handler thread.
  unsigned deliverData(const Message& dataMessage_, const Field& commandId_)
  {
    unsigned messagesDelivered = 0;
    size_t hval = commandId_.hash_function();
    if(_previousCommandId == hval &&
       _lookupGenerationCount == _generationCount)
    {
      messagesDelivered += _previousHandler.deliverData(dataMessage_);
    }
    else
    {
      Lock<Mutex> lock(_lock);
      RouteMap::iterator it = _routes.find(commandId_);
      if(it != _routes.end())
      {
        _previousCommandId = hval;
        _lookupGenerationCount = _generationCount;
        _previousHandler = it->second;
        messagesDelivered += it->second.deliverData(dataMessage_);
      }
    }
    return messagesDelivered;
  }

  void invalidateCache(void)
  {
      _previousCommandId = 0;
  }

  void unsubscribeAll(void)
  {
    AMPS_FETCH_ADD(&_generationCount,1);
    std::vector<Field> removeIds;
    Lock<Mutex> lock(_lock);
    for (RouteMap::iterator it = _routes.begin(); it != _routes.end(); ++it)
    {
        if (it->second.isTerminationAck(0)) removeIds.push_back(it->first);
    }
    for (size_t i=0; i<removeIds.size(); ++i)
    {
        RouteMap::iterator it = _routes.find(removeIds[i]);
        // Make a non-const copy of Field and clear it, which will clear i as well
        Field f = it->first;
        f.clear();
        _routes.erase(it);
    }
  }

private:
  typedef std::map<Field, MessageRoute> RouteMap;
  RouteMap _routes;
  mutable Mutex _lock;

  MessageRoute          _previousHandler;
  size_t                _previousCommandId;
  mutable ATOMIC_TYPE   _lookupGenerationCount;
  mutable ATOMIC_TYPE   _generationCount;

  // Deliver the ack to any waiting handlers
  // AND remove the route if it's a termination ack
  unsigned _deliverAck(const Message& ackMessage_, unsigned ackType_, Field& commandId_)
  {
    Lock<Mutex> lock(_lock);
    unsigned messagesDelivered = 0;
    RouteMap::iterator it = _routes.find(commandId_);
    if(it!=_routes.end())
    {
      MessageRoute& route = it->second;
      messagesDelivered += route.deliverAck(ackMessage_, ackType_);
      if(route.isTerminationAck(ackType_))
      {
        _removeRoute(it);
        ++messagesDelivered;
      }
    }
    return messagesDelivered;
  }
  unsigned _processAckForRemoval(unsigned ackType_, Field& commandId_)
  {
    Lock<Mutex> lock(_lock);
    RouteMap::iterator it = _routes.find(commandId_);
    if(it!=_routes.end())
    {
      MessageRoute& route = it->second;
      if(route.isTerminationAck(ackType_))
      {
        _removeRoute(it);
        return 1U;
      }
    }
    return 0U;
  }

  // returns true if a route was removed.
  bool _removeRoute(RouteMap::iterator& it_)
  {
    // Called with lock already held
    AMPS_FETCH_ADD(&_generationCount,1);
    // Make a non-const copy of Field and clear it, which will clear i as well
    Field f = it_->first;
    f.clear();
    _routes.erase(it_);
    return true;
  }

};


}

#endif