func (eventsSourceProvider) Events(address string, cert *cert.Info) (Client, EventsSource, error) {
	rawClient, err := client.New(
		fmt.Sprintf("https://%s", address),
		client.WithTLSServerCert(string(cert.PublicKey())),
		client.WithTLSClientCert(string(cert.PublicKey())),
		client.WithTLSClientKey(string(cert.PrivateKey())),
		client.WithUserAgent("cluster-notifier"),
	)
	if err != nil {
		return nil, nil, errors.WithStack(err)
	}
	return rawClient, events.NewEvents(rawClient), nil
}
Connect configures the client with the correct parameters for
node-to-node communication.
package events
import (
	"context"
	"fmt"
	"time"
	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/gorilla/websocket"
	"github.com/pkg/errors"
	"github.com/spoke-d/thermionic/internal/cert"
	"github.com/spoke-d/thermionic/internal/clock"
	"github.com/spoke-d/thermionic/internal/db"
	"github.com/spoke-d/task"
	"github.com/spoke-d/thermionic/pkg/client"
	"github.com/spoke-d/thermionic/pkg/events"
)
type Cluster interface {
	db.ClusterTransactioner
}
type Endpoints interface {
	NetworkAddress() string
	NetworkCert() *cert.Info
}
type EventHook func(int64, interface{})
type Client interface {
	Query(string, string, interface{}, string) (*client.Response, string, error)
	Websocket(path string) (*websocket.Conn, error)
}
type EventsSourceProvider interface {
	Events(string, *cert.Info) (Client, EventsSource, error)
}
type EventsSource interface {
	GetEvents() (*events.EventListener, error)
}
const Interval = 1
type Events struct {
	endpoints            Endpoints
	cluster              Cluster
	hook                 EventHook
	eventsSourceProvider EventsSourceProvider
	clock                clock.Clock
	logger               log.Logger
}
func New(endpoints Endpoints, cluster Cluster, hook EventHook, options ...Option) *Events {
	opts := newOptions()
	for _, option := range options {
		option(opts)
	}
	return &Events{
		endpoints:            endpoints,
		cluster:              cluster,
		hook:                 hook,
		eventsSourceProvider: opts.eventsSourceProvider,
		clock:                opts.clock,
		logger:               opts.logger,
	}
}
func (e *Events) Run() (task.Func, task.Schedule) {
	listeners := make(map[int64]*events.EventListener)
	eventsWrapper := func(ctx context.Context) {
		ch := make(chan struct{})
		go func() {
			e.updateListeners(listeners)
			ch <- struct{}{}
		}()
		select {
		case <-ch:
		case <-ctx.Done():
		}
	}
	schedule := task.Every(time.Duration(Interval) * time.Second)
	return eventsWrapper, schedule
}
func (e *Events) updateListeners(listeners map[int64]*events.EventListener) {
	var nodes []db.NodeInfo
	var offlineThreshold time.Duration
	if err := e.cluster.Transaction(func(tx *db.ClusterTx) error {
		var err error
		if nodes, err = tx.Nodes(); err != nil {
			return errors.WithStack(err)
		}
		if offlineThreshold, err = tx.NodeOfflineThreshold(); err != nil {
			return errors.WithStack(err)
		}
		return nil
	}); err != nil {
		level.Warn(e.logger).Log("msg", "failed to get current cluster nodes", "err", err)
		return
	}
	if len(nodes) == 1 {
		return
	}
	address := e.endpoints.NetworkAddress()
	ids := make([]int, len(nodes))
	for i, node := range nodes {
		ids[i] = int(node.ID)
		if node.IsOffline(e.clock, offlineThreshold) || node.Address == address {
			continue
		}
		_, ok := listeners[node.ID]
		if ok {
			if listeners[node.ID].IsActive() {
				continue
			}
			delete(listeners, node.ID)
		}
		_, source, err := e.eventsSourceProvider.Events(node.Address, e.endpoints.NetworkCert())
		if err != nil {
			continue
		}
		listener, err := source.GetEvents()
		if err != nil {
			level.Warn(e.logger).Log("msg", "failed to get events from node", "address", node.Address, "err", err)
			continue
		}
		level.Debug(e.logger).Log("msg", "listening for events on node", "address", node.Address)
		listener.AddHandler(nil, func(event interface{}) {
			e.hook(node.ID, event)
		})
		listeners[node.ID] = listener
	}
	for id, listener := range listeners {
		if !contains(ids, int(id)) {
			listener.Disconnect()
			delete(listeners, id)
		}
	}
}
func contains(ids []int, id int) bool {
	for _, v := range ids {
		if v == id {
			return true
		}
	}
	return false
}
type eventsSourceProvider struct{}