code: func (adapter *Registry) PropagateChanges(ctx context.Context, txData map[string]datasync.ChangeValue) error {
  	var events []func(done chan error)
  
  	for _, sub := range adapter.subscriptions {
  		var changes []datasync.ProtoWatchResp
  
  		for _, prefix := range sub.KeyPrefixes {
  			for key, val := range txData {
  				if !strings.HasPrefix(key, prefix) {
  					continue
  				}
  
  				var (
  					prev   datasync.KeyVal
  					curRev int64
  				)
  
  				if val.GetChangeType() == datasync.Delete {
  					if _, prev = adapter.lastRev.Del(key); prev != nil {
  						curRev = prev.GetRevision() + 1
  					} else {
  						continue
  					}
  				} else {
  					_, prev, curRev = adapter.lastRev.Put(key, val)
  				}
  
  				changes = append(changes, &ChangeResp{
  					Key:        key,
  					ChangeType: val.GetChangeType(),
  					CurrVal:    val,
  					CurrRev:    curRev,
  					PrevVal:    prev,
  				})
  			}
  		}
  
  		if len(changes) > 0 {
  			sendTo := func(sub *Subscription) func(done chan error) {
  				return func(done chan error) {
  					sub.ChangeChan <- &ChangeEvent{
  						ctx:      ctx,
  						Changes:  changes,
  						delegate: &DoneChannel{done},
  					}
  				}
  			}
  			events = append(events, sendTo(sub))
  		}
  	}
  
  	done := make(chan error, 1)
  	go AggregateDone(events, done)
  
  	select {
  	case err := <-done:
  		if err != nil {
  			return err
  		}
  	case <-time.After(PropagateChangesTimeout):
  		logrus.DefaultLogger().Warnf("Timeout of aggregated data-change callbacks (%v)",
  			PropagateChangesTimeout)
  	}
  
  	return nil
  }
docs: // PropagateChanges fills registered channels with the data.