// AddEvents sends the events to a chan buffer that eventually is flushed in big groups
960858
func (forwarder *BufferedForwarder) AddEvents(ctx context.Context, events []*event.Event) error {
	if forwarder.checker.CtxFlagCheck.HasFlag(ctx) {
		forwarder.cdim.With(ctx, forwarder.logger).Log("Events call received in buffered forwarder")
	}
	atomic.AddInt64(&forwarder.stats.totalEventsBuffered, int64(len(events)))
	if *forwarder.config.MaxTotalEvents <= atomic.LoadInt64(&forwarder.stats.totalEventsBuffered) {
		atomic.AddInt64(&forwarder.stats.totalEventsBuffered, int64(-len(events)))
		return errEBufferFull(forwarder.identifier)
	}
	select {
	case forwarder.eChan <- events:
		return nil
	case <-forwarder.stopContext.Done():
		return forwarder.stopContext.Err()
	case <-ctx.Done():
		return ctx.Err()
	}
}
301393
func (form FormImpl) AddEventHandler(evt string, f HandlerFunc) Form {
	if form.trans == nil {
		form.trans = map[string]HandlerFunc{}
	}
	form.trans[evt] = f
	return form
}
1054824
func (s *AutoRollbackConfiguration) SetEvents(v []*string) *AutoRollbackConfiguration {
	s.Events = v
	return s
}
1332943
func (f *Forwarder) AddEvents(ctx context.Context, events []*event.Event) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	atomic.AddInt64(&f.stats.totalEventsForwarded, int64(len(events)))
	for _, e := range events {
		if _, err := f.writeString(f.file, e.String()+"\n"); err != nil {
			return errors.Annotate(err, "cannot write event to string")
		}
	}
	return f.file.Sync()
}
608212
func (s *GetWorkflowExecutionHistoryOutput) SetEvents(v []*HistoryEvent) *GetWorkflowExecutionHistoryOutput {
	s.Events = v
	return s
}
1334578
func (a *AsyncMultiTokenSink) AddEventsWithToken(token string, events []*event.Event) (err error) {
	var channelID int64
	if channelID, err = a.getChannel(token, len(a.evChannels)); err == nil {
		var worker = a.evChannels[channelID]
		_ = atomic.AddInt64(&a.evBuffered, int64(len(events)))
		var m = &evMsg{
			token: token,
			data:  events,
		}
		select {
		// check if the sink is closing and return if so
		// reading from a.closing will only return a value if the a.closing channel is closed
		case <-a.closing:
			err = fmt.Errorf("unable to add events: the worker has been stopped")
		default:
			select {
			case worker.input <- m:
				atomic.AddInt64(&a.stats.TotalEventsBuffered, int64(len(events)))
			default:
				err = fmt.Errorf("unable to add events: the input buffer is full")
			}
		}

	} else {
		err = fmt.Errorf("unable to add events: there was an error while hashing the token to a worker. %v", err)
	}

	return
}
413164
func addFieldsToEvent(fields map[string]string, ev *event.Event) {
	for k, v := range fields {
		// Try to convert values to integer types where sensible, and extract
		// timestamp for event
		switch k {
		case "session_id", "pid", "session_line_number":
			if typed, err := strconv.Atoi(v); err == nil {
				ev.Data[k] = typed
			} else {
				ev.Data[k] = v
			}
		case "timestamp", "timestamp_millis":
			if timestamp, err := time.Parse("2006-01-02 15:04:05.999 MST", v); err == nil {
				ev.Timestamp = timestamp
			} else {
				logrus.WithField("timestamp", v).WithError(err).Debug("Error parsing query timestamp")
			}
		case "timestamp_unix":
			if typed, err := strconv.Atoi(v); err == nil {
				// Convert millisecond-resolution Unix timestamp to time.Time
				// object
				timestamp := time.Unix(int64(typed/1000), int64((1000*1000)*(typed%1000))).UTC()
				ev.Timestamp = timestamp
			} else {
				logrus.WithField("timestamp", v).WithError(err).Debug("Error parsing query timestamp")
			}
		default:
			ev.Data[k] = v
		}
	}
}
894799
func (em *EventManager) AddEvent(eo EventSubs) bool {
	id := GetEventID(eo)

	if em.HasEvent(id) {
		return false
	}

	em.ro.Lock()
	em.events[id] = eo
	em.ro.Unlock()
	return true
}
1340673
func (s *GetExecutionHistoryOutput) SetEvents(v []*HistoryEvent) *GetExecutionHistoryOutput {
	s.Events = v
	return s
}
1079634
func (e *EmptyService) Events(ctx context.Context, events chan events.ContainerEvent) error {
	return nil
}
541636
func (q *Queue) Events() EventMask {
	ret := EventMask(0)

	q.mu.RLock()
	for e := q.list.Front(); e != nil; e = e.Next() {
		ret |= e.mask
	}
	q.mu.RUnlock()

	return ret
}
426087
func (b *EventBus) AddHandler(m eh.EventMatcher, h eh.EventHandler) {
	ch := b.channel(m, h, false)
	go b.handle(m, h, ch)
}
626497
func processEvents(inboundEvents chan []byte, parsedEvents chan *Event) {
	go func(inboundEvents chan []byte, parsedEvents chan *Event) {
		for event := range inboundEvents {
			var e Event
			json.Unmarshal(event, &e)
			parsedEvents <- &e
		}
	}(inboundEvents, parsedEvents)
}
1159227
func (s *CloudFunctionConfiguration) SetEvents(v []*string) *CloudFunctionConfiguration {
	s.Events = v
	return s
}
667992
func (s *GetLogEventsOutput) SetEvents(v []*OutputLogEvent) *GetLogEventsOutput {
	s.Events = v
	return s
}
1318212
func (s *ActivityService) ListEvents(ctx context.Context, opt *ListOptions) ([]*Event, *Response, error) {
	u, err := addOptions("events", opt)
	if err != nil {
		return nil, nil, err
	}

	req, err := s.client.NewRequest("GET", u, nil)
	if err != nil {
		return nil, nil, err
	}

	var events []*Event
	resp, err := s.client.Do(ctx, req, &events)
	if err != nil {
		return nil, resp, err
	}

	return events, resp, nil
}
1282531
func appendEvent(events []riemanngo.Event, sink *RiemannSink, host, name string, value interface{}, labels map[string]string, timestamp int64) []riemanngo.Event {
	event := riemanngo.Event{
		Time:        timestamp,
		Service:     name,
		Host:        host,
		Description: "",
		Attributes:  labels,
		Metric:      value,
		Ttl:         sink.config.Ttl,
		State:       sink.config.State,
		Tags:        sink.config.Tags,
	}
	// state everywhere
	events = append(events, event)
	if len(events) >= sink.config.BatchSize {
		err := riemannCommon.SendData(sink.client, events)
		if err != nil {
			glog.Warningf("Error sending events to Riemann: %v", err)
			// client will reconnect later
			sink.client = nil
		}
		events = nil
	}
	return events
}