// sync tries to retrieve existing data and send them to process.
// The revision to watch will be set to the revision in response.
// All events sent will have isCreated=true
426638
func (wc *watchChan) sync() error {
	opts := []clientv3.OpOption{}
	if wc.recursive {
		opts = append(opts, clientv3.WithPrefix())
	}
	getResp, err := wc.watcher.client.Get(wc.ctx, wc.key, opts...)
	if err != nil {
		return err
	}
	wc.initialRev = getResp.Header.Revision
	for _, kv := range getResp.Kvs {
		wc.sendEvent(parseKV(kv))
	}
	return nil
}
1124102
func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
	found := make([]interface{}, 0, len(items))
	for _, item := range items {
		found = append(found, item)
	}
	return r.store.Replace(found, resourceVersion)
}
251063
func (media *StoryMedia) Sync() error {
	insta := media.inst
	query := []trayRequest{
		{"SUPPORTED_SDK_VERSIONS", "9.0,10.0,11.0,12.0,13.0,14.0,15.0,16.0,17.0,18.0,19.0,20.0,21.0,22.0,23.0,24.0"},
		{"FACE_TRACKER_VERSION", "10"},
		{"segmentation", "segmentation_enabled"},
		{"COMPRESSION", "ETC2_COMPRESSION"},
	}
	qjson, err := json.Marshal(query)
	if err != nil {
		return err
	}

	id := media.Pk.(string)
	data, err := insta.prepareData(
		map[string]interface{}{
			"user_ids":                   []string{id},
			"supported_capabilities_new": b2s(qjson),
		},
	)
	if err != nil {
		return err
	}

	body, err := insta.sendRequest(
		&reqOptions{
			Endpoint: urlReelMedia,
			Query:    generateSignature(data),
			IsPost:   true,
		},
	)
	if err == nil {
		resp := trayResp{}
		err = json.Unmarshal(body, &resp)
		if err == nil {
			m, ok := resp.Reels[id]
			if ok {
				media.Items = m.Items
				media.setValues()
				return nil
			}
			err = fmt.Errorf("cannot find %s structure in response", id)
		}
	}
	return err
}
835210
func (w *Watcher) Watch() {
	eventSent := false

	for {
		select {
		case event := <-w.watcher.Events:
			// discard chmod events
			if event.Op&fsnotify.Chmod != fsnotify.Chmod {
				// test files do not need a rebuild
				if isTestFile(event.Name) {
					continue
				}
				if !isWatchedFileType(event.Name) {
					continue
				}
				if eventSent {
					continue
				}
				eventSent = true
				// prevent consequent builds
				go func() {
					w.update <- struct{}{}
					time.Sleep(watchDelta)
					eventSent = false
				}()

			}
		case err := <-w.watcher.Errors:
			if err != nil {
				log.Fatalf("Watcher error: %s", err)
			}
			return
		}
	}
}
743993
func (ws *watcherState) syncWithRuntime() {
	var wg sync.WaitGroup

	timeoutCtx, cancel := ctx.WithTimeout(ctx.Background(), 10*time.Second)
	defer cancel()

	cList, err := Client().workloadIDsList(timeoutCtx)
	if err != nil {
		log.WithError(err).Error("Failed to retrieve the container list")
		return
	}
	for _, contID := range cList {
		if ignoredContainer(contID) {
			continue
		}

		if alreadyHandled := ws.handlingContainerID(contID); !alreadyHandled {
			log.WithFields(logrus.Fields{
				logfields.ContainerID: shortContainerID(contID),
			}).Debug("Found unwatched container")

			wg.Add(1)
			go func(wg *sync.WaitGroup, id string) {
				defer wg.Done()
				Client().handleCreateWorkload(id, false)
			}(&wg, contID)
		}
	}

	// Wait for all spawned go routines handling container creations to exit
	wg.Wait()
}
1326286
func (comments *Comments) Sync() {
	endpoint := fmt.Sprintf(urlCommentSync, comments.item.ID)
	comments.endpoint = endpoint
	return
}
1243524
func (s *LDAPGroupSyncer) Sync() ([]*userv1.Group, []error) {
	openshiftGroups := []*userv1.Group{}
	var errors []error

	// determine what to sync
	klog.V(1).Infof("Listing with %v", s.GroupLister)
	ldapGroupUIDs, err := s.GroupLister.ListGroups()
	if err != nil {
		errors = append(errors, err)
		return nil, errors
	}
	klog.V(1).Infof("Sync ldapGroupUIDs %v", ldapGroupUIDs)

	for _, ldapGroupUID := range ldapGroupUIDs {
		klog.V(1).Infof("Checking LDAP group %v", ldapGroupUID)

		// get membership data
		memberEntries, err := s.GroupMemberExtractor.ExtractMembers(ldapGroupUID)
		if err != nil {
			fmt.Fprintf(s.Err, "Error determining LDAP group membership for %q: %v.\n", ldapGroupUID, err)
			errors = append(errors, err)
			continue
		}

		// determine OpenShift Users' usernames for LDAP group members
		usernames, err := s.determineUsernames(memberEntries)
		if err != nil {
			fmt.Fprintf(s.Err, "Error determining usernames for LDAP group %q: %v.\n", ldapGroupUID, err)
			errors = append(errors, err)
			continue
		}
		klog.V(1).Infof("Has OpenShift users %v", usernames)

		// update the OpenShift Group corresponding to this record
		openshiftGroup, err := s.makeOpenShiftGroup(ldapGroupUID, usernames)
		if err != nil {
			fmt.Fprintf(s.Err, "Error building OpenShift group for LDAP group %q: %v.\n", ldapGroupUID, err)
			errors = append(errors, err)
			continue
		}
		openshiftGroups = append(openshiftGroups, openshiftGroup)

		if !s.DryRun {
			fmt.Fprintf(s.Out, "group/%s\n", openshiftGroup.Name)
			if err := s.updateOpenShiftGroup(openshiftGroup); err != nil {
				fmt.Fprintf(s.Err, "Error updating OpenShift group %q for LDAP group %q: %v.\n", openshiftGroup.Name, ldapGroupUID, err)
				errors = append(errors, err)
				continue
			}
		}
	}

	return openshiftGroups, errors
}
965156
func (lbc *loadBalancerController) sync(dryRun bool) error {
	if !lbc.epController.HasSynced() || !lbc.svcController.HasSynced() {
		time.Sleep(100 * time.Millisecond)
		return errDeferredSync
	}
	httpSvc, httpsTermSvc, tcpSvc := lbc.getServices()
	if len(httpSvc) == 0 && len(httpsTermSvc) == 0 && len(tcpSvc) == 0 {
		return nil
	}
	if err := lbc.cfg.write(
		map[string][]service{
			"http":      httpSvc,
			"httpsTerm": httpsTermSvc,
			"tcp":       tcpSvc,
		}, dryRun); err != nil {
		return err
	}
	if dryRun {
		return nil
	}
	return lbc.cfg.reload()
}
1178075
func (media *FeedMedia) Sync() error {
	id := media.ID()
	insta := media.inst

	data, err := insta.prepareData(
		map[string]interface{}{
			"media_id": id,
		},
	)
	if err != nil {
		return err
	}

	body, err := insta.sendRequest(
		&reqOptions{
			Endpoint: fmt.Sprintf(urlMediaInfo, id),
			Query:    generateSignature(data),
			IsPost:   false,
		},
	)
	if err != nil {
		return err
	}

	m := FeedMedia{}
	err = json.Unmarshal(body, &m)
	*media = m
	media.endpoint = urlMediaInfo
	media.inst = insta
	media.NextID = id
	media.setValues()
	return err
}
1109722
func (w *Watch) update(newRevision int64){
	w.rwl.Lock()
	defer w.rwl.Unlock()
	w.revision = newRevision
	close(w.cond)
	w.cond = make(chan struct{})
}
536816
func (pb *PingBatcher) Sync() error {
	request := make(chan struct{})
	select {
	case pb.syncChan <- request:
		select {
		case <-request:
			return nil
		case <-pb.tomb.Dying():
			break
		}
	case <-pb.tomb.Dying():
		break
	}
	if err := pb.tomb.Err(); err == nil {
		return errors.Errorf("PingBatcher is stopped")
	} else {
		return err
	}
}
246635
func (api *PublicDownloaderAPI) Syncing(ctx context.Context) (*rpc.Subscription, error) {
	notifier, supported := rpc.NotifierFromContext(ctx)
	if !supported {
		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
	}

	rpcSub := notifier.CreateSubscription()

	go func() {
		statuses := make(chan interface{})
		sub := api.SubscribeSyncStatus(statuses)

		for {
			select {
			case status := <-statuses:
				notifier.Notify(rpcSub.ID, status)
			case <-rpcSub.Err():
				sub.Unsubscribe()
				return
			case <-notifier.Closed():
				sub.Unsubscribe()
				return
			}
		}
	}()

	return rpcSub, nil
}
1266225
func (s *PerUserKeyring) sync(m MetaContext, upak *keybase1.UserPlusAllKeys, deviceID keybase1.DeviceID, decryptionKey GenericKey) (err error) {
	defer m.Trace("PerUserKeyring#sync", func() error { return err })()

	m.Debug("PerUserKeyring#sync(%v, %v)", m.LoginContext() != nil, upak != nil)

	s.Lock()
	defer s.Unlock()

	box, prevs, err := s.fetchBoxesLocked(m, deviceID)
	if err != nil {
		return err
	}

	if upak == nil {
		upak, err = s.getUPAK(m, upak)
		if err != nil {
			return err
		}
	}

	checker := newPerUserKeyChecker(upak)
	newKeys, err := s.importLocked(m, box, prevs, decryptionKey, checker)
	if err != nil {
		return err

	}
	s.mergeLocked(newKeys, checker.seqgen)
	return nil
}
1126164
func (sc *syncContext) doApplySync(syncTasks []syncTask, dryRun, force, update bool) bool {
	syncSuccessful := true

	var createTasks []syncTask
	var pruneTasks []syncTask
	for _, syncTask := range syncTasks {
		if syncTask.targetObj == nil {
			pruneTasks = append(pruneTasks, syncTask)
		} else {
			createTasks = append(createTasks, syncTask)
		}
	}

	var wg sync.WaitGroup
	for _, task := range pruneTasks {
		wg.Add(1)
		go func(t syncTask) {
			defer wg.Done()
			resDetails := sc.pruneObject(t.liveObj, sc.syncOp.Prune, dryRun)
			if !resDetails.Status.Successful() {
				syncSuccessful = false
			}
			if update || !resDetails.Status.Successful() {
				sc.setResourceDetails(&resDetails)
			}
		}(task)
	}
	wg.Wait()

	processCreateTasks := func(tasks []syncTask) {
		var createWg sync.WaitGroup
		for i := range tasks {
			if dryRun && tasks[i].skipDryRun {
				continue
			}
			createWg.Add(1)
			go func(t syncTask) {
				defer createWg.Done()
				if hookutil.IsHook(t.targetObj) {
					return
				}
				resDetails := sc.applyObject(t.targetObj, dryRun, force)
				if !resDetails.Status.Successful() {
					syncSuccessful = false
				}
				if update || !resDetails.Status.Successful() {
					sc.setResourceDetails(&resDetails)
				}
			}(tasks[i])
		}
		createWg.Wait()
	}

	var tasksGroup []syncTask
	for _, task := range createTasks {
		//Only wait if the type of the next task is different than the previous type
		if len(tasksGroup) > 0 && tasksGroup[0].targetObj.GetKind() != task.targetObj.GetKind() {
			processCreateTasks(tasksGroup)
			tasksGroup = []syncTask{task}
		} else {
			tasksGroup = append(tasksGroup, task)
		}
	}
	if len(tasksGroup) > 0 {
		processCreateTasks(tasksGroup)
	}
	return syncSuccessful
}
196191
func (s *Bool) Watch(watch BoolWatch) {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	s.watches = append(s.watches, watch)
}
890595
func (s *Logging) Sync() error {
	s.mutex.RLock()
	w := s.writer
	s.mutex.RUnlock()

	return w.Sync()
}
153145
func (e *etcdClient) Watch(w *Watcher) {
	localCache := watcherCache{}
	listSignalSent := false

	scopedLog := e.getLogger().WithFields(logrus.Fields{
		fieldWatcher: w,
		fieldPrefix:  w.prefix,
	})
	<-e.Connected()

reList:
	for {
		e.limiter.Wait(ctx.TODO())
		res, err := e.client.Get(ctx.Background(), w.prefix, client.WithPrefix(),
			client.WithSerializable())
		if err != nil {
			scopedLog.WithError(Hint(err)).Warn("Unable to list keys before starting watcher")
			continue
		}

		nextRev := res.Header.Revision + 1
		scopedLog.Debugf("List response from etcd len=%d: %+v", res.Count, res)

		if res.Count > 0 {
			for _, key := range res.Kvs {
				t := EventTypeCreate
				if localCache.Exists(key.Key) {
					t = EventTypeModify
				}

				localCache.MarkInUse(key.Key)
				scopedLog.Debugf("Emitting list result as %v event for %s=%v", t, key.Key, key.Value)

				queueStart := spanstat.Start()
				w.Events <- KeyValueEvent{
					Key:   string(key.Key),
					Value: key.Value,
					Typ:   t,
				}
				trackEventQueued(string(key.Key), t, queueStart.End(true).Total())
			}
		}

		// More keys to be read, call Get() again
		if res.More {
			continue
		}

		// Send out deletion events for all keys that were deleted
		// between our last known revision and the latest revision
		// received via Get
		localCache.RemoveDeleted(func(k string) {
			event := KeyValueEvent{
				Key: k,
				Typ: EventTypeDelete,
			}

			scopedLog.Debugf("Emitting EventTypeDelete event for %s", k)
			queueStart := spanstat.Start()
			w.Events <- event
			trackEventQueued(k, EventTypeDelete, queueStart.End(true).Total())
		})

		// Only send the list signal once
		if !listSignalSent {
			w.Events <- KeyValueEvent{Typ: EventTypeListDone}
			listSignalSent = true
		}

	recreateWatcher:
		scopedLog.WithField(fieldRev, nextRev).Debug("Starting to watch a prefix")

		e.limiter.Wait(ctx.TODO())
		etcdWatch := e.client.Watch(ctx.Background(), w.prefix,
			client.WithPrefix(), client.WithRev(nextRev))
		for {
			select {
			case <-w.stopWatch:
				close(w.Events)
				w.stopWait.Done()
				return

			case r, ok := <-etcdWatch:
				if !ok {
					time.Sleep(50 * time.Millisecond)
					goto recreateWatcher
				}

				scopedLog := scopedLog.WithField(fieldRev, r.Header.Revision)

				if err := r.Err(); err != nil {
					// We tried to watch on a compacted
					// revision that may no longer exist,
					// recreate the watcher and try to
					// watch on the next possible revision
					if err == v3rpcErrors.ErrCompacted {
						scopedLog.WithError(Hint(err)).Debug("Tried watching on compacted revision")
					}

					// mark all local keys in state for
					// deletion unless the upcoming GET
					// marks them alive
					localCache.MarkAllForDeletion()

					goto reList
				}

				nextRev = r.Header.Revision + 1
				scopedLog.Debugf("Received event from etcd: %+v", r)

				for _, ev := range r.Events {
					event := KeyValueEvent{
						Key:   string(ev.Kv.Key),
						Value: ev.Kv.Value,
					}

					switch {
					case ev.Type == client.EventTypeDelete:
						event.Typ = EventTypeDelete
						localCache.RemoveKey(ev.Kv.Key)
					case ev.IsCreate():
						event.Typ = EventTypeCreate
						localCache.MarkInUse(ev.Kv.Key)
					default:
						event.Typ = EventTypeModify
						localCache.MarkInUse(ev.Kv.Key)
					}

					scopedLog.Debugf("Emitting %v event for %s=%v", event.Typ, event.Key, event.Value)

					queueStart := spanstat.Start()
					w.Events <- event
					trackEventQueued(string(ev.Kv.Key), event.Typ, queueStart.End(true).Total())
				}
			}
		}
	}
}