base

Documentation

Overview

Package base Secondary level of core library.

Index

• Variables
• type ConditionManager
• type MemoryManager
• type Message
  • func (o *Message) GetContext() context.Context
  • func (o *Message) GetError() error
  • func (o *Message) GetIgnored() bool
  • func (o *Message) Release()
  • func (o *Message) SetBody(b []byte) *Message
  • func (o *Message) SetContext(c context.Context) *Message
  • func (o *Message) SetDuration(d float64) *Message
  • func (o *Message) SetError(e error) *Message
  • func (o *Message) SetIgnored(i bool) *Message
• type Notification
  • func (o *Notification) Parse(s string) error
  • func (o *Notification) Release()
• type Payload
  • func (o *Payload) GetContext() context.Context
  • func (o *Payload) GetError() error
  • func (o *Payload) GetIgnored() bool
  • func (o *Payload) GetMessageId() string
  • func (o *Payload) Release()
  • func (o *Payload) SetContext(c context.Context) *Payload
  • func (o *Payload) SetDuration(d float64) *Payload
  • func (o *Payload) SetError(e error) *Payload
  • func (o *Payload) SetIgnored(b bool) *Payload
  • func (o *Payload) SetMessageId(s string) *Payload
• type PoolManager
• type Registry
• type ResultInterface
• type ResultValidator
• type Subscriber
  • func NewSubscriber(m *models.Task, t SubscriberType) (s *Subscriber)
• type SubscriberProtocol
• type SubscriberResponseType
• type SubscriberType
• type Task
  • func (o *Task) EnNotificationFailed() bool
  • func (o *Task) EnNotificationSucceed() bool
  • func (o *Task) IsNotification() bool
  • func (o *Task) IsNotificationFailed() bool
  • func (o *Task) IsNotificationSucceed() bool

Variables

var (

	// SubscriberProtocolDefault
	// default subscription protocol.
	SubscriberProtocolDefault = "http"

	// SubscriberProtocolList
	// subscription protocol list.
	SubscriberProtocolList = map[string]SubscriberProtocol{
		"http":  SubscriberProtocolHttp,
		"https": SubscriberProtocolHttp,
		"rpc":   SubscriberProtocolRpc,
		"grpc":  SubscriberProtocolRpc,
		"tcp":   SubscriberProtocolTcp,
		"ws":    SubscriberProtocolWebsocket,
		"wss":   SubscriberProtocolWebsocket,
	}

	// SubscriberProtocolPort
	// subscription port mapping.
	SubscriberProtocolPort = map[string]int{
		"http":  80,
		"https": 443,
	}
)

Types

type ConditionManager

type ConditionManager interface {
	// Expression
	// return registered expression string.
	//
	// Defined in follow columns.
	//   - task.handler_condition
	//   - task.failed_condition
	//   - task.succeed_condition
	Expression() string

	// MatchJsonString
	// return result after matched json string.
	//
	// Return true if matched with json string, otherwise
	// false return.
	MatchJsonString(s string) (ignored bool, err error)
}

ConditionManager interface of condition manager.

type MemoryManager

type MemoryManager interface {
	// GetRegistries
	// return registry map in memory.
	GetRegistries() map[int]*Registry

	// GetRegistry
	// return registry instance in memory by id.
	GetRegistry(id int) *Registry

	// GetRegistryByName
	// return registry instance in memory by topic name
	// and tag.
	GetRegistryByName(topic, tag string) *Registry

	// GetTask
	// return task instance in memory by id.
	GetTask(id int) *Task

	// GetTaskFromBean
	// return task instance use task id.
	GetTaskFromBean(ctx context.Context, id int) (task *Task, err error)

	// GetTasks
	// return task instance map in memory.
	GetTasks() map[int]*Task

	// Reload
	// read from database into memory.
	Reload() error
}

MemoryManager interface of memory manager.

var (
	// Memory
	// instance of memory manager.
	Memory MemoryManager
)

type Message

type Message struct {
	Dequeue          int
	Keyword          string
	MessageBody      string
	MessageId        string
	MessageTime      int64
	PayloadMessageId string
	TaskId           int
	// contains filtered or unexported fields
}

Message struct for message consumed properties.

func (*Message) GetContext

func (o *Message) GetContext() context.Context

func (*Message) GetError

func (o *Message) GetError() error

func (*Message) GetIgnored

func (o *Message) GetIgnored() bool

func (*Message) Release

func (o *Message) Release()

func (*Message) SetBody

func (o *Message) SetBody(b []byte) *Message

func (*Message) SetContext

func (o *Message) SetContext(c context.Context) *Message

func (*Message) SetDuration

func (o *Message) SetDuration(d float64) *Message

func (*Message) SetError

func (o *Message) SetError(e error) *Message

func (*Message) SetIgnored

func (o *Message) SetIgnored(i bool) *Message

type Notification

