Merge branch 'main' of github.com:linkall-labs/vanus into clientstream-v2

Signed-off-by: jyjiangkai <jyjiangkai@163.com>

Author: hwjiangkai

README.md

@@ -1,17 +1,79 @@
-# Vanus
 
+<div align="center">
+<h1>Vanus</h1>
+<!-- 
+<img src="https://user-images.githubusercontent.capabilitiesom/68597908/206148625-43f14f58-f3c0-4042-82a0-9f9421c270fa.png" width="288" >
+-->
+<p>
+Vanus is an open-source message queue with built-in event processing capabilities.
+<!--
+<br>
+It helps users build event pipelines between SaaS, cloud services, and cloud functions in minutes.
+-->
+</p>
+
+[![stars](https://img.shields.io/github/stars/linkall-labs/vanus.svg?style=flat&logo=github&colorB=blueviolet&label=stars)](https://github.com/linkall-labs/vanus)
 [![License](https://img.shields.io/badge/License-Apache_2.0-green.svg)](https://github.com/linkall-labs/vanus/blob/main/LICENSE)
-[![Language](https://img.shields.io/badge/Language-Go-blue.svg)](https://golang.org/)
 [![codecov](https://codecov.io/gh/linkall-labs/vanus/branch/main/graph/badge.svg?token=RSXSIMEY4V)](https://codecov.io/gh/linkall-labs/vanus)
-
-## What is Vanus
-
-Vanus is a Serverless event platform for easily building Event-Driven Architecture (EDA) applications.
-It provides enterprises and organizations an innovative platform for collecting, storing, distributing, and processing events at scale.
-
-## Quick Start
-
-See [quick-start](https://docs.linkall.com/getting-started/quick-start)
+[![Language](https://img.shields.io/github/go-mod/go-version/linkall-labs/vanus?logo=go)](https://golang.org/)
+[![playground](https://img.shields.io/badge/Playground-Try%20it%20%20free-red)](https://play.linkall.com)
+[![playground](https://img.shields.io/badge/Docs-online-green)](https://linkall.com/docs)
+
+
+<p>
+    <a href="https://join.slack.com/t/vanusworkspace/shared_invite/zt-1jilbbfo2-NxiFG0VOo8ABGCCNaeNfcA"><img src="https://img.shields.io/badge/slack-join-3CC798?style=social&logo=slack" height=23></a>   
+    &nbsp;
+    <a href="https://twitter.com/Vanus_dev"><img src="https://img.shields.io/badge/-Twitter-red?style=social&logo=twitter" height=23></a>
+   <!-- <a href="https://twitter.com/Vanus_dev"><img src="https://img.shields.io/twitter/follow/vanus_dev?style=social" height=23></a> -->
+    &nbsp;
+    <a href="https://www.youtube.com/channel/UC7rd6IgjfNYTOXf2FerFNyA"><img src="https://img.shields.io/badge/-Youtube-red?style=social&logo=youtube" height=23></a>
+   <!-- <a href="https://www.youtube.com/channel/UC7rd6IgjfNYTOXf2FerFNyA"><img src="https://img.shields.io/youtube/channel/views/UC7rd6IgjfNYTOXf2FerFNyA?style=social" height=23></a>  -->
+    &nbsp;
+</p>
+
+<!--
+<h3 align="center">
+  <a href="https://featbit.gitbook.io/docs/installation">Installation</a>
+  <span> · </span>
+  <a href="https://featbit.gitbook.io/">Getting Started</a>
+  <span> · </span>
+  <a href="https://join.slack.com/t/featbit/shared_invite/zt-1ew5e2vbb-x6Apan1xZOaYMnFzqZkGNQ">Online Support</a>
+  <span> · </span>
+  <a href="https://featbit.gitbook.io/">Documentation</a>  
+  <span> · </span>
+  <a href="https://github.com/featbit/featbit/discussions/categories/announcements">Milestones</a>
+</h3>
+-->
+</div>
+
+--------------------------------------------------
+
+## Introduction
+
+Vanus helps users build event pipelines between SaaS, cloud services, and cloud functions in minutes.
+
+**1. Build the event-driven system**
+- Get events from cloud services and SaaS, and deliver them to cloud functions or microservices.
+- Synchronize changed data or transfer data to the data lake.
+- Obtain events generated by SaaS and send them to other SaaS.
+
+**2. Out-of-the-box event computing capabilities**
+- Real-time processing during event transmissions, such as filtering and transformation.
+- Natively support the CloudEvents specification, and can directly send events to workloads that support CloudEvent.
+
+**3. 100% open source, Super easy to use**
+- One-click deployment, the installation is completed within 1 minute, and developers without MQ experience can also use it.
+- Message queues and connectors are 100% open source, a one-stop open-source solution.
+
+## Getting Started
+
+You can install Vanus with a single command within 1 minute. Check out our [website](https://linkall.com) for detailed information.
+
+```shell
+kubectl apply -f https://download.linkall.com/all-in-one/v0.5.4.yml
+```
+
+We also provide an [interactive Kubernetes environment](https://play.linkall.com) to simply deploy and try Vanus in your browser.
 
 ## Community
 
@@ -26,17 +88,6 @@ We have a few channels for contact:
 
 See [here](CONTRIBUTING.md) for how to contribute to Vanus.
 
-## Understanding Vanus
-
-### Project Layout
-
-This Project follows [golang-standards/project-layout](https://github.com/golang-standards/project-layout), see the
-project for understanding well vanus' codebase.
-
-### Architecture
-
-![architecture](docs/architecture.jpg)
-
 ## License
 
 Vanus is under the Apache 2.0 license. See the [LICENSE](LICENSE) file for details.

client/internal/vanus/store/block_store.go

@@ -18,8 +18,8 @@ import (
 	// standard libraries
 	"context"
 	"io"
-	"math/rand"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/linkall-labs/vanus/observability/tracing"
@@ -66,17 +66,18 @@ func newBlockStore(endpoint string) (*BlockStore, error) {
 	return s, nil
 }
 
-type streamstate string
+type streamState string
 
 var (
-	stateRunning streamstate = "running"
-	stateCLosed  streamstate = "closed"
+	stateRunning streamState = "running"
+	stateClosed  streamState = "closed"
 )
 
 type appendStreamCache struct {
+	opaqueID  uint64
 	stream    segpb.SegmentServer_AppendToBlockStreamClient
 	callbacks sync.Map
-	state     streamstate
+	state     streamState
 	once      sync.Once
 }
 
@@ -85,7 +86,7 @@ func (a *appendStreamCache) isRunning() bool {
 }
 
 func (a *appendStreamCache) isClosed() bool {
-	return a.state == stateCLosed
+	return a.state == stateClosed
 }
 
 func (a *appendStreamCache) release() {
@@ -96,27 +97,26 @@ func (a *appendStreamCache) release() {
 func (a *appendStreamCache) releaseStream() {
 	a.once.Do(func() {
 		a.stream.CloseSend()
-		a.state = stateCLosed
+		a.state = stateClosed
 	})
 }
 
 func (a *appendStreamCache) releaseCallbacks() {
 	a.callbacks.Range(func(key, value interface{}) bool {
-		if value != nil {
-			value.(appendCallback)(&segpb.AppendToBlockStreamResponse{
-				ResponseCode: errpb.ErrorCode_CLOSED,
-				ResponseMsg:  "append stream closed",
-				Offsets:      []int64{},
-			})
-		}
+		value.(appendCallback)(&segpb.AppendToBlockStreamResponse{
+			ResponseCode: errpb.ErrorCode_CLOSED,
+			ResponseMsg:  "append stream closed",
+			Offsets:      []int64{},
+		})
 		return true
 	})
 }
 
 type readStreamCache struct {
+	opaqueID  uint64
 	stream    segpb.SegmentServer_ReadFromBlockStreamClient
 	callbacks sync.Map
-	state     streamstate
+	state     streamState
 	once      sync.Once
 }
 
@@ -125,7 +125,7 @@ func (r *readStreamCache) isRunning() bool {
 }
 
 func (r *readStreamCache) isClosed() bool {
-	return r.state == stateCLosed
+	return r.state == stateClosed
 }
 
 func (r *readStreamCache) release() {
@@ -136,21 +136,19 @@ func (r *readStreamCache) release() {
 func (r *readStreamCache) releaseStream() {
 	r.once.Do(func() {
 		r.stream.CloseSend()
-		r.state = stateCLosed
+		r.state = stateClosed
 	})
 }
 
 func (r *readStreamCache) releaseCallbacks() {
 	r.callbacks.Range(func(key, value interface{}) bool {
-		if value != nil {
-			value.(readCallback)(&segpb.ReadFromBlockStreamResponse{
-				ResponseCode: errpb.ErrorCode_CLOSED,
-				ResponseMsg:  "read stream closed",
-				Events: &cepb.CloudEventBatch{
-					Events: []*cepb.CloudEvent{},
-				},
-			})
-		}
+		value.(readCallback)(&segpb.ReadFromBlockStreamResponse{
+			ResponseCode: errpb.ErrorCode_CLOSED,
+			ResponseMsg:  "read stream closed",
+			Events: &cepb.CloudEventBatch{
+				Events: []*cepb.CloudEvent{},
+			},
+		})
 		return true
 	})
 }
@@ -335,7 +333,7 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 	}
 
 	// generate unique opaqueID
-	opaqueID := rand.New(rand.NewSource(time.Now().UnixNano())).Uint64()
+	atomic.AddUint64(&append.opaqueID, 1)
 
 	//TODO(jiangkai): delete the reference of CloudEvents/v2 in Vanus
 	eventpbs := make([]*cepb.CloudEvent, len(events))
@@ -348,13 +346,13 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 	}
 
 	donec := make(chan struct{})
-	append.callbacks.Store(opaqueID, appendCallback(func(res *segpb.AppendToBlockStreamResponse) {
+	append.callbacks.Store(append.opaqueID, appendCallback(func(res *segpb.AppendToBlockStreamResponse) {
 		resp = res
 		close(donec)
 	}))
 
 	req := &segpb.AppendToBlockStreamRequest{
-		Id:      opaqueID,
+		Id:      append.opaqueID,
 		BlockId: block,
 		Events: &cepb.CloudEventBatch{
 			Events: eventpbs,
@@ -369,10 +367,10 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 		append.releaseStream()
 		// reset new stream connections
 		s.connectAppendStream(ctx)
-		c, _ := append.callbacks.LoadAndDelete(opaqueID)
+		c, _ := append.callbacks.LoadAndDelete(append.opaqueID)
 		if c != nil {
 			c.(appendCallback)(&segpb.AppendToBlockStreamResponse{
-				Id:           opaqueID,
+				Id:           append.opaqueID,
 				ResponseCode: errpb.ErrorCode_CLOSED,
 				ResponseMsg:  "append stream closed",
 				Offsets:      []int64{},
@@ -384,10 +382,10 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 	select {
 	case <-donec:
 	case <-_ctx.Done():
-		c, _ := append.callbacks.LoadAndDelete(opaqueID)
+		c, _ := append.callbacks.LoadAndDelete(append.opaqueID)
 		if c != nil {
 			c.(appendCallback)(&segpb.AppendToBlockStreamResponse{
-				Id:           opaqueID,
+				Id:           append.opaqueID,
 				ResponseCode: errpb.ErrorCode_CONTEXT_CANCELED,
 				ResponseMsg:  "append stream context canceled",
 				Offsets:      []int64{},
@@ -465,10 +463,10 @@ func (s *BlockStore) ReadStream(
 	}
 
 	// generate unique RequestId
-	opaqueID := rand.New(rand.NewSource(time.Now().UnixNano())).Uint64()
+	atomic.AddUint64(&read.opaqueID, 1)
 
 	donec := make(chan struct{})
-	read.callbacks.Store(opaqueID, readCallback(func(res *segpb.ReadFromBlockStreamResponse) {
+	read.callbacks.Store(read.opaqueID, readCallback(func(res *segpb.ReadFromBlockStreamResponse) {
 		resp = res
 		close(donec)
 	}))
@@ -486,10 +484,10 @@ func (s *BlockStore) ReadStream(
 		})
 		read.releaseStream()
 		s.connectReadStream(ctx)
-		c, _ := read.callbacks.LoadAndDelete(opaqueID)
+		c, _ := read.callbacks.LoadAndDelete(read.opaqueID)
 		if c != nil {
 			c.(readCallback)(&segpb.ReadFromBlockStreamResponse{
-				Id:           opaqueID,
+				Id:           read.opaqueID,
 				ResponseCode: errpb.ErrorCode_CLOSED,
 				ResponseMsg:  "read stream closed",
 				Events: &cepb.CloudEventBatch{
@@ -503,10 +501,10 @@ func (s *BlockStore) ReadStream(
 	select {
 	case <-donec:
 	case <-_ctx.Done():
-		c, _ := read.callbacks.LoadAndDelete(opaqueID)
+		c, _ := read.callbacks.LoadAndDelete(read.opaqueID)
 		if c != nil {
 			c.(readCallback)(&segpb.ReadFromBlockStreamResponse{
-				Id:           opaqueID,
+				Id:           read.opaqueID,
 				ResponseCode: errpb.ErrorCode_CONTEXT_CANCELED,
 				ResponseMsg:  "read stream context canceled",
 				Events: &cepb.CloudEventBatch{

client/pkg/eventlog/log_segment.go

@@ -239,7 +239,7 @@ func (s *segment) Read(ctx context.Context, from int64, size int16, pollingTimeo
 		}
 		off, ok := v.(int32)
 		if !ok {
-			return events, errors.ErrInvalidRequest.WithMessage("corrupted event")
+			return events, errors.ErrCorruptedEvent.WithMessage("corrupted event")
 		}
 		offset := s.startOffset + int64(off)
 		buf := make([]byte, 8)

internal/controller/eventbus/controller.go

@@ -153,7 +153,7 @@ func (ctrl *controller) createEventBus(ctx context.Context,
 	ctrl.mutex.Lock()
 	defer ctrl.mutex.Unlock()
 	if !ctrl.isReady(ctx) {
-		return nil, errors.ErrServerNotRunning.WithMessage("the cluster isn't ready to create eventbus")
+		return nil, errors.ErrResourceNotReady.WithMessage("the cluster isn't ready to create eventbus")
 	}
 	logNum := req.LogNumber
 	if logNum == 0 {

internal/controller/eventbus/controller_test.go

@@ -95,7 +95,7 @@ func TestController_CreateEventBus(t *testing.T) {
 			So(res, ShouldBeNil)
 			et, ok := err.(*errors.ErrorType)
 			So(ok, ShouldBeTrue)
-			So(et.Code, ShouldEqual, errpb.ErrorCode_RESOURCE_EXIST)
+			So(et.Code, ShouldEqual, errpb.ErrorCode_RESOURCE_ALREADY_EXIST)
 			So(et.Description, ShouldEqual, "resource already exist")
 			So(et.Message, ShouldEqual, "the eventbus already exist")
 		})

internal/controller/eventbus/eventlog/eventlog.go

@@ -670,7 +670,7 @@ func (mgr *eventlogManager) createSegment(ctx context.Context, el *eventlog) (*S
 		ins := mgr.volMgr.GetVolumeInstanceByID(seg.GetLeaderBlock().VolumeID)
 		srv := ins.GetServer()
 		if srv == nil {
-			return nil, errors.ErrServerNotRunning.WithMessage("volume instance no server")
+			return nil, errors.ErrResourceNotReady.WithMessage("volume instance no server")
 		}
 		_, err = srv.GetClient().ActivateSegment(ctx, &segment.ActivateSegmentRequest{
 			EventLogId:     seg.EventLogID.Uint64(),
@@ -888,7 +888,7 @@ func (el *eventlog) add(ctx context.Context, seg *Segment) error {
 	el.mutex.Lock()
 	defer el.mutex.Unlock()
 	if !seg.isReady() {
-		return errors.ErrServerNotRunning.WithMessage("the segment not ready")
+		return errors.ErrResourceNotReady.WithMessage("the segment not ready")
 	}
 	if el.get(seg.ID) != nil {
 		return nil

internal/controller/eventbus/server/instance.go

@@ -64,7 +64,7 @@ func (ins *volumeInstance) CreateBlock(ctx context.Context, capacity int64) (*me
 		VolumeID: ins.md.ID,
 	}
 	if ins.srv == nil {
-		return nil, errors.ErrServerNotRunning.WithMessage("volume instance no server")
+		return nil, errors.ErrResourceNotReady.WithMessage("volume instance no server")
 	}
 	_, err = ins.srv.GetClient().CreateBlock(ctx, &segpb.CreateBlockRequest{
 		Size: blk.Capacity,
@@ -83,7 +83,7 @@ func (ins *volumeInstance) CreateBlock(ctx context.Context, capacity int64) (*me
 
 func (ins *volumeInstance) DeleteBlock(ctx context.Context, id vanus.ID) error {
 	if ins.srv == nil {
-		return errors.ErrServerNotRunning.WithMessage("volume instance no server")
+		return errors.ErrResourceNotReady.WithMessage("volume instance no server")
 	}
 	blk := ins.md.Blocks[id.Uint64()]
 	if blk == nil {

internal/controller/trigger/controller.go

@@ -119,7 +119,7 @@ func (ctrl *controller) ResetOffsetToTimestamp(ctx context.Context,
 		return nil, errors.ErrResourceNotFound.WithMessage("subscription not exist")
 	}
 	if sub.Phase != metadata.SubscriptionPhaseRunning {
-		return nil, errors.ErrServerNotRunning.WithMessage("subscription is not running")
+		return nil, errors.ErrResourceNotReady.WithMessage("subscription is not running")
 	}
 	tWorker := ctrl.workerManager.GetTriggerWorker(sub.TriggerWorker)
 	if tWorker == nil {

internal/store/block/block.go

@@ -44,8 +44,10 @@ type Reader interface {
 	Read(ctx context.Context, seq int64, num int) ([]Entry, error)
 }
 
+type AppendCallback = func(seqs []int64, err error)
+
 type Appender interface {
-	Append(ctx context.Context, entries []Entry, cb func([]int64, error))
+	Append(ctx context.Context, entries []Entry, cb AppendCallback)
 }
 
 type Block interface {