pod/

Apr 8, 2020

Changes in this version

+ const AdminURI

+ const RedisRequestConfigKey

+ const SpcKeyReplace

+ var DefaultRequestConfig = &RequestConfig

+ func NetlocFromString(s string) (*netloc.Netloc, error)

+ type MergeFunc func(*request.Request, *RequestConfig, bool)

+ type NetlocRequestConfig struct

+ Netloc *netloc.Netloc

+ Rule *RequestConfig

type QueueBox

+ func (box *QueueBox) Cleanup() error

+ func (box *QueueBox) Setup() (err error)

+ type RequestConfig struct

+ Body string

+ Cookies map[string]string

+ Headers map[string]string

+ Meta map[string]interface{}

+ Method string

+ type RequestWrapper struct

+ func NewRequestWrapper(core *Core) *RequestWrapper

+ func (wrapper *RequestWrapper) Add(nlc *netloc.Netloc, reqConfig *RequestConfig) (*netloc.Netloc, *RequestConfig, error)

+ func (wrapper *RequestWrapper) Cleanup() error

+ func (wrapper *RequestWrapper) Delete(nlc *netloc.Netloc) (*RequestConfig, error)

+ func (wrapper *RequestWrapper) Get(nlc *netloc.Netloc) (*netloc.Netloc, *RequestConfig)

+ func (wrapper *RequestWrapper) GetAll() []*NetlocRequestConfig

+ func (wrapper *RequestWrapper) Load() error

+ func (wrapper *RequestWrapper) MatchURI(uri string) (*netloc.Netloc, *RequestConfig)

+ func (wrapper *RequestWrapper) Merge(req *request.Request, reqConfig *RequestConfig, force bool)

+ func (wrapper *RequestWrapper) Setup() error

+ func (wrapper *RequestWrapper) Wrap(req *request.Request) (*netloc.Netloc, *RequestConfig)

Apr 2, 2020

Changes in this version

type QueueStatus

+ func (i QueueStatus) String() string

Mar 31, 2020

Changes in this version

+ type ExistError string

+ func (e ExistError) Error() string

Mar 31, 2020

Changes in this version

type Core

+ func New(conf *viper.Viper, client *redis.Client) (core *Core)

Mar 31, 2020

Changes in this version

+ type Core struct

+ QueueBox *QueueBox

+ func (core *Core) Info() (sth.Result, error)

+ func (core *Core) OnStart() (err error)

+ func (core *Core) OnStop() error

+ func (core *Core) Switch(pauseOrResume bool) (sth.Result, error)

type Queue

+ func (queue *Queue) ItemID() uint64

+ func (queue *Queue) Pop() (req *request.Request, qsize int64, err error)

+ func (queue *Queue) Push(request *request.Request) (result sth.Result, err error)

type QueueBox

+ func (box *QueueBox) Load() (err error)

+ func (box *QueueBox) PauseQueue(qid sth.QueueID) (sth.Result, error)

+ func (box *QueueBox) PopRequest(qid sth.QueueID) (req *request.Request, err error)

+ func (box *QueueBox) PushRequest(raw *request.RawRequest) (result sth.Result, err error)

+ func (box *QueueBox) ResumeQueue(qid sth.QueueID) (sth.Result, error)

Mar 26, 2020

Changes in this version

+ type NotExistError string

+ func (e NotExistError) Error() string

type QueueBox

+ func (box *QueueBox) QueuesCount(status QueueStatus) int

Mar 24, 2020

Changes in this version

+ var StatusMap = map[string]Status

type QueueStatus

+ func (s *QueueStatus) UnmarshalJSON(b []byte) error

+ func (s QueueStatus) MarshalJSON() ([]byte, error)

type Status

+ func (s *Status) UnmarshalJSON(b []byte) error

+ func (s Status) MarshalJSON() ([]byte, error)

Mar 23, 2020

Changes in this version

type Queue

+ func (queue *Queue) SetStatusOn(newStatus QueueStatus, on QueueStatus) (err error)

Mar 22, 2020

Changes in this version

+ const RedisPausedQueuesKey

type QueueBox