type Notification struct {
	MessageBody string `json:"__gmd__message_body_"`
	TaskId      int    `json:"__gmd__task_id_"`
}

Notification struct for notification produce.

func (*Notification) Parse

func (o *Notification) Parse(s string) error

Parse unmarshal json string into instance.

func (*Notification) Release

func (o *Notification) Release()

Release instance to pool.

type Payload

type Payload struct {
	FilterTag        string
	Hash             string
	Keyword          string
	MessageBody      string
	MessageMessageId string
	MessageTaskId    int
	Offset           int
	RegistryId       int
	TopicName        string
	TopicTag         string
	// contains filtered or unexported fields
}

Payload struct for message publish properties.

func (*Payload) GetContext

func (o *Payload) GetContext() context.Context

func (*Payload) GetError

func (o *Payload) GetError() error

func (*Payload) GetIgnored

func (o *Payload) GetIgnored() bool

func (*Payload) GetMessageId

func (o *Payload) GetMessageId() string

func (*Payload) Release

func (o *Payload) Release()

func (*Payload) SetContext

func (o *Payload) SetContext(c context.Context) *Payload

func (*Payload) SetDuration

func (o *Payload) SetDuration(d float64) *Payload

func (*Payload) SetError

func (o *Payload) SetError(e error) *Payload

func (*Payload) SetIgnored

func (o *Payload) SetIgnored(b bool) *Payload

func (*Payload) SetMessageId

func (o *Payload) SetMessageId(s string) *Payload

type PoolManager

type PoolManager interface {
	// AcquireMessage
	// acquire message instance from pool.
	AcquireMessage() *Message

	// AcquireNotification
	// acquire notification instance from pool.
	AcquireNotification() *Notification

	// AcquirePayload
	// acquire payload instance from pool.
	AcquirePayload() *Payload

	// ReleaseMessage
	// release message instance to pool.
	ReleaseMessage(x *Message)

	// ReleaseNotification
	// release notification instance to pool.
	ReleaseNotification(x *Notification)

	// ReleasePayload
	// release payload instance to pool.
	ReleasePayload(x *Payload)
}

PoolManager interface of pool manager.

var (
	// Pool
	// instance of pool manager.
	Pool PoolManager
)

type Registry

type Registry struct {
	Id                             int
	FilterTag, TopicTag, TopicName string
}

Registry memory registry relation.

type ResultInterface

type ResultInterface interface {
	// Acquire
	// acquire validator instance from registered pool.
	Acquire() ResultValidator

	// Register
	// register validator instance override default.
	Register(v func() ResultValidator)

	// Release
	// release validator instance to pool.
	Release(x ResultValidator)
}

ResultInterface interface of result.

var (
	// Result
	// instance of result.
	Result ResultInterface
)

type ResultValidator

type ResultValidator interface {
	// After
	// called when release to pool.
	After()

	// Before
	// called when acquired from pool.
	Before()

	// Parse
	// parse dispatched result and return status.
	Parse(body []byte) (code string, err error)
}

ResultValidator instance of result validator.

type Subscriber

type Subscriber struct {
	Host, Addr, Method string
	Port, Timeout      int

	Condition    ConditionManager
	IgnoreCodes  []string
	Protocol     SubscriberProtocol
	ResponseType SubscriberResponseType
}

Subscriber struct for subscription.

func NewSubscriber

func NewSubscriber(m *models.Task, t SubscriberType) (s *Subscriber)

NewSubscriber create and return subscriber instance.

type SubscriberProtocol

type SubscriberProtocol int

const (
	SubscriberProtocolHttp SubscriberProtocol
	SubscriberProtocolRpc
	SubscriberProtocolTcp
	SubscriberProtocolWebsocket
)

type SubscriberResponseType

type SubscriberResponseType int

const (
	SubscriberResponseTypeErrnoIsZero SubscriberResponseType
)

type SubscriberType

type SubscriberType int

const (
	SubscriberTypeHandler SubscriberType
	SubscriberTypeFailed
	SubscriberTypeSucceed
)

type Task

type Task struct {
	Id      int
	Title   string
	Updated int64

	Parallels    int
	Concurrency  int32
	MaxRetry     int
	DelaySeconds int
	Broadcasting bool

	RegistryId int
	TopicName  string
	TopicTag   string
	FilterTag  string

	HandlerSubscriber *Subscriber
	FailedSubscriber  *Subscriber
	SucceedSubscriber *Subscriber
	// contains filtered or unexported fields
}

Task memory subscription task.

func (*Task) EnNotificationFailed

func (o *Task) EnNotificationFailed() bool

func (*Task) EnNotificationSucceed

func (o *Task) EnNotificationSucceed() bool

func (*Task) IsNotification

func (o *Task) IsNotification() bool

func (*Task) IsNotificationFailed

func (o *Task) IsNotificationFailed() bool

func (*Task) IsNotificationSucceed

func (o *Task) IsNotificationSucceed() bool