func (a *Agent) shouldProcessUserEvent(msg *UserEvent) bool {
	if msg.Version > userEventMaxVersion {
		a.logger.Printf("[WARN] agent: Event version %d may have unsupported features (%s)",
			msg.Version, msg.Name)
	}
	if msg.NodeFilter != "" {
		re, err := regexp.Compile(msg.NodeFilter)
		if err != nil {
			a.logger.Printf("[ERR] agent: Failed to parse node filter '%s' for event '%s': %v",
				msg.NodeFilter, msg.Name, err)
			return false
		}
		if !re.MatchString(a.config.NodeName) {
			return false
		}
	}
	if msg.ServiceFilter != "" {
		re, err := regexp.Compile(msg.ServiceFilter)
		if err != nil {
			a.logger.Printf("[ERR] agent: Failed to parse service filter '%s' for event '%s': %v",
				msg.ServiceFilter, msg.Name, err)
			return false
		}
		var tagRe *regexp.Regexp
		if msg.TagFilter != "" {
			re, err := regexp.Compile(msg.TagFilter)
			if err != nil {
				a.logger.Printf("[ERR] agent: Failed to parse tag filter '%s' for event '%s': %v",
					msg.TagFilter, msg.Name, err)
				return false
			}
			tagRe = re
		}
		services := a.state.Services()
		found := false
	OUTER:
		for name, info := range services {
			if !re.MatchString(name) {
				continue
			}
			if tagRe == nil {
				found = true
				break
			}
			for _, tag := range info.Tags {
				if !tagRe.MatchString(tag) {
					continue
				}
				found = true
				break OUTER
			}
		}
		if !found {
			return false
		}
	}
	return true
}
shouldProcessUserEvent checks if an event makes it through our filters
package agent
import (
	"fmt"
	"regexp"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/consul/consul/structs"
)
const (
	userEventMaxVersion = 1
	remoteExecName = "_rexec"
)
type UserEvent struct {
	ID string
	Name string `codec:"n"`
	Payload []byte `codec:"p,omitempty"`
	NodeFilter string `codec:"nf,omitempty"`
	ServiceFilter string `codec:"sf,omitempty"`
	TagFilter string `codec:"tf,omitempty"`
	Version int `codec:"v"`
	LTime uint64 `codec:"-"`
}
func validateUserEventParams(params *UserEvent) error {
	if params.Name == "" {
		return fmt.Errorf("User event missing name")
	}
	if params.TagFilter != "" && params.ServiceFilter == "" {
		return fmt.Errorf("Cannot provide tag filter without service filter")
	}
	if params.NodeFilter != "" {
		if _, err := regexp.Compile(params.NodeFilter); err != nil {
			return fmt.Errorf("Invalid node filter: %v", err)
		}
	}
	if params.ServiceFilter != "" {
		if _, err := regexp.Compile(params.ServiceFilter); err != nil {
			return fmt.Errorf("Invalid service filter: %v", err)
		}
	}
	if params.TagFilter != "" {
		if _, err := regexp.Compile(params.TagFilter); err != nil {
			return fmt.Errorf("Invalid tag filter: %v", err)
		}
	}
	return nil
}
func (a *Agent) UserEvent(dc string, params *UserEvent) error {
	if err := validateUserEventParams(params); err != nil {
		return err
	}
	params.ID = generateUUID()
	params.Version = userEventMaxVersion
	payload, err := encodeMsgPack(&params)
	if err != nil {
		return fmt.Errorf("UserEvent encoding failed: %v", err)
	}
	if dc == "" || dc == a.config.Datacenter {
		if a.server != nil {
			return a.server.UserEvent(params.Name, payload)
		} else {
			return a.client.UserEvent(params.Name, payload)
		}
	} else {
		args := structs.EventFireRequest{
			Datacenter: dc,
			Name:       params.Name,
			Payload:    payload,
		}
		args.AllowStale = true
		var out structs.EventFireResponse
		return a.RPC("Internal.EventFire", &args, &out)
	}
}
func (a *Agent) handleEvents() {
	for {
		select {
		case e := <-a.eventCh:
			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)
			if !a.shouldProcessUserEvent(msg) {
				continue
			}
			a.ingestUserEvent(msg)
		case <-a.shutdownCh:
			return
		}
	}
}