// NewSync creates a new trie data download scheduler.
190336
func NewSync(root common.Hash, database ethdb.Reader, callback LeafCallback) *Sync {
	ts := &Sync{
		database: database,
		membatch: newSyncMemBatch(),
		requests: make(map[common.Hash]*request),
		queue:    prque.New(nil),
	}
	ts.AddSubTrie(root, 0, common.Hash{}, callback)
	return ts
}
322165
func NewStreamC(dataShards, parityShards int, conReads, conWrites bool, o ...Option) (StreamEncoder, error) {
	enc, err := New(dataShards, parityShards, o...)
	if err != nil {
		return nil, err
	}
	rs := enc.(*reedSolomon)
	r := rsStream{r: rs, bs: 4 << 20}
	r.readShards = readShards
	r.writeShards = writeShards
	if conReads {
		r.readShards = cReadShards
	}
	if conWrites {
		r.writeShards = cWriteShards
	}
	return &r, err
}
75063
func NewSyncConnection(spec *ConnectionSpec) (c SyncConnection, err Error) {
	defer func() {
		if e := recover(); e != nil {
			connerr := e.(error)
			err = newSystemErrorWithCause("NewSyncConnection", connerr)
		}
	}()

	connHdl := newConnHdl(spec)
	connHdl.connect()
	c = connHdl
	return
}
1099939
func NewSync(token, environment, codeVersion, serverHost, serverRoot string) *Client {
	configuration := createConfiguration(token, environment, codeVersion, serverHost, serverRoot)
	transport := NewSyncTransport(token, configuration.endpoint)
	return &Client{
		Transport:     transport,
		configuration: configuration,
	}
}
826832
func NewStateSync(root common.Hash, database ethdb.Reader) *trie.Sync {
	var syncer *trie.Sync
	callback := func(leaf []byte, parent common.Hash) error {
		var obj Account
		if err := rlp.Decode(bytes.NewReader(leaf), &obj); err != nil {
			return err
		}
		syncer.AddSubTrie(obj.Root, 64, parent, nil)
		syncer.AddRawEntry(common.BytesToHash(obj.CodeHash), 64, parent)
		return nil
	}
	syncer = trie.NewSync(root, database, callback)
	return syncer
}
526852
func (helper *CacheHelper) DoResync(resyncEv datasync.ResyncEvent) error {
	var wasError error
	//idx.Put()
	ifaces, found := resyncEv.GetValues()[helper.Prefix]
	if found {
		// Step 1: fill the existing things
		resyncNames := map[string]interface{}{}
		for {
			item, stop := ifaces.GetNext()
			if stop {
				break
			}
			ifaceName, err := helper.ParseName(item.GetKey())
			if err != nil {
				wasError = err
			} else {
				current := proto.Clone(helper.DataPrototype)
				item.GetValue(current)
				helper.IDX.Put(ifaceName, current)
				resyncNames[ifaceName] = nil
			}
		}

		// Step 2:
		existingNames := []string{} //TODO
		for _, existingName := range existingNames {
			if _, found := resyncNames[existingName]; !found {
				helper.IDX.Delete(existingName)
			}
		}
	}
	return wasError
}
643983
func NewSyncedEventStream(stream EventStream) SelectableEventStream {
	var (
		sync syncedEventStream

		in   <-chan Event
		subs <-chan EventWriter
		halt <-chan eventStreamHaltReq
	)

	closeChans := func() func() {
		var (
			inCh   = make(chan Event)
			subsCh = make(chan EventWriter)
			haltCh = make(chan eventStreamHaltReq)
		)

		sync.in, in = inCh, inCh
		sync.subs, subs = subsCh, subsCh
		sync.halt, halt = haltCh, haltCh

		return func() {
			close(inCh)
			close(subsCh)
			close(haltCh)
		}
	}()

	go func() {
		var haltReq eventStreamHaltReq

	communication:
		for {
			select {
			case e := <-in:
				stream.Write(e)

			case w := <-subs:
				stream.Subscribe(w)

			case haltReq = <-halt:
				break communication
			}
		}

		closeChans()
		haltReq.hasHalted <- stream
	}()

	return sync
}
440351
func (n *Stan) PublishAsync(subject string, data PayloadWithContext, ackH stan.AckHandler) (string, error) {
	n.updateContext(data, RequestType_PUB)
	bData, err := proto.Marshal(data)
	if err != nil {
		return "", err
	}
	return n.Con.PublishAsync(subject, bData, ackH)
}
1040996
func NewSchedule(clock clock.Clock) *Schedule {
	return &Schedule{
		time: clock,
		m:    make(map[interface{}]*scheduleItem),
	}
}
222369
func NewStream(dataShards, parityShards int, o ...Option) (StreamEncoder, error) {
	enc, err := New(dataShards, parityShards, o...)
	if err != nil {
		return nil, err
	}
	rs := enc.(*reedSolomon)
	r := rsStream{r: rs, bs: 4 << 20}
	r.readShards = readShards
	r.writeShards = writeShards
	return &r, err
}
303310
func NewSource(options ...Option) *Source {
	src := &Source{
		records:     make(chan record),
		maxRetries:  3,
		concurrency: 1,
	}

	for _, opt := range options {
		opt(src)
	}
	go src.getRecords()
	return src
}
261924
func (f *DeltaFIFO) Resync() error {
	f.lock.Lock()
	defer f.lock.Unlock()

	if f.knownObjects == nil {
		return nil
	}

	keys := f.knownObjects.ListKeys()
	for _, k := range keys {
		if err := f.syncKeyLocked(k); err != nil {
			return err
		}
	}
	return nil
}
577687
func NewScheduler() *Scheduler {
	return &Scheduler{
		Sink:             NewHTTPSink(),
		Timer:            timekeeper.RealTime{},
		ErrorHandler:     DefaultErrorHandler,
		ReportingDelayNs: DefaultReportingDelay.Nanoseconds(),
		callbackMap:      make(map[string]*callbackPair),
	}
}
8782
func (l *LinuxMonitor) Resync(ctx context.Context) error {
	return l.proc.Resync(ctx, nil)
}
819347
func newStateSync(d *Downloader, root common.Hash) *stateSync {
	return &stateSync{
		d:       d,
		sched:   state.NewStateSync(root, d.stateDB),
		keccak:  sha3.NewLegacyKeccak256(),
		tasks:   make(map[common.Hash]*stateTask),
		deliver: make(chan *stateReq),
		cancel:  make(chan struct{}),
		done:    make(chan struct{}),
	}
}
143878
func newSyncMemBatch() *syncMemBatch {
	return &syncMemBatch{
		batch: make(map[common.Hash][]byte),
		order: make([]common.Hash, 0, 256),
	}
}
1352858
func NewSource(opts ...SrcOption) (*Source, error) {
	s := &Source{
		records: make(chan record, 100),
	}
	for _, opt := range opts {
		err := opt(s)
		if err != nil {
			return nil, err
		}
	}
	go s.run()
	return s, nil
}