rungroup

README

rungroup

rungroup is inspired by the run package.

rungroup is a mechanism for launching a group of goroutines and managing their life cycle based on errgroup

Create a RunGroup, and then add actors to it. Actors are defined as a pair of functions: an execute function, which should run synchronously; and an interrupt function, which, when invoked, should cause the execute function to return. Finally, invoke Wait, which waits until the first actor exits, invokes the interrupt functions, and finally returns control to the caller only once all actors have returned. This general-purpose API allows callers to model pretty much any runnable task, and achieve well-defined lifecycle semantics for the group. Can be used to gracefully shutdown a service.

Examples

Create the rungroup

g := rungroup.NewRunGroup()

Add actor with interrupt function

ctx, cancel := context.WithCancel(context.Background())
g.Add(func() error {
	... main actor processed...
	return nil
}, func() error {
	cancel() // terminate main process
	return nil
})

Add actor with cancellation context

g.AddCtx(func(ctx context.Context) error {
	for {
		select {
			case <-ctx.Done():
				return
			default:
			   ...main process...
		}
	}
})

Add actor with net.Listener

ln, _ := net.Listen("tcp", ":8080")
g.Add(func() error {
	return http.Serve(ln, nil)
}, func() error {
	ln.Close()
})

Add actor with io.ReadCloser

var conn io.ReadCloser = ...
g.Add(func() error {
	s := bufio.NewScanner(conn)
	for s.Scan() {
		println(s.Text())
	}
	return s.Err()
}, func(error) {
	conn.Close()
})

Add http actor

httpHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("Hello, World!"))
})

g.Add(rungroup.HttpServerActors(&http.Server{
	Addr:    ":8000",
	Handler: httpHandler,
}))

Add grpc actor

grpcServer := grpc.NewServer()
listener, err := net.Listen("tcp", addr)
g.Add(GrpcServerActors(grpcServer, listener))

Add signal listener actor

g.Add(rungroup.SignalHandlerActors(syscall.SIGINT, syscall.SIGTERM))

Run the group of actors

if err := g.Run(); err != nil {
	print("terminate with error: %s", err.Error())
}

Documentation

Index

• func GrpcServerActors(grpcServer GrpcServer, grpcListener net.Listener, ...) (func() error, func() error)
• func HttpServerActors(httpServer *http.Server, opts ...func(server *wrapServer)) (func() error, func() error)
• func SignalHandlerActors(sig ...os.Signal) (func() error, func() error)
• func WithShutdownTimeout(shutdownTimeout time.Duration) func(w *wrapServer)
• type GrpcServer
• type RunGroup
  • func NewRunGroup() *RunGroup
  • func (g *RunGroup) Add(execute func() error, interrupt func() error)
  • func (g *RunGroup) AddCtx(execute func(ctx context.Context) error)
  • func (g *RunGroup) Run() error

Functions

func GrpcServerActors

func GrpcServerActors(
	grpcServer GrpcServer,
	grpcListener net.Listener,
	opts ...func(server *wrapServer),
) (func() error, func() error)

func HttpServerActors

func HttpServerActors(httpServer *http.Server, opts ...func(server *wrapServer)) (func() error, func() error)

func SignalHandlerActors

func SignalHandlerActors(sig ...os.Signal) (func() error, func() error)

func WithShutdownTimeout

func WithShutdownTimeout(shutdownTimeout time.Duration) func(w *wrapServer)

Types

type GrpcServer

type GrpcServer interface {
	Serve(lis net.Listener) error
	GracefulStop()
	Stop()
}

type RunGroup

type RunGroup struct {
	*errgroup.Group
	// contains filtered or unexported fields
}

func NewRunGroup

func NewRunGroup() *RunGroup

NewRunGroup create actors group

func (*RunGroup) Add

func (g *RunGroup) Add(execute func() error, interrupt func() error)

Add is function add a new execute and interrupt actor

func (*RunGroup) AddCtx

func (g *RunGroup) AddCtx(execute func(ctx context.Context) error)

AddCtx is function add a new execute with cancellation context

func (*RunGroup) Run

func (g *RunGroup) Run() error

Run is function run all added actors