// WatchOperationPut watches PUT operations for DDL lock operation.
// If want to watch all operations matching, pass empty string for `task`, `source`, `upSchema` and `upTable`.
// This function can be called by DM-worker and DM-master.
WatchOperationPut watches PUT operations for DDL lock operation.
If want to watch all operations matching, pass empty string for `task`, `source`, `upSchema` and `upTable`.
This function can be called by DM-worker and DM-master.
// WatchOperationPut watches PUT operations for DDL lock operation . // If want to watch all operations matching pass empty string for ` task ` ` source ` ` upSchema ` and ` upTable ` . // This function can be called by DM - worker and DM - master .
func WatchOperationPut(ctx context.Context, cli *clientv3.Client,
	task, source, upSchema, upTable string, revision int64,
	outCh chan<- Operation, errCh chan<- error) {
	var ch clientv3.WatchChan
	wCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	if upTable == "" {
		ch = cli.Watch(wCtx, common.ShardDDLOptimismOperationKeyAdapter.Path(), clientv3.WithPrefix(),
			clientv3.WithRev(revision))
	} else {
		ch = cli.Watch(wCtx, common.ShardDDLOptimismOperationKeyAdapter.Encode(task, source, upSchema, upTable),
			clientv3.WithRev(revision))
	}
	for {
		select {
		case <-ctx.Done():
			return
		case resp, ok := <-ch:
			if !ok {
				return
			}
			if resp.Canceled {
				select {
				case errCh <- resp.Err():
				case <-ctx.Done():
				}
				return
			}
			for _, ev := range resp.Events {
				if ev.Type != mvccpb.PUT {
					continue
				}
				op, err := operationFromJSON(string(ev.Kv.Value))
				op.Revision = ev.Kv.ModRevision
				if err != nil {
					select {
					case errCh <- err:
					case <-ctx.Done():
						return
					}
				} else {
					select {
					case outCh <- op:
					case <-ctx.Done():
						return
					}
				}
			}
		}
	}
}
func WatchOperationPut ( ctx context . Context , cli * clientv3 . Client , task , source , upSchema , upTable string , revision int64 , outCh chan <- Operation , errCh chan <- error ) { var ch clientv3 . WatchChan wCtx , cancel := context . WithCancel ( ctx ) defer cancel ( ) if upTable == " " { ch = cli . Watch ( wCtx , common . ShardDDLOptimismOperationKeyAdapter . Path ( ) , clientv3 . WithPrefix ( ) , clientv3 . WithRev ( revision ) ) } else { ch = cli . Watch ( wCtx , common . ShardDDLOptimismOperationKeyAdapter . Encode ( task , source , upSchema , upTable ) , clientv3 . WithRev ( revision ) ) } for { select { case <- ctx . Done ( ) : return case resp , ok := <- ch : if ! ok { return } if resp . Canceled { select { case errCh <- resp . Err ( ) : case <- ctx . Done ( ) : } return } for _ , ev := range resp . Events { if ev . Type != mvccpb . PUT { continue } op , err := operationFromJSON ( string ( ev . Kv . Value ) ) op . Revision = ev . Kv . ModRevision if err != nil { select { case errCh <- err : case <- ctx . Done ( ) : return } } else { select { case outCh <- op : case <- ctx . Done ( ) : return } } } } } }
WatchOperationPut watches PUT operations for DDL lock operation.
WatchOperationPut watches PUT operations for DDL lock operation .
// caller may use empty keys to expect a prefix watch