// Dispatch events to handlers in async mode
493419
func (self *FSock) dispatchEvent(event string) {
	eventName := headerVal(event, "Event-Name")
	if eventName == "BACKGROUND_JOB" { // for bgapi BACKGROUND_JOB
		go self.doBackgroudJob(event)
		return
	}

	if eventName == "CUSTOM" {
		eventSubclass := headerVal(event, "Event-Subclass")
		if len(eventSubclass) != 0 {
			eventName += " " + urlDecode(eventSubclass)
		}
	}

	for _, handleName := range []string{eventName, "ALL"} {
		if _, hasHandlers := self.eventHandlers[handleName]; hasHandlers {
			// We have handlers, dispatch to all of them
			for _, handlerFunc := range self.eventHandlers[handleName] {
				go handlerFunc(event, self.connId)
			}
			return
		}
	}
	if self.logger != nil {
		fmt.Printf("No dispatcher, event name: %s, handlers: %+v\n", eventName, self.eventHandlers)
		self.logger.Warning(fmt.Sprintf("<FSock> No dispatcher for event: <%+v>", event))
	}
}
328674
func (e *EventServer) processEvent(ctx context.Context, eventInfo *common.EventInfo) (err error) {

	if e.registerer == nil {
		return fmt.Errorf("No registered handlers")
	}

	f, err := e.registerer.GetHandler(eventInfo.PUType, eventInfo.EventType)
	if err != nil {
		return fmt.Errorf("Handler not found: %s", err)
	}

	return f(ctx, eventInfo)
}
1315426
func (plugin *ExamplePlugin) asyncEventHandler() {
	plugin.Log.Info("Started Kafka async event handler...")
	msgCounter := 0
	asyncMsgSucc := 0
	if messageCountNum == 0 {
		// Set as done
		plugin.asyncSuccess = true
		plugin.asyncRecv = true
	}
	for {
		select {
		// Channel subscribed with watcher
		case message := <-plugin.asyncSubscription:
			plugin.Log.Infof("Received async Kafka Message, topic '%s', partition '%v', offset '%v', key: '%s', ",
				message.GetTopic(), message.GetPartition(), message.GetOffset(), message.GetKey())
			// mark the offset
			plugin.kafkaWatcher.MarkOffset(message, "")
			msgCounter++
			if msgCounter == messageCountNum {
				plugin.asyncRecv = true
			}
		case message := <-plugin.asyncSuccessChannel:
			plugin.Log.Infof("Async message successfully delivered, topic '%s', partition '%v', offset '%v', key: '%s', ",
				message.GetTopic(), message.GetPartition(), message.GetOffset(), message.GetKey())
			asyncMsgSucc++
			if asyncMsgSucc == messageCountNum {
				plugin.asyncSuccess = true
			}
			// Error callback channel
		case err := <-plugin.asyncErrorChannel:
			plugin.Log.Errorf("Failed to publish async message, %v", err)
		}
	}
}
1255148
func WithEventHandler(handler events.Handler) Option {
	return func(c *Config) Option {
		old := c.handler
		c.handler = handler
		return WithEventHandler(old)
	}
}
1358702
func (c *Client) callOnDisconnectHandlers() {
	c.m.RLock()
	defer c.m.RUnlock()

	for _, handler := range c.onDisconnectHandlers {
		func() {
			defer nopRecover()
			handler()
		}()
	}
}
1174486
func (a *AsyncEventEmitter) EmitEvent(event Event) {
	a.listenersLock.RLock()
	for _, listener := range a.listeners {
		go listener.HandleEvent(event)
	}
	a.listenersLock.RUnlock()
}
483124
func (ed *Dispatcher) HandleConnectEvent(e esdispatcher.Event) {
	evt := e.(*ConnectEvent)

	if ed.connection != nil {
		// Already connected. No error.
		evt.ErrCh <- nil
		return
	}

	eventch, err := ed.EventCh()
	if err != nil {
		evt.ErrCh <- err
		return
	}

	peers, err := ed.discoveryService.GetPeers()
	if err != nil {
		evt.ErrCh <- err
		return
	}

	if len(peers) == 0 {
		evt.ErrCh <- errors.New("no peers to connect to")
		return
	}

	peer, err := ed.peerResolver.Resolve(peers)
	if err != nil {
		evt.ErrCh <- err
		return
	}

	conn, err := ed.connectionProvider(ed.context, ed.chConfig, peer)
	if err != nil {
		logger.Warnf("error creating connection: %s", err)
		evt.ErrCh <- errors.WithMessagef(err, "could not create client conn")
		return
	}

	ed.connection = conn
	ed.setConnectedPeer(peer)

	go ed.connection.Receive(eventch)

	evt.ErrCh <- nil
}
493904
func (c *Client) SendEvent(event Event, opts ...interface{}) (res StatusMessageRes, err error) {
	return c.SendEvents([]Event{event}, opts...)
}
71955
func (ed *Dispatcher) EventCh() (chan<- interface{}, error) {
	state := ed.getState()
	if state == dispatcherStateStarted {
		return ed.eventch, nil
	}
	return nil, errors.Errorf("dispatcher not started - Current state [%d]", state)
}
1029582
func (v *version) Events() EventInformer {
	return &eventInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
}
1307078
func (self *RawDispatcher) Dispatch(mux *ServeMux, w http.ResponseWriter, r *http.Request) *ServeMux {
	//check the prefix for sanity
	pre := self.Prefix + "/"
	path := r.URL.Path
	if strings.HasSuffix(path, "/") && path != "/" {
		path = path[0 : len(path)-1]
	}
	if self.Prefix != "" {
		if !strings.HasPrefix(path, pre) {
			panic(fmt.Sprintf("expected prefix %s on the URL path but not found on: %s", pre, path))
		}
		path = path[len(pre):]
	}
	parts := strings.Split(path, "/")
	bundle, err := self.IO.BundleHook(w, r, self.SessionMgr)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to create parameter bundle:%s", err), http.StatusInternalServerError)
		return nil
	}
	self.DispatchSegment(mux, w, r, parts, self.Root, bundle)
	return nil
}
172796
func (d *dispatcher) Dispatch(g grim.GrimReaper, conn net.Conn) {

	// Create buffer and read connection type
	buf := make([]byte, 1)
	if _, err := conn.Read(buf); err != nil {
		d.logger.Warn("error reading StreamType",
			"remote", conn.RemoteAddr().String, "err", err)
		conn.Close()
		return
	}

	// Dispatch connection based on first byte
	ctype := StreamType(buf[0])
	if handler, ok := d.handlers[ctype]; ok {
		g.SpawnFunc(func(c context.Context) {
			handler.Handle(c, conn)
		})
		return
	}

	// Process unknown streams if there is an unknown stream handler
	if d.unknown != nil {
		g.SpawnFunc(func(c context.Context) {
			d.unknown.Handle(c, conn)
		})
		return
	}

	// If there is not an unknown stream handler, close the stream.
	d.logger.Warn("closing connection: unknown stream type",
		"type", ctype, "remote", conn.RemoteAddr().String())
	conn.Close()
}
25052
func (c *Container) Dispatch(httpWriter http.ResponseWriter, httpRequest *http.Request) {
	if httpWriter == nil {
		panic("httpWriter cannot be nil")
	}
	if httpRequest == nil {
		panic("httpRequest cannot be nil")
	}
	c.dispatch(httpWriter, httpRequest)
}
1153868
func (ed *Dispatcher) HandleDisconnectedEvent(e esdispatcher.Event) {
	evt := e.(*DisconnectedEvent)

	logger.Debugf("Disconnecting from event server: %s", evt.Err)

	if ed.connection != nil {
		ed.connection.Close()
		ed.connection = nil
	}

	if ed.connectionRegistration != nil {
		logger.Debugf("Disconnected from event server: %s", evt.Err)
		select {
		case ed.connectionRegistration.Eventch <- NewConnectionEvent(false, evt.Err):
		default:
			logger.Warn("Unable to send to connection event channel.")
		}
	} else {
		logger.Warnf("Disconnected from event server: %s", evt.Err)
	}

	if ed.peerMonitorDone != nil {
		close(ed.peerMonitorDone)
		ed.peerMonitorDone = nil
	}
}
82391
func (a *Agent) handleEvents() {
	for {
		select {
		case e := <-a.eventCh:
			// Decode the event
			msg := new(UserEvent)
			if err := decodeMsgPack(e.Payload, msg); err != nil {
				a.logger.Printf("[ERR] agent: Failed to decode event: %v", err)
				continue
			}
			msg.LTime = uint64(e.LTime)

			// Skip if we don't pass filtering
			if !a.shouldProcessUserEvent(msg) {
				continue
			}

			// Ingest the event
			a.ingestUserEvent(msg)

		case <-a.shutdownCh:
			return
		}
	}
}
281557
func (t *nonrecursiveTree) dispatch(c <-chan EventInfo) {
	for ei := range c {
		dbgprintf("dispatching %v on %q", ei.Event(), ei.Path())
		go func(ei EventInfo) {
			var nd node
			var isrec bool
			dir, base := split(ei.Path())
			fn := func(it node, isbase bool) error {
				isrec = isrec || it.Watch.IsRecursive()
				if isbase {
					nd = it
				} else {
					it.Watch.Dispatch(ei, recursive)
				}
				return nil
			}
			t.rw.RLock()
			// Notify recursive watchpoints found on the path.
			if err := t.root.WalkPath(dir, fn); err != nil {
				dbgprint("dispatch did not reach leaf:", err)
				t.rw.RUnlock()
				return
			}
			// Notify parent watchpoint.
			nd.Watch.Dispatch(ei, 0)
			isrec = isrec || nd.Watch.IsRecursive()
			// If leaf watchpoint exists, notify it.
			if nd, ok := nd.Child[base]; ok {
				isrec = isrec || nd.Watch.IsRecursive()
				nd.Watch.Dispatch(ei, 0)
			}
			t.rw.RUnlock()
			// If the event describes newly leaf directory created within
			if !isrec || ei.Event() != Create {
				return
			}
			if ok, err := ei.(isDirer).isDir(); !ok || err != nil {
				return
			}
			t.rec <- ei
		}(ei)
	}
}
495621
func (r *Renderer) RenderEvents(eventCh <-chan *TaskEvent) {
	for event := range eventCh {
		r.RenderEvent(event)
	}
}