dispatch

README

dispatch

A powerful read and handle workflow dispatcher for golang.

• Install
• Examples

---

Install

With a correctly configured Go toolchain:

go get -u github.com/searKing/dispatch

---

Examples

Let's start registering a couple of URL paths and handlers:

func main() {
    var conn chan int
    workflow := dispatch.NewDispatcher(
    	dispatch.ReaderFunc(func() (interface{}, error) {
    		return ReadMessage(conn)
    	}), 
    	dispatch.HandlerFunc(func(msg interface{}) error {
    		m := msg.(*int)
    		return HandleMessage(m)
    	}))
    workflow.Start()
}

Here we can set the workflow joinable.

    workflow := dispatch.NewDispatcher(nil, nil).Joinable()
    go workflow.Start()
    workflow.Join()

Here we can cancel the workflow.

    workflow := dispatch.NewDispatcher(nil, nil).Joinable()
    go workflow.Start()
	workflow.Context().Done()
	workflow.Join()

And this is all you need to know about the basic usage. More advanced options are explained below.

SEE example

---

License

MIT licensed. See the LICENSE file for details.

Documentation

Index

• type ChanDispatch
  • func NewChanDispatch(handler Handler, concurrentReadMax int) *ChanDispatch
  • func NewChanDispatch3(handler Handler, concurrentReadMax int, concurrentHandleMax int) *ChanDispatch
  • func (thiz *ChanDispatch) SendMessage(message any) bool
• type Dispatch
  • func NewDispatch(reader Reader, handler Handler) *Dispatch
  • func NewDispatch3(reader Reader, handler Handler, concurrentMax int) *Dispatch
  • func (d *Dispatch) AllowHandleInGroutine() bool
  • func (d *Dispatch) Context() context.Context
  • func (d *Dispatch) GetHandleGoroutine() bool
  • func (d *Dispatch) Handle(msg any) error
  • func (d *Dispatch) Join() *Dispatch
  • func (d *Dispatch) Joinable() *Dispatch
  • func (d *Dispatch) PutHandleGoroutine()
  • func (d *Dispatch) Read() (any, error)
  • func (d *Dispatch) Start() *Dispatch
  • func (d *Dispatch) UnJoinable() *Dispatch
  • func (d *Dispatch) WithContext(ctx context.Context) *Dispatch
• type Handler
• type HandlerFunc
  • func (f HandlerFunc) Handle(ctx context.Context, msg any) error
• type Reader
• type ReaderFunc
  • func (f ReaderFunc) Read(ctx context.Context) (msg any, err error)
• type WaitGroup

Examples

• Dispatch

Types

type ChanDispatch

type ChanDispatch struct {
	*Dispatch
	// contains filtered or unexported fields
}

func NewChanDispatch

func NewChanDispatch(handler Handler, concurrentReadMax int) *ChanDispatch

func NewChanDispatch3

func NewChanDispatch3(handler Handler, concurrentReadMax int, concurrentHandleMax int) *ChanDispatch

func (*ChanDispatch) SendMessage

func (thiz *ChanDispatch) SendMessage(message any) bool

type Dispatch

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

Dispatch is a middleman between the Reader and Processor.

Example

package main

import (
	"context"
	"errors"
	"log"

	"github.com/searKing/golang/go/x/dispatch"
)

type DispatchMsg struct {
	data int
}

func main() {
	var conn chan DispatchMsg
	dispatch.NewDispatch(
		dispatch.ReaderFunc(func(ctx context.Context) (any, error) {
			return ReadMessage(conn)
		}),
		dispatch.HandlerFunc(func(ctx context.Context, msg any) error {
			m := msg.(*DispatchMsg)
			return HandleMessage(m)
		})).Start()
}

func ReadMessage(c <-chan DispatchMsg) (any, error) {
	var msg DispatchMsg
	var ok bool

	if msg, ok = <-c; ok {
		log.Println("Recv From Channel Failed")
		return nil, errors.New("recv from channel failed")
	}
	log.Printf("Recv From Channel Success: %v\n", msg.data)
	return &msg, nil
}

// just print what's received
func HandleMessage(msg *DispatchMsg) error {
	if msg == nil {
		log.Println("Handle From Channel Failed")
		return errors.New("handle from channel failed")
	}
	log.Printf("Handle From Channel Success: %v\n", msg.data)
	return nil
}

func NewDispatch

func NewDispatch(reader Reader, handler Handler) *Dispatch

func NewDispatch3

func NewDispatch3(reader Reader, handler Handler, concurrentMax int) *Dispatch

func (*Dispatch) AllowHandleInGroutine

func (d *Dispatch) AllowHandleInGroutine() bool

func (*Dispatch) Context

func (d *Dispatch) Context() context.Context

func (*Dispatch) GetHandleGoroutine

func (d *Dispatch) GetHandleGoroutine() bool

func (*Dispatch) Handle

func (d *Dispatch) Handle(msg any) error

func (*Dispatch) Join

func (d *Dispatch) Join() *Dispatch

wait until all recv and handle workflows finished, such as join in Thread

func (*Dispatch) Joinable

func (d *Dispatch) Joinable() *Dispatch

make Dispatch joinable Join() blocks until all recv and handle workflows started after Join() is finished RECOMMECD : call Joinable() before Start() to join all workflows

func (*Dispatch) PutHandleGoroutine

func (d *Dispatch) PutHandleGoroutine()

func (*Dispatch) Read

func (d *Dispatch) Read() (any, error)

func (*Dispatch) Start

func (d *Dispatch) Start() *Dispatch

遍历读取消息，并进行分发处理

func (*Dispatch) UnJoinable

func (d *Dispatch) UnJoinable() *Dispatch

make Dispatch unjoinable, as Join() return immediately when called

func (*Dispatch) WithContext

func (d *Dispatch) WithContext(ctx context.Context) *Dispatch

type Handler

type Handler interface {
	Handle(ctx context.Context, msg any) error
}

type HandlerFunc

type HandlerFunc func(ctx context.Context, msg any) error

func (HandlerFunc) Handle

func (f HandlerFunc) Handle(ctx context.Context, msg any) error

type Reader

type Reader interface {
	Read(ctx context.Context) (msg any, err error)
}

type ReaderFunc

type ReaderFunc func(ctx context.Context) (msg any, err error)

func (ReaderFunc) Read

func (f ReaderFunc) Read(ctx context.Context) (msg any, err error)

type WaitGroup

type WaitGroup interface {
	Add(delta int)
	Done()
	Wait()
}