// PropagateResync fills registered channels with the data.
10386
func (adapter *Registry) PropagateResync(ctx context.Context, txData map[string]datasync.ChangeValue) error {
	var events []func(done chan error)
	adapter.lastRev.Cleanup()

	for _, sub := range adapter.subscriptions {
		items := map[string]datasync.KeyValIterator{}

		for _, prefix := range sub.KeyPrefixes {
			var kvs []datasync.KeyVal

			for key, val := range txData {
				if strings.HasPrefix(key, prefix) {
					// TODO: call Put only once for each key (different subscriptions)
					adapter.lastRev.PutWithRevision(key, val)

					kvs = append(kvs, &KeyVal{
						key:       key,
						LazyValue: val,
						rev:       val.GetRevision(),
					})
				}
			}
			items[prefix] = NewKVIterator(kvs)
		}

		sendTo := func(sub *Subscription) func(done chan error) {
			return func(done chan error) {
				sub.ResyncChan <- &ResyncEventDB{
					ctx:         ctx,
					its:         items,
					DoneChannel: NewDoneChannel(done),
				}
			}
		}
		events = append(events, sendTo(sub))
	}

	done := make(chan error, 1)
	go AggregateDone(events, done)

	select {
	case err := <-done:
		if err != nil {
			return err
		}

	// TODO: maybe higher timeout for resync event?
	case <-time.After(PropagateChangesTimeout):
		logrus.DefaultLogger().Warnf("Timeout of aggregated resync callbacks (%v)",
			PropagateChangesTimeout)
	}

	return nil
}
1351739
func (keys *watchBrokerKeys) watchResync(resyncReg resync.Registration) {
	for resyncStatus := range resyncReg.StatusChan() {
		if resyncStatus.ResyncStatus() == resync.Started {
			err := keys.resync()
			if err != nil {
				// We are not able to propagate it somewhere else.
				logrus.DefaultLogger().Errorf("getting resync data failed: %v", err)
				// TODO NICE-to-HAVE publish the err using the transport asynchronously
			}
		}
		resyncStatus.Ack()
	}
}
617620
func (n *ExecNode) Snapshots() (map[string][]byte, error) {
	if n.client == nil {
		return nil, errors.New("RPC not started")
	}
	var snapshots map[string][]byte
	return snapshots, n.client.Call(&snapshots, "simulation_snapshot")
}
1136011
func (pgb *ChainDBRPC) SyncChainDBAsync(ctx context.Context, res chan dbtypes.SyncResult,
	client rpcutils.MasterBlockGetter, updateAllAddresses, updateAllVotes, newIndexes bool,
	updateExplorer chan *chainhash.Hash, barLoad chan *dbtypes.ProgressBarLoad) {
	// Allowing db to be nil simplifies logic in caller.
	if pgb == nil {
		res <- dbtypes.SyncResult{
			Height: -1,
			Error:  fmt.Errorf("ChainDB (psql) disabled"),
		}
		return
	}
	pgb.ChainDB.SyncChainDBAsync(ctx, res, client, updateAllAddresses,
		updateAllVotes, newIndexes, updateExplorer, barLoad)
}
1116806
func NewBinaryPropagator(tracer *Tracer) *BinaryPropagator {
	return &BinaryPropagator{
		tracer:  tracer,
		buffers: sync.Pool{New: func() interface{} { return &bytes.Buffer{} }},
	}
}
1232120
func (conn *ProtoConnection) NewAsyncPublisher(topic string, successClb func(messaging.ProtoMessage), errorClb func(messaging.ProtoMessageErr)) (messaging.ProtoPublisher, error) {
	return &protoAsyncPublisherKafka{conn, topic, successClb, errorClb}, nil
}
1227033
func (c *NodeClient) listenAndForward(ctx context.Context, ln net.Listener, remoteAddr string) {
	defer ln.Close()
	defer c.Close()

	for {
		// Accept connections from the client.
		conn, err := ln.Accept()
		if err != nil {
			log.Errorf("Port forwarding failed: %v.", err)
			break
		}

		// Proxy the connection to the remote address.
		go func() {
			err := c.proxyConnection(ctx, conn, remoteAddr)
			if err != nil {
				log.Warnf("Failed to proxy connection: %v.", err)
			}
		}()
	}
}
376782
func (t *Client) UploadAsync(datasetName string, tableName string, data interface{}) (err error) {
	if t.ensureDataset {
		if _, exists := t.ensureDatasets.LoadOrStore(datasetName, true); !exists {
			if _, err = t.EnsureDataset(datasetName); err != nil {
				return
			}
		}
	}
	if t.ensureTable {
		key := datasetName + "/" + tableName
		if _, exists := t.ensureTables.LoadOrStore(key, true); !exists {
			if _, err = t.EnsureTable(datasetName, tableName, data); err != nil {
				return
			}
		}
	}

	dataset := t.client.Dataset(datasetName)
	table := dataset.Table(tableName)
	t.msgQueue <- uploadMessage{
		table: table,
		data:  data,
	}
	return nil
}
37214
func (sc *conn) PublishAsync(subject string, data []byte, ah AckHandler) (string, error) {
	return sc.publishAsync(subject, data, ah, nil)
}
921435
func (p *pipelineAfterChannel) PushBack(h AfterHandlerTempl) {
	p.Handlers = append(p.Handlers, h)
}
501190
func NewSync(dsn string) (*Handler, error) {
	handler, err := New(dsn)
	if err != nil {
		return nil, err
	}

	handler.sendSynchronously = true

	return handler, nil
}
859483
func (w *Wallet) rescanProgressHandler() {
	quit := w.quitChan()
out:
	for {
		// These can't be processed out of order since both chans are
		// unbuffured and are sent from same context (the batch
		// handler).
		select {
		case msg := <-w.rescanProgress:
			n := msg.Notification
			log.Infof("Rescanned through block %v (height %d)",
				n.Hash, n.Height)

		case msg := <-w.rescanFinished:
			n := msg.Notification
			addrs := msg.Addresses
			noun := pickNoun(len(addrs), "address", "addresses")
			log.Infof("Finished rescan for %d %s (synced to block "+
				"%s, height %d)", len(addrs), noun, n.Hash,
				n.Height)

			go w.resendUnminedTxs()

		case <-quit:
			break out
		}
	}
	w.wg.Done()
}
818902
func propagateTearDown(handler tearDowner, t *tomb.Tomb) {
	if err := handler.TearDown(); err != nil {
		t.Kill(err)
	}
}
463428
func PropagationWait(d time.Duration) Option {
	return func(c *Client) error {
		c.propagationWait = d
		return nil
	}
}
262391
func NewSyncProducer(addrs []string, config *Config) (SyncProducer, error) {
	if config == nil {
		config = NewConfig()
		config.Producer.Return.Successes = true
	}

	if err := verifyProducerConfig(config); err != nil {
		return nil, err
	}

	p, err := NewAsyncProducer(addrs, config)
	if err != nil {
		return nil, err
	}
	return newSyncProducerFromAsyncProducer(p.(*asyncProducer)), nil
}
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)
}
89340
func (mux *Multiplexer) propagateMessage(msg *client.ConsumerMessage) {
	mux.rwlock.RLock()
	defer mux.rwlock.RUnlock()

	if msg == nil {
		return
	}

	// Find subscribed topics. Note: topic can be subscribed for both dynamic and manual consuming
	for _, subscription := range mux.mapping {
		if msg.Topic == subscription.topic {
			// Clustered mode - message is consumed only on right partition and offset
			if subscription.manual {
				if msg.Partition == subscription.partition && msg.Offset >= subscription.offset {
					mux.Debug("offset ", msg.Offset, string(msg.Value), string(msg.Key), msg.Partition)
					subscription.byteConsMsg(msg)
				}
			} else {
				// Non-manual mode
				// if we are not able to write into the channel we should skip the receiver
				// and report an error to avoid deadlock
				mux.Debug("offset ", msg.Offset, string(msg.Value), string(msg.Key), msg.Partition)
				subscription.byteConsMsg(msg)
			}
		}
	}
}