+ func (box *QueueBox) LoadPaused() (err error)

+ func (box *QueueBox) SetStatus(qid QueueID, newStatus QueueStatus) (Result, error)

type QueueID

+ func QueueIDFromString(key string) (qid QueueID, err error)

+ type ResultAndErrorFunc func() (Result, error)

Mar 20, 2020

Changes in this version

type QueueBox

+ func (box *QueueBox) SetQueueStatus(qid QueueID, status QueueStatus) (Result, error)

+ type ReturnResultAndError func() (Result, error)

Mar 19, 2020

Changes in this version

type QueueStatus

+ const QueueInit

+ const QueueRemoved

type Status

+ const Init

Mar 19, 2020

Changes in this version

+ func MustNewMinuteWindow() *average.SlidingWindow

+ func WindowTotal(window *average.SlidingWindow, second time.Duration) int64

+ type CallByQueue func(*Queue) (Result, error)

type QueueBox

+ func (box *QueueBox) AddRequest(qid QueueID, req *request.Request) (result Result, err error)

+ func (box *QueueBox) ClearQueue(qid QueueID) (result Result, err error)

+ func (box *QueueBox) DeleteIdleQueue(qid QueueID) (err error)

+ func (box *QueueBox) DeleteQueue(qid QueueID) (result Result, err error)

+ func (box *QueueBox) GetRequest(qid QueueID) (result Result, err error)

+ func (box *QueueBox) LoadQueues() (err error)

+ func (box *QueueBox) QueueInfo(qid QueueID) (result Result, err error)

+ func (box *QueueBox) QueuesNum(status QueueStatus) int

+ func (box *QueueBox) SyncQueue(qid QueueID, force bool) (result Result, err error)

+ func (box *QueueBox) ViewQueue(qid QueueID, start int64, end int64) (result Result, err error)

Mar 17, 2020

Changes in this version

+ var QueueStatusMap = map[string]QueueStatus

type Pod

+ func (pod *Pod) Queues(k int) (result Result, err error)

+ func (pod *Pod) ViewQueues(k int, start int, status QueueStatus) Result

type QueueBox

+ func (box *QueueBox) Queues(k int) Result

+ func (box *QueueBox) ViewQueues(k int, start int, status QueueStatus) Result

Mar 14, 2020

Changes in this version

+ const RedisConfigKey

+ var QueueStatusList = []QueueStatus

+ type ExceedLimitError string

+ func (e ExceedLimitError) Error() string

+ type Limiter struct

+ func NewLimiter(pod *Pod) *Limiter

+ func (limiter *Limiter) AllowedNewRequest(request *request.Request) error

type Pod

+ func (pod *Pod) LoadQueues() (err error)

+ func (pod *Pod) OnStart() (err error)

+ func (pod *Pod) OnStop() (err error)

type QueueID

+ func (qid QueueID) ItemID() uint64

type QueueStatus

+ const QueueUndefined

type Status

+ const Stopped

Mar 10, 2020

Changes in this version

+ const QueueNotExist

+ const RedisQueueKeyPrefix

+ func LoadQueues(pod *Pod) (err error)

+ func RedisKeyFromQueueID(qid QueueID) string

+ type CallByQueueID struct

+ func NewCallByQueueIDOperation(pod *Pod, f FuncWithQueueID) *CallByQueueID

+ func (opt *CallByQueueID) Operate(id QueueID) (result Result, err error)

+ type FuncWithQueueID func(QueueID) (Result, error)

+ type IdxQueue struct

+ Idx int

+ func NewIdxQueue(pod *Pod, qid QueueID, status QueueStatus, idx int) *IdxQueue

+ type Pod struct

+ Client *redis.Client

+ Process *process.Process

+ func NewPod(conf *viper.Viper, client *redis.Client) (pod *Pod, err error)

+ func (pod *Pod) AddRequest(rawReq *request.RawRequest) (result Result, err error)

+ func (pod *Pod) ClearQueue(qid QueueID) (Result, error)

+ func (pod *Pod) Conf() *viper.Viper

+ func (pod *Pod) DeleteQueue(qid QueueID) (Result, error)

