asynctask

README

async task client in go

This project is a package used to support async task client of golang, support multi backend and broker, include Redis, AMQP e.g.

quick start

Here is a sample code to help you quick start, the api is easily to use, enjoy it!

sample code

package test

import (
	"fmt"
	"github.com/go-redis/redis"
	rBackend "github.com/gukz/asynctask/backend/redis"
	rBroker "github.com/gukz/asynctask/broker/redis"
)
func init() {
	asynctask.RegisteHandler(map[string]interface{}{
		"async_func": func(name string, a int64) (string, int64, error) {
			fmt.Println("function is running", name, a)
			return name + "is finished", a * a, nil
		},
	})
}

func main(){
	// defin a redis client
	client := redis.NewClient(&redis.Options{
		Addr:	  "127.0.0.1:6379",
		Password: "",
		DB:	      0,
	})

	backend, _ := rBackend.NewBackend(client, "backend_", 10*time.Minute)
	broker, _ := rBroker.NewBroker(client, "broker_")

	// create a async base object
	asyncBase := asynctask.NewAsyncTask("test_async_queue", broker, backend)

	go func() {
		var a = 1
		for {
			// build the async call
			msg := asynctask.NewMessage(
				"test_func", []asynctask.TypeValue{
					{Name: "name1", Type: "string", Value: "hello"},
					{Name: "age", Type: "int64", Value: a++},
				})
			// send this async call
			_ = asyncBase.GetProducer().Send(msg)
			// wait it to be process
			time.Sleep(2 * time.Second)
			// check the result of the async call
			res, _ := asyncBase.GetProducer().GetResult(msg.TaskId)
		}
	}()

	// start handle async call with concurrent of 10.
	asyncBase.GetWorker().Serve(10)
}

more feature comes later...

• support middleware(used to monitor task failure).
• support task with context.
• support more backend and broker.

Documentation

Index

• Constants
• func OriginalValue(data reflect.Value) (interface{}, error)
• func ReflectValue(valueType string, value interface{}) (reflect.Value, error)
• func RegisteHandler(h map[string]interface{})
• func TypeValue2ReflectValue(data []TypeValue) []reflect.Value
• type AsyncBase
  • func NewAsyncTask(queue string, broker Broker, backend Backend) *AsyncBase
  • func (t *AsyncBase) GetProducer() *producer
  • func (t *AsyncBase) GetWorker() *worker
• type Backend
• type Broker
• type BrokerMessage
• type Logger
  • func (t *Logger) Error(format string, args ...interface{}) error
  • func (t *Logger) Info(format string, args ...interface{})
  • func (t *Logger) Warn(format string, args ...interface{})
• type Message
  • func NewMessage(name string, args []TypeValue) *Message
• type Result
• type ResultValue
• type TypeValue

Constants

const NoResultFound string = "No errors found"

Functions

func OriginalValue

func OriginalValue(data reflect.Value) (interface{}, error)

func ReflectValue

func ReflectValue(valueType string, value interface{}) (reflect.Value, error)

func RegisteHandler

func RegisteHandler(h map[string]interface{})

func TypeValue2ReflectValue

func TypeValue2ReflectValue(data []TypeValue) []reflect.Value

Types

type AsyncBase

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

func NewAsyncTask

func NewAsyncTask(queue string, broker Broker, backend Backend) *AsyncBase

func (*AsyncBase) GetProducer

func (t *AsyncBase) GetProducer() *producer

func (*AsyncBase) GetWorker

func (t *AsyncBase) GetWorker() *worker

type Backend

type Backend interface {
	CheckHealth() bool
	SetResult(taskId string, result []byte) error
	GetResult(taskId string) ([]byte, error)
	Close() error
}

Backend used to manage task's status and result

type Broker

type Broker interface {
	CheckHealth() bool
	PopMessage(string) (*BrokerMessage, error)
	Ack(ackAll bool, message *BrokerMessage) error
	PushMessage(string, []byte) error
	Close() error
}

Broker used to manage the task queue

type BrokerMessage

type BrokerMessage struct {
	Body   []byte
	Origin interface{}
}

type Logger

type Logger struct {
}

func (*Logger) Error

func (t *Logger) Error(format string, args ...interface{}) error

func (*Logger) Info

func (t *Logger) Info(format string, args ...interface{})

func (*Logger) Warn

func (t *Logger) Warn(format string, args ...interface{})

type Message

type Message struct {
	TaskId string
	Name   string
	Args   []TypeValue
}

func NewMessage

func NewMessage(name string, args []TypeValue) *Message

type Result

type Result struct {
	ReturnValues []*ResultValue
	HasError     bool
	Error        string
}

type ResultValue

type ResultValue struct {
	Type  string
	Value interface{}
}

type TypeValue

type TypeValue struct {
	Name  string      `bson:"name"`
	Type  string      `bson:"type"`
	Value interface{} `bson:"value"`
}