feat: add eventbus state check

client/client.go

@@ -28,26 +28,20 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	// this project.
-	eb "github.com/vanus-labs/vanus/client/internal/vanus/eventbus"
+	eb "github.com/vanus-labs/vanus/client/internal/eventbus"
 	"github.com/vanus-labs/vanus/client/pkg/api"
 	"github.com/vanus-labs/vanus/client/pkg/eventbus"
 )
 
-type Client interface {
-	Eventbus(ctx context.Context, opts ...api.EventbusOption) api.Eventbus
-	Disconnect(ctx context.Context)
-}
-
 type client struct {
-	// Endpoints is a list of URLs.
-	Endpoints     []string
-	eventbusCache sync.Map
+	endpoints []string
+	cache     sync.Map
 
 	mu     sync.RWMutex
 	tracer *tracing.Tracer
 }
 
-func (c *client) Eventbus(ctx context.Context, opts ...api.EventbusOption) api.Eventbus {
+func (c *client) Eventbus(ctx context.Context, opts ...api.EventbusOption) (api.Eventbus, error) {
 	_, span := c.tracer.Start(ctx, "EventbusService")
 	defer span.End()
 
@@ -56,19 +50,18 @@ func (c *client) Eventbus(ctx context.Context, opts ...api.EventbusOption) api.E
 		apply(defaultOpts)
 	}
 
-	err := GetEventbusIDIfNotSet(ctx, c.Endpoints, defaultOpts)
+	err := GetEventbusIDIfNotSet(ctx, c.endpoints, defaultOpts)
 	if err != nil {
 		log.Error(ctx).Err(err).
 			Str("eventbus_name", defaultOpts.Name).
 			Uint64("eventbus_id", defaultOpts.ID).
 			Msg("get eventbus id failed")
-		return nil
+		return nil, err
 	}
 
 	bus := func() api.Eventbus {
-		c.mu.RLock()
-		defer c.mu.RUnlock()
-		if value, ok := c.eventbusCache.Load(defaultOpts.ID); ok {
+		if value, ok := c.cache.Load(defaultOpts.ID); ok {
+			value.(*eventbus.Eventbus).Acquire()
 			return value.(api.Eventbus)
 		} else {
 			return nil
@@ -78,37 +71,39 @@ func (c *client) Eventbus(ctx context.Context, opts ...api.EventbusOption) api.E
 	if bus == nil {
 		c.mu.Lock()
 		defer c.mu.Unlock()
-		if value, ok := c.eventbusCache.Load(defaultOpts.ID); ok { // double check
-			return value.(api.Eventbus)
+		if value, ok := c.cache.Load(defaultOpts.ID); ok { // double check
+			return value.(api.Eventbus), nil
 		} else {
 			cfg := &eb.Config{
-				Endpoints: c.Endpoints,
+				Endpoints: c.endpoints,
 				ID:        defaultOpts.ID,
 			}
-			newEventbus := eventbus.NewEventbus(cfg)
-			c.eventbusCache.Store(defaultOpts.ID, newEventbus)
-			return newEventbus
+			newEventbus := eventbus.NewEventbus(cfg, c.close)
+			newEventbus.Acquire()
+			c.cache.Store(defaultOpts.ID, newEventbus)
+			return newEventbus, nil
 		}
 	}
-	return bus
+	return bus, nil
 }
 
 func (c *client) Disconnect(ctx context.Context) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.eventbusCache.Range(func(key, value interface{}) bool {
+	c.cache.Range(func(key, value interface{}) bool {
 		value.(api.Eventbus).Close(ctx)
-		c.eventbusCache.Delete(key)
 		return true
 	})
 }
 
-func Connect(endpoints []string) Client {
+func (c *client) close(id uint64) {
+	c.cache.Delete(id)
+}
+
+func Connect(endpoints []string) api.Client {
 	if len(endpoints) == 0 {
 		return nil
 	}
 	return &client{
-		Endpoints: endpoints,
+		endpoints: endpoints,
 	}
 }
 

client/examples/eventbus/append/main.go

@@ -43,7 +43,10 @@ func main() {
 	if err != nil {
 		panic("invalid id")
 	}
-	bus := c.Eventbus(ctx, api.WithName("quick-start"), api.WithID(eventbusID.Uint64()))
+	bus, err := c.Eventbus(ctx, api.WithID(eventbusID.Uint64()))
+	if err != nil {
+		panic(err.Error())
+	}
 	w := bus.Writer()
 	// Create an Event.
 	event := ce.NewEvent()

client/examples/eventbus/read/main.go

@@ -37,7 +37,10 @@ func main() {
 	if err != nil {
 		panic("invalid id")
 	}
-	eb := c.Eventbus(ctx, api.WithID(eventbusID.Uint64()))
+	eb, err := c.Eventbus(ctx, api.WithID(eventbusID.Uint64()))
+	if err != nil {
+		panic(err.Error())
+	}
 	ls, err := eb.ListLog(ctx)
 	if err != nil {
 		log.Print(err.Error())

client/internal/vanus/net/connection.go → client/internal/net/connection.go

@@ -14,6 +14,6 @@
 
 package net
 
-import "github.com/vanus-labs/vanus/client/internal/vanus/net/connection"
+import "github.com/vanus-labs/vanus/client/internal/net/connection"
 
 var Connect = connection.Connect

client/internal/vanus/net/rpc/bare/client.go → client/internal/net/rpc/bare/client.go

@@ -31,8 +31,8 @@ import (
 	"github.com/vanus-labs/vanus/pkg/errors"
 
 	// this project.
-	"github.com/vanus-labs/vanus/client/internal/vanus/net/connection"
-	"github.com/vanus-labs/vanus/client/internal/vanus/net/rpc"
+	"github.com/vanus-labs/vanus/client/internal/net/connection"
+	"github.com/vanus-labs/vanus/client/internal/net/rpc"
 )
 
 const (

client/internal/vanus/store/block_store.go → client/internal/store/block_store.go

@@ -29,8 +29,8 @@ import (
 	segpb "github.com/vanus-labs/vanus/proto/pkg/segment"
 
 	// this project.
-	"github.com/vanus-labs/vanus/client/internal/vanus/net/rpc"
-	"github.com/vanus-labs/vanus/client/internal/vanus/net/rpc/bare"
+	"github.com/vanus-labs/vanus/client/internal/net/rpc"
+	"github.com/vanus-labs/vanus/client/internal/net/rpc/bare"
 	"github.com/vanus-labs/vanus/client/pkg/primitive"
 )
 

client/pkg/api/client.go

@@ -23,6 +23,13 @@ import (
 	"github.com/vanus-labs/vanus/proto/pkg/codec"
 )
 
+type CloseFunc func(id uint64)
+
+type Client interface {
+	Eventbus(ctx context.Context, opts ...EventbusOption) (Eventbus, error)
+	Disconnect(ctx context.Context)
+}
+
 type Eventbus interface {
 	Writer(opts ...WriteOption) BusWriter
 	Reader(opts ...ReadOption) BusReader