content: /*
   * Copyright 1999-2018 Alibaba Group Holding Ltd.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.alibaba.nacos.common.notify;
  
  import com.alibaba.nacos.common.notify.listener.Subscriber;
  import com.alibaba.nacos.common.utils.ConcurrentHashSet;
  
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  
  /**
   * The default share event publisher implementation for slow event.
   *
   * @author zongtanghu
   */
  public class DefaultSharePublisher extends DefaultPublisher implements ShardedEventPublisher {
      
      private final Map<Class<? extends SlowEvent>, Set<Subscriber>> subMappings = new ConcurrentHashMap<>();
      
      private final Lock lock = new ReentrantLock();
      
      @Override
      public void addSubscriber(Subscriber subscriber, Class<? extends Event> subscribeType) {
          // Actually, do a classification based on the slowEvent type.
          Class<? extends SlowEvent> subSlowEventType = (Class<? extends SlowEvent>) subscribeType;
          // For stop waiting subscriber, see {@link DefaultPublisher#openEventHandler}.
          subscribers.add(subscriber);
          
          lock.lock();
          try {
              Set<Subscriber> sets = subMappings.get(subSlowEventType);
              if (sets == null) {
                  Set<Subscriber> newSet = new ConcurrentHashSet<>();
                  newSet.add(subscriber);
                  subMappings.put(subSlowEventType, newSet);
                  return;
              }
              sets.add(subscriber);
          } finally {
              lock.unlock();
          }
      }
      
      @Override
      public void removeSubscriber(Subscriber subscriber, Class<? extends Event> subscribeType) {
          // Actually, do a classification based on the slowEvent type.
          Class<? extends SlowEvent> subSlowEventType = (Class<? extends SlowEvent>) subscribeType;
          // For removing to parent class attributes synchronization.
          subscribers.remove(subscriber);
          
          lock.lock();
          try {
              Set<Subscriber> sets = subMappings.get(subSlowEventType);
              
              if (sets != null) {
                  sets.remove(subscriber);
              }
          } finally {
              lock.unlock();
          }
      }
      
      @Override
      public void receiveEvent(Event event) {
          
          final long currentEventSequence = event.sequence();
          // get subscriber set based on the slow EventType.
          final Class<? extends SlowEvent> slowEventType = (Class<? extends SlowEvent>) event.getClass();
          
          // Get for Map, the algorithm is O(1).
          Set<Subscriber> subscribers = subMappings.get(slowEventType);
          if (null == subscribers) {
              LOGGER.debug("[NotifyCenter] No subscribers for slow event {}", slowEventType.getName());
              return;
          }
          
          // Notification single event subscriber
          for (Subscriber subscriber : subscribers) {
              // Whether to ignore expiration events
              if (subscriber.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {
                  LOGGER.debug("[NotifyCenter] the {} is unacceptable to this subscriber, because had expire",
                          event.getClass());
                  continue;
              }
              
              // Notify single subscriber for slow event.
              notifySubscriber(subscriber, event);
          }
      }
  }
  