// EnableEventListener watches for docker events. Performs the plumbing for the
// containers started or dead.
503613
func (d *dockerClient) EnableEventListener() (chan<- *EventMessage, error) {
	if d == nil {
		log.Debug("Not enabling docker event listener because dockerClient is nil")
		return nil, nil
	}
	log.Info("Enabling docker event listener")

	ws := newWatcherState()

	since := time.Now()
	eo := dTypes.EventsOptions{Since: strconv.FormatInt(since.Unix(), 10)}
	r, err := d.Events(ctx.Background(), eo)

	go d.listenForDockerEvents(ws, r, err)

	log.Debug("Started to listen for docker events")
	return nil, nil
}
702503
func (t *NotificationConfig) AddEvents(events ...NotificationEventType) {
	t.Events = append(t.Events, events...)
}
831803
func (c *ClusterManager) AddEventListener(listener cluster.ClusterListener) error {
	logrus.Printf("Adding cluster event listener: %s", listener.String())
	c.listeners.PushBack(listener)
	return nil
}
1103977
func (e EventListener) Event(event *core.Event) {

	configValue := ConfigFactory.GetConfigurationByKey(event.Key)
	openlogging.GetLogger().Infof("config value ", event.Key, " | ", configValue)
}
928504
func (remote *RemoteDebugger) DomainEvents(domain string, enable bool) error {
	method := domain

	if enable {
		method += ".enable"
	} else {
		method += ".disable"
	}

	_, err := remote.SendRequest(method, nil)
	return err
}
671260
func (b *serverBuilder) GetEventHandler(event ServerBuildEvents) *ServerBuildEventHandler {
	if handler, ok := b.buildEventHandlers[event]; ok {
		return &handler
	}
	return nil
}
529894
func (b *serverBuilder) SetEventHandler(event ServerBuildEvents, handler ServerBuildEventHandler) {
	b.buildEventHandlers[event] = handler
}
671672
func (d *DockerMonitor) handleStartEvent(ctx context.Context, event *events.Message) error {

	container, err := d.retrieveDockerInfo(ctx, event)
	if err != nil {
		return err
	}

	if !container.State.Running {
		return nil
	}

	puID, err := puIDFromDockerID(container.ID)
	if err != nil {
		return err
	}

	runtime, err := d.extractMetadata(container)
	if err != nil {
		return err
	}

	// If it is a host container, we need to activate it as a Linux process. We will
	// override the options that the metadata extractor provided.
	if container.HostConfig.NetworkMode == constants.DockerHostMode {
		options := hostModeOptions(container)
		options.PolicyExtensions = runtime.Options().PolicyExtensions
		runtime.SetOptions(*options)
		runtime.SetPUType(common.LinuxProcessPU)
	}

	runtime.SetOptions(runtime.Options())

	if err = d.config.Policy.HandlePUEvent(ctx, puID, tevents.EventStart, runtime); err != nil {
		if d.killContainerOnPolicyError {
			timeout := 0 * time.Second
			if err1 := d.dockerClient.ContainerStop(ctx, event.ID, &timeout); err1 != nil {
				zap.L().Warn("Unable to stop illegal container",
					zap.String("dockerID", event.ID),
					zap.Error(err1),
				)
			}
			d.config.Collector.CollectContainerEvent(&collector.ContainerRecord{
				ContextID: event.ID,
				IPAddress: nil,
				Tags:      nil,
				Event:     collector.ContainerFailed,
			})
			return fmt.Errorf("unable to start container because of policy: container %s killed: %s", event.ID, err)
		}
		return fmt.Errorf("unable to set policy: container %s kept alive per policy: %s", puID, err)
	}

	// if the container has hostnet set to true or is linked
	// to container with hostnet set to true, program the cgroup.
	if isHostNetworkContainer(runtime) {
		if err = d.setupHostMode(puID, runtime, container); err != nil {
			return fmt.Errorf("unable to setup host mode for container %s: %s", puID, err)
		}
	}
	return nil
}
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()
}
960290
func (s *QueueConfigurationDeprecated) SetEvent(v string) *QueueConfigurationDeprecated {
	s.Event = &v
	return s
}
1142973
func (r *marathonClient) AddEventsListener(filter int) (EventsChannel, error) {
	r.Lock()
	defer r.Unlock()

	// step: someone has asked to start listening to event, we need to register for events
	// if we haven't done so already
	if err := r.registerSubscription(); err != nil {
		return nil, err
	}

	channel := make(EventsChannel)
	r.listeners[channel] = EventsChannelContext{
		filter:     filter,
		done:       make(chan struct{}, 1),
		completion: &sync.WaitGroup{},
	}
	return channel, nil
}
1064011
func (b *DiskBuilder) SetEventHandler(event DiskBuildEvents, handler DiskBuildEventHandler) {
	b.buildEventHandlers[event] = handler
}
856168
func (sd *etcdServiceDiscovery) AddListener(listener SDListener) {
	sd.listeners = append(sd.listeners, listener)
}
984378
func listenAndDispatch(c *Client, em *clientEventMonitor, r *io.PipeReader, w *io.PipeWriter) {
	listener := make(chan *dockerclient.APIEvents)
	c.dc.AddEventListener(listener)
	for {
		evt := <-listener
		if evt != nil {
			em.dispatch(evt)
		}
	}
}
666580
func (d *DockerDiscovery) listenerForContainer(cntnr *docker.Container) *ChangeListener {
	// See if the container has the SidecarListener label, which
	// will tell us the ServicePort of the port that should be
	// subscribed to Sidecar events.
	svcPortStr, ok := cntnr.Config.Labels["SidecarListener"]
	if !ok {
		return nil
	}

	// Be careful about ID matching
	id := cntnr.ID
	if len(id) > 12 {
		id = id[:12]
	}

	svc := d.findServiceByID(id)
	if svc == nil {
		return nil
	}

	listenPort := portForServicePort(svc, svcPortStr, "tcp") // We only do HTTP (TCP)
	// -1 is returned when there is no match
	if listenPort == nil {
		log.Warnf(
			"SidecarListener label found on %s, but no matching ServicePort! '%s'",
			svc.ID, svcPortStr,
		)
		return nil
	}

	return &ChangeListener{
		Name: svc.ListenerName(),
		Url:  fmt.Sprintf("http://%s:%d/sidecar/update", listenPort.IP, listenPort.Port),
	}
}
634810
func EnableEventListener() (eventsCh chan<- *EventMessage, err error) {
	if Client() == nil {
		return nil, nil
	}
	return Client().EnableEventListener()
}
424208
func (d *eventDispatcher) RemoveEventListener(typ string, listener EventListener) {
	d.Lock()
	defer d.Unlock()

	// Grab a reference to the function pointer once.
	ptr := reflect.ValueOf(listener).Pointer()

	// Find listener by pointer and remove it.
	listeners := d.listeners[typ]
	for i, l := range listeners {
		if reflect.ValueOf(l).Pointer() == ptr {
			d.listeners[typ] = append(listeners[:i], listeners[i+1:]...)
		}
	}
}