workgroup

README

go-workgroup

Forked from: da440dil/go-workgroup

Synchronization for groups of related goroutines.

Example HTTP server

// Create workgroup
var wg workgroup.Group
// Add function to cancel execution using os signal
wg.Add(workgroup.Signal())
// Create http server
srv := http.Server{Addr: "127.0.0.1:8080"}
// Add function to start and stop http server
wg.Add(workgroup.Server(
	func() error {
		fmt.Printf("Server listen at %v\n", srv.Addr)
		err := srv.ListenAndServe()
		fmt.Printf("Server stopped listening with error: %v\n", err)
		if err != http.ErrServerClosed {
			return err
		}
		return nil
	},
	func() error {
		fmt.Println("Server is about to shutdown")
		ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
		defer cancel()

		err := srv.Shutdown(ctx)
		fmt.Printf("Server shutdown with error: %v\n", err)
		return err
	},
))
// Create context to cancel execution after 5 seconds
ctx, cancel := context.WithCancel(context.Background())
go func() {
	time.Sleep(time.Second * 5)
	fmt.Println("Context canceled")
	cancel()
}()
// Add function to cancel execution using context
wg.Add(workgroup.Context(ctx))
// Execute each function
err := wg.Run()
fmt.Printf("Workgroup run stopped with error: %v\n", err)
// Output in case execution has been canceled with context:
// Server listen at 127.0.0.1:8080
// Context canceled
// Server is about to shutdown
// Server stopped listening with error: http: Server closed
// Server shutdown with error: <nil>
// Workgroup run stopped with error: context canceled
//
// Output in case execution has been interrupted with os signal:
// Server listen at 127.0.0.1:8080
// Server is about to shutdown
// Server stopped listening with error: http: Server closed
// Server shutdown with error: <nil>
// Workgroup run stopped with error: <nil>

Example gRPC server

// Create workgroup
var wg workgroup.Group
// Add function to cancel execution using os signal
wg.Add(workgroup.Signal())
// Create grpc server
srv := grpc.NewServer()
// Add function to start and stop grpc server
wg.Add(workgroup.Server(
	func() error {
		lis, err := net.Listen("tcp", "127.0.0.1:50051")
		if err != nil {
			return err
		}
		fmt.Printf("Server listen at %v\n", lis.Addr())
		err = srv.Serve(lis)
		fmt.Printf("Server stopped listening with error: %v\n", err)
		return err
	},
	func() error {
		fmt.Println("Server is about to shutdown")
		ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
		defer cancel()

		var wg workgroup.Group
		wg.Add(workgroup.Context(ctx))
		wg.Add(func(stop <-chan struct{}) error {
			srv.GracefulStop()
			return nil
		})
		err := wg.Run()
		fmt.Printf("Server shutdown with error: %v\n", err)
		return err
	},
))
// Create context to cancel execution after 5 seconds
ctx, cancel := context.WithCancel(context.Background())
go func() {
	time.Sleep(time.Second * 5)
	fmt.Println("Context canceled")
	cancel()
}()
// Add function to cancel execution using context
wg.Add(workgroup.Context(ctx))
// Execute each function
err := wg.Run()
fmt.Printf("Workgroup run stopped with error: %v\n", err)
// Output in case execution has been canceled with context:
// Server listen at 127.0.0.1:50051
// Context canceled
// Server is about to shutdown
// Server shutdown with error: <nil>
// Server stopped listening with error: <nil>
// Workgroup run stopped with error: context canceled
//
// Output in case execution has been interrupted with os signal:
// Server listen at 127.0.0.1:50051
// Server is about to shutdown
// Server shutdown with error: <nil>
// Server stopped listening with error: <nil>
// Workgroup run stopped with error: <nil>

Documentation

Overview

Package workgroup provides synchronization for groups of related goroutines.

Index

• type Group
  • func (g *Group) Add(fn RunFunc)
  • func (g *Group) Run() error
• type RunFunc
  • func Context(ctx context.Context) RunFunc
  • func Server(serve func() error, shutdown func() error) RunFunc
  • func Signal(sig ...os.Signal) RunFunc

Examples

• Group

Types

type Group

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

Group is a group of related goroutines. The zero value for a Group is fully usable without initialization.

Example

var g workgroup.Group

wait := make(chan struct{})

g.Add(func(<-chan struct{}) error {
	<-wait
	fmt.Println("one")
	return errors.New("three")
})
g.Add(func(stop <-chan struct{}) error {
	<-stop
	fmt.Println("two")
	return errors.New("four")
})

result := make(chan error)
go func() {
	result <- g.Run()
}()

close(wait)
fmt.Printf("%v\n", <-result)

Output:
one
two
three

func (*Group) Add

func (g *Group) Add(fn RunFunc)

Add adds a function to the Group. The function will be exectuted in its own goroutine when Run is called. Add must be called before Run.

func (*Group) Run

func (g *Group) Run() error

Run executes each function registered via Add in its own goroutine. Run blocks until all functions have returned, then returns the first non-nil error (if any) from them. The first function to return will trigger the closure of the channel passed to each function, which should in turn, return.

type RunFunc

type RunFunc func(<-chan struct{}) error

RunFunc is a function to execute with other related functions in its own goroutine. The closure of the channel passed to RunFunc should trigger return.

func Context

func Context(ctx context.Context) RunFunc

Context creates function for canceling execution using context.

func Server

func Server(serve func() error, shutdown func() error) RunFunc

Server creates function for canceling execution. First function passed in should block. Second function passed in should unblock first function.

func Signal

func Signal(sig ...os.Signal) RunFunc

Signal creates function for canceling execution using os signal.