bee

README

bee

An asynchronous task execution framework based on the producer consumer model.

Features

supported middleware

• redis
• pulsar
• rocketmq
• rabbitmq

Install

go get -u github.com/bee-org/bee

Quick start

Redis example:

package main

import (
	"context"
	"fmt"
	"github.com/bee-org/bee"
	"github.com/bee-org/bee/broker/redis"
	"os"
)

func printHandler(c *bee.Context) error {
	var result int64
	err := c.Parse(&result)
	fmt.Println("printHandler", result, err)
	return nil
}

func main() {
	b, err := redis.NewBroker(redis.Config{
		URL:   os.Getenv("REDIS_URL"),
		Topic: "bee",
	})
	if err != nil {
		panic(err)
	}
	b.Register("print", printHandler)
	if err = b.Worker(); err != nil {
		panic(err)
	}

	b.Send(context.TODO(), "print", 1) // producer
	// output: printHandler 1 <nil>
}

RabbitMQ example:

package main

import (
	"context"
	"fmt"
	"github.com/bee-org/bee"
	"github.com/bee-org/bee/broker/amqp"
	"os"
)

func printHandler(c *bee.Context) error {
	var result int64
	err := c.Parse(&result)
	fmt.Println("printHandler", result, err)
	return nil
}

func main() {
	b, err := amqp.NewBroker(amqp.Config{
		URL:   os.Getenv("RABBIT_URL"),
		Queue: "bee",
	})
	if err != nil {
		panic(err)
	}
	b.Register("print", printHandler)
	if err = b.Worker(); err != nil {
		panic(err)
	}

	b.Send(context.TODO(), "print", 1) // producer
	// output: printHandler 1 <nil>
}

Message Protocol

bit:0-------8------16------24----------------n
    ┌───────┬───────┬───────┬────────────────┐
    │version│ retry │length │     name       │ header
    ├───────┴───────┴───────┴────────────────┤
    │                  body                  │
    └────────────────────────────────────────┘

• version(8bit): Protocol Version number
• retry(8bit): The number of times a message was redelivered
• length(8bit): The length of the function name that follows
• name(n-bit): The function name
• body(n-bit): Data required for function execution

Documentation

Index

• type Context
  • func NewCtx(ctx context.Context, m message.Message) *Context
  • func (c *Context) Deadline() (deadline time.Time, ok bool)
  • func (c *Context) Done() <-chan struct{}
  • func (c *Context) Err() error
  • func (c *Context) Message() message.Message
  • func (c *Context) Name() string
  • func (c *Context) Parse(v interface{}) error
  • func (c *Context) Req() interface{}
  • func (c *Context) Value(key interface{}) interface{}
• type Handler
• type Option

Types

type Context

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

func NewCtx

func NewCtx(ctx context.Context, m message.Message) *Context

func (*Context) Deadline

func (c *Context) Deadline() (deadline time.Time, ok bool)

func (*Context) Done

func (c *Context) Done() <-chan struct{}

func (*Context) Err

func (c *Context) Err() error

func (*Context) Message

func (c *Context) Message() message.Message

func (*Context) Name

func (c *Context) Name() string

func (*Context) Parse

func (c *Context) Parse(v interface{}) error

func (*Context) Req

func (c *Context) Req() interface{}

Req must be called after the Parse, Return the req recorded at parsing time.

func (*Context) Value

func (c *Context) Value(key interface{}) interface{}

type Handler

type Handler func(ctx *Context) error

type Option

type Option interface {
}