mbus

Documentation

Index

• type AddressTable
• type MetricsRecorder
  • func NewMetricsRecorder(clock clock.Clock) *MetricsRecorder
  • func (r *MetricsRecorder) GetMaxSinceLastInterval() (float64, error)
  • func (r *MetricsRecorder) RecordMessageTransitTime(unixTimeNS int64)
• type NatsConn
• type NatsConnProvider
• type NatsConnWithUrlProvider
  • func (p *NatsConnWithUrlProvider) Connection(opts ...nats.Option) (NatsConn, error)
• type RegistryMessage
• type ServiceDiscoveryStartMessage
• type Subscriber
  • func NewSubscriber(natsConnProvider NatsConnProvider, subOpts SubscriberOpts, ...) *Subscriber
  • func (s *Subscriber) Close()
  • func (s *Subscriber) Run(signals <-chan os.Signal, ready chan<- struct{}) error
  • func (s *Subscriber) RunOnce() error
• type SubscriberOpts

Types

type AddressTable

type AddressTable interface {
	Add(infraNames []string, ip string)
	Remove(infraNames []string, ip string)
	PausePruning()
	ResumePruning()
	SetWarm()
}

type MetricsRecorder

type MetricsRecorder struct {
	sync.RWMutex

	Clock clock.Clock
	// contains filtered or unexported fields
}

func NewMetricsRecorder

func NewMetricsRecorder(clock clock.Clock) *MetricsRecorder

func (*MetricsRecorder) GetMaxSinceLastInterval

func (r *MetricsRecorder) GetMaxSinceLastInterval() (float64, error)

func (*MetricsRecorder) RecordMessageTransitTime

func (r *MetricsRecorder) RecordMessageTransitTime(unixTimeNS int64)

type NatsConn

type NatsConn interface {
	PublishMsg(m *nats.Msg) error
	Close()
	Flush() error
	ConnectedUrl() string
	Subscribe(string, nats.MsgHandler) (*nats.Subscription, error)
}

type NatsConnProvider

type NatsConnProvider interface {
	Connection(opts ...nats.Option) (NatsConn, error)
}

type NatsConnWithUrlProvider

type NatsConnWithUrlProvider struct {
	Url string
}

func (*NatsConnWithUrlProvider) Connection

func (p *NatsConnWithUrlProvider) Connection(opts ...nats.Option) (NatsConn, error)

type RegistryMessage

type RegistryMessage struct {
	IP                string   `json:"host"`
	InfraNames        []string `json:"uris"`
	EndpointUpdatedAt int64    `json:"endpoint_updated_at_ns"`
}

type ServiceDiscoveryStartMessage

type ServiceDiscoveryStartMessage struct {
	Id                               string `json:"id"`
	Host                             string `json:"host"`
	MinimumRegisterIntervalInSeconds int    `json:"minimumRegisterIntervalInSeconds"`
	PruneThresholdInSeconds          int    `json:"pruneThresholdInSeconds"`
}

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

func NewSubscriber

func NewSubscriber(
	natsConnProvider NatsConnProvider,
	subOpts SubscriberOpts,
	warmingDuration time.Duration,
	table AddressTable,
	localIP string,
	recorder routeMessageRecorder,
	logger lager.Logger,
	metricsSender metricsSender,
	clock clock.Clock,
) *Subscriber

func (*Subscriber) Close

func (s *Subscriber) Close()

func (*Subscriber) Run

func (s *Subscriber) Run(signals <-chan os.Signal, ready chan<- struct{}) error

func (*Subscriber) RunOnce

func (s *Subscriber) RunOnce() error

type SubscriberOpts

type SubscriberOpts struct {
	ID                               string
	MinimumRegisterIntervalInSeconds int
	PruneThresholdInSeconds          int
	AcceptTLS                        bool
}