.gitignore

@@ -1,6 +1,8 @@
 # Develop tools
 /.vscode/
 /.idea/
+.idea
+.vscode
 
 # Binaries for programs and plugins
 *.exe
@@ -13,6 +15,7 @@
 _obj
 _test
 _build
+.DS_Store
 
 # Architecture specific extensions/prefixes
 *.[568vq]

broker/broker.go

@@ -4,6 +4,7 @@ package broker // import "go.unistack.org/micro/v3/broker"
 import (
 	"context"
 	"errors"
+	"time"
 
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/metadata"
@@ -17,6 +18,8 @@ var (
 	ErrNotConnected = errors.New("broker not connected")
 	// ErrDisconnected returns when broker disconnected
 	ErrDisconnected = errors.New("broker disconnected")
+	// DefaultGracefulTimeout
+	DefaultGracefulTimeout = 5 * time.Second
 )
 
 // Broker is an interface used for asynchronous messaging.

broker/options.go

@@ -9,6 +9,7 @@ import (
 	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/meter"
 	"go.unistack.org/micro/v3/register"
+	"go.unistack.org/micro/v3/sync"
 	"go.unistack.org/micro/v3/tracer"
 )
 
@@ -36,17 +37,22 @@ type Options struct {
 	Name string
 	// Addrs holds the broker address
 	Addrs []string
+
+	Wait *sync.WaitGroup
+
+	GracefulTimeout time.Duration
 }
 
 // NewOptions create new Options
 func NewOptions(opts ...Option) Options {
 	options := Options{
-		Register: register.DefaultRegister,
-		Logger:   logger.DefaultLogger,
-		Context:  context.Background(),
-		Meter:    meter.DefaultMeter,
-		Codec:    codec.DefaultCodec,
-		Tracer:   tracer.DefaultTracer,
+		Register:        register.DefaultRegister,
+		Logger:          logger.DefaultLogger,
+		Context:         context.Background(),
+		Meter:           meter.DefaultMeter,
+		Codec:           codec.DefaultCodec,
+		Tracer:          tracer.DefaultTracer,
+		GracefulTimeout: DefaultGracefulTimeout,
 	}
 	for _, o := range opts {
 		o(&options)