Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

add skipUntil config to skip events until that time #104

Merged
merged 2 commits into from
Aug 16, 2024
Merged

add skipUntil config to skip events until that time #104

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1c47f1c
add skipUntil config to skip events until that day
ErenArslan Aug 14, 2024
12d55be
handle skipUntil config nil
ErenArslan Aug 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion config/dcp.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,8 @@ type DCPGroup struct {
}

type DCPListener struct {
BufferSize uint `yaml:"bufferSize"`
BufferSize uint `yaml:"bufferSize"`
SkipUntil *time.Time `yaml:"skipUntil"`
}

type ExternalDcpConfig struct {
Expand Down
31 changes: 28 additions & 3 deletions couchbase/observer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,13 @@ func (so *observer) canForward(vbID uint16, seqNo uint64) bool {
return !so.needCatchup(vbID, seqNo)
}

func (so *observer) isBeforeSkipWindow(eventTime time.Time) bool {
if so.config.Dcp.Listener.SkipUntil == nil {
return false
}
return so.config.Dcp.Listener.SkipUntil.Before(eventTime)
}

func (so *observer) convertToCollectionName(collectionID uint32) string {
if name, ok := so.collectionIDs[collectionID]; ok {
return name
Expand Down Expand Up @@ -173,6 +180,12 @@ func (so *observer) Mutation(mutation gocbcore.DcpMutation) { //nolint:dupl
return
}

eventTime := time.Unix(int64(mutation.Cas/1000000000), 0)

if so.isBeforeSkipWindow(eventTime) {
return
}

if currentSnapshot, ok := so.currentSnapshots.Load(mutation.VbID); ok && currentSnapshot != nil {
vbUUID, _ := so.uuIDMap.Load(mutation.VbID)

Expand All @@ -185,7 +198,7 @@ func (so *observer) Mutation(mutation gocbcore.DcpMutation) { //nolint:dupl
SeqNo: mutation.SeqNo,
},
CollectionName: so.convertToCollectionName(mutation.CollectionID),
EventTime: time.Unix(int64(mutation.Cas/1000000000), 0),
EventTime: eventTime,
},
})
}
Expand All @@ -204,6 +217,12 @@ func (so *observer) Deletion(deletion gocbcore.DcpDeletion) { //nolint:dupl
return
}

eventTime := time.Unix(int64(deletion.Cas/1000000000), 0)

if so.isBeforeSkipWindow(eventTime) {
return
}

if currentSnapshot, ok := so.currentSnapshots.Load(deletion.VbID); ok && currentSnapshot != nil {
vbUUID, _ := so.uuIDMap.Load(deletion.VbID)

Expand All @@ -216,7 +235,7 @@ func (so *observer) Deletion(deletion gocbcore.DcpDeletion) { //nolint:dupl
SeqNo: deletion.SeqNo,
},
CollectionName: so.convertToCollectionName(deletion.CollectionID),
EventTime: time.Unix(int64(deletion.Cas/1000000000), 0),
EventTime: eventTime,
},
})
}
Expand All @@ -235,6 +254,12 @@ func (so *observer) Expiration(expiration gocbcore.DcpExpiration) { //nolint:dup
return
}

eventTime := time.Unix(int64(expiration.Cas/1000000000), 0)

if so.isBeforeSkipWindow(eventTime) {
return
}

if currentSnapshot, ok := so.currentSnapshots.Load(expiration.VbID); ok && currentSnapshot != nil {
vbUUID, _ := so.uuIDMap.Load(expiration.VbID)

Expand All @@ -247,7 +272,7 @@ func (so *observer) Expiration(expiration gocbcore.DcpExpiration) { //nolint:dup
SeqNo: expiration.SeqNo,
},
CollectionName: so.convertToCollectionName(expiration.CollectionID),
EventTime: time.Unix(int64(expiration.Cas/1000000000), 0),
EventTime: eventTime,
},
})
}
Expand Down
Loading