// AsyncEvents returns a new AsyncEventStream that will buffer upto 100 events
// before applying backpressure.
180896
func AsyncEvents(e EventStream) EventStream {
	s := &asyncEventStream{
		e:      e,
		events: make(chan Event, 100),
	}
	go s.start()
	return s
}
1203008
func (s *StdoutClient) SendEvents(events map[string]event.Event) error {
	var n int
	var stats = make([]string, 0)

	for _, e := range events {
		for _, stat := range e.Stats() {

			stat = fmt.Sprintf("%s%s", s.prefix, strings.Replace(stat, "%HOST%", Hostname, 1))
			_n := n + len(stat) + 1

			if _n > UDPPayloadSize {
				// with this last event, the UDP payload would be too big
				if _, err := fmt.Fprintf(s.FD, strings.Join(stats, "\n")); err != nil {
					return err
				}
				// reset payload after flushing, and add the last event
				stats = []string{stat}
				n = len(stat)
				continue
			}

			// can fit more into the current payload
			n = _n
			stats = append(stats, stat)
		}
	}

	if len(stats) != 0 {
		if _, err := fmt.Fprintf(s.FD, strings.Join(stats, "\n")); err != nil {
			return err
		}
	}

	return nil
}
596483
func handleEvents() {
	for {
		event, ok := <-eventsIn
		if !ok {
			log.Println()
			log.Println("Stopped event handler!")
			break
		}

		bee := GetBee(event.Bee)
		(*bee).LogEvent()

		log.Println()
		log.Println("Event received:", event.Bee, "/", event.Name, "-", GetEventDescriptor(&event).Description)
		for _, v := range event.Options {
			log.Println("\tOptions:", v)
		}

		go func() {
			defer func() {
				if e := recover(); e != nil {
					log.Printf("Fatal chain event: %s %s", e, debug.Stack())
				}
			}()

			execChains(&event)
		}()
	}
}
600428
func (this *sinkManager) ExportEvents(data *core.EventBatch) {
	var wg sync.WaitGroup
	for _, sh := range this.sinkHolders {
		wg.Add(1)
		go func(sh sinkHolder, wg *sync.WaitGroup) {
			defer wg.Done()
			glog.V(2).Infof("Pushing events to: %s", sh.sink.Name())
			select {
			case sh.eventBatchChannel <- data:
				glog.V(2).Infof("Data events completed: %s", sh.sink.Name())
				// everything ok
			case <-time.After(this.exportEventsTimeout):
				glog.Warningf("Failed to events data to sink: %s", sh.sink.Name())
			}
		}(sh, &wg)
	}
	// Wait for all pushes to complete or timeout.
	wg.Wait()
}
88634
func (c *Client) EventCount() (count int, err error) {
	conn := c.Pool.Get()
	defer conn.Close()

	count, err = redis.Int(conn.Do("ZCARD", db.EventsCollection))
	if err != nil {
		return 0, err
	}

	return count, nil
}
950928
func (v *Stream) EventAddCallback(events StreamEventType, callback StreamEventCallback) error {
	callbackID := registerCallbackId(callback)

	var err C.virError
	ret := C.virStreamEventAddCallbackWrapper(v.ptr, (C.int)(events), (C.int)(callbackID), &err)
	if ret == -1 {
		return makeError(&err)
	}

	return nil
}
20199
func (c *Client) ListAppEvents(eventType string) ([]AppEventEntity, error) {
	return c.ListAppEventsByQuery(eventType, nil)
}
1081614
func (e *RateLimitErrorLogging) AddEvents(ctx context.Context, points []*event.Event, next Sink) error {
	err := next.AddEvents(ctx, points)
	e.throttleLog(err)
	return err
}
1293602
func (s *ClientConfigChangerSink) AddEvents(ctx context.Context, events []*event.Event) error {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.Destination.AddEvents(ctx, events)
}
413154
func (b *eventBatch) events() []wrappedEvent {
	sort.Sort(byTimestamp(b.batch))
	return b.batch
}
1174486
func (a *AsyncEventEmitter) EmitEvent(event Event) {
	a.listenersLock.RLock()
	for _, listener := range a.listeners {
		go listener.HandleEvent(event)
	}
	a.listenersLock.RUnlock()
}
1081659
func (c *CircularBuffer) Events() []Event {
	c.Lock()
	defer c.Unlock()
	return c.eventsCopy()
}
1099567
func SendEvents(c Client, e *[]Event) (*proto.Msg, error) {
	buff := make(
		[]*proto.Event, len(*e),
	)

	for i, elem := range *e {
		epb, err := EventToProtocolBuffer(
			&elem,
		)

		if err != nil {
			return nil, err
		}

		buff[i] = epb
	}

	message := new(proto.Msg)
	message.Events = buff

	return c.Send(message)
}
530292
func (u *UneventfulForwarder) AddEvents(ctx context.Context, events []*event.Event) error {
	return nil
}
62916
func (lf ListenerFunc) OnEvent(ctx context.Context, event Event) {
	lf(ctx, event)
}
213597
func (prod *AwsCloudwatchLogs) numEvents(messages []*core.Message, checkFn ...indexNumFn) int {
	index := maxBatchEvents
	for _, fn := range checkFn {
		result := fn(messages)
		if result < index {
			index = result
		}
	}
	return index
}
542663
func (c *BackgroundService) ClearEvents(service string) (*gcdmessage.ChromeResponse, error) {
	var v BackgroundServiceClearEventsParams
	v.Service = service
	return c.ClearEventsWithParams(&v)
}