Repository: vanus-labs/vanus
Pull request #457 (2023-02-24 05:48:41)

feat(trigger): split Number type of transform to Int and Float
<!--

Thank you for contributing to Vanus!

PR Title Format: 

feat/fix/refactor/docs/...: what's changed

Note: see https://github.com/linkall-labs/vanus/blob/main/CONTRIBUTING.md to know details about title format
-->

### What problem does this PR solve?
<!--

Please create an issue first to describe the problem

There MUST be one line starting with "Issue Number:  ".

-->

Issue Number: close #xxx

### Problem Summary

### What is changed and how does it work?

### Check List

<!-- At least one of them must be included. -->

#### Tests

- [x] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No code


codecov[bot]: # [Codecov](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) Report
> Merging [#457](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) (d82f767) into [main](https://codecov.io/gh/vanus-labs/vanus/commit/2a78cb5c98e21ede2d084739bfadd4151ac48054?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) (2a78cb5) will **decrease** coverage by `0.56%`.
> The diff coverage is `35.10%`.

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://codecov.io/gh/vanus-labs/vanus/pull/457/graphs/tree.svg?width=650&height=150&src=pr&token=RSXSIMEY4V&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs)](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs)

```diff
@@            Coverage Diff             @@
##             main     #457      +/-   ##
==========================================
- Coverage   57.71%   57.15%   -0.56%     
==========================================
  Files         180      184       +4     
  Lines       14614    15005     +391     
==========================================
+ Hits         8434     8576     +142     
- Misses       5552     5797     +245     
- Partials      628      632       +4     
```


| [Impacted Files](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) | Coverage Δ | |
|---|---|---|
| [...ernal/controller/trigger/validation/subscripton.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvY29udHJvbGxlci90cmlnZ2VyL3ZhbGlkYXRpb24vc3Vic2NyaXB0b24uZ28=) | `69.29% <ø> (+0.85%)` | :arrow_up: |
| [internal/gateway/proxy/deadletter.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvZ2F0ZXdheS9wcm94eS9kZWFkbGV0dGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [internal/gateway/proxy/direct.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvZ2F0ZXdheS9wcm94eS9kaXJlY3QuZ28=) | `73.33% <0.00%> (-5.24%)` | :arrow_down: |
| [internal/store/segment/server.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvc3RvcmUvc2VnbWVudC9zZXJ2ZXIuZ28=) | `14.10% <0.00%> (-0.03%)` | :arrow_down: |
| [internal/trigger/client/gcloud\_functions.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvdHJpZ2dlci9jbGllbnQvZ2Nsb3VkX2Z1bmN0aW9ucy5nbw==) | `0.00% <0.00%> (ø)` | |
| [internal/trigger/client/grpc.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvdHJpZ2dlci9jbGllbnQvZ3JwYy5nbw==) | `0.00% <0.00%> (ø)` | |
| [internal/trigger/client/http.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvdHJpZ2dlci9jbGllbnQvaHR0cC5nbw==) | `0.00% <0.00%> (ø)` | |
| [internal/trigger/client/interface.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvdHJpZ2dlci9jbGllbnQvaW50ZXJmYWNlLmdv) | `0.00% <0.00%> (-19.24%)` | :arrow_down: |
| [internal/trigger/client/lambda.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvdHJpZ2dlci9jbGllbnQvbGFtYmRhLmdv) | `0.00% <0.00%> (ø)` | |
| [internal/gateway/proxy/proxy.go](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs#diff-aW50ZXJuYWwvZ2F0ZXdheS9wcm94eS9wcm94eS5nbw==) | `23.08% <6.52%> (-2.01%)` | :arrow_down: |
| ... and [29 more](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs). Last update [1da484d...d82f767](https://codecov.io/gh/vanus-labs/vanus/pull/457?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=vanus-labs).

</details>

Patch:
diff --git a/internal/controller/eventbus/eventlog/eventlog.go b/internal/controller/eventbus/eventlog/eventlog.go
index 301a9e5..38045ed 100644
--- a/internal/controller/eventbus/eventlog/eventlog.go
+++ b/internal/controller/eventbus/eventlog/eventlog.go
@@ -52,14 +52,12 @@ type Manager interface {
 	AcquireEventLog(ctx context.Context, eventbusID vanus.ID, eventbusName string) (*metadata.Eventlog, error)
 	GetEventLog(ctx context.Context, id vanus.ID) *metadata.Eventlog
 	DeleteEventlog(ctx context.Context, id vanus.ID)
-	GetEventLogSegmentList(elID vanus.ID) []*Segment
-	GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog,
-		num int) ([]*Segment, error)
-	UpdateSegment(ctx context.Context, m map[string][]Segment)
-	GetSegmentByBlockID(block *metadata.Block) (*Segment, error)
 	GetBlock(id vanus.ID) *metadata.Block
-	GetSegment(id vanus.ID) *Segment
 	UpdateSegmentReplicas(ctx context.Context, segID vanus.ID, term uint64) error
+	GetEventLogSegmentList(elID vanus.ID) []Segment
+	GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]Segment, error)
+	UpdateSegment(ctx context.Context, m map[string][]Segment)
+	GetSegmentByBlockID(block *metadata.Block) (Segment, error)
 }
 
 var mgr = &eventlogManager{
@@ -135,15 +133,11 @@ func (mgr *eventlogManager) Run(ctx context.Context, kvClient kv.Client, startTa
 			return err2
 		}
 		mgr.eventLogMap.Store(elMD.ID.Key(), el)
-		ptr := el.segmentList.Front()
-		if ptr != nil {
-			seg, _ := ptr.Value.(*Segment)
-			for seg != nil {
-				mgr.globalSegmentMap.Store(seg.ID.Key(), seg)
-				for _, v := range seg.Replicas.Peers {
-					mgr.globalBlockMap.Store(v.ID.Key(), v)
-				}
-				seg = el.nextOf(seg)
+
+		for _, ptr := range el.getAllSegments() {
+			mgr.globalSegmentMap.Store(ptr.ID.Key(), ptr)
+			for _, v := range ptr.Replicas.Peers {
+				mgr.globalBlockMap.Store(v.ID.Key(), v)
 			}
 		}
 	}
@@ -250,9 +244,8 @@ func (mgr *eventlogManager) DeleteEventlog(ctx context.Context, id vanus.ID) {
 }
 
 func (mgr *eventlogManager) GetAppendableSegment(ctx context.Context,
-	eli *metadata.Eventlog, num int) ([]*Segment, error) {
-	result := make([]*Segment, 0)
-
+	eli *metadata.Eventlog, num int) ([]Segment, error) {
+	result := make([]Segment, 0, num)
 	if eli == nil || num == 0 {
 		return result, nil
 	}
@@ -286,14 +279,13 @@ func (mgr *eventlogManager) GetAppendableSegment(ctx context.Context,
 		s = el.currentAppendableSegment()
 	}
 
-	for len(result) < num && s != nil {
-		result = append(result, s)
-		s = el.nextOf(s)
-	}
-	return result, nil
+	return el.listOfRight(s, true), nil
 }
 
 func (mgr *eventlogManager) UpdateSegment(ctx context.Context, m map[string][]Segment) {
+	mgr.mutex.Lock()
+	defer mgr.mutex.Unlock()
+
 	// iterate eventlog
 	for eventlogID, segments := range m {
 		v, exist := mgr.eventLogMap.Load(eventlogID)
@@ -336,17 +328,20 @@ func (mgr *eventlogManager) UpdateSegment(ctx context.Context, m map[string][]Se
 	}
 }
 
-func (mgr *eventlogManager) GetEventLogSegmentList(elID vanus.ID) []*Segment {
-	result := make([]*Segment, 0)
+func (mgr *eventlogManager) GetEventLogSegmentList(elID vanus.ID) []Segment {
+	result := make([]Segment, 0)
 	v, exist := mgr.eventLogMap.Load(elID.Key())
 	if !exist {
 		return result
 	}
+
 	el, _ := v.(*eventlog)
-	s := el.head()
-	for s != nil {
-		result = append(result, s)
-		s = el.nextOf(s)
+	list := el.getAllSegments()
+	el.lock()
+	defer el.unlock()
+
+	for _, v := range list {
+		result = append(result, v.Copy())
 	}
 	return result
 }
@@ -359,7 +354,7 @@ func (mgr *eventlogManager) GetBlock(id vanus.ID) *metadata.Block {
 	return v.(*metadata.Block)
 }
 
-func (mgr *eventlogManager) GetSegment(id vanus.ID) *Segment {
+func (mgr *eventlogManager) getSegment(id vanus.ID) *Segment {
 	v, exist := mgr.globalSegmentMap.Load(id.Key())
 	if !exist {
 		return nil
@@ -373,7 +368,7 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 		return errors.ErrBlockNotFound
 	}
 
-	seg := mgr.GetSegment(blk.SegmentID)
+	seg := mgr.getSegment(blk.SegmentID)
 	if seg == nil {
 		return errors.ErrSegmentNotFound
 	}
@@ -382,6 +377,13 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 		return nil
 	}
 
+	el := mgr.getEventLog(seg.EventLogID)
+	if el == nil {
+		return errors.ErrEventLogNotFound
+	}
+	el.lock()
+	defer el.unlock()
+
 	seg.Replicas.Leader = leaderID.Uint64()
 	seg.Replicas.Term = term
 	data, _ := json.Marshal(seg)
@@ -396,20 +398,22 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 	return nil
 }
 
-func (mgr *eventlogManager) GetSegmentByBlockID(block *metadata.Block) (*Segment, error) {
+func (mgr *eventlogManager) GetSegmentByBlockID(block *metadata.Block) (Segment, error) {
 	v, exist := mgr.eventLogMap.Load(block.EventlogID.Key())
 	if !exist {
-		return nil, errors.ErrEventLogNotFound
+		return Segment{}, errors.ErrEventLogNotFound
 	}
 	el, _ := v.(*eventlog)
-	return el.get(block.SegmentID), nil
+	el.rLock()
+	defer el.rUnlock()
+	return el.get(block.SegmentID).Copy(), nil
 }
 
 func (mgr *eventlogManager) stop() {
 	mgr.cancel()
 }
 
-func (mgr *eventlogManager) getSegmentTopology(_ context.Context, seg *Segment) map[uint64]string {
+func (mgr *eventlogManager) getSegmentTopology(_ context.Context, seg Segment) map[uint64]string {
 	var addrs = map[uint64]string{}
 	for _, v := range seg.Replicas.Peers {
 		ins := mgr.volMgr.GetVolumeInstanceByID(v.VolumeID)
@@ -742,7 +746,7 @@ func (mgr *eventlogManager) createSegment(ctx context.Context, el *eventlog) (*S
 	_, err = srv.GetClient().ActivateSegment(ctx, &segment.ActivateSegmentRequest{
 		EventLogId:     seg.EventLogID.Uint64(),
 		ReplicaGroupId: seg.Replicas.ID.Uint64(),
-		Replicas:       mgr.getSegmentTopology(ctx, seg),
+		Replicas:       mgr.getSegmentTopology(ctx, *seg),
 	})
 
 	if err != nil {
@@ -945,12 +949,8 @@ func (el *eventlog) appendableSegmentNumber() int {
 	if cur == nil {
 		return 0
 	}
-	count := 0
-	for cur != nil {
-		count++
-		cur = el.nextOf(cur)
-	}
-	return count
+
+	return len(el.listOfRight(cur, true))
 }
 
 func (el *eventlog) currentAppendableSegment() *Segment {
@@ -959,6 +959,7 @@ func (el *eventlog) currentAppendableSegment() *Segment {
 	if el.size() == 0 {
 		return nil
 	}
+
 	if el.writePtr == nil {
 		head := el.segmentList.Front()
 		for head != nil {
@@ -989,6 +990,7 @@ func (el *eventlog) currentAppendableSegment() *Segment {
 func (el *eventlog) add(ctx context.Context, seg *Segment) error {
 	el.mutex.Lock()
 	defer el.mutex.Unlock()
+
 	if !seg.isReady() {
 		return errors.ErrInvalidSegment
 	}
@@ -1091,37 +1093,69 @@ func (el *eventlog) size() int {
 	return el.segmentList.Len()
 }
 
-func (el *eventlog) nextOf(seg *Segment) *Segment {
-	if seg == nil {
-		return nil
-	}
+func (el *eventlog) getAllSegments() []*Segment {
 	el.mutex.RLock()
 	defer el.mutex.RUnlock()
 
-	node := el.segmentList.Get(seg.ID.Uint64())
-	if node == nil {
-		return nil
+	segs := make([]*Segment, 0)
+
+	if el.size() == 0 {
+		return segs
 	}
-	next := node.Next()
-	if next == nil {
-		return nil
+	ptr := el.segmentList.Front()
+	for ptr != nil {
+		next, _ := ptr.Value.(*Segment)
+		segs = append(segs, next)
+		ptr = ptr.Next()
 	}
-	return next.Value.(*Segment)
+	return segs
 }
 
-func (el *eventlog) previousOf(seg *Segment) *Segment {
+func (el *eventlog) listOfRight(seg *Segment, includeSelf bool) []Segment {
+	el.mutex.RLock()
+	defer el.mutex.RUnlock()
+	segs := make([]Segment, 0)
 	if seg == nil {
-		return nil
+		return segs
+	}
+	if includeSelf {
+		segs = append(segs, *seg)
 	}
+	node := el.segmentList.Get(seg.ID.Uint64())
+	if node == nil {
+		return segs
+	}
+	v := node.Next()
+	for v != nil {
+		next, _ := v.Value.(*Segment)
+		segs = append(segs, *next)
+		v = v.Next()
+	}
+
+	return segs
+}
+
+func (el *eventlog) listOfPrevious(seg *Segment) []*Segment { //nolint:unused // ok
 	el.mutex.RLock()
 	defer el.mutex.RUnlock()
-
+	segs := make([]*Segment, 0)
 	node := el.segmentList.Get(seg.ID.Uint64())
-	prev := node.Prev()
-	if prev == nil {
-		return nil
+	if node == nil {
+		return segs
+	}
+	v := node.Prev()
+	for v != nil {
+		next, _ := v.Value.(*Segment)
+		segs = append(segs, next)
+		v = v.Prev()
+	}
+
+	// reverse slice
+	for i, j := 0, len(segs)-1; i < j; i, j = i+1, j-1 {
+		segs[i], segs[j] = segs[j], segs[i]
 	}
-	return prev.Value.(*Segment)
+
+	return segs
 }
 
 func (el *eventlog) deleteHead(ctx context.Context) error {
@@ -1192,3 +1226,11 @@ func (el *eventlog) lock() {
 func (el *eventlog) unlock() {
 	el.mutex.Unlock()
 }
+
+func (el *eventlog) rLock() {
+	el.mutex.RLock()
+}
+
+func (el *eventlog) rUnlock() {
+	el.mutex.RUnlock()
+}
diff --git a/internal/controller/eventbus/eventlog/mock_eventlog.go b/internal/controller/eventbus/eventlog/mock_eventlog.go
index bcac927..aeae26f 100644
--- a/internal/controller/eventbus/eventlog/mock_eventlog.go
+++ b/internal/controller/eventbus/eventlog/mock_eventlog.go
@@ -65,10 +65,10 @@ func (mr *MockManagerMockRecorder) DeleteEventlog(ctx, id interface{}) *gomock.C
 }
 
 // GetAppendableSegment mocks base method.
-func (m *MockManager) GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]*Segment, error) {
+func (m *MockManager) GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]Segment, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetAppendableSegment", ctx, eli, num)
-	ret0, _ := ret[0].([]*Segment)
+	ret0, _ := ret[0].([]Segment)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
@@ -108,10 +108,10 @@ func (mr *MockManagerMockRecorder) GetEventLog(ctx, id interface{}) *gomock.Call
 }
 
 // GetEventLogSegmentList mocks base method.
-func (m *MockManager) GetEventLogSegmentList(elID vanus.ID) []*Segment {
+func (m *MockManager) GetEventLogSegmentList(elID vanus.ID) []Segment {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetEventLogSegmentList", elID)
-	ret0, _ := ret[0].([]*Segment)
+	ret0, _ := ret[0].([]Segment)
 	return ret0
 }
 
@@ -121,25 +121,11 @@ func (mr *MockManagerMockRecorder) GetEventLogSegmentList(elID interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEventLogSegmentList", reflect.TypeOf((*MockManager)(nil).GetEventLogSegmentList), elID)
 }
 
-// GetSegment mocks base method.
-func (m *MockManager) GetSegment(id vanus.ID) *Segment {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetSegment", id)
-	ret0, _ := ret[0].(*Segment)
-	return ret0
-}
-
-// GetSegment indicates an expected call of GetSegment.
-func (mr *MockManagerMockRecorder) GetSegment(id interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetSegment", reflect.TypeOf((*MockManager)(nil).GetSegment), id)
-}
-
 // GetSegmentByBlockID mocks base method.
-func (m *MockManager) GetSegmentByBlockID(block *metadata.Block) (*Segment, error) {
+func (m *MockManager) GetSegmentByBlockID(block *metadata.Block) (Segment, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetSegmentByBlockID", block)
-	ret0, _ := ret[0].(*Segment)
+	ret0, _ := ret[0].(Segment)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
diff --git a/internal/controller/eventbus/eventlog/segment.go b/internal/controller/eventbus/eventlog/segment.go
index c02c4fb..112fa20 100644
--- a/internal/controller/eventbus/eventlog/segment.go
+++ b/internal/controller/eventbus/eventlog/segment.go
@@ -143,7 +143,7 @@ type ReplicaGroup struct {
 	DestroyAt time.Time                  `json:"destroy_at"`
 }
 
-func Convert2ProtoSegment(ctx context.Context, ins ...*Segment) []*metapb.Segment {
+func Convert2ProtoSegment(ctx context.Context, ins ...Segment) []*metapb.Segment {
 	segs := make([]*metapb.Segment, len(ins))
 	for idx := 0; idx < len(ins); idx++ {
 		seg := ins[idx]
diff --git a/internal/primitive/transform/action/condition/condition_if_action.go b/internal/primitive/transform/action/condition/condition_if_action.go
index 49801c1..c49b7f2 100644
--- a/internal/primitive/transform/action/condition/condition_if_action.go
+++ b/internal/primitive/transform/action/condition/condition_if_action.go
@@ -57,7 +57,7 @@ func (a *conditionIfAction) Execute(ceCtx *context.EventContext) error {
 	} else {
 		switch op {
 		case ">=", ">", "<=", "<":
-			a.ArgTypes = []common.Type{common.Number, common.String, common.Number, common.Any, common.Any}
+			a.ArgTypes = []common.Type{common.Float, common.String, common.Float, common.Any, common.Any}
 		default:
 			return errors.Errorf("not support op [%s]", op)
 		}
diff --git a/internal/primitive/transform/action/strings/split_with_delimiter.go b/internal/primitive/transform/action/strings/split_with_delimiter.go
new file mode 100644
index 0000000..ac05f4d
--- /dev/null
+++ b/internal/primitive/transform/action/strings/split_with_delimiter.go
@@ -0,0 +1,44 @@
+// Copyright 2023 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package strings
+
+import (
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/common"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/function"
+)
+
+type splitWithDelimiterAction struct {
+	action.FunctionAction
+}
+
+// NewSplitWithDelimiterAction ["split_with_delimiter","sourceJsonPath", "delimiter", "targetJsonPath"].
+func NewSplitWithDelimiterAction() action.Action {
+	a := &splitWithDelimiterAction{}
+	a.CommonAction = action.CommonAction{
+		ActionName: "SPLIT_WITH_DELIMITER",
+		FixedArgs:  []arg.TypeList{arg.EventList, []arg.Type{arg.Constant}, []arg.Type{arg.EventData}},
+		Fn:         function.SplitWithSepFunction,
+	}
+	return a
+}
+
+func (a *splitWithDelimiterAction) Init(args []arg.Arg) error {
+	a.TargetArg = args[2]
+	a.Args = args[:2]
+	a.ArgTypes = []common.Type{common.String, common.String}
+	return nil
+}
diff --git a/internal/primitive/transform/common/cast.go b/internal/primitive/transform/common/cast.go
index 7218e83..3f562d6 100644
--- a/internal/primitive/transform/common/cast.go
+++ b/internal/primitive/transform/common/cast.go
@@ -29,7 +29,7 @@ func Cast(val interface{}, target Type) (interface{}, error) {
 		switch value := val.(type) {
 		case int32: // ce attribute
 			return strconv.FormatInt(int64(value), 10), nil
-		case int64: // ce attribute
+		case int64:
 			return strconv.FormatInt(value, 10), nil
 		case float64: // ce data json marshal
 			return fmt.Sprintf("%v", val), nil
@@ -38,7 +38,7 @@ func Cast(val interface{}, target Type) (interface{}, error) {
 		}
 		// Casting to string is always defined
 		return fmt.Sprintf("%v", val), nil
-	case Number:
+	case Float:
 		switch value := val.(type) {
 		case string:
 			v, err := strconv.ParseFloat(value, 64)
@@ -54,6 +54,22 @@ func Cast(val interface{}, target Type) (interface{}, error) {
 			return float64(value), nil
 		}
 		return 0, fmt.Errorf("undefined cast from %v to %v", TypeFromVal(val), target)
+	case Int:
+		switch value := val.(type) {
+		case string:
+			v, err := strconv.ParseInt(value, 10, 64)
+			if err != nil {
+				err = fmt.Errorf("cannot cast from String to Int: %w", err)
+			}
+			return int(v), err
+		case float64:
+			return int(value), nil
+		case int32:
+			return int(value), nil
+		case int64:
+			return int(value), nil
+		}
+		return 0, fmt.Errorf("undefined cast from %v to %v", TypeFromVal(val), target)
 	case Bool:
 		if value, ok := val.(string); ok {
 			lowerCase := strings.ToLower(value)
diff --git a/internal/primitive/transform/common/types.go b/internal/primitive/transform/common/types.go
index 8698cef..4599963 100644
--- a/internal/primitive/transform/common/types.go
+++ b/internal/primitive/transform/common/types.go
@@ -18,7 +18,8 @@ type Type uint8
 
 const (
 	String Type = iota
-	Number
+	Float
+	Int
 	Bool
 	Object
 	Array
@@ -30,8 +31,10 @@ func (t Type) String() string {
 	switch t {
 	case String:
 		return "string"
-	case Number:
-		return "number"
+	case Float:
+		return "float"
+	case Int:
+		return "int"
 	case Bool:
 		return "bool"
 	case Object:
@@ -59,7 +62,9 @@ func TypeFromVal(val interface{}) Type {
 	case string:
 		return String
 	case float64:
-		return Number
+		return Float
+	case int:
+		return Int
 	case bool:
 		return Bool
 	case map[string]interface{}:
diff --git a/internal/primitive/transform/function/datatime_functions.go b/internal/primitive/transform/function/datatime_functions.go
index 908befb..157a852 100644
--- a/internal/primitive/transform/function/datatime_functions.go
+++ b/internal/primitive/transform/function/datatime_functions.go
@@ -44,10 +44,10 @@ var DateFormatFunction = function{
 
 var UnixTimeFormatFunction = function{
 	name:         "UNIX_TIME_FORMAT",
-	fixedArgs:    []common.Type{common.Number, common.String},
+	fixedArgs:    []common.Type{common.Int, common.String},
 	variadicArgs: common.TypePtr(common.String),
 	fn: func(args []interface{}) (interface{}, error) {
-		t := time.Unix(int64(args[0].(float64)), 0)
+		t := time.Unix(int64(args[0].(int)), 0)
 		loc := time.UTC
 		if len(args) > 2 && args[2].(string) != "" {
 			var err error
diff --git a/internal/primitive/transform/function/math_functions.go b/internal/primitive/transform/function/math_functions.go
index e18dd46..ea0e306 100644
--- a/internal/primitive/transform/function/math_functions.go
+++ b/internal/primitive/transform/function/math_functions.go
@@ -22,8 +22,8 @@ import (
 
 var MathAddFunction = function{
 	name:         "MATH_ADD",
-	fixedArgs:    []common.Type{common.Number, common.Number},
-	variadicArgs: common.TypePtr(common.Number),
+	fixedArgs:    []common.Type{common.Float, common.Float},
+	variadicArgs: common.TypePtr(common.Float),
 	fn: func(args []interface{}) (interface{}, error) {
 		var sum float64
 		for i := 0; i < len(args); i++ {
@@ -36,7 +36,7 @@ var MathAddFunction = function{
 
 var MathSubFunction = function{
 	name:      "MATH_SUB",
-	fixedArgs: []common.Type{common.Number, common.Number},
+	fixedArgs: []common.Type{common.Float, common.Float},
 	fn: func(args []interface{}) (interface{}, error) {
 		return args[0].(float64) - args[1].(float64), nil
 	},
@@ -44,8 +44,8 @@ var MathSubFunction = function{
 
 var MathMulFunction = function{
 	name:         "MATH_MUL",
-	fixedArgs:    []common.Type{common.Number, common.Number},
-	variadicArgs: common.TypePtr(common.Number),
+	fixedArgs:    []common.Type{common.Float, common.Float},
+	variadicArgs: common.TypePtr(common.Float),
 	fn: func(args []interface{}) (interface{}, error) {
 		sum := float64(1)
 		for i := 0; i < len(args); i++ {
@@ -58,7 +58,7 @@ var MathMulFunction = function{
 
 var MathDivFunction = function{
 	name:      "MATH_DIV",
-	fixedArgs: []common.Type{common.Number, common.Number},
+	fixedArgs: []common.Type{common.Float, common.Float},
 	fn: func(args []interface{}) (interface{}, error) {
 		if args[1].(float64) == 0 {
 			return nil, fmt.Errorf("division by zero")
diff --git a/internal/primitive/transform/function/strings_functions.go b/internal/primitive/transform/function/strings_functions.go
index e087bf9..2846d27 100644
--- a/internal/primitive/transform/function/strings_functions.go
+++ b/internal/primitive/transform/function/strings_functions.go
@@ -73,24 +73,24 @@ var AddSuffixFunction = function{
 var SplitWithSepFunction = function{
 	name:         "SPLIT_WITH_SEP",
 	fixedArgs:    []common.Type{common.String, common.String},
-	variadicArgs: common.TypePtr(common.Number),
+	variadicArgs: common.TypePtr(common.Int),
 	fn: func(args []interface{}) (interface{}, error) {
 		s, _ := args[0].(string)
 		sep, _ := args[1].(string)
 		if len(args) == 2 {
 			return strings.Split(s, sep), nil
 		}
-		return strings.SplitN(s, sep, int(args[2].(float64))), nil
+		return strings.SplitN(s, sep, args[2].(int)), nil
 	},
 }
 
 var ReplaceBetweenPositionsFunction = function{
 	name:      "REPLACE_BETWEEN_POSITIONS",
-	fixedArgs: []common.Type{common.String, common.Number, common.Number, common.String},
+	fixedArgs: []common.Type{common.String, common.Int, common.Int, common.String},
 	fn: func(args []interface{}) (interface{}, error) {
 		path, _ := args[0].(string)
-		startPosition := int(args[1].(float64))
-		endPosition := int(args[2].(float64))
+		startPosition, _ := args[1].(int)
+		endPosition, _ := args[2].(int)
 		targetValue, _ := args[3].(string)
 		if startPosition >= len(path) {
 			return nil, fmt.Errorf("start position must be less than the length of the string")
@@ -125,21 +125,18 @@ var CapitalizeWord = function{
 	fixedArgs: []common.Type{common.String},
 	fn: func(args []interface{}) (interface{}, error) {
 		value, _ := args[0].(string)
-		if len(value) == 0 {
-			return value, nil
-		}
-		if len(value) == 1 {
-			return strings.ToUpper(string(value[0])), nil
-		}
-		capWords := make([]rune, 0, len([]rune(value)))
-		prev := ' '
-		for _, v := range value {
-			if v != ' ' && prev == ' ' {
-				v = unicode.ToUpper(v)
+		rs := []rune(value)
+		inWord := false
+		for i, r := range rs {
+			if !unicode.IsSpace(r) {
+				if !inWord {
+					rs[i] = unicode.ToTitle(r)
+				}
+				inWord = true
+			} else {
+				inWord = false
 			}
-			capWords = append(capWords, v)
-			prev = v
 		}
-		return string(capWords), nil
+		return string(rs), nil
 	},
 }
diff --git a/internal/primitive/transform/runtime/init.go b/internal/primitive/transform/runtime/init.go
index ec19d0b..390cd85 100644
--- a/internal/primitive/transform/runtime/init.go
+++ b/internal/primitive/transform/runtime/init.go
@@ -54,6 +54,7 @@ func init() {
 		strings.NewCapitalizeSentenceAction,
 		strings.NewCheckCustomValuesAction,
 		strings.NewCapitalizeWordAction,
+		strings.NewSplitWithDelimiterAction,
 		// condition
 		condition.NewConditionIfAction,
 		// array


Test patch:
diff --git a/internal/controller/eventbus/eventlog/eventlog_test.go b/internal/controller/eventbus/eventlog/eventlog_test.go
index 57fe3b1..3d740d0 100644
--- a/internal/controller/eventbus/eventlog/eventlog_test.go
+++ b/internal/controller/eventbus/eventlog/eventlog_test.go
@@ -511,14 +511,14 @@ func TestEventlogManager_CreateAndGetEventlog(t *testing.T) {
 			So(blockObj.VolumeID, ShouldEqual, vol1.ID)
 			So(blockObj.SegmentID, ShouldEqual, segments[0].ID)
 
-			seg := utMgr.GetSegment(segments2[1].ID)
+			seg := utMgr.getSegment(segments2[1].ID)
 			So(seg, ShouldNotBeNil)
 			So(seg.EventLogID, ShouldEqual, logMD.ID)
 
 			blockObj = utMgr.GetBlock(vanus.NewIDFromUint64(segments[1].Replicas.Leader))
 			segAnother, err := utMgr.GetSegmentByBlockID(blockObj)
 			So(err, ShouldBeNil)
-			So(segAnother, ShouldEqual, seg)
+			So(segAnother, ShouldResemble, *seg)
 		})
 	})
 }
@@ -1064,15 +1064,12 @@ func TestEventlog_All(t *testing.T) {
 			So(el.indexAt(4), ShouldBeNil)
 			So(el.indexAt(999), ShouldBeNil)
 
-			So(el.nextOf(el.indexAt(0)), ShouldEqual, seg2)
-			So(el.nextOf(el.indexAt(1)), ShouldEqual, seg3)
-			So(el.nextOf(el.indexAt(2)), ShouldEqual, seg4)
-			So(el.nextOf(el.indexAt(3)), ShouldBeNil)
+			list := el.getAllSegments()
 
-			So(el.previousOf(el.indexAt(0)), ShouldBeNil)
-			So(el.previousOf(el.indexAt(1)), ShouldEqual, seg1)
-			So(el.previousOf(el.indexAt(2)), ShouldEqual, seg2)
-			So(el.previousOf(el.indexAt(3)), ShouldEqual, seg3)
+			So(list[0], ShouldEqual, seg1)
+			So(list[1], ShouldEqual, seg2)
+			So(list[2], ShouldEqual, seg3)
+			So(list[3], ShouldEqual, seg4)
 		})
 	})
 }
diff --git a/internal/controller/eventbus/eventlog/segment_test.go b/internal/controller/eventbus/eventlog/segment_test.go
index e7a9594..0bada72 100644
--- a/internal/controller/eventbus/eventlog/segment_test.go
+++ b/internal/controller/eventbus/eventlog/segment_test.go
@@ -38,7 +38,7 @@ func TestConvert2ProtoSegment(t *testing.T) {
 		block1 := vanus.NewTestID()
 		block2 := vanus.NewTestID()
 		block3 := vanus.NewTestID()
-		seg := &Segment{
+		seg := Segment{
 			ID:                segID,
 			Capacity:          64 * 1024 * 1024,
 			EventLogID:        eID,
diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go
index 5d464be..c0de7ca 100644
--- a/internal/gateway/gateway_test.go
+++ b/internal/gateway/gateway_test.go
@@ -166,7 +166,6 @@ func TestGateway_EventID(t *testing.T) {
 		ControllerAddr: controllers,
 	}
 	ga := NewGateway(cfg)
-	defer ga.Stop()
 
 	ga.client = mockClient
 	ctx, cancel := context.WithCancel(context.Background())
diff --git a/internal/primitive/transform/action/strings/capitalize_word_test.go b/internal/primitive/transform/action/strings/capitalize_word_test.go
index 6c4ae36..8108e4d 100644
--- a/internal/primitive/transform/action/strings/capitalize_word_test.go
+++ b/internal/primitive/transform/action/strings/capitalize_word_test.go
@@ -86,4 +86,16 @@ func TestCapitalizeWordAction(t *testing.T) {
 		So(err, ShouldBeNil)
 		So(e.Extensions()["test"], ShouldEqual, "A")
 	})
+	Convey("test capitalize word: leading symbols", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", "let 'em go")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, "Let 'em Go")
+	})
 }
diff --git a/internal/primitive/transform/action/strings/split_with_delimiter_test.go b/internal/primitive/transform/action/strings/split_with_delimiter_test.go
new file mode 100644
index 0000000..19af4f2
--- /dev/null
+++ b/internal/primitive/transform/action/strings/split_with_delimiter_test.go
@@ -0,0 +1,45 @@
+// Copyright 2023 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package strings_test
+
+import (
+	"testing"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/strings"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/runtime"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestSplitWithDelimiterAction(t *testing.T) {
+	funcName := strings.NewSplitWithDelimiterAction().Name()
+	Convey("test split with delimiter", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test", ",", "$.data.target"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		data := map[string]interface{}{}
+		e.SetExtension("test", "one,two,three")
+		ceCtx := &context.EventContext{
+			Event: &e,
+			Data:  data,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		res, ok := data["target"]
+		So(ok, ShouldBeTrue)
+		So(res, ShouldResemble, []string{"one", "two", "three"})
+	})
+}
diff --git a/internal/primitive/transform/runtime/action_bench_test.go b/internal/primitive/transform/runtime/action_bench_test.go
index b54f5f4..0ba903b 100644
--- a/internal/primitive/transform/runtime/action_bench_test.go
+++ b/internal/primitive/transform/runtime/action_bench_test.go
@@ -79,4 +79,5 @@ func BenchmarkAction(b *testing.B) {
 	b.Run("replace_with_regex", actionBenchmark([]interface{}{"replace_with_regex", "$.data.str", "a", "Aa"}))
 	b.Run("capitalize_sentence", actionBenchmark([]interface{}{"capitalize_sentence", "$.data.str"}))
 	b.Run("check_custom_values", actionBenchmark([]interface{}{"check_custom_values", "$.data.str", "value", "$.data.target", "true", "false"}))
+	b.Run("split_with_delimiter", actionBenchmark([]interface{}{"split_with_delimiter", "$.data.str", "a", "$.data.target"}))
 }
diff --git a/test/regression/pubsub/main.go b/test/regression/pubsub/main.go
index 3995149..0246e0e 100644
--- a/test/regression/pubsub/main.go
+++ b/test/regression/pubsub/main.go
@@ -32,7 +32,7 @@ var (
 	sentFailed     int64
 	receivedNumber int64
 
-	totalNumber        = int64(1 << 15) // 1024*1024
+	totalNumber        = int64(1 << 20) // 1024*1024
 	parallelism        = 4
 	batchSize          = 8
 	maximumPayloadSize = 4 * 1024