setMetaData and getMetaData may used internally by dcp clients. Expects send/receive of opaque []byte data. May be invoked from multiple goroutines, so need to manage synchronization. Setting mustPersist=true bypasses checkpoint threshold checks and forces persistence as long as persistCheckpoints
(vbucketId uint16, value []byte, mustPersist bool)
| 102 | // Setting mustPersist=true bypasses checkpoint threshold checks and forces persistence as long as |
| 103 | // persistCheckpoints=true |
| 104 | func (c *DCPCommon) setMetaData(vbucketId uint16, value []byte, mustPersist bool) error { |
| 105 | |
| 106 | c.m.Lock() |
| 107 | defer c.m.Unlock() |
| 108 | |
| 109 | c.meta[vbucketId] = value |
| 110 | |
| 111 | // Check persistMeta to avoids persistence if the only feed events we've seen are the DCP echo of DCP checkpoint docs |
| 112 | if c.persistCheckpoints && (mustPersist || c.updatesSinceCheckpoint[vbucketId] >= kCheckpointThreshold) { |
| 113 | |
| 114 | // Don't checkpoint more frequently than kCheckpointTimeThreshold |
| 115 | if !mustPersist && time.Since(c.lastCheckpointTime[vbucketId]) < kCheckpointTimeThreshold { |
| 116 | return nil |
| 117 | } |
| 118 | |
| 119 | err := c.persistCheckpoint(vbucketId, value) |
| 120 | if err != nil { |
| 121 | WarnfCtx(c.loggingCtx, "Unable to persist DCP metadata - will retry next snapshot. Error: %v", err) |
| 122 | return fmt.Errorf("Unable to persist DCP metadata") |
| 123 | } |
| 124 | c.updatesSinceCheckpoint[vbucketId] = 0 |
| 125 | c.lastCheckpointTime[vbucketId] = time.Now() |
| 126 | } |
| 127 | return nil |
| 128 | } |
| 129 | |
| 130 | func (c *DCPCommon) getMetaData(vbucketId uint16) ( |
| 131 | value []byte, lastSeq uint64, err error) { |