cron

Documentation

Overview

Package cron implements delayed message execution.

This package provides a queue implementation for scheduling message for execution in the future and weave.Ticker compatible task runner.

Index

• Variables
• func NewTaskResultBucket() orm.ModelBucket
• func RegisterQuery(qr weave.QueryRouter)
• type Authenticator
  • func (Authenticator) GetConditions(ctx weave.Context) []weave.Condition
  • func (a Authenticator) HasAddress(ctx weave.Context, addr weave.Address) bool
• type Scheduler
  • func NewScheduler(enc TaskMarshaler) *Scheduler
  • func (s *Scheduler) Delete(db weave.KVStore, taskID []byte) error
  • func (s *Scheduler) Schedule(db weave.KVStore, runAt time.Time, auth []weave.Condition, msg weave.Msg) ([]byte, error)
• type TaskMarshaler
• type TaskResult
  • func (t *TaskResult) Copy() orm.CloneableData
  • func (*TaskResult) Descriptor() ([]byte, []int)
  • func (m *TaskResult) GetExecHeight() int64
  • func (m *TaskResult) GetExecTime() github_com_iov_one_weave.UnixTime
  • func (m *TaskResult) GetInfo() string
  • func (m *TaskResult) GetMetadata() *weave.Metadata
  • func (m *TaskResult) GetSuccessful() bool
  • func (m *TaskResult) Marshal() (dAtA []byte, err error)
  • func (m *TaskResult) MarshalTo(dAtA []byte) (int, error)
  • func (*TaskResult) ProtoMessage()
  • func (m *TaskResult) Reset()
  • func (m *TaskResult) Size() (n int)
  • func (m *TaskResult) String() string
  • func (m *TaskResult) Unmarshal(dAtA []byte) error
  • func (t *TaskResult) Validate() error
  • func (m *TaskResult) XXX_DiscardUnknown()
  • func (m *TaskResult) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
  • func (m *TaskResult) XXX_Merge(src proto.Message)
  • func (m *TaskResult) XXX_Size() int
  • func (m *TaskResult) XXX_Unmarshal(b []byte) error
• type Ticker
  • func NewTicker(h weave.Handler, enc TaskMarshaler) *Ticker
  • func (t *Ticker) Tick(ctx context.Context, db store.CacheableKVStore) weave.TickResult

Variables

var (
	ErrInvalidLengthCodec = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowCodec   = fmt.Errorf("proto: integer overflow")
)

Functions

func NewTaskResultBucket

func NewTaskResultBucket() orm.ModelBucket

NewTaskResultBucket returns a bucket for storing Task results.

func RegisterQuery

func RegisterQuery(qr weave.QueryRouter)

Types

type Authenticator

type Authenticator struct{}

Authenticator implements an x.Authenticator interface that should be used to authorize cron task execution. Use it together with WithAuth function to control the cron task execution authentication.

func (Authenticator) GetConditions

func (Authenticator) GetConditions(ctx weave.Context) []weave.Condition

GetConditions implements x.Authenticator interface.

func (Authenticator) HasAddress

func (a Authenticator) HasAddress(ctx weave.Context, addr weave.Address) bool

HasAddress implements x.Authenticator interface.

type Scheduler

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

Scheduler is the weave.Scheduler implementation.

func NewScheduler

func NewScheduler(enc TaskMarshaler) *Scheduler

NewScheduler returns a scheduler implementation that is using given encoding for serializing data. Returned scheduler implements weave.Scheduler interface.

Always use the same marshaler for ticker and scheduler.

func (*Scheduler) Delete

func (s *Scheduler) Delete(db weave.KVStore, taskID []byte) error

Delete implements weave.Scheduler interface.

func (*Scheduler) Schedule

func (s *Scheduler) Schedule(db weave.KVStore, runAt time.Time, auth []weave.Condition, msg weave.Msg) ([]byte, error)

Schedule implements weave.Scheduler interface.

Due to the implementation details, transaction is guaranteed to be executed after given time, but not exactly at given time. If another transaction is already scheduled for the exact same time, execution of this transaction is delayed until the next free slot.

Time granularity is second.

type TaskMarshaler

type TaskMarshaler interface {
	// MarshalTask serialize given data into its binary format.
	MarshalTask(auth []weave.Condition, msg weave.Msg) ([]byte, error)

	// UnmarshalTask deserialize data (created using MarshalTask method)
	// from its binary representation into Go structures.
	UnmarshalTask([]byte) (auth []weave.Condition, msg weave.Msg, err error)
}

TaskMarshaler represents an encoded that is used to marshal and unmarshal a task. This interface is to be implemented by this package user.

type TaskResult

type TaskResult struct {
	Metadata *weave.Metadata `protobuf:"bytes,1,opt,name=metadata,proto3" json:"metadata,omitempty"`
	// Successful is set to true if the task was successfully executed.
	Successful bool `protobuf:"varint,2,opt,name=successful,proto3" json:"successful,omitempty"`
	// Info contains any additinal information that might be useful to lean more
	// about the task execution.
	Info string `protobuf:"bytes,3,opt,name=info,proto3" json:"info,omitempty"`
	// Exec time hold the information of when the task was executed.
	ExecTime github_com_iov_one_weave.UnixTime `` /* 129-byte string literal not displayed */
	// Exec height holds the block height value at the time the task was executed.
	ExecHeight int64 `protobuf:"varint,5,opt,name=exec_height,json=execHeight,proto3" json:"exec_height,omitempty"`
}

TaskResult is a publicly available information about task execution result. It is only created for those tasks that were executed.

Due to a bug in tendermint we must store this information ourselves instead of relying on the usual search via tag. https://github.com/tendermint/tendermint/issues/3665

func (*TaskResult) Copy

func (t *TaskResult) Copy() orm.CloneableData

func (*TaskResult) Descriptor

func (*TaskResult) Descriptor() ([]byte, []int)

func (*TaskResult) GetExecHeight

func (m *TaskResult) GetExecHeight() int64

func (*TaskResult) GetExecTime

func (m *TaskResult) GetExecTime() github_com_iov_one_weave.UnixTime

func (*TaskResult) GetInfo

func (m *TaskResult) GetInfo() string

func (*TaskResult) GetMetadata

func (m *TaskResult) GetMetadata() *weave.Metadata

func (*TaskResult) GetSuccessful

func (m *TaskResult) GetSuccessful() bool

func (*TaskResult) Marshal

func (m *TaskResult) Marshal() (dAtA []byte, err error)

func (*TaskResult) MarshalTo

func (m *TaskResult) MarshalTo(dAtA []byte) (int, error)

func (*TaskResult) ProtoMessage

func (*TaskResult) ProtoMessage()

func (*TaskResult) Reset

func (m *TaskResult) Reset()

func (*TaskResult) Size

func (m *TaskResult) Size() (n int)

func (*TaskResult) String

func (m *TaskResult) String() string

func (*TaskResult) Unmarshal

func (m *TaskResult) Unmarshal(dAtA []byte) error

func (*TaskResult) Validate

func (t *TaskResult) Validate() error

func (*TaskResult) XXX_DiscardUnknown

func (m *TaskResult) XXX_DiscardUnknown()

func (*TaskResult) XXX_Marshal

func (m *TaskResult) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*TaskResult) XXX_Merge

func (m *TaskResult) XXX_Merge(src proto.Message)

func (*TaskResult) XXX_Size

func (m *TaskResult) XXX_Size() int

func (*TaskResult) XXX_Unmarshal

func (m *TaskResult) XXX_Unmarshal(b []byte) error

type Ticker

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

Ticker allows to execute messages queued for future execution. It does this by implementing weave.Ticker interface.

func NewTicker

func NewTicker(h weave.Handler, enc TaskMarshaler) *Ticker

NewTicker returns a cron runner instance that is using given handler to process all queued messages that execution time is due. All serialization is done using provided marshaler.

Always use the same marshaler for ticker and scheduler.

func (*Ticker) Tick

func (t *Ticker) Tick(ctx context.Context, db store.CacheableKVStore) weave.TickResult

Tick implementes weave.Ticker interface.

Tick can process any number of messages suitable for execution. All changes are done atomically and apply only on success.