proc

Documentation

Overview

Package proc provides actual actions to be performed once the sync has filtered all events

Index

• func MergeNodes(left *TreeNode, right *TreeNode, diff *SourceDiff)
• type ChildrenCursor
  • func (c *ChildrenCursor) Next() *TreeNode
• type Merger
  • func NewMerger(ctx context.Context) *Merger
  • func (b *Merger) AddRequeueChannel(source PathSyncSource, channel chan EventInfo)
  • func (b *Merger) Logger() *zap.Logger
  • func (b *Merger) ProcessBatches()
  • func (b *Merger) RegisterEventChannel(out chan ProcessorEvent)
  • func (b *Merger) Shutdown()
• type ProcessFunc
• type SourceDiff
  • func ComputeSourcesDiff(ctx context.Context, left common.PathSyncSource, right common.PathSyncSource, ...) (diff *SourceDiff, err error)
  • func (diff *SourceDiff) FilterMissing(source common.PathSyncSource, target common.PathSyncTarget, in []*tree.Node, ...) (out map[string]*filters.BatchedEvent)
  • func (diff *SourceDiff) Stats() map[string]int
  • func (diff *SourceDiff) String() string
  • func (diff *SourceDiff) ToBidirectionalBatch(leftTarget common.PathSyncTarget, rightTarget common.PathSyncTarget) (batch *filters.BidirectionalBatch, err error)
  • func (diff *SourceDiff) ToUnidirectionalBatch(direction string) (batch *filters.Batch, err error)
• type TreeNode
  • func NewTreeNode(n *tree.Node) *TreeNode
  • func TreeNodeFromSource(source common.PathSyncSource) (*TreeNode, error)
  • func (t *TreeNode) AddChild(n *TreeNode)
  • func (t *TreeNode) Enqueue(missings []*tree.Node) []*tree.Node
  • func (t *TreeNode) GetCursor() *ChildrenCursor
  • func (t *TreeNode) GetHash() string
  • func (t *TreeNode) GetLevel() int
  • func (t *TreeNode) Label() string
  • func (t *TreeNode) ParentPath() string
  • func (t *TreeNode) PrintOut()
  • func (t *TreeNode) SortedChildren() []*TreeNode

Functions

func MergeNodes

func MergeNodes(left *TreeNode, right *TreeNode, diff *SourceDiff)

MergeNodes will recursively detect differences between two hash trees.

Types

type ChildrenCursor

type ChildrenCursor struct {
	// contains filtered or unexported fields
}

ChildrenCursor provides a Nexter for browsing a node children

func (*ChildrenCursor) Next

func (c *ChildrenCursor) Next() *TreeNode

Next sends the next child or nil

type Merger

type Merger struct {
	BatchesChannel  chan *filters.Batch
	RequeueChannels map[PathSyncSource]chan EventInfo
	LocksChannel    chan filters.LockEvent
	UnlocksChannel  chan filters.UnlockEvent
	JobsInterrupt   chan bool
	GlobalContext   context.Context
	// contains filtered or unexported fields
}

func NewMerger

func NewMerger(ctx context.Context) *Merger

func (*Merger) AddRequeueChannel

func (b *Merger) AddRequeueChannel(source PathSyncSource, channel chan EventInfo)

func (*Merger) Logger

func (b *Merger) Logger() *zap.Logger

func (*Merger) ProcessBatches

func (b *Merger) ProcessBatches()

func (*Merger) RegisterEventChannel

func (b *Merger) RegisterEventChannel(out chan ProcessorEvent)

func (*Merger) Shutdown

func (b *Merger) Shutdown()

type ProcessFunc

type ProcessFunc func(event *filters.BatchedEvent, operationId string) error

type SourceDiff

type SourceDiff struct {
	Left         common.PathSyncSource
	Right        common.PathSyncSource
	MissingLeft  []*tree.Node
	MissingRight []*tree.Node
	Context      context.Context
}

SourceDiff represents basic differences between two sources. It can be then transformed to Batch, depending on the sync being unidirectional (transform to Creates and Deletes) or bidirectional (transform only to Creates)

func ComputeSourcesDiff

func ComputeSourcesDiff(ctx context.Context, left common.PathSyncSource, right common.PathSyncSource, strong bool, statusChan chan filters.BatchProcessStatus) (diff *SourceDiff, err error)

ComputeSourceDiff loads the diff by crawling the sources in parallel, filling up a Hash Tree and performing the merge.

func (*SourceDiff) FilterMissing

func (diff *SourceDiff) FilterMissing(source common.PathSyncSource, target common.PathSyncTarget, in []*tree.Node, folders bool, removes bool) (out map[string]*filters.BatchedEvent)

FilterMissing transforms Missing slices to BatchEvents

func (*SourceDiff) Stats

func (diff *SourceDiff) Stats() map[string]int

func (*SourceDiff) String

func (diff *SourceDiff) String() string

String provides a string representation of this diff

func (*SourceDiff) ToBidirectionalBatch

func (diff *SourceDiff) ToBidirectionalBatch(leftTarget common.PathSyncTarget, rightTarget common.PathSyncTarget) (batch *filters.BidirectionalBatch, err error)

ToBidirectionalBatch transforms this diff to a batch

func (*SourceDiff) ToUnidirectionalBatch

func (diff *SourceDiff) ToUnidirectionalBatch(direction string) (batch *filters.Batch, err error)

ToUnidirectionalBatch transforms this diff to a batch

type TreeNode

type TreeNode struct {
	tree.Node
	sync.Mutex
	// contains filtered or unexported fields
}

TreeNode builds a Merkle Tree but with N children and the ability to compute the COLLECTION Nodes hashes to detect changes in branches more rapidly

func NewTreeNode

func NewTreeNode(n *tree.Node) *TreeNode

NewTreeNode creates a new node from a tree.Node. Can be a root, a COLL or a LEAF.

func TreeNodeFromSource

func TreeNodeFromSource(source common.PathSyncSource) (*TreeNode, error)

TreeNodeFromSource populates a hash tree with leafs and folders by walking a source. When it comes accross a LEAF without Etag value, it asks the source to recompute it in a parallel fashion with throttling (max 15 at the same time). At the end of the operation, the tree should be fully loaded with all LEAF etags (but not COLL etags).

func (*TreeNode) AddChild

func (t *TreeNode) AddChild(n *TreeNode)

AddChild appends a child to the children map (with lock)

func (*TreeNode) Enqueue

func (t *TreeNode) Enqueue(missings []*tree.Node) []*tree.Node

Enqueue recursively appends al tree.Node and the children's tree.Node to a slice

func (*TreeNode) GetCursor

func (t *TreeNode) GetCursor() *ChildrenCursor

GetCursor gives a cursor to crawl the current node children

func (*TreeNode) GetHash

func (t *TreeNode) GetHash() string

GetHash returns the Etag of the node. For leaf it should be available, for Folders if it is not already computed, it will compute an etag from the children recursively, using their name and Etag.

func (*TreeNode) GetLevel

func (t *TreeNode) GetLevel() int

GetLevel computes the current level of this node (depth)

func (*TreeNode) Label

func (t *TreeNode) Label() string

Label returns the basename of the path

func (*TreeNode) ParentPath

func (t *TreeNode) ParentPath() string

ParentPath returns the parent Dir path

func (*TreeNode) PrintOut

func (t *TreeNode) PrintOut()

PrintOut sends to fmt.Println a tree version of this node

func (*TreeNode) SortedChildren

func (t *TreeNode) SortedChildren() []*TreeNode

SortedChildren sorts children by their labels. An internal flag avoids resorting if it was already sorted once.