server

Documentation

Index

• Constants
• Variables
• func Main(port int, dbpath string) error
• type BoltMessageBucket
  • func (b *BoltMessageBucket) Del(id MessageID) error
  • func (b *BoltMessageBucket) DelBucket() error
  • func (b *BoltMessageBucket) Get(id MessageID) (*Message, error)
  • func (b *BoltMessageBucket) Put(msg *Message) error
  • func (b *BoltMessageBucket) Walk(walkFunc func(msg *Message) error) error
• type BoltStore
  • func NewBoltStore(path string) *BoltStore
  • func (bs *BoltStore) Close() error
  • func (bs *BoltStore) MessageBucket(name string) MessageBucket
  • func (bs *BoltStore) Open() error
• type Broker
  • func NewBroker(ctx context.Context, dbpath string) *Broker
  • func (b *Broker) Done()
  • func (b *Broker) FinishMessage(name string, id MessageID) error
  • func (b *Broker) GetMessage(name string, id MessageID) (*Message, error)
  • func (b *Broker) Init() error
  • func (b *Broker) NewID() MessageID
  • func (b *Broker) PushMessage(name string, job *config.Job) (*Message, error)
  • func (b *Broker) ReportJob(ctx context.Context, req *pb.ReportJobRequest) (res *pb.ReportJobResponse, err error)
  • func (b *Broker) ReportJobDone(ctx context.Context, req *pb.ReportJobDoneRequest) (res *pb.ReportJobDoneResponse, err error)
  • func (b *Broker) SubscribeJob(ctx context.Context, req *pb.SubscribeJobRequest) (*pb.SubscribeJobResponse, error)
  • func (b *Broker) Topic(name string) *Topic
• type Json
• type LQueue
  • func NewLQueue() *LQueue
  • func (q *LQueue) Len() int
  • func (q *LQueue) List() []interface{}
  • func (q *LQueue) Pop() interface{}
  • func (q *LQueue) Push(elem interface{})
• type Message
  • func DecodeMessage(v []byte) (*Message, error)
  • func NewMessage(id MessageID, job *config.Job) *Message
  • func (m *Message) Encode() []byte
  • func (m *Message) JSON() Json
  • func (m *Message) SetResults(workerId string, tasks map[string]interface{})
• type MessageBucket
• type MessageID
  • func GetMessageID(v []byte) MessageID
  • func (id MessageID) Bytes() []byte
  • func (id MessageID) String() string
• type MessageIDGenerator
• type Queue
• type Store
  • func NewTopicStore(storeType string, path string, topic string) (Store, error)
• type TaskResults
  • func NewTaskResults(workerId string, tasks map[string]interface{}) *TaskResults
• type Topic
  • func NewTopic(ctx context.Context, name string, retryCheckDuration time.Duration, ...) *Topic
  • func (t *Topic) FinishMessage(id MessageID) error
  • func (t *Topic) Init() error
  • func (t *Topic) PopMessage() *Message
  • func (t *Topic) PushMessage(msg *Message)

Constants

const (
	MSG_PENDING = iota
	MSG_RECEIVED
	MSG_SUCCESS
	MSG_FAILURE
)

const (
	MsgPendingState  = "PENDING"
	MsgReceivedState = "RECEIVED"
	MsgSuccessState  = "SUCCESS"
	MsgFailureState  = "FAILURE"
)

const (
	MessageBucketName                = "messages"
	MessagePendingBucketName         = "pendingMessages"
	WorkerPerMessageBucketNamePrefix = "worker:"
)

const MsgIdLen = 16

Variables

var (
	ErrTopicNotFound = errors.New("Topic not found")
	ErrMsgNotFound   = errors.New("Message not found")
)

var ErrNotSupportedStoreType = errors.New("this store type is not supported")

var ErrSequenceExpired = errors.New("sequence expired")

var ErrTimeBackwards = errors.New("time has gone backwards")

var (
	MsgStates map[int]string
)

Functions

func Main

func Main(port int, dbpath string) error

Types

type BoltMessageBucket

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

func (*BoltMessageBucket) Del

func (b *BoltMessageBucket) Del(id MessageID) error

func (*BoltMessageBucket) DelBucket

func (b *BoltMessageBucket) DelBucket() error

func (*BoltMessageBucket) Get

func (b *BoltMessageBucket) Get(id MessageID) (*Message, error)

func (*BoltMessageBucket) Put

func (b *BoltMessageBucket) Put(msg *Message) error

func (*BoltMessageBucket) Walk

