routine

README

routine

go routine control with context, support: Main, Go, Pool and some useful Executors.

Main

package main

import (
	"context"
	"errors"
	"log"
	
	"github.com/x-mod/routine"
)

func foo(ctx context.Context) error {
	log.Println("foo begin")
	log.Println("foo end")
	return errors.New("foo done")
}

func main(){
	err := routine.Main(
		context.TODO(),
		routine.Concurrent(4, routine.ExecutorFunc(foo)),
		routine.Cleanup(routine.ExecutorFunc(func(ctx context.Context) error {
			log.Println("clear")
			return nil
		})),
	)
	if err != nil {
		log.Println("failed:", err)
	}
}

Routine

import "github.com/x-mod/routine"

err := routine.New(execute).Execute(ctx)

Executors

import "github.com/x-mod/routine"

crontab := routine.Crontab("* * * * *", execute)

Enjoy

Documentation

Index

• Variables
• func ArgumentsFrom(ctx context.Context) ([]interface{}, bool)
• func FromCrontab(ctx context.Context) time.Time
• func FromRepeat(ctx context.Context) int
• func FromRetry(ctx context.Context) int
• func Main(ctx context.Context, exec Executor, opts ...Opt) error
• func WithArgments(ctx context.Context, args ...interface{}) context.Context
• func WithParent(ctx context.Context, parent *Routine) context.Context
• type AppendExecutor
  • func Append(execs ...Executor) *AppendExecutor
  • func (ae *AppendExecutor) Append(execs ...Executor)
  • func (ae *AppendExecutor) Execute(ctx context.Context) error
• type CancelInterruptor
  • func NewCancelInterruptor(sig syscall.Signal) *CancelInterruptor
  • func (c *CancelInterruptor) Interrupt() InterruptHandler
  • func (c *CancelInterruptor) Signal() syscall.Signal
• type CommandExecutor
  • func Command(cmd string, opts ...CommandOpt) *CommandExecutor
  • func (cmd *CommandExecutor) Execute(ctx context.Context) error
• type CommandOpt
  • func ARG(arg string) CommandOpt
  • func ENV(env string) CommandOpt
  • func Stderr(wr io.Writer) CommandOpt
  • func Stdin(rd io.Reader) CommandOpt
  • func Stdout(wr io.Writer) CommandOpt
• type ConcurrentExecutor
  • func Concurrent(c int, exec Executor) *ConcurrentExecutor
  • func (ce *ConcurrentExecutor) Execute(ctx context.Context) error
• type CrontabExecutor
  • func Crontab(plan string, exec Executor) *CrontabExecutor
  • func (c *CrontabExecutor) Everyday(flag bool)
  • func (c *CrontabExecutor) Execute(ctx context.Context) error
  • func (c *CrontabExecutor) IsTimeMuted(tm time.Time) bool
  • func (c *CrontabExecutor) Mute(begin time.Time, end time.Time)
  • func (c *CrontabExecutor) Weekend(flag bool)
  • func (c *CrontabExecutor) Workday(flag bool)
• type CrontabMute
• type DeadlineExecutor
  • func Deadline(d time.Time, exec Executor) *DeadlineExecutor
  • func (tm *DeadlineExecutor) Execute(ctx context.Context) error
• type Executor
  • func UseExecutorMiddleware(exec Executor, middleware ...ExecutorMiddleware) Executor
• type ExecutorFunc
  • func (f ExecutorFunc) Execute(ctx context.Context) error
• type ExecutorMiddleware
• type GuaranteeExecutor
  • func Guarantee(exec Executor) *GuaranteeExecutor
  • func (g *GuaranteeExecutor) Execute(ctx context.Context) error
• type InterruptHandler
• type Interruptor
• type Opt
  • func Arguments(args ...interface{}) Opt
  • func Cleanup(exec Executor) Opt
  • func Interrupts(ints ...Interruptor) Opt
  • func Prepare(exec Executor) Opt
• type ParallelExecutor
  • func Parallel(execs ...Executor) *ParallelExecutor
  • func (pe *ParallelExecutor) Execute(ctx context.Context) error
