<commit_before>package main

import (
	"errors"
	"fmt"
	"strconv"
	"strings"
	"sync"

	"github.com/flynn/flynn/Godeps/_workspace/src/gopkg.in/inconshreveable/log15.v2"
	ct "github.com/flynn/flynn/controller/types"
)

// ErrEventBufferOverflow is returned to clients when the in-memory event
// buffer is full due to clients not reading events quickly enough.
var ErrEventBufferOverflow = errors.New("event stream buffer overflow")

// eventBufferSize is the amount of events to buffer in memory.
const eventBufferSize = 1000

// EventSubscriber receives app events from the EventListener loop and maintains
// it's own loop to forward those events to the Events channel.
type EventSubscriber struct {
	Events  chan *ct.AppEvent
	Err     error
	errOnce sync.Once

	l          *EventListener
	queue      chan *ct.AppEvent
	appID      string
	objectType string
	objectID   string

	stop     chan struct{}
	stopOnce sync.Once
}

// Notify filters the event based on it's type and objectID and then pushes
// it to the event queue.
func (e *EventSubscriber) Notify(event *ct.AppEvent) {
	if e.objectType != "" && e.objectType != string(event.ObjectType) {
		return
	}
	if e.objectID != "" && e.objectID != event.ObjectID {
		return
	}
	select {
	case e.queue <- event:
	default:
		// Run in a goroutine to avoid deadlock with Notify
		go e.CloseWithError(ErrEventBufferOverflow)
	}
}

// loop pops events off the queue and sends them to the Events channel.
func (e *EventSubscriber) loop() {
	defer close(e.Events)
	for {
		select {
		case <-e.stop:
			return
		case event := <-e.queue:
			e.Events <- event
		}
	}
}

// Close unsubscribes from the EventListener and stops the loop.
func (e *EventSubscriber) Close() {
	e.l.Unsubscribe(e)
	e.stopOnce.Do(func() { close(e.stop) })
}

// CloseWithError sets the Err field and then closes the subscriber.
func (e *EventSubscriber) CloseWithError(err error) {
	e.errOnce.Do(func() { e.Err = err })
	e.Close()
}

func newEventListener(a *AppRepo) *EventListener {
	return &EventListener{
		appRepo:     a,
		subscribers: make(map[string]map[*EventSubscriber]struct{}),
	}
}

// EventListener creates a postgres Listener for app events and forwards them
// to subscribers.
type EventListener struct {
	appRepo *AppRepo

	subscribers map[string]map[*EventSubscriber]struct{}
	subMtx      sync.RWMutex

	closed    bool
	closedMtx sync.RWMutex
}

// Subscribe creates and returns an EventSubscriber for the given app, type and object.
func (e *EventListener) Subscribe(appID, objectType, objectID string) (*EventSubscriber, error) {
	e.subMtx.Lock()
	defer e.subMtx.Unlock()
	if e.IsClosed() {
		return nil, errors.New("event listener closed")
	}
	s := &EventSubscriber{
		Events:     make(chan *ct.AppEvent),
		l:          e,
		queue:      make(chan *ct.AppEvent, eventBufferSize),
		stop:       make(chan struct{}),
		appID:      appID,
		objectType: objectType,
		objectID:   objectID,
	}
	go s.loop()
	if _, ok := e.subscribers[appID]; !ok {
		e.subscribers[appID] = make(map[*EventSubscriber]struct{})
	}
	e.subscribers[appID][s] = struct{}{}
	return s, nil
}

// Unsubscribe unsubscribes the given subscriber.
func (e *EventListener) Unsubscribe(s *EventSubscriber) {
	e.subMtx.Lock()
	defer e.subMtx.Unlock()
	if subs, ok := e.subscribers[s.appID]; ok {
		delete(subs, s)
		if len(subs) == 0 {
			delete(e.subscribers, s.appID)
		}
	}
}

// Listen creates a postgres listener for app events and starts a goroutine to
// forward the events to subscribers.
func (e *EventListener) Listen() error {
	log := log15.New("component", "controller", "fn", "EventListener.Listen")
	listener, err := e.appRepo.db.Listen("app_events", log)
	if err != nil {
		e.SetClosed()
		return err
	}
	go func() {
		for {
			n, ok := <-listener.Notify
			if !ok {
				e.CloseWithError(listener.Err)
				return
			}
			idApp := strings.SplitN(n.Extra, ":", 2)
			if len(idApp) != 2 {
				log.Error(fmt.Sprintf("invalid app event notification: %q", n.Extra))
				continue
			}
			id, err := strconv.ParseInt(idApp[0], 10, 64)
			if err != nil {
				log.Error(fmt.Sprintf("invalid app event notification: %q", n.Extra), "err", err)
				continue
			}
			event, err := e.appRepo.GetEvent(id)
			if err != nil {
				log.Error(fmt.Sprintf("invalid app event notification: %q", n.Extra), "err", err)
				continue
			}
			e.Notify(event)
		}
	}()
	return nil
}

// Notify notifies all sbscribers of the given event.
func (e *EventListener) Notify(event *ct.AppEvent) {
	e.subMtx.RLock()
	defer e.subMtx.RUnlock()
	if subs, ok := e.subscribers[event.AppID]; ok {
		for sub := range subs {
			go sub.Notify(event)
		}<commit_msg>controller: Don’t notify app events using goroutines<commit_after>package main

import (
	"errors"
	"fmt"
	"strconv"
	"strings"
	"sync"

	"github.com/flynn/flynn/Godeps/_workspace/src/gopkg.in/inconshreveable/log15.v2"
	ct "github.com/flynn/flynn/controller/types"
)

// ErrEventBufferOverflow is returned to clients when the in-memory event
// buffer is full due to clients not reading events quickly enough.
var ErrEventBufferOverflow = errors.New("event stream buffer overflow")

// eventBufferSize is the amount of events to buffer in memory.
const eventBufferSize = 1000

// EventSubscriber receives app events from the EventListener loop and maintains
// it's own loop to forward those events to the Events channel.
type EventSubscriber struct {
	Events  chan *ct.AppEvent
	Err     error
	errOnce sync.Once

	l          *EventListener
	queue      chan *ct.AppEvent
	appID      string
	objectType string
	objectID   string

	stop     chan struct{}
	stopOnce sync.Once
}

// Notify filters the event based on it's type and objectID and then pushes
// it to the event queue.
func (e *EventSubscriber) Notify(event *ct.AppEvent) {
	if e.objectType != "" && e.objectType != string(event.ObjectType) {
		return
	}
	if e.objectID != "" && e.objectID != event.ObjectID {
		return
	}
	select {
	case e.queue <- event:
	default:
		// Run in a goroutine to avoid deadlock with Notify
		go e.CloseWithError(ErrEventBufferOverflow)
	}
}

// loop pops events off the queue and sends them to the Events channel.
func (e *EventSubscriber) loop() {
	defer close(e.Events)
	for {
		select {
		case <-e.stop:
			return
		case event := <-e.queue:
			e.Events <- event
		}
	}
}

// Close unsubscribes from the EventListener and stops the loop.
func (e *EventSubscriber) Close() {
	e.l.Unsubscribe(e)
	e.stopOnce.Do(func() { close(e.stop) })
}

// CloseWithError sets the Err field and then closes the subscriber.
func (e *EventSubscriber) CloseWithError(err error) {
	e.errOnce.Do(func() { e.Err = err })
	e.Close()
}

func newEventListener(a *AppRepo) *EventListener {
	return &EventListener{
		appRepo:     a,
		subscribers: make(map[string]map[*EventSubscriber]struct{}),
	}
}

// EventListener creates a postgres Listener for app events and forwards them
// to subscribers.
type EventListener struct {
	appRepo *AppRepo

	subscribers map[string]map[*EventSubscriber]struct{}
	subMtx      sync.RWMutex

	closed    bool
	closedMtx sync.RWMutex
}

// Subscribe creates and returns an EventSubscriber for the given app, type and object.
func (e *EventListener) Subscribe(appID, objectType, objectID string) (*EventSubscriber, error) {
	e.subMtx.Lock()
	defer e.subMtx.Unlock()
	if e.IsClosed() {
		return nil, errors.New("event listener closed")
	}
	s := &EventSubscriber{
		Events:     make(chan *ct.AppEvent),
		l:          e,
		queue:      make(chan *ct.AppEvent, eventBufferSize),
		stop:       make(chan struct{}),
		appID:      appID,
		objectType: objectType,
		objectID:   objectID,
	}
	go s.loop()
	if _, ok := e.subscribers[appID]; !ok {
		e.subscribers[appID] = make(map[*EventSubscriber]struct{})
	}
	e.subscribers[appID][s] = struct{}{}
	return s, nil
}

// Unsubscribe unsubscribes the given subscriber.
func (e *EventListener) Unsubscribe(s *EventSubscriber) {
	e.subMtx.Lock()
	defer e.subMtx.Unlock()
	if subs, ok := e.subscribers[s.appID]; ok {
		delete(subs, s)
		if len(subs) == 0 {
			delete(e.subscribers, s.appID)
		}
	}
}

// Listen creates a postgres listener for app events and starts a goroutine to
// forward the events to subscribers.
func (e *EventListener) Listen() error {
	log := log15.New("component", "controller", "fn", "EventListener.Listen")
	listener, err := e.appRepo.db.Listen("app_events", log)
	if err != nil {
		e.SetClosed()
		return err
	}
	go func() {
		for {
			n, ok := <-listener.Notify
			if !ok {
				e.CloseWithError(listener.Err)
				return
			}
			idApp := strings.SplitN(n.Extra, ":", 2)
			if len(idApp) != 2 {
				log.Error(fmt.Sprintf("invalid app event notification: %q", n.Extra))
				continue
			}
			id, err := strconv.ParseInt(idApp[0], 10, 64)
			if err != nil {
				log.Error(fmt.Sprintf("invalid app event notification: %q", n.Extra), "err", err)
				continue
			}
			event, err := e.appRepo.GetEvent(id)
			if err != nil {
				log.Error(fmt.Sprintf("invalid app event notification: %q", n.Extra), "err", err)
				continue
			}
			e.Notify(event)
		}
	}()
	return nil
}

// Notify notifies all sbscribers of the given event.
func (e *EventListener) Notify(event *ct.AppEvent) {
	e.subMtx.RLock()
	defer e.subMtx.RUnlock()
	if subs, ok := e.subscribers[event.AppID]; ok {
		for sub := range subs {
			sub.Notify(event)
		}
{'fraction_non_alphanumeric': 0.12438238453276047, 'fraction_numerical': 0.003651987110633727, 'mean_word_length': 3.550830889540567, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2807219', 'n_tokens_mistral': 3428, 'n_tokens_neox': 3183, 'n_words': 923}