func (b *BoltMessageBucket) Walk(walkFunc func(msg *Message) error) error

type BoltStore

type BoltStore struct {
	Path string
	// contains filtered or unexported fields
}

func NewBoltStore

func NewBoltStore(path string) *BoltStore

func (*BoltStore) Close

func (bs *BoltStore) Close() error

func (*BoltStore) MessageBucket

func (bs *BoltStore) MessageBucket(name string) MessageBucket

func (*BoltStore) Open

func (bs *BoltStore) Open() error

type Broker

type Broker struct {
	ID     int64
	DBPath string
	Topics map[string]*Topic
	// contains filtered or unexported fields
}

func NewBroker

func NewBroker(ctx context.Context, dbpath string) *Broker

func (*Broker) Done

func (b *Broker) Done()

func (*Broker) FinishMessage

func (b *Broker) FinishMessage(name string, id MessageID) error

func (*Broker) GetMessage

func (b *Broker) GetMessage(name string, id MessageID) (*Message, error)

func (*Broker) Init

func (b *Broker) Init() error

func (*Broker) NewID

func (b *Broker) NewID() MessageID

This method is implemented to MessageIDGenerator

func (*Broker) PushMessage

func (b *Broker) PushMessage(name string, job *config.Job) (*Message, error)

func (*Broker) ReportJob

func (b *Broker) ReportJob(ctx context.Context, req *pb.ReportJobRequest) (res *pb.ReportJobResponse, err error)

func (*Broker) ReportJobDone

func (b *Broker) ReportJobDone(ctx context.Context, req *pb.ReportJobDoneRequest) (res *pb.ReportJobDoneResponse, err error)

func (*Broker) SubscribeJob

func (b *Broker) SubscribeJob(ctx context.Context, req *pb.SubscribeJobRequest) (*pb.SubscribeJobResponse, error)

twirp rpc interface

func (*Broker) Topic

func (b *Broker) Topic(name string) *Topic

type Json

type Json map[string]interface{}

type LQueue

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

func NewLQueue

func NewLQueue() *LQueue

func (*LQueue) Len

func (q *LQueue) Len() int

func (*LQueue) List

func (q *LQueue) List() []interface{}

func (*LQueue) Pop

func (q *LQueue) Pop() interface{}

func (*LQueue) Push

func (q *LQueue) Push(elem interface{})

type Message

type Message struct {
	ID      MessageID
	Job     *config.Job
	Created time.Time
	State   int
	Results *TaskResults
}

func DecodeMessage

func DecodeMessage(v []byte) (*Message, error)

func NewMessage

func NewMessage(id MessageID, job *config.Job) *Message

func (*Message) Encode

func (m *Message) Encode() []byte

func (*Message) JSON

func (m *Message) JSON() Json

func (*Message) SetResults

func (m *Message) SetResults(workerId string, tasks map[string]interface{})

type MessageBucket

type MessageBucket interface {
	Get(id MessageID) (*Message, error)
	Put(msg *Message) error
	Del(id MessageID) error
	Walk(walkFunc func(*Message) error) error
	DelBucket() error
}

type MessageID

type MessageID [MsgIdLen]byte

func GetMessageID

func GetMessageID(v []byte) MessageID

func (MessageID) Bytes

func (id MessageID) Bytes() []byte

func (MessageID) String

func (id MessageID) String() string

type MessageIDGenerator

type MessageIDGenerator interface {
	NewID() MessageID
}

type Queue

type Queue interface {
	Push(element interface{})
	Pop() interface{}
}

type Store

type Store interface {
	Open() error
	Close() error
	MessageBucket(name string) MessageBucket
}

func NewTopicStore

func NewTopicStore(storeType string, path string, topic string) (Store, error)

type TaskResults

type TaskResults struct {
	WorkerId string                 `json:"worker"`
	Tasks    map[string]interface{} `json:"tasks"`
}

func NewTaskResults

func NewTaskResults(workerId string, tasks map[string]interface{}) *TaskResults

type Topic

type Topic struct {
	Name  string
	Queue Queue
	// contains filtered or unexported fields
}

func NewTopic

func NewTopic(ctx context.Context, name string, retryCheckDuration time.Duration, store Store) *Topic

func (*Topic) FinishMessage

func (t *Topic) FinishMessage(id MessageID) error

func (*Topic) Init

func (t *Topic) Init() error

func (*Topic) PopMessage

func (t *Topic) PopMessage() *Message

func (*Topic) PushMessage

func (t *Topic) PushMessage(msg *Message)