structs

Documentation

Index

• Variables
• type Await
  • func NewAwait(sendIDs []uuid.UUID) (aw *Await)
  • func (aw *Await) Close()
  • func (aw *Await) Send(tr *TaskResponse) (bool, error)
• type ClientControl
• type ClientResponse
• type ConnTransport
• type IndexerClienter
• type SimpleWorkerInfo
• type StreamAccess
  • func NewStreamAccess(transport ConnTransport, managerID string, conn *WorkerInfo) *StreamAccess
  • func (sa *StreamAccess) Close() error
  • func (sa *StreamAccess) Ping(ctx context.Context) (time.Duration, error)
  • func (sa *StreamAccess) Reconnect(ctx context.Context, logger *zap.Logger) error
  • func (sa *StreamAccess) Recv(tr *TaskResponse) error
  • func (sa *StreamAccess) Req(tr TaskRequest, aw *Await) error
  • func (sa *StreamAccess) Run(ctx context.Context, logger *zap.Logger) error
• type StreamState
• type TaskError
• type TaskErrorType
• type TaskRequest
• type TaskResponse
• type WorkerAddress
• type WorkerCompositeKey
• type WorkerConnection
• type WorkerInfo

Variables

var ErrStreamNotOnline = errors.New("Stream is not Online")

Types

type Await

type Await struct {
	sync.RWMutex
	Created        time.Time
	State          StreamState
	Resp           chan *TaskResponse
	ReceivedFinals int
	Uids           []uuid.UUID
}

func NewAwait

func NewAwait(sendIDs []uuid.UUID) (aw *Await)

func (*Await) Close

func (aw *Await) Close()

func (*Await) Send

func (aw *Await) Send(tr *TaskResponse) (bool, error)

type ClientControl

type ClientControl struct {
	Type string
	Resp chan ClientResponse
}

type ClientResponse

type ClientResponse struct {
	OK    bool
	Error string
	Time  time.Duration
}

type ConnTransport

type ConnTransport interface {
	Run(ctx context.Context, logger *zap.Logger, stream *StreamAccess)
	Type() string
}

type IndexerClienter

type IndexerClienter interface {
	RegisterStream(ctx context.Context, stream *StreamAccess) error
}

type SimpleWorkerInfo

type SimpleWorkerInfo struct {
	Network string
	Version string
	ChainID string
	ID      string
}

type StreamAccess

type StreamAccess struct {
	State           StreamState
	StreamID        uuid.UUID
	ResponseMap     map[uuid.UUID]*Await
	RequestListener chan TaskRequest

	ManagerID string

	ClientControl chan ClientControl

	CancelConnection context.CancelFunc

	WorkerInfo *WorkerInfo
	Transport  ConnTransport
	// contains filtered or unexported fields
}

func NewStreamAccess

func NewStreamAccess(transport ConnTransport, managerID string, conn *WorkerInfo) *StreamAccess

func (*StreamAccess) Close

func (sa *StreamAccess) Close() error

func (*StreamAccess) Ping

func (sa *StreamAccess) Ping(ctx context.Context) (time.Duration, error)

func (*StreamAccess) Reconnect

func (sa *StreamAccess) Reconnect(ctx context.Context, logger *zap.Logger) error

func (*StreamAccess) Recv

func (sa *StreamAccess) Recv(tr *TaskResponse) error

func (*StreamAccess) Req

func (sa *StreamAccess) Req(tr TaskRequest, aw *Await) error

func (*StreamAccess) Run

func (sa *StreamAccess) Run(ctx context.Context, logger *zap.Logger) error

type StreamState

type StreamState int

const (
	StreamUnknown StreamState = iota
	StreamOnline
	StreamError
	StreamReconnecting
	StreamClosing
	StreamOffline
)

type TaskError

type TaskError struct {
	Msg  string
	Type TaskErrorType
}

type TaskErrorType

type TaskErrorType string

type TaskRequest

type TaskRequest struct {
	ID      uuid.UUID
	Network string
	ChainID string
	Version string

	Type    string
	Payload json.RawMessage
}

type TaskResponse

type TaskResponse struct {
	ID      uuid.UUID
	Version string
	Type    string
	Order   int64
	Final   bool
	Error   TaskError
	Payload json.RawMessage
}

type WorkerAddress

type WorkerAddress struct {
	IP      net.IP `json:"ip"`
	Address string `json:"address"`
}

type WorkerCompositeKey

type WorkerCompositeKey struct {
	Network string
	ChainID string
	Version string
}

type WorkerConnection

type WorkerConnection struct {
	Version   string          `json:"version"`
	Type      string          `json:"type"`
	Addresses []WorkerAddress `json:"addresses"`
}

type WorkerInfo

type WorkerInfo struct {
	NodeSelfID     string             `json:"node_id"`
	Type           string             `json:"type"`
	ChainID        string             `json:"chain_id"`
	State          StreamState        `json:"state"`
	ConnectionInfo []WorkerConnection `json:"connection"`
	LastCheck      time.Time          `json:"last_check"`
}