aggmq

Documentation

Index

• Constants
• Variables
• type AggComm
  • func (comm *AggComm) Destroy()
  • func (comm *AggComm) RecvAgg() <-chan AggMqMsg
  • func (comm *AggComm) RecvCmd() <-chan Command
  • func (comm *AggComm) SendAgg(agg AggMqMsg)
  • func (comm *AggComm) SendCmd(cmd int, seq uint64) error
  • func (comm *AggComm) SendResp(err error)
• type AggMqMsg
• type AggProcessorConfig
• type AggProcessorReq
• type Command
• type MQ
  • func NewMQ() *MQ
  • func (mq *MQ) AskProcessor(apc AggProcessorConfig) (*AggComm, error)
• type Response

Constants

const (

	// CmdWaitSlaveSync is command given to slave syncer.
	// It indicates to the syncer that we want to wait for slave syncer to finish it's work
	CmdWaitSlaveSync

	// CmdRestartSlaveSyncer is command to restart the slave syncer
	CmdRestartSlaveSyncer

	// CmdKillMe is command to kill the slave syncer
	CmdKillMe
)

Variables

var (
	// ErrSendCmdTimeout returned when the command doesn't get any response after
	// some amount of time
	ErrSendCmdTimeout = errors.New("send command timed out")
)

var (
	// ErrSlaveSyncTimeout returned when slave syncer doesn't give response
	// until some amount of time
	ErrSlaveSyncTimeout = errors.New("slave sync timed out")
)

Types

type AggComm

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

AggComm defines communication channels between aggregation producer and consumer/processor

func (*AggComm) Destroy

func (comm *AggComm) Destroy()

Destroy destroys this communication channel. It also means destroying the consumer

func (*AggComm) RecvAgg

func (comm *AggComm) RecvAgg() <-chan AggMqMsg

RecvAgg recv aggregation from this comm object

func (*AggComm) RecvCmd

func (comm *AggComm) RecvCmd() <-chan Command

RecvCmd receive command from this comm object

func (*AggComm) SendAgg

func (comm *AggComm) SendAgg(agg AggMqMsg)

SendAgg send aggregation via this comm object

func (*AggComm) SendCmd

func (comm *AggComm) SendCmd(cmd int, seq uint64) error

SendCmd send command using this comm object

func (*AggComm) SendResp

func (comm *AggComm) SendResp(err error)

SendResp send response using this comm object

type AggMqMsg

type AggMqMsg []byte

AggMqMsg defines message given to the aggregation processor

type AggProcessorConfig

type AggProcessorConfig struct {
	VdiskID      string
	DataShards   int
	ParityShards int
	PrivKey      string
}

AggProcessorConfig defines config given to the aggregation processor

type AggProcessorReq

type AggProcessorReq struct {
	Comm   *AggComm
	Config AggProcessorConfig
}

AggProcessorReq defines request to the processor provider

type Command

type Command struct {
	Type int
	Seq  uint64
}

Command defines command from producer to consumer/processor

type MQ

type MQ struct {
	NeedProcessorCh   chan AggProcessorReq
	NeedProcessorResp chan error
	Comms             map[string]*AggComm
	// contains filtered or unexported fields
}

MQ defines this message queue

func NewMQ

func NewMQ() *MQ

NewMQ creates new MQ

func (*MQ) AskProcessor

func (mq *MQ) AskProcessor(apc AggProcessorConfig) (*AggComm, error)

AskProcessor as for aggregation processor. we currently only have slave syncer

type Response

type Response struct {
	Err error
}

Response defines response from consumer/processor to producer