ipc_event_stream.go

package agent

import (
	"fmt"
	"github.com/hashicorp/serf/serf"
	"log"
)

type streamClient interface {
	Send(*responseHeader, interface{}) error
	RegisterQuery(*serf.Query) uint64
}

// eventStream is used to stream events to a client over IPC
type eventStream struct {
	client  streamClient
	eventCh chan serf.Event
	filters []EventFilter
	logger  *log.Logger
	seq     uint64
}

func newEventStream(client streamClient, filters []EventFilter, seq uint64, logger *log.Logger) *eventStream {
	es := &eventStream{
		client:  client,
		eventCh: make(chan serf.Event, 512),
		filters: filters,
		logger:  logger,
		seq:     seq,
	}
	go es.stream()
	return es
}

func (es *eventStream) HandleEvent(e serf.Event) {
	// Check the event
	for _, f := range es.filters {
		if f.Invoke(e) {
			goto HANDLE
		}
	}
	return

	// Do a non-blocking send
HANDLE:
	select {
	case es.eventCh <- e:
	default:
		es.logger.Printf("[WARN] agent.ipc: Dropping event to %v", es.client)
	}
}

func (es *eventStream) Stop() {
	close(es.eventCh)
}

func (es *eventStream) stream() {
	var err error
	for event := range es.eventCh {
		switch e := event.(type) {
		case serf.MemberEvent:
			err = es.sendMemberEvent(e)
		case serf.UserEvent:
			err = es.sendUserEvent(e)
		case *serf.Query:
			err = es.sendQuery(e)
		default:
			err = fmt.Errorf("Unknown event type: %s", event.EventType().String())
		}
		if err != nil {
			es.logger.Printf("[ERR] agent.ipc: Failed to stream event to %v: %v",
				es.client, err)
			return
		}
	}
}

// sendMemberEvent is used to send a single member event
func (es *eventStream) sendMemberEvent(me serf.MemberEvent) error {
	members := make([]Member, 0, len(me.Members))
	for _, m := range me.Members {
		sm := Member{
			Name:        m.Name,
			Addr:        m.Addr,
			Port:        m.Port,
			Tags:        m.Tags,
			Status:      m.Status.String(),
			ProtocolMin: m.ProtocolMin,
			ProtocolMax: m.ProtocolMax,
			ProtocolCur: m.ProtocolCur,
			DelegateMin: m.DelegateMin,
			DelegateMax: m.DelegateMax,
			DelegateCur: m.DelegateCur,
		}
		members = append(members, sm)
	}

	header := responseHeader{
		Seq:   es.seq,
		Error: "",
	}
	rec := memberEventRecord{
		Event:   me.String(),
		Members: members,
	}
	return es.client.Send(&header, &rec)
}

// sendUserEvent is used to send a single user event
func (es *eventStream) sendUserEvent(ue serf.UserEvent) error {
	header := responseHeader{
		Seq:   es.seq,
		Error: "",
	}
	rec := userEventRecord{
		Event:    ue.EventType().String(),
		LTime:    ue.LTime,
		Name:     ue.Name,
		Payload:  ue.Payload,
		Coalesce: ue.Coalesce,
	}
	return es.client.Send(&header, &rec)
}

// sendQuery is used to send a single query event
func (es *eventStream) sendQuery(q *serf.Query) error {
	id := es.client.RegisterQuery(q)

	header := responseHeader{
		Seq:   es.seq,
		Error: "",
	}
	rec := queryEventRecord{
		Event:   q.EventType().String(),
		ID:      id,
		LTime:   q.LTime,
		Name:    q.Name,
		Payload: q.Payload,
	}
	return es.client.Send(&header, &rec)
}