This repository contains go implementation of a Couchbase Database Change Protocol (DCP) client. You can find more information by looking at our docs.
package main
import (
"github.com/Trendyol/go-dcp"
"github.com/Trendyol/go-dcp/logger"
"github.com/Trendyol/go-dcp/models"
)
func listener(ctx *models.ListenerContext) {
switch event := ctx.Event.(type) {
case models.DcpMutation:
logger.Log.Info(
"mutated(vb=%v,eventTime=%v) | id: %v, value: %v | isCreated: %v",
event.VbID, event.EventTime, string(event.Key), string(event.Value), event.IsCreated(),
)
case models.DcpDeletion:
logger.Log.Info(
"deleted(vb=%v,eventTime=%v) | id: %v",
event.VbID, event.EventTime, string(event.Key),
)
case models.DcpExpiration:
logger.Log.Info(
"expired(vb=%v,eventTime=%v) | id: %v",
event.VbID, event.EventTime, string(event.Key),
)
}
ctx.Ack()
}
func main() {
connector, err := dcp.NewDcp("config.yml", listener)
if err != nil {
panic(err)
}
defer connector.Close()
connector.Start()
}
$ go get github.com/Trendyol/go-dcp
| Variable | Type | Required | Default | Description |
|---|---|---|---|---|
hosts |
[]string | yes | - | Couchbase host like localhost:8091. |
username |
string | yes | - | Couchbase username. |
password |
string | yes | - | Couchbase password. |
bucketName |
string | yes | - | Couchbase DCP bucket. |
dcp.group.name |
string | yes | DCP group name for vbuckets. | |
scopeName |
string | no | _default | Couchbase scope name. |
collectionNames |
[]string | no | _default | Couchbase collection names. |
connectionBufferSize |
uint, string | no | 20mb | Source Bucket tcp connection buffer size (x Node Count). Check this if you get OOM Killed. |
maxQueueSize |
int | no | 2048 | The maximum number of requests that can be queued waiting to be sent to a node. Check this if you get queue overflowed or queue full. |
connectionTimeout |
time.Duration | no | 1m | Couchbase connection timeout. |
secureConnection |
bool | no | false | Enable TLS connection of Couchbase. |
rootCAPath |
string | no | *not set | if secureConnection set true this field is required. |
debug |
bool | no | false | For debugging purpose. |
dcp.bufferSize |
int | no | 16mb | DCP internal queue buffer size (x Node Count). Check this if you get OOM Killed. |
dcp.mode |
string | no | infinite | Set DCP mode finite If you want to listen to DCP events until now. Set DCP mode infinite If you want to listen to DCP events infinitely. |
dcp.connectionBufferSize |
uint, string | no | 20mb | DCP tcp connection buffer size (x Node Count). Check this if you get OOM Killed. |
dcp.connectionTimeout |
time.Duration | no | 1m | DCP connection timeout. |
dcp.maxQueueSize |
int | no | 2048 | The maximum number of requests that can be queued waiting to be sent to a node. Check this if you get queue overflowed or queue full. |
dcp.listener.skipUntil |
time.Time | no | Set this if you want to skip events until certain time. | |
dcp.group.membership.type |
string | no | DCP membership types. couchbase, kubernetesHa, kubernetesStatefulSet, static, dynamic or lilCouchbase. Check examples for details. |
|
dcp.group.membership.memberNumber |
int | no | 1 | Set this if membership is static. Other methods will ignore this field. |
dcp.group.membership.totalMembers |
int | no | 1 | Set this if membership is static or kubernetesStatefulSet. Other methods will ignore this field. |
dcp.group.membership.rebalanceDelay |
time.Duration | no | 30s | Works for autonomous mode. If membership is dynamic, it is ignored and set to 0s. |
dcp.group.membership.config |
map[string]string | no | *not set | Set key-values of config. expirySeconds,heartbeatInterval,heartbeatToleranceDuration,monitorInterval,timeout for couchbase type |
dcp.config.disableChangeStreams |
bool | no | false | Set this to true if you did not want to get older versions of changes for Couchbase Server 7.2.0+ using Magma storage buckets |
dcp.config.priority |
string | no | low | DCP stream priority requested from the Couchbase cluster. Allowed values: low, medium, high. Maps to gocbcore's DcpAgentPriority (set_priority DCP control). |
leaderElection.enabled |
bool | no | false | Set this true for memberships kubernetesHa. |
leaderElection.type |
string | no | kubernetes | Leader Election types. kubernetes |
leaderElection.config |
map[string]string | no | *not set | Set key-values of config. leaseLockName,leaseLockNamespace, leaseDuration, renewDeadline, retryPeriod for kubernetes type. |
leaderElection.rpc.port |
int | no | 8081 | This field is usable for kubernetesStatefulSet membership. |
checkpoint.type |
string | no | auto | Set checkpoint type auto or manual. |
checkpoint.autoReset |
string | no | earliest | Set checkpoint start point to earliest or latest. |
checkpoint.interval |
time.Duration | no | 1m | Checkpoint checking interval. |
checkpoint.timeout |
time.Duration | no | 1m | Checkpoint checking timeout. |
| `healthChec |
$ claude mcp add go-dcp \
-- python -m otcore.mcp_server <graph>