zrok/controller/metrics/bridge.go

package metrics

import (
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

type BridgeConfig struct {
	Source interface{}
	Sink   interface{}
}

type Bridge struct {
	src     ZitiEventJsonSource
	srcJoin chan struct{}
	snk     ZitiEventJsonSink
	events  chan ZitiEventMsg
	close   chan struct{}
	join    chan struct{}
}

func NewBridge(cfg *BridgeConfig) (*Bridge, error) {
	b := &Bridge{
		events: make(chan ZitiEventMsg),
		join:   make(chan struct{}),
		close:  make(chan struct{}),
	}
	if v, ok := cfg.Source.(ZitiEventJsonSource); ok {
		b.src = v
	} else {
		return nil, errors.New("invalid source type")
	}
	if v, ok := cfg.Sink.(ZitiEventJsonSink); ok {
		b.snk = v
	} else {
		return nil, errors.New("invalid sink type")
	}
	return b, nil
}

func (b *Bridge) Start() (join chan struct{}, err error) {
	if b.srcJoin, err = b.src.Start(b.events); err != nil {
		return nil, err
	}

	go func() {
		logrus.Info("started")
		defer logrus.Info("stopped")
		defer close(b.join)

	eventLoop:
		for {
			select {
			case eventJson := <-b.events:
				logrus.Info(eventJson)
				if err := b.snk.Handle(eventJson.Data()); err == nil {
					logrus.Infof("-> %v", eventJson.Data())
				} else {
					logrus.Error(err)
				}
				eventJson.Ack()

			case <-b.close:
				logrus.Info("received close signal")
				break eventLoop
			}
		}
	}()

	return b.join, nil
}

func (b *Bridge) Stop() {
	b.src.Stop()
	close(b.close)
	<-b.srcJoin
	<-b.join
}