async

Documentation

Index

• Constants
• Variables
• type Delegate
• type Option
  • func WithName(name string) Option
  • func WithQSize(qSize int) Option
  • func WithWaitGroup(wg *sync.WaitGroup) Option
• type Proc
• type ProcChan
  • func NewProcChan(opts ...Option) *ProcChan
  • func (c *ProcChan) AsyncProc(ctx context.Context, proc Proc) (interface{}, error)
  • func (c *ProcChan) Run()
  • func (c *ProcChan) Size() int
  • func (c *ProcChan) Stop()
  • func (c *ProcChan) WaitStop()
• type Q
  • func NewQ(size int) *Q
  • func (a *Q) Add(req interface{}) error
  • func (a *Q) AddAnyway(req interface{}, ts time.Duration) error
  • func (a *Q) AddPrior(req interface{}) error
  • func (a *Q) Close()
  • func (a *Q) IsClosed() bool
  • func (a *Q) Pop() (interface{}, error)
  • func (a *Q) PopAnyway() (interface{}, error)
  • func (a *Q) Size() int
• type RunnerQ
  • func NewRunnerQ(opts ...Option) *RunnerQ
  • func (c *RunnerQ) AsyncCall(fn interface{}, ctx context.Context, arg interface{}) (interface{}, error)
  • func (c *RunnerQ) AsyncDelegate(ctx context.Context, delegate Delegate) (interface{}, error)
  • func (c *RunnerQ) AsyncProc(ctx context.Context, proc Proc) (interface{}, error)
  • func (c *RunnerQ) Run()
  • func (c *RunnerQ) Size() int
  • func (c *RunnerQ) Stop()
  • func (c *RunnerQ) WaitStop()

Constants

const (
	//DefaultQSize default runner queue size
	DefaultQSize = 1024 * 8
)

Variables

var (
	//ErrClosed -- msg queue closed
	ErrClosed = status.Error(codes.Unavailable, "async.queue.closed")
	//ErrFull -- msg queue is full
	ErrFull = status.Error(codes.ResourceExhausted, "async.queue.full")
	//ErrSync it should never gonna happen
	ErrSync = status.Error(codes.OutOfRange, "async.never.gonna.happen.crazy")
)

Types

type Delegate

type Delegate func(ctx context.Context) (interface{}, error)

Delegate : proc delegate function

type Option

type Option func(option *optionT)

Option : only qSize option

func WithName

func WithName(name string) Option

WithName : setup name

func WithQSize

func WithQSize(qSize int) Option

WithQSize : setup qSize

func WithWaitGroup

func WithWaitGroup(wg *sync.WaitGroup) Option

WithWaitGroup : setup outside wait group controller If this value be set, the wait group will be increase when runner loop go routine exits.

type Proc

type Proc interface {
	//Do : do proc
	Do(ctx context.Context) (interface{}, error)
}

Proc : proc interface

type ProcChan

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

ProcChan : async proc chan

func NewProcChan

func NewProcChan(opts ...Option) *ProcChan

NewProcChan : new async proc queue

func (*ProcChan) AsyncProc

func (c *ProcChan) AsyncProc(ctx context.Context, proc Proc) (interface{}, error)

AsyncProc : async proc ctx -- context.Context proc -- proc interface

func (*ProcChan) Run

func (c *ProcChan) Run()

Run : run all queue msg handler

func (*ProcChan) Size

func (c *ProcChan) Size() int

Size : get queue size

func (*ProcChan) Stop

func (c *ProcChan) Stop()

Stop : stop

func (*ProcChan) WaitStop

func (c *ProcChan) WaitStop()

WaitStop : wait runner loop exits

type Q

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

Q actor queue structure define

func NewQ

func NewQ(size int) *Q

NewQ new queue size: queue size, if set 0, it means that no limitation of size, it's an infinity queue.

func (*Q) Add

func (a *Q) Add(req interface{}) error

Add : add normal request to the normal queue end place.

func (*Q) AddAnyway

func (a *Q) AddAnyway(req interface{}, ts time.Duration) error

AddAnyway add normal request to the normal queue end place anyway if queue full, sleep then try

func (*Q) AddPrior

func (a *Q) AddPrior(req interface{}) error

AddPrior add prior request to queue first place.

func (*Q) Close

func (a *Q) Close()

Close : close the queue

func (*Q) IsClosed

func (a *Q) IsClosed() bool

IsClosed return queue is closed or not

func (*Q) Pop

func (a *Q) Pop() (interface{}, error)

Pop consume an item, if list is empty, it's been blocked

func (*Q) PopAnyway

func (a *Q) PopAnyway() (interface{}, error)

PopAnyway consume an item like Pop, but it can consume even the queue is closed.

func (*Q) Size

func (a *Q) Size() int

Size return queue size

type RunnerQ

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

RunnerQ : async runner

func NewRunnerQ

func NewRunnerQ(opts ...Option) *RunnerQ

NewRunnerQ : new async runner queue

func (*RunnerQ) AsyncCall

func (c *RunnerQ) AsyncCall(fn interface{}, ctx context.Context, arg interface{}) (interface{}, error)

AsyncCall : async call ctx -- context.Context callCtxT -- call context

func (*RunnerQ) AsyncDelegate

func (c *RunnerQ) AsyncDelegate(ctx context.Context, delegate Delegate) (interface{}, error)

AsyncDelegate : async delegate ctx -- context.Context delegate -- delegate function

func (*RunnerQ) AsyncProc

func (c *RunnerQ) AsyncProc(ctx context.Context, proc Proc) (interface{}, error)

AsyncProc : async proc ctx -- context.Context proc -- proc interface

func (*RunnerQ) Run

func (c *RunnerQ) Run()

Run : run all queue msg handler

func (*RunnerQ) Size

func (c *RunnerQ) Size() int

Size : get queue size

func (*RunnerQ) Stop

func (c *RunnerQ) Stop()

Stop : stop

func (*RunnerQ) WaitStop

func (c *RunnerQ) WaitStop()

WaitStop : wait runner loop exits