connectivity

Documentation

Index

• Variables
• type ConnectivityInfo
• type Manager
  • func NewManager(id string, logger *zap.Logger) *Manager
  • func (m *Manager) AddTransport(c structs.ConnTransport) error
  • func (m *Manager) AttachToMux(mux *http.ServeMux)
  • func (m *Manager) GetAllWorkers() map[string]WorkerNetworkStatic
  • func (m *Manager) GetWorkers(kind string) []structs.WorkerInfo
  • func (m *Manager) Register(id, network, chain string, connInfo structs.WorkerConnection) error
  • func (m *Manager) Send(trs []structs.TaskRequest) (*structs.Await, error)
  • func (m *Manager) Unregister(id, kind, chain, version string) error
• type Network
• type PingInfo
• type RoundRobinWorkers
  • func NewRoundRobinWorkers() *RoundRobinWorkers
  • func (rrw *RoundRobinWorkers) AddWorker(id string, stream *structs.StreamAccess) error
  • func (rrw *RoundRobinWorkers) BringOnline(id string) error
  • func (rrw *RoundRobinWorkers) Close(id string) error
  • func (rrw *RoundRobinWorkers) GetWorker(id string) (twi TaskWorkerInfo, ok bool)
  • func (rrw *RoundRobinWorkers) GetWorkers() TaskWorkerRecordInfo
  • func (rrw *RoundRobinWorkers) Ping(ctx context.Context, id string) (time.Duration, error)
  • func (rrw *RoundRobinWorkers) Reconnect(ctx context.Context, logger *zap.Logger, id string) error
  • func (rrw *RoundRobinWorkers) SendNext(tr structs.TaskRequest, aw *structs.Await) (failedWorkerID string, err error)
  • func (rrw *RoundRobinWorkers) SendToWoker(id string, tr structs.TaskRequest, aw *structs.Await) error
• type TaskWorkerInfo
• type TaskWorkerInfoAwait
• type TaskWorkerRecord
• type TaskWorkerRecordInfo
• type WorkerInfoStatic
• type WorkerNetworkStatic
• type WorkersPool

Variables

var (
	ErrStreamOffline       = errors.New("stream is Offline")
	ErrNoWorkersAvailable  = errors.New("no workers available")
	ErrWorkerDoesNotExists = errors.New("workers does not exists")
)

Types

type ConnectivityInfo

type ConnectivityInfo struct {
	Address string `json:"address"`
	Version string `json:"version"`
	Type    string `json:"type"`
}

type Manager

type Manager struct {
	ID string
	// contains filtered or unexported fields
}

func NewManager

func NewManager(id string, logger *zap.Logger) *Manager

NewManager is Manager constructor

func (*Manager) AddTransport

func (m *Manager) AddTransport(c structs.ConnTransport) error

AddTransport for connectivity

func (*Manager) AttachToMux

func (m *Manager) AttachToMux(mux *http.ServeMux)

func (*Manager) GetAllWorkers

func (m *Manager) GetAllWorkers() map[string]WorkerNetworkStatic

GetAllWorkers returns static list of workers

func (*Manager) GetWorkers

func (m *Manager) GetWorkers(kind string) []structs.WorkerInfo

GetWorkers gets workers of kind

func (*Manager) Register

func (m *Manager) Register(id, network, chain string, connInfo structs.WorkerConnection) error

Register new worker in Manager

func (*Manager) Send

func (m *Manager) Send(trs []structs.TaskRequest) (*structs.Await, error)

Send sends a set of requests

func (*Manager) Unregister

func (m *Manager) Unregister(id, kind, chain, version string) error

Unregister unregistring worker

type Network

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

type PingInfo

type PingInfo struct {
	ID           string           `json:"id"`
	Network      string           `json:"network"`
	ChainID      string           `json:"chain_id"`
	Connectivity ConnectivityInfo `json:"connectivity"`
}

PingInfo contract is defined here

type RoundRobinWorkers

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

func NewRoundRobinWorkers

func NewRoundRobinWorkers() *RoundRobinWorkers

func (*RoundRobinWorkers) AddWorker

func (rrw *RoundRobinWorkers) AddWorker(id string, stream *structs.StreamAccess) error

func (*RoundRobinWorkers) BringOnline

func (rrw *RoundRobinWorkers) BringOnline(id string) error

BringOnline Brings worker Online, removing duplicates

func (*RoundRobinWorkers) Close

func (rrw *RoundRobinWorkers) Close(id string) error

Close closes worker of given id

func (*RoundRobinWorkers) GetWorker

func (rrw *RoundRobinWorkers) GetWorker(id string) (twi TaskWorkerInfo, ok bool)

func (*RoundRobinWorkers) GetWorkers

func (rrw *RoundRobinWorkers) GetWorkers() TaskWorkerRecordInfo

GetWorkers returns current workers information

func (*RoundRobinWorkers) Ping

func (rrw *RoundRobinWorkers) Ping(ctx context.Context, id string) (time.Duration, error)

func (*RoundRobinWorkers) Reconnect

func (rrw *RoundRobinWorkers) Reconnect(ctx context.Context, logger *zap.Logger, id string) error

Reconnect reconnects stream if exists

func (*RoundRobinWorkers) SendNext

func (rrw *RoundRobinWorkers) SendNext(tr structs.TaskRequest, aw *structs.Await) (failedWorkerID string, err error)

func (*RoundRobinWorkers) SendToWoker

func (rrw *RoundRobinWorkers) SendToWoker(id string, tr structs.TaskRequest, aw *structs.Await) error

SendToWoker sends task to worker

type TaskWorkerInfo

type TaskWorkerInfo struct {
	WorkerID string    `json:"worker_id"`
	LastSend time.Time `json:"last_send"`

	StreamState structs.StreamState `json:"stream_state"`
	StreamID    uuid.UUID           `json:"stream_id"`
	ResponseMap map[uuid.UUID]TaskWorkerInfoAwait
}

type TaskWorkerInfoAwait

type TaskWorkerInfoAwait struct {
	Created        time.Time           `json:"created"`
	State          structs.StreamState `json:"state"`
	ReceivedFinals int                 `json:"received_finals"`
	Uids           []uuid.UUID         `json:"expected"`
}

type TaskWorkerRecord

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

type TaskWorkerRecordInfo

type TaskWorkerRecordInfo struct {
	Workers []TaskWorkerInfo `json:"workers"`
	All     int              `json:"all"`
	Active  int              `json:"active"`
}

type WorkerInfoStatic

type WorkerInfoStatic struct {
	structs.WorkerInfo
	TaskWorkerInfo TaskWorkerInfo `json:"tasks"`
}

type WorkerNetworkStatic

type WorkerNetworkStatic struct {
	Workers map[string]WorkerInfoStatic `json:"workers"`
	All     int                         `json:"all"`
	Active  int                         `json:"active"`
}

type WorkersPool

type WorkersPool interface {
	AddWorker(id string, stream *structs.StreamAccess) error
	SendNext(tr structs.TaskRequest, aw *structs.Await) (failedWorkerID string, err error)
	Ping(ctx context.Context, id string) (time.Duration, error)
	Close(id string) error
	BringOnline(id string) error
	Reconnect(ctx context.Context, logger *zap.Logger, id string) error
	GetWorker(id string) (twi TaskWorkerInfo, ok bool)
	GetWorkers() TaskWorkerRecordInfo
	SendToWoker(id string, tr structs.TaskRequest, aw *structs.Await) error
}