zrok/controller/metrics/bridge.go
2023-05-01 13:45:45 -05:00

79 lines
1.4 KiB
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
}