+ func (pod *Pod) ForceSyncQueue(qid QueueID) (result Result, err error)

+ func (pod *Pod) GetRequest(qid QueueID) (result Result, err error)

+ func (pod *Pod) Info() (result Result, err error)

+ func (pod *Pod) OrderedQueues(k int64, start int64, status QueueStatus) Result

+ func (pod *Pod) Pause() (Result, error)

+ func (pod *Pod) PauseQueue(qid QueueID) (result Result, err error)

+ func (pod *Pod) QueueInfo(qid QueueID) (Result, error)

+ func (pod *Pod) RandomQueues(k int64, status QueueStatus) Result

+ func (pod *Pod) Resume() (Result, error)

+ func (pod *Pod) ResumeQueue(qid QueueID) (result Result, err error)

+ func (pod *Pod) SyncQueue(qid QueueID) (result Result, err error)

+ func (pod *Pod) ViewQueue(qid QueueID, start int64, end int64) (result Result, err error)

+ type Queue struct

+ ID QueueID

+ func NewQueue(pod *Pod, id QueueID, status QueueStatus) *Queue

+ func (queue *Queue) Clear() (result Result, err error)

+ func (queue *Queue) Dequeuing() int64

+ func (queue *Queue) Get() (result Result, err error)

+ func (queue *Queue) Idle() bool

+ func (queue *Queue) Info() (result Result, err error)

+ func (queue *Queue) Put(request *request.Request) (Result, error)

+ func (queue *Queue) QueueSize() int64

+ func (queue *Queue) Queuing() int64

+ func (queue *Queue) SetStatus(status QueueStatus) (result Result, err error)

+ func (queue *Queue) Status() QueueStatus

+ func (queue *Queue) Sync(lock bool) (result Result, err error)

+ func (queue *Queue) View(start int64, end int64) (result Result, err error)

+ type QueueBox struct

+ func NewQueueBox(pod *Pod) *QueueBox

+ func (box *QueueBox) AddQueue(qid QueueID) (queue *IdxQueue, err error)

+ func (box *QueueBox) GetQueue(qid QueueID) (*IdxQueue, bool)

+ func (box *QueueBox) Info() (result Result)

+ func (box *QueueBox) OrderedQueues(k int64, start int64, status QueueStatus) Result

+ func (box *QueueBox) QueueNum(status QueueStatus) int

+ func (box *QueueBox) RandomQueues(k int64, status QueueStatus) Result

+ func (box *QueueBox) RemoveQueue(qid QueueID) error

+ func (box *QueueBox) UpdateQueueStatus(qid QueueID, status QueueStatus) (result Result, err error)

+ type QueueID struct

+ Host string

+ Port string

+ Scheme string

+ func CreateQueueID(host, port, scheme string) QueueID

+ func QueueIDFromRedisKey(key string) (qid QueueID, err error)

+ func QueueIDFromRequest(req *request.Request) QueueID

+ func (qid QueueID) MarshalJSON() ([]byte, error)

+ func (qid QueueID) String() string

+ type QueueNotExistError string

+ func (e QueueNotExistError) Error() string

+ type QueueOperation interface

+ Operate func(id QueueID) (Result, error)

+ type QueueStatus string

+ const QueueNilStatus

+ const QueuePaused

+ const QueueWorking

+ type Result map[string]interface

+ type Stats struct

+ func NewStats() *Stats

+ func (stats *Stats) IncrRequestNum(d int64) int64

+ func (stats *Stats) RequestNum() int64

+ func (stats *Stats) Stats() Result

+ func (stats *Stats) UpdateInput(d int64)

+ func (stats *Stats) UpdateOutput(d int64)

+ type Status string

+ const Paused

+ const Preparing

+ const Stopping

+ const Working

+ type UnavailableError string

+ func (e UnavailableError) Error() string

+ type ViewQueue struct

+ End int64

+ R *Pod

+ Start int64

+ func NewViewQueueOperation(pod *Pod, start, end int64) *ViewQueue

+ func (opt *ViewQueue) Operate(id QueueID) (Result, error)