订阅接口级配置

@param config   provider/consumer config
@param listener config listener
8597
protected void subscribeConfig(final AbstractInterfaceConfig config, ConfigListener listener) {
        try {
            if (configObserver == null) { // 初始化
                configObserver = new ZookeeperConfigObserver();
            }
            configObserver.addConfigListener(config, listener);
            final String configPath = buildConfigPath(rootPath, config);
            // 监听配置节点下 子节点增加、子节点删除、子节点Data修改事件
            PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, configPath, true);
            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client1, PathChildrenCacheEvent event) throws Exception {
                    if (LOGGER.isDebugEnabled(config.getAppName())) {
                        LOGGER.debug("Receive zookeeper event: " + "type=[" + event.getType() + "]");
                    }
                    switch (event.getType()) {
                        case CHILD_ADDED: //新增接口级配置
                            configObserver.addConfig(config, configPath, event.getData());
                            break;
                        case CHILD_REMOVED: //删除接口级配置
                            configObserver.removeConfig(config, configPath, event.getData());
                            break;
                        case CHILD_UPDATED:// 更新接口级配置
                            configObserver.updateConfig(config, configPath, event.getData());
                            break;
                        default:
                            break;
                    }
                }
            });
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            INTERFACE_CONFIG_CACHE.put(configPath, pathChildrenCache);
            configObserver.updateConfigAll(config, configPath, pathChildrenCache.getCurrentData());
        } catch (Exception e) {
            throw new SofaRpcRuntimeException("Failed to subscribe provider config from zookeeperRegistry!", e);
        }
    }
510705
protected ClientTransportConfig providerToClientConfig(ProviderInfo providerInfo) {
        return new ClientTransportConfig()
            .setConsumerConfig(consumerConfig)
            .setProviderInfo(providerInfo)
            .setContainer(consumerConfig.getProtocol())
            .setConnectTimeout(consumerConfig.getConnectTimeout())
            .setInvokeTimeout(consumerConfig.getTimeout())
            .setDisconnectTimeout(consumerConfig.getDisconnectTimeout())
            .setConnectionNum(consumerConfig.getConnectionNum())
            .setChannelListeners(consumerConfig.getOnConnect());
    }
903890
public synchronized void addServletReferenceListener(ServletReferenceListener listener) {
        if (this.cacheWrappers == null) {
            cacheWrappers = new ArrayList();
        }
        this.cacheWrappers.add(listener);
    }
1283217
public void initialize(Context context, ResponseListener listener) {
        this.context = context;
        this.listener = (listener != null) ? listener : new ResponseListener() {
            final String message = "ResponseListener is not specified. Defaulting to empty listener.";

            @Override
            public void onSuccess(Response response) {
                logger.debug(message);
            }

            @Override
            public void onFailure(Response response, Throwable t, JSONObject extendedInfo) {
                logger.debug(message);
            }
        };

        logger.debug("AuthorizationRequestAgent is initialized.");
    }
430707
public void removeListener(HttpListener listener)
    {
        if (listener==null)
            return;
        
        for (int l=0;l<_listeners.size();l++)
        {
            if (listener.equals(_listeners.get(l)))
            {
                _listeners.remove(l);
                removeComponent(listener);
                if (listener.isStarted())
                    try{listener.stop();}catch(InterruptedException e){log.warn(LogSupport.EXCEPTION,e);}
                listener.setHttpServer(null);
            }
        }
    }
90256
@Override
	public final void setConsumer(StreamConsumer<T> consumer) {
		checkNotNull(consumer);
		checkState(this.consumer == null, "Consumer has already been set");

		checkState(getCapabilities().contains(LATE_BINDING) || eventloop.tick() == createTick,
				LATE_BINDING_ERROR_MESSAGE, this);
		this.consumer = consumer;
		onWired();
		consumer.getAcknowledgement()
				.whenException(this::close)
				.whenComplete(acknowledgement);
	}
375905
public void addListener(ComObjectListener listener) {
        if(listener==null)
            throw new IllegalArgumentException("listener is null");
        if(listeners.contains(listener))
            throw new IllegalArgumentException("can't register the same listener twice");
        listeners.add(listener);
    }
87847
@SuppressWarnings("unchecked")
  public <T> Listener<T> onEvent(String event, Consumer listener) {
    Set<Consumer> listeners = eventListeners.computeIfAbsent(event, e -> new CopyOnWriteArraySet<>());
    listeners.add(listener);
    return new Listener<T>() {
      @Override
      public void accept(T event) {
        listener.accept(event);
      }
      @Override
      public void close() {
        listeners.remove(listener);
      }
    };
  }
448013
public Listener<Session.State> onStateChange(Consumer<Session.State> callback) {
    Listener<Session.State> listener = new Listener<Session.State>() {
      @Override
      public void accept(Session.State state) {
        callback.accept(state);
      }
      @Override
      public void close() {
        changeListeners.remove(this);
      }
    };
    changeListeners.add(listener);
    return listener;
  }
310156
public void addSipApplicationListener(String listener) {
    	if(logger.isDebugEnabled()) {
			logger.debug("addSipApplicationListener " + getName());
		}

    	sipApplicationListeners.add(listener);
    	fireContainerEvent("addSipApplicationListener", listener);

        // FIXME - add instance if already started?

    }
24021
public void registerListener(IPluginEvent listener) {
        if (pluginEventListeners1 == null) {
            pluginEventListeners1 = new ArrayList<>();
        }

        if (!pluginEventListeners1.contains(listener)) {
            pluginEventListeners1.add(listener);
        }
    }
805059
function (config, options) {
    //获取Mq的配置
    Mq.call(this, config, options);
    this.exchange = options.exchange;
    this.exchangeType = options.type,
        this.consumers = [];
    this.isAlive = false;
    this.options = options.options || {};
}
199227
public void registerConsumer( LocalMessageConsumer consumer )
    {
    	consumersLock.writeLock().lock();
    	try
    	{
    		localConsumers.add(consumer);
    	}
    	finally
    	{
    		consumersLock.writeLock().unlock();
    	}
    }
527134
public synchronized void registerSSLConfigChangeListener(SSLConfigChangeListener listener, SSLConfigChangeEvent event) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            Tr.entry(tc, "registerSSLConfigChangeListener", new Object[] { listener, event });

        List<SSLConfigChangeListener> listenerList = sslConfigListenerMap.get(event.getAlias());

        if (listenerList != null) {
            // used to hold sslconfig -> list of listener references
            listenerList.add(listener);
            sslConfigListenerMap.put(event.getAlias(), listenerList);
        } else {
            listenerList = new ArrayList<SSLConfigChangeListener>();
            listenerList.add(listener);
            sslConfigListenerMap.put(event.getAlias(), listenerList);
        }

        // used to hold listener -> listener event references
        sslConfigListenerEventMap.put(listener, event);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            Tr.exit(tc, "registerSSLConfigChangeListener");
    }
130771
protected final <T extends EventListener> void addListener(Class<T> listenerType, T listener) {
		if (this.listeners == null) {
			this.listeners = new ListenerCollection<>();
		}
		this.listeners.add(listenerType, listener);
	}
958260
public Closeable onTickEvent(final BiConsumer<BitfinexTickerSymbol, BitfinexTick> listener) {
        tickConsumers.offer(listener);
        return () -> tickConsumers.remove(listener);
    }
77818
private void notifyToListener(ProviderInfoListener listener, ComposeUserData mergedResult) {

        if ("".equalsIgnoreCase(mergedResult.getLocalZone()) ||
            DEFAULT_ZONE.equalsIgnoreCase(mergedResult.getLocalZone())) {
            listener.updateProviders(new ProviderGroup(flatComposeData(mergedResult)));
        } else {
            final Map<String, List<ProviderInfo>> zoneData = mergedResult.getZoneData();

            List<ProviderGroup> result = new ArrayList<ProviderGroup>();

            for (Map.Entry<String, List<ProviderInfo>> dataEntry : zoneData.entrySet()) {

                //localZone 的特殊放到 default 分组一份.为了在目标 zone 不可用的情况下兜底
                if (dataEntry.getKey().equalsIgnoreCase(mergedResult.getLocalZone())) {
                    result.add(new ProviderGroup(dataEntry.getValue()));
                }
                //其他 zone 的正常放
                result.add(new ProviderGroup(dataEntry.getKey(), dataEntry.getValue()));

            }

            listener.updateAllProviders(result);

        }
    }