// AddListener adds a channel of ListenerEvents on behalf of a listener.
// The listener channel must be buffered.
338262
func (cb *Breaker) AddListener(listener chan ListenerEvent) {
	cb.listeners = append(cb.listeners, listener)
}
846817
func (state *ServicesState) AddListener(listener Listener) {
	if listener.Chan() == nil {
		log.Errorf("Refusing to add listener %s with nil channel!", listener.Name())
		return
	}

	if cap(listener.Chan()) < 1 {
		log.Errorf("Refusing to add blocking channel as listener: %s", listener.Name())
		return
	}
	state.Lock()
	defer state.Unlock()

	state.listeners[listener.Name()] = listener
	log.Debugf("AddListener(): added %s, new count %d", listener.Name(), len(state.listeners))
}
1163828
func (_m *DBClient) AddEvent(e models.Event) (string, error) {
	ret := _m.Called(e)

	var r0 string
	if rf, ok := ret.Get(0).(func(models.Event) string); ok {
		r0 = rf(e)
	} else {
		r0 = ret.Get(0).(string)
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(models.Event) error); ok {
		r1 = rf(e)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}
1328139
func (s *ModifyListenerInput) SetListenerArn(v string) *ModifyListenerInput {
	s.ListenerArn = &v
	return s
}
1312514
func NewListener(inner net.Listener, config *Config) net.Listener {
	l := new(listener)
	l.Listener = inner
	l.config = config
	return l
}
121968
func (s *Server) AddAppListener(listener AppListener) {
	s.appListeners = append(s.appListeners, listener)
}
1155450
func (s *CreateLoadBalancerListenersInput) SetListeners(v []*Listener) *CreateLoadBalancerListenersInput {
	s.Listeners = v
	return s
}
934945
func (s *LoadBalancerDescription) SetListeners(v []*Listener) *LoadBalancerDescription {
	s.Listeners = v
	return s
}
422717
func (rtm *Client) AddEventListener(event Event, handler EventListener) {
	rtm.EventListeners[event] = append(rtm.EventListeners[event], handler)
}
856168
func (sd *etcdServiceDiscovery) AddListener(listener SDListener) {
	sd.listeners = append(sd.listeners, listener)
}
245184
func (s *ListenerDescription) SetListener(v *Listener) *ListenerDescription {
	s.Listener = v
	return s
}
921745
func (s *UpdateListenerOutput) SetListener(v *Listener) *UpdateListenerOutput {
	s.Listener = v
	return s
}
832898
func (s *XDSServer) AddListener(name string, kind policy.L7ParserType, port uint16, isIngress bool, wg *completion.WaitGroup) {
	log.Debugf("Envoy: %s AddListener %s", kind, name)

	s.mutex.Lock()
	listener := s.listeners[name]
	if listener == nil {
		listener = &Listener{}
		s.listeners[name] = listener
	}
	listener.count++
	listener.mutex.Lock() // needed for other than 'count'
	if listener.count > 1 && !listener.nacked {
		log.Debugf("Envoy: Reusing listener: %s", name)
		if !listener.acked {
			// Listener not acked yet, add a completion to the waiter's list
			log.Debugf("Envoy: Waiting for a non-acknowledged reused listener: %s", name)
			listener.waiters = append(listener.waiters, wg.AddCompletion())
		}
		listener.mutex.Unlock()
		s.mutex.Unlock()
		return
	}
	// Try again after a NACK, potentially with a different port number, etc.
	if listener.nacked {
		listener.acked = false
		listener.nacked = false
	}
	listener.mutex.Unlock() // Listener locked again in callbacks below

	clusterName := egressClusterName
	if isIngress {
		clusterName = ingressClusterName
	}

	// Fill in the listener-specific parts.
	listenerConf := proto.Clone(s.listenerProto).(*envoy_api_v2.Listener)
	if kind == policy.ParserTypeHTTP {
		listenerConf.FilterChains = append(listenerConf.FilterChains, proto.Clone(s.httpFilterChainProto).(*envoy_api_v2_listener.FilterChain))
		// listenerConf.FilterChains[0].Filters[1].ConfigType.(*envoy_api_v2_listener.Filter_Config).Config.Fields["http_filters"].GetListValue().Values[0].GetStructValue().Fields["config"].GetStructValue().Fields["policy_name"] = &structpb.Value{Kind: &structpb.Value_StringValue{StringValue: endpointPolicyName}}
		routes := listenerConf.FilterChains[0].Filters[1].ConfigType.(*envoy_api_v2_listener.Filter_Config).Config.Fields["route_config"].GetStructValue().Fields["virtual_hosts"].GetListValue().Values[0].GetStructValue().Fields["routes"].GetListValue().Values
		routes[0].GetStructValue().Fields["route"].GetStructValue().Fields["cluster"] = &structpb.Value{Kind: &structpb.Value_StringValue{StringValue: clusterName}}
		routes[1].GetStructValue().Fields["route"].GetStructValue().Fields["cluster"] = &structpb.Value{Kind: &structpb.Value_StringValue{StringValue: clusterName}}
	} else {
		listenerConf.FilterChains = append(listenerConf.FilterChains, proto.Clone(s.tcpFilterChainProto).(*envoy_api_v2_listener.FilterChain))
		// listenerConf.FilterChains[0].Filters[0].ConfigType.(*envoy_api_v2_listener.Filter_Config).Config.Fields["policy_name"] = &structpb.Value{Kind: &structpb.Value_StringValue{StringValue: endpointPolicyName}}
		// listenerConf.FilterChains[0].Filters[0].ConfigType.(*envoy_api_v2_listener.Filter_Config).Config.Fields["l7_proto"] = &structpb.Value{Kind: &structpb.Value_StringValue{StringValue: kind.String()}}
		listenerConf.FilterChains[0].Filters[1].ConfigType.(*envoy_api_v2_listener.Filter_Config).Config.Fields["cluster"] = &structpb.Value{Kind: &structpb.Value_StringValue{StringValue: clusterName}}
	}

	listenerConf.Name = name
	listenerConf.Address.GetSocketAddress().PortSpecifier = &envoy_api_v2_core.SocketAddress_PortValue{PortValue: uint32(port)}
	if isIngress {
		listenerConf.ListenerFilters[0].ConfigType.(*envoy_api_v2_listener.ListenerFilter_Config).Config.Fields["is_ingress"].GetKind().(*structpb.Value_BoolValue).BoolValue = true
	}

	s.listenerMutator.Upsert(ListenerTypeURL, name, listenerConf, []string{"127.0.0.1"},
		wg.AddCompletionWithCallback(func(err error) {
			// listener might have already been removed, so we can't look again
			// but we still need to complete all the completions in case
			// someone is still waiting!
			listener.mutex.Lock()
			if err == nil {
				// Allow future users to not need to wait
				listener.acked = true
			} else {
				// Prevent further reuse of a failed listener
				listener.nacked = true
			}
			// Pass the completion result to all the additional waiters.
			for _, waiter := range listener.waiters {
				waiter.Complete(err)
			}
			listener.waiters = nil
			listener.mutex.Unlock()
		}))
	s.mutex.Unlock()
}
1294235
func NewListener(listener net.Listener, allowedIPs string, headerReadTimeout int) (net.Listener, error) {
	ppl, err := newListener(listener, allowedIPs, headerReadTimeout)
	if err == nil {
		go ppl.acceptLoop()
	}
	return ppl, err
}
1336754
func (dis *dispatcher) RegisterListener(listenerObj core.EventListener, keys ...string) error {
	if listenerObj == nil {
		err := errors.New("nil listener")
		openlogging.GetLogger().Error("nil listener supplied:" + err.Error())
		return errors.New("nil listener")
	}

	for _, key := range keys {
		listenerList, ok := dis.listeners[key]
		if !ok {
			listenerList = make([]core.EventListener, 0)
		}

		// for duplicate registration
		for _, listener := range listenerList {
			if listener == listenerObj {
				return nil
			}
		}

		// append new listener
		listenerList = append(listenerList, listenerObj)

		// assign latest listener list
		dis.listeners[key] = listenerList
	}
	return nil
}
494447
func WithListener(l net.Listener) JSONSourceOption {
	return func(j *JSONSource) {
		j.listener = l
		j.addr = l.Addr().String()
	}
}
487581
func (s *DescribeListenerOutput) SetListener(v *Listener) *DescribeListenerOutput {
	s.Listener = v
	return s
}