• type RepeatExecutor
  • func Repeat(repeat int, interval time.Duration, exec Executor) *RepeatExecutor
  • func (r *RepeatExecutor) Execute(ctx context.Context) error
• type RetryExecutor
  • func Retry(retry int, exec Executor) *RetryExecutor
  • func (retry *RetryExecutor) Execute(ctx context.Context) error
• type Routine
  • func New(exec Executor, opts ...Opt) *Routine
  • func ParentFrom(ctx context.Context) (*Routine, bool)
  • func (r *Routine) Execute(ctx context.Context) error
  • func (r *Routine) Go(ctx context.Context, exec Executor) chan error
  • func (r *Routine) Stop()
• type TimeoutExecutor
  • func Timeout(d time.Duration, exec Executor) *TimeoutExecutor
  • func (tm *TimeoutExecutor) Execute(ctx context.Context) error

Variables

var (
	//ErrNoneExecutor error
	ErrNoneExecutor = errors.New("none executor")
	//ErrNoneContext error
	ErrNoneContext = errors.New("none context")
	//ErrNonePlan error
	ErrNonePlan = errors.New("none plan")
)

var DefaultCancelInterruptors []Interruptor

DefaultCancelInterruptors include INT/TERM/KILL signals

Functions

func ArgumentsFrom

func ArgumentsFrom(ctx context.Context) ([]interface{}, bool)

ArgumentsFrom extract from context

func FromCrontab

func FromCrontab(ctx context.Context) time.Time

FromCrontab current crontab time

func FromRepeat

func FromRepeat(ctx context.Context) int

FromRepeat current repeated times

func FromRetry

func FromRetry(ctx context.Context) int

FromRetry current retied times

func Main

func Main(ctx context.Context, exec Executor, opts ...Opt) error

func WithArgments

func WithArgments(ctx context.Context, args ...interface{}) context.Context

WithArgments inject into context

func WithParent

func WithParent(ctx context.Context, parent *Routine) context.Context

Types

type AppendExecutor

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

AppendExecutor

func Append

func Append(execs ...Executor) *AppendExecutor

Append new

func (*AppendExecutor) Append

func (ae *AppendExecutor) Append(execs ...Executor)

func (*AppendExecutor) Execute

func (ae *AppendExecutor) Execute(ctx context.Context) error

type CancelInterruptor

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

CancelInterruptor definition

func NewCancelInterruptor

func NewCancelInterruptor(sig syscall.Signal) *CancelInterruptor

NewCancelInterruptor if fn is nil will cancel context

func (*CancelInterruptor) Interrupt

func (c *CancelInterruptor) Interrupt() InterruptHandler

Interrupt inplement the interface

func (*CancelInterruptor) Signal

func (c *CancelInterruptor) Signal() syscall.Signal

Signal inplement the interface

type CommandExecutor

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

CommandExecutor struct

func Command

func Command(cmd string, opts ...CommandOpt) *CommandExecutor

Command new

func (*CommandExecutor) Execute

func (cmd *CommandExecutor) Execute(ctx context.Context) error

Execute implement Executor

type CommandOpt

type CommandOpt func(*CommandExecutor)

func ARG

func ARG(arg string) CommandOpt

func ENV

func ENV(env string) CommandOpt

func Stderr

func Stderr(wr io.Writer) CommandOpt

func Stdin

func Stdin(rd io.Reader) CommandOpt

func Stdout

func Stdout(wr io.Writer) CommandOpt

type ConcurrentExecutor

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

ConcurrentExecutor struct

func Concurrent

func Concurrent(c int, exec Executor) *ConcurrentExecutor

Concurrent new

func (*ConcurrentExecutor) Execute

func (ce *ConcurrentExecutor) Execute(ctx context.Context) error

Execute implement Executor

type CrontabExecutor

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

CrontabExecutor struct

func Crontab

func Crontab(plan string, exec Executor) *CrontabExecutor

Crontab new

func (*CrontabExecutor) Everyday

func (c *CrontabExecutor) Everyday(flag bool)

func (*CrontabExecutor) Execute

func (c *CrontabExecutor) Execute(ctx context.Context) error

Execute implement Executor

func (*CrontabExecutor) IsTimeMuted

func (c *CrontabExecutor) IsTimeMuted(tm time.Time) bool

func (*CrontabExecutor) Mute

func (c *CrontabExecutor) Mute(begin time.Time, end time.Time)

func (*CrontabExecutor) Weekend

func (c *CrontabExecutor) Weekend(flag bool)

func (*CrontabExecutor) Workday

func (c *CrontabExecutor) Workday(flag bool)

type CrontabMute

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

type DeadlineExecutor

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

DeadlineExecutor struct

func Deadline

func Deadline(d time.Time, exec Executor) *DeadlineExecutor

Deadline new

func (*DeadlineExecutor) Execute

func (tm *DeadlineExecutor) Execute(ctx context.Context) error

Execute implement Executor

type Executor

type Executor interface {
	//Execute before stopping make sure all subroutines stopped
	Execute(context.Context) error
}

Executor interface definition

func UseExecutorMiddleware

func UseExecutorMiddleware(exec Executor, middleware ...ExecutorMiddleware) Executor

UseExecutorMiddleware wraps a Executor in one or more middleware.

type ExecutorFunc

type ExecutorFunc func(context.Context) error

ExecutorFunc definition

func (ExecutorFunc) Execute

func (f ExecutorFunc) Execute(ctx context.Context) error

Execute ExecutorFunc implemention of Executor

type ExecutorMiddleware

type ExecutorMiddleware func(Executor) Executor

ExecutorMiddleware is a function that middlewares can implement to be able to chain.

type GuaranteeExecutor

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

GuaranteeExecutor struct, make sure of none error return

func Guarantee

func Guarantee(exec Executor) *GuaranteeExecutor

Guarantee insure exec NEVER PANIC

func (*GuaranteeExecutor) Execute

func (g *GuaranteeExecutor) Execute(ctx context.Context) error

Execute implement Executor interface

type InterruptHandler

type InterruptHandler func(ctx context.Context) (exit bool)

InterruptHandler definition

type Interruptor

type Interruptor interface {
	Signal() syscall.Signal
	Interrupt() InterruptHandler
}

Interruptor definition

type Opt

type Opt func(*options)

Opt interface

func Arguments

func Arguments(args ...interface{}) Opt

Arguments Opt for Main

func Cleanup

func Cleanup(exec Executor) Opt

Cleanup Opt for Main

func Interrupts

func Interrupts(ints ...Interruptor) Opt

Interrupts Opt for Main

func Prepare

func Prepare(exec Executor) Opt

Prepare Opt for Main

type ParallelExecutor

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

ParallelExecutor

func Parallel

func Parallel(execs ...Executor) *ParallelExecutor

Parallel new

func (*ParallelExecutor) Execute

func (pe *ParallelExecutor) Execute(ctx context.Context) error

Execute implement Executor

type RepeatExecutor

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

RepeatExecutor struct

func Repeat

func Repeat(repeat int, interval time.Duration, exec Executor) *RepeatExecutor

Repeat new

func (*RepeatExecutor) Execute

func (r *RepeatExecutor) Execute(ctx context.Context) error

Execute implement Executor

type RetryExecutor

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

RetryExecutor struct

func Retry

func Retry(retry int, exec Executor) *RetryExecutor

Retry new

func (*RetryExecutor) Execute

func (retry *RetryExecutor) Execute(ctx context.Context) error

Execute implement Executor interface

type Routine

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

func New

func New(exec Executor, opts ...Opt) *Routine

func ParentFrom

func ParentFrom(ctx context.Context) (*Routine, bool)

func (*Routine) Execute

func (r *Routine) Execute(ctx context.Context) error

func (*Routine) Go

func (r *Routine) Go(ctx context.Context, exec Executor) chan error

func (*Routine) Stop

func (r *Routine) Stop()

type TimeoutExecutor

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

TimeoutExecutor struct

func Timeout

func Timeout(d time.Duration, exec Executor) *TimeoutExecutor

Timeout new

func (*TimeoutExecutor) Execute

func (tm *TimeoutExecutor) Execute(ctx context.Context) error