Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2017 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include <cmn/agent_cmn.h>
       7             : #include <cmn/event_notifier.h>
       8             : 
       9             : //Event registeration handle
      10           0 : EventNotifyHandle::EventNotifyHandle(KeyPtr key, Callback cb) :
      11           0 :     key_(key), cb_(cb) {
      12           0 : }
      13             : 
      14           0 : EventNotifyHandle::~EventNotifyHandle() {
      15           0 : }
      16             : 
      17             : //Event notify manager
      18           2 : EventNotifier::EventNotifier(Agent *agent) : map_(),
      19           2 :   work_queue_(agent->task_scheduler()->GetTaskId(kEventNotifierTask), 0,
      20             :               boost::bind(&EventNotifier::Process, this, _1)),
      21           2 :   agent_(agent) {
      22           2 :       work_queue_.set_name("Event Notify Manager");
      23           2 : }
      24             : 
      25           4 : EventNotifier::~EventNotifier() {
      26           2 :     work_queue_.Shutdown();
      27           2 :     assert(map_.empty() == true);
      28           4 : }
      29             : 
      30           0 : bool EventNotifier::Enqueue(WorkQueueMessage::Ptr data) {
      31           0 :     work_queue_.Enqueue(data);
      32           0 :     return true;
      33             : }
      34             : 
      35           0 : bool EventNotifier::Process(WorkQueueMessage::Ptr data) {
      36           0 :     if (data->type_ == WorkQueueMessage::PUBLISHER) {
      37           0 :         NotifyInternal(data->handle_ptr_->key());
      38           0 :     } else if (data->type_ == WorkQueueMessage::REGISTER_SUBSCRIBER) {
      39           0 :         RegisterSubscriberInternal(data->handle_ptr_);
      40           0 :     } else if (data->type_ == WorkQueueMessage::DEREGISTER_SUBSCRIBER) {
      41           0 :         DeRegisterSubscriberInternal(data->handle_ptr_);
      42             :     }
      43           0 :     return true;
      44             : }
      45             : 
      46           0 : void EventNotifier::Notify(EventNotifyKey *key) {
      47           0 :     EventNotifyHandle::Ptr handle(new EventNotifyHandle(KeyPtr(key), Callback()));
      48             :     WorkQueueMessage::Ptr data(new WorkQueueMessage(WorkQueueMessage::PUBLISHER,
      49           0 :                                handle));
      50           0 :     work_queue_.Enqueue(data);
      51           0 : }
      52             : 
      53           0 : void EventNotifier::NotifyInternal(KeyPtr key) {
      54           0 :     EventNotifier::NotifyMapIter map_it = map_.find(key);
      55           0 :     if (map_it == map_.end()) {
      56             :         //No subscribers
      57           0 :         return;
      58             :     }
      59           0 :     EventNotifier::SubscribersList list = map_it->second;
      60           0 :     for (EventNotifier::SubscribersListIter it = list.begin();
      61           0 :          it != list.end(); it++) {
      62           0 :         (*it)->Notify();
      63             :     }
      64           0 : }
      65             : 
      66             : EventNotifyHandle::Ptr
      67           0 : EventNotifier::RegisterSubscriber(EventNotifyKey *key,
      68             :                                        Callback cb) {
      69           0 :     EventNotifyHandle::Ptr handle(new EventNotifyHandle(KeyPtr(key), cb));
      70             :     WorkQueueMessage::Ptr data(new WorkQueueMessage
      71           0 :                                (WorkQueueMessage::REGISTER_SUBSCRIBER, handle));
      72           0 :     work_queue_.Enqueue(data);
      73           0 :     return handle;
      74           0 : }
      75             : 
      76             : void
      77           0 : EventNotifier::RegisterSubscriberInternal(EventNotifyHandle::Ptr handle) {
      78             :     //It is not ensured that same subscriber is not registered multiple times.
      79           0 :     map_[handle->key()].push_back(handle);
      80           0 : }
      81             : 
      82           0 : void EventNotifier::DeregisterSubscriber(EventNotifyHandle::Ptr handle) {
      83             :     WorkQueueMessage::Ptr data(new WorkQueueMessage
      84           0 :                                (WorkQueueMessage::DEREGISTER_SUBSCRIBER, handle));
      85           0 :     work_queue_.Enqueue(data);
      86           0 : }
      87             : 
      88             : void
      89           0 : EventNotifier::DeRegisterSubscriberInternal(EventNotifyHandle::Ptr handle) {
      90           0 :     if (handle == NULL)
      91           0 :         return;
      92             : 
      93           0 :     KeyPtr key_ptr = static_cast<KeyPtr>(handle->key());
      94           0 :     EventNotifier::NotifyMapIter map_it = map_.find(key_ptr);
      95           0 :     if (map_it == map_.end()) {
      96           0 :         return;
      97             :     }
      98             : 
      99             :     EventNotifier::SubscribersListIter it =
     100           0 :         std::find(map_it->second.begin(), map_it->second.end(), handle);
     101           0 :     if (it == map_it->second.end()) {
     102           0 :         return;
     103             :     }
     104           0 :     map_it->second.erase(it);
     105             : 
     106           0 :     if (map_it->second.empty()) {
     107           0 :         map_.erase(map_it);
     108             :     }
     109             : 
     110           0 :     return;
     111           0 : }
     112             : 
     113           0 : EventNotifier::WorkQueueMessage::WorkQueueMessage(Type type,
     114           0 :                                           EventNotifyHandle::Ptr ptr) :
     115           0 :     type_(type), handle_ptr_(ptr) {
     116           0 : }
     117             : 
     118             : //EventNotifyKey
     119           0 : EventNotifyKey::EventNotifyKey(Type type) : type_(type) {
     120           0 :     ref_count_ = 0;
     121           0 : }
     122             : 
     123           0 : EventNotifyKey::~EventNotifyKey() {
     124           0 : }
     125             : 
     126           0 : bool EventNotifyKey::IsLess(const EventNotifyKey &rhs) const {
     127           0 :     if (type() != rhs.type()) {
     128           0 :         return (type() < rhs.type());
     129             :     }
     130           0 :     return false;
     131             : }