// newEventQueue returns a queue to the provided sink. If the updater is non-
// nil, it will be called to update pending metrics on ingress and egress.
515576
func newEventQueue(sink Sink, listeners ...eventQueueListener) *eventQueue {
	eq := eventQueue{
		sink:      sink,
		events:    list.New(),
		listeners: listeners,
	}

	eq.cond = sync.NewCond(&eq.mu)
	go eq.run()
	return &eq
}
1242460
func NewWatchUpdateQueue() WatchUpdateQueue {
	mtx := &sync.Mutex{}
	return &watchQueue{
		m:       mtx,
		cv:      sync.NewCond(mtx),
		updates: make([]*watchUpdate, 0)}
}
1089275
func NewConnectionEvent(connected bool, err DisconnectedError) *ConnectionEvent {
	return &ConnectionEvent{Connected: connected, Err: err}
}
579856
func NewChannelQueue(capacity int) (Queue, error) {
	cap := int32(capacity)
	if capacity <= 0 {
		return nil, fmt.Errorf("Invalid size for queue: %d", capacity)
	}
	qChan := make(chan interface{}, cap)

	return &chanQueue{capacity: cap, qChan: qChan}, nil
}
1026436
func NewEventPipeline(streams ...EventStream) EventStream {
	switch len(streams) {
	case 0:
		return nil
	case 1:
		return streams[0]
	default:
	}

	streams[0].Subscribe(streams[1])

	return struct {
		EventWriter
		EventEmitter
	}{
		streams[0],
		NewEventPipeline(streams[1:]...),
	}

}
909933
func NewPlanQueue() (*PlanQueue, error) {
	q := &PlanQueue{
		enabled: false,
		stats:   new(QueueStats),
		ready:   make([]*pendingPlan, 0, 16),
		waitCh:  make(chan struct{}, 1),
	}
	return q, nil
}
609133
func NewPriorityQueue(hint int, allowDuplicates bool) *PriorityQueue {
	return &PriorityQueue{
		items:           make(priorityItems, 0, hint),
		itemMap:         make(map[Item]struct{}, hint),
		allowDuplicates: allowDuplicates,
	}
}
671586
func processEventQueue(xu *xgbutil.XUtil, pingBefore, pingAfter chan struct{}) {
	for !Empty(xu) {
		if Quitting(xu) {
			return
		}

		// We send the ping *before* the next event is dequeued.
		// This is so the queue doesn't present a misrepresentation of which
		// events haven't been processed yet.
		if pingBefore != nil && pingAfter != nil {
			pingBefore <- struct{}{}
		}
		ev, err := Dequeue(xu)

		// If we gobbled up an error, send it to the error event handler
		// and move on the next event/error.
		if err != nil {
			ErrorHandlerGet(xu)(err)
			if pingBefore != nil && pingAfter != nil {
				pingAfter <- struct{}{}
			}
			continue
		}

		// We know there isn't an error. If there isn't an event either,
		// then there's a bug somewhere.
		if ev == nil {
			xgbutil.Logger.Fatal("BUG: Expected an event but got nil.")
		}

		hooks := getHooks(xu)
		for _, hook := range hooks {
			if !hook.Run(xu, ev) {
				goto END
			}
		}

		switch event := ev.(type) {
		case xproto.KeyPressEvent:
			e := KeyPressEvent{&event}

			// If we're redirecting key events, this is the place to do it!
			if wid := RedirectKeyGet(xu); wid > 0 {
				e.Event = wid
			}

			xu.TimeSet(e.Time)
			runCallbacks(xu, e, KeyPress, e.Event)
		case xproto.KeyReleaseEvent:
			e := KeyReleaseEvent{&event}

			// If we're redirecting key events, this is the place to do it!
			if wid := RedirectKeyGet(xu); wid > 0 {
				e.Event = wid
			}

			xu.TimeSet(e.Time)
			runCallbacks(xu, e, KeyRelease, e.Event)
		case xproto.ButtonPressEvent:
			e := ButtonPressEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, ButtonPress, e.Event)
		case xproto.ButtonReleaseEvent:
			e := ButtonReleaseEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, ButtonRelease, e.Event)
		case xproto.MotionNotifyEvent:
			e := MotionNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, MotionNotify, e.Event)
		case xproto.EnterNotifyEvent:
			e := EnterNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, EnterNotify, e.Event)
		case xproto.LeaveNotifyEvent:
			e := LeaveNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, LeaveNotify, e.Event)
		case xproto.FocusInEvent:
			e := FocusInEvent{&event}
			runCallbacks(xu, e, FocusIn, e.Event)
		case xproto.FocusOutEvent:
			e := FocusOutEvent{&event}
			runCallbacks(xu, e, FocusOut, e.Event)
		case xproto.KeymapNotifyEvent:
			e := KeymapNotifyEvent{&event}
			runCallbacks(xu, e, KeymapNotify, NoWindow)
		case xproto.ExposeEvent:
			e := ExposeEvent{&event}
			runCallbacks(xu, e, Expose, e.Window)
		case xproto.GraphicsExposureEvent:
			e := GraphicsExposureEvent{&event}
			runCallbacks(xu, e, GraphicsExposure, xproto.Window(e.Drawable))
		case xproto.NoExposureEvent:
			e := NoExposureEvent{&event}
			runCallbacks(xu, e, NoExposure, xproto.Window(e.Drawable))
		case xproto.VisibilityNotifyEvent:
			e := VisibilityNotifyEvent{&event}
			runCallbacks(xu, e, VisibilityNotify, e.Window)
		case xproto.CreateNotifyEvent:
			e := CreateNotifyEvent{&event}
			runCallbacks(xu, e, CreateNotify, e.Parent)
		case xproto.DestroyNotifyEvent:
			e := DestroyNotifyEvent{&event}
			runCallbacks(xu, e, DestroyNotify, e.Window)
		case xproto.UnmapNotifyEvent:
			e := UnmapNotifyEvent{&event}
			runCallbacks(xu, e, UnmapNotify, e.Window)
		case xproto.MapNotifyEvent:
			e := MapNotifyEvent{&event}
			runCallbacks(xu, e, MapNotify, e.Event)
		case xproto.MapRequestEvent:
			e := MapRequestEvent{&event}
			runCallbacks(xu, e, MapRequest, e.Window)
			runCallbacks(xu, e, MapRequest, e.Parent)
		case xproto.ReparentNotifyEvent:
			e := ReparentNotifyEvent{&event}
			runCallbacks(xu, e, ReparentNotify, e.Window)
		case xproto.ConfigureNotifyEvent:
			e := ConfigureNotifyEvent{&event}
			runCallbacks(xu, e, ConfigureNotify, e.Window)
		case xproto.ConfigureRequestEvent:
			e := ConfigureRequestEvent{&event}
			runCallbacks(xu, e, ConfigureRequest, e.Window)
			runCallbacks(xu, e, ConfigureRequest, e.Parent)
		case xproto.GravityNotifyEvent:
			e := GravityNotifyEvent{&event}
			runCallbacks(xu, e, GravityNotify, e.Window)
		case xproto.ResizeRequestEvent:
			e := ResizeRequestEvent{&event}
			runCallbacks(xu, e, ResizeRequest, e.Window)
		case xproto.CirculateNotifyEvent:
			e := CirculateNotifyEvent{&event}
			runCallbacks(xu, e, CirculateNotify, e.Window)
		case xproto.CirculateRequestEvent:
			e := CirculateRequestEvent{&event}
			runCallbacks(xu, e, CirculateRequest, e.Window)
		case xproto.PropertyNotifyEvent:
			e := PropertyNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, PropertyNotify, e.Window)
		case xproto.SelectionClearEvent:
			e := SelectionClearEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, SelectionClear, e.Owner)
		case xproto.SelectionRequestEvent:
			e := SelectionRequestEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, SelectionRequest, e.Requestor)
		case xproto.SelectionNotifyEvent:
			e := SelectionNotifyEvent{&event}
			xu.TimeSet(e.Time)
			runCallbacks(xu, e, SelectionNotify, e.Requestor)
		case xproto.ColormapNotifyEvent:
			e := ColormapNotifyEvent{&event}
			runCallbacks(xu, e, ColormapNotify, e.Window)
		case xproto.ClientMessageEvent:
			e := ClientMessageEvent{&event}
			runCallbacks(xu, e, ClientMessage, e.Window)
		case xproto.MappingNotifyEvent:
			e := MappingNotifyEvent{&event}
			runCallbacks(xu, e, MappingNotify, NoWindow)
		case shape.NotifyEvent:
			e := ShapeNotifyEvent{&event}
			runCallbacks(xu, e, ShapeNotify, e.AffectedWindow)
		default:
			if event != nil {
				xgbutil.Logger.Printf("ERROR: UNSUPPORTED EVENT TYPE: %T",
					event)
			}
		}

	END:

		if pingBefore != nil && pingAfter != nil {
			pingAfter <- struct{}{}
		}
	}
}
586961
func NewConcurrentQueue(bufferSize int) *ConcurrentQueue {
	return &ConcurrentQueue{
		chanIn:   make(chan interface{}),
		chanOut:  make(chan interface{}, bufferSize),
		overflow: list.New(),
		quit:     make(chan struct{}),
	}
}
319359
func (b *Bus) Enqueue(e Event) error {
	if b.stopped {
		return nil
	}

	// Callers can filter events from being added
	b.mutex.RLock()
	for _, filterInfo := range b.enqueueFilters {
		if filterInfo.filter(e) {
			b.mutex.RUnlock()
			return nil
		}
	}
	b.mutex.RUnlock()

	select {
	case b.events <- e:
		return nil
	default:
		return ErrBusFull
	}
}
992094
func NewQueue(service *Service, workers uint) *Queue {
	// unbuffered channels
	q := &Queue{
		service:       service,
		notifications: make(chan notification),
		Responses:     make(chan Response),
	}
	// startup workers to send notifications
	for i := uint(0); i < workers; i++ {
		go worker(q)
	}
	return q
}
1029631
func NewNFQueue(queueId uint16, maxPacketsInQueue uint32, packetSize uint32) (*NFQueue, error) {
	var nfq = NFQueue{}
	var err error
	var ret C.int

	if nfq.h, err = C.nfq_open(); err != nil {
		return nil, fmt.Errorf("Error opening NFQueue handle: %v\n", err)
	}

	if ret, err = C.nfq_unbind_pf(nfq.h, AF_INET); err != nil || ret < 0 {
		return nil, fmt.Errorf("Error unbinding existing NFQ handler from AF_INET protocol family: %v\n", err)
	}

	if ret, err = C.nfq_unbind_pf(nfq.h, AF_INET6); err != nil || ret < 0 {
		return nil, fmt.Errorf("Error unbinding existing NFQ handler from AF_INET6 protocol family: %v\n", err)
	}

	if ret, err := C.nfq_bind_pf(nfq.h, AF_INET); err != nil || ret < 0 {
		return nil, fmt.Errorf("Error binding to AF_INET protocol family: %v\n", err)
	}

	if ret, err := C.nfq_bind_pf(nfq.h, AF_INET6); err != nil || ret < 0 {
		return nil, fmt.Errorf("Error binding to AF_INET6 protocol family: %v\n", err)
	}

	nfq.packets = make(chan NFPacket)
	nfq.idx = uint32(time.Now().UnixNano())
	theTabeLock.Lock()
	theTable[nfq.idx] = &nfq.packets
	theTabeLock.Unlock()
	if nfq.qh, err = C.CreateQueue(nfq.h, C.u_int16_t(queueId), C.u_int32_t(nfq.idx)); err != nil || nfq.qh == nil {
		C.nfq_close(nfq.h)
		return nil, fmt.Errorf("Error binding to queue: %v\n", err)
	}

	if ret, err = C.nfq_set_queue_maxlen(nfq.qh, C.u_int32_t(maxPacketsInQueue)); err != nil || ret < 0 {
		C.nfq_destroy_queue(nfq.qh)
		C.nfq_close(nfq.h)
		return nil, fmt.Errorf("Unable to set max packets in queue: %v\n", err)
	}

	if C.nfq_set_mode(nfq.qh, C.u_int8_t(2), C.uint(packetSize)) < 0 {
		C.nfq_destroy_queue(nfq.qh)
		C.nfq_close(nfq.h)
		return nil, fmt.Errorf("Unable to set packets copy mode: %v\n", err)
	}

	if nfq.fd, err = C.nfq_fd(nfq.h); err != nil {
		C.nfq_destroy_queue(nfq.qh)
		C.nfq_close(nfq.h)
		return nil, fmt.Errorf("Unable to get queue file-descriptor. %v\n", err)
	}

	go nfq.run()

	return &nfq, nil
}
103593
func NewNodeMonitor(queueSize int) *NodeMonitor {
	nm := &NodeMonitor{
		queue: make(chan []byte, queueSize),
	}

	go nm.eventDrainer()

	return nm
}
549127
func New() (*Event, error) {
	id, err := manifold.NewID(idtype.ActivityEvent)
	if err != nil {
		return nil, err
	}

	evt := &Event{
		ID:            id,
		StructType:    "event",
		StructVersion: 1,
	}

	return evt, nil
}
580729
func NewEventAggregator(lruCacheSize int, keyFunc EventAggregatorKeyFunc, messageFunc EventAggregatorMessageFunc,
	maxEvents int, maxIntervalInSeconds int, clock clock.Clock) *EventAggregator {
	return &EventAggregator{
		cache:                lru.New(lruCacheSize),
		keyFunc:              keyFunc,
		messageFunc:          messageFunc,
		maxEvents:            uint(maxEvents),
		maxIntervalInSeconds: uint(maxIntervalInSeconds),
		clock:                clock,
	}
}
764705
func (eq *eventQueue) next() []Event {
	eq.mu.Lock()
	defer eq.mu.Unlock()

	for eq.events.Len() < 1 {
		if eq.closed {
			eq.cond.Broadcast()
			return nil
		}

		eq.cond.Wait()
	}

	front := eq.events.Front()
	block := front.Value.([]Event)
	eq.events.Remove(front)

	return block
}
351677
func (eb *eventBuffer) QueueEvent(event *lpEvent) error {
	if event == nil {
		return errors.New("event was nil")
	}
	// Cull our buffer if we're at max capacity
	if eb.List.Len() >= eb.MaxBufferSize {
		oldestEvent := eb.List.Back()
		if oldestEvent != nil {
			eb.List.Remove(oldestEvent)
		}
	}
	// Add event to front of our list
	eb.List.PushFront(event)
	// Update oldestEventTime with the time of our least recent event (at back)
	// keeping track of this allows for more efficient event TTL expiration purges
	if lastElement := eb.List.Back(); lastElement != nil {
		lastEvent, ok := lastElement.Value.(*lpEvent)
		if !ok {
			return fmt.Errorf("Found non-event type in event buffer.")
		}
		eb.oldestEventTime = lastEvent.Timestamp
	}
	return nil
}