// sendSubsToRoute will send over our subject interest to
// the remote side. For each account we will send the
// complete interest for all subjects, both normal as a binary
// and queue group weights.
513696
func (s *Server) sendSubsToRoute(route *client) {
	s.mu.Lock()
	// Estimated size of all protocols. It does not have to be accurate at all.
	eSize := 0
	// Send over our account subscriptions.
	// copy accounts into array first
	accs := make([]*Account, 0, 32)
	s.accounts.Range(func(k, v interface{}) bool {
		a := v.(*Account)
		accs = append(accs, a)
		a.mu.RLock()
		// Proto looks like: "RS+ <account name> <subject>[ <queue weight>]\r\n"
		// If we wanted to have better estimates (or even accurate), we would
		// collect the subs here instead of capturing the accounts and then
		// later going over each account.
		eSize += len(a.rm) * (4 + len(a.Name) + 256)
		a.mu.RUnlock()
		return true
	})
	s.mu.Unlock()

	sendSubs := func(accs []*Account) {
		var raw [32]*subscription
		var closed bool

		route.mu.Lock()
		for _, a := range accs {
			subs := raw[:0]

			a.mu.RLock()
			c := a.randomClient()
			if c == nil {
				a.mu.RUnlock()
				continue
			}
			for key, n := range a.rm {
				// FIXME(dlc) - Just pass rme around.
				// Construct a sub on the fly. We need to place
				// a client (or im) to properly set the account.
				var subj, qn []byte
				s := strings.Split(key, " ")
				subj = []byte(s[0])
				if len(s) > 1 {
					qn = []byte(s[1])
				}
				// TODO(dlc) - This code needs to change, but even if left alone could be more
				// efficient with these tmp subs.
				sub := &subscription{client: c, subject: subj, queue: qn, qw: n}
				subs = append(subs, sub)

			}
			a.mu.RUnlock()

			closed = route.sendRouteSubProtos(subs, false, func(sub *subscription) bool {
				return route.canImport(string(sub.subject))
			})

			if closed {
				route.mu.Unlock()
				return
			}
		}
		route.mu.Unlock()
		if !closed {
			route.Debugf("Sent local subscriptions to route")
		}
	}
	// Decide if we call above function in go routine or in place.
	if eSize > sendRouteSubsInGoRoutineThreshold {
		s.startGoRoutine(func() {
			sendSubs(accs)
			s.grWG.Done()
		})
	} else {
		sendSubs(accs)
	}
}
1022911
func (s *DefaultSubscriber) SubscribeAppEvents(appID string, eventType types.EventType, handler AppEventHandler) error {
	key := ApplicationKey{appID, AppEvents, string(eventType)}
	deliveries, err := s.subscribe(key.String())
	if err != nil {
		return err
	}
	go func() {
		for letter := range deliveries {
			handler(s, appID, types.EventType(eventType), letter.Body)
		}
	}()
	return nil
}
305176
func (c *EncodedConn) QueueSubscribe(subject, queue string, cb Handler) (*Subscription, error) {
	return c.subscribe(subject, queue, cb)
}
181250
func (c *client) processRemoteUnsub(arg []byte) (err error) {
	srv := c.srv
	if srv == nil {
		return nil
	}
	accountName, subject, _, err := c.parseUnsubProto(arg)
	if err != nil {
		return fmt.Errorf("processRemoteUnsub %s", err.Error())
	}
	// Lookup the account
	acc, _ := c.srv.LookupAccount(accountName)
	if acc == nil {
		c.Debugf("Unknown account %q for subject %q", accountName, subject)
		// Mark this account as not interested since we received a RS- and we
		// do not have any record of it.
		return nil
	}

	c.mu.Lock()
	if c.nc == nil {
		c.mu.Unlock()
		return nil
	}

	updateGWs := false
	// We store local subs by account and subject and optionally queue name.
	// RS- will have the arg exactly as the key.
	key := string(arg)
	sub, ok := c.subs[key]
	if ok {
		delete(c.subs, key)
		acc.sl.Remove(sub)
		c.removeReplySubTimeout(sub)
		updateGWs = srv.gateway.enabled
	}
	c.mu.Unlock()

	if updateGWs {
		srv.gatewayUpdateSubInterest(accountName, sub, -1)
	}

	// Now check on leafnode updates.
	srv.updateLeafNodes(acc, sub, -1)

	if c.opts.Verbose {
		c.sendOK()
	}
	return nil
}
1154733
func (c *client) sendLeafNodeSubUpdate(key string, n int32) {
	_b := [64]byte{}
	b := bytes.NewBuffer(_b[:0])
	c.writeLeafSub(b, key, n)
	c.sendProto(b.Bytes(), false)
}
860532
func (s *Server) broadcastUnSubscribe(sub *subscription) {
	trace := atomic.LoadInt32(&s.logging.trace) == 1
	s.mu.Lock()
	subs := []*subscription{sub}
	for _, route := range s.routes {
		route.mu.Lock()
		route.sendRouteUnSubProtos(subs, trace, func(sub *subscription) bool {
			return route.canImport(string(sub.subject))
		})
		route.mu.Unlock()
	}
	s.mu.Unlock()
}
971453
func sendPairs(w *wallet.Wallet, amounts map[string]btcutil.Amount,
	account uint32, minconf int32, feeSatPerKb btcutil.Amount) (string, error) {

	outputs, err := makeOutputs(amounts, w.ChainParams())
	if err != nil {
		return "", err
	}
	tx, err := w.SendOutputs(outputs, account, minconf, feeSatPerKb)
	if err != nil {
		if err == txrules.ErrAmountNegative {
			return "", ErrNeedPositiveAmount
		}
		if waddrmgr.IsError(err, waddrmgr.ErrLocked) {
			return "", &ErrWalletUnlockNeeded
		}
		switch err.(type) {
		case btcjson.RPCError:
			return "", err
		}

		return "", &btcjson.RPCError{
			Code:    btcjson.ErrRPCInternal.Code,
			Message: err.Error(),
		}
	}

	txHashStr := tx.TxHash().String()
	log.Infof("Successfully sent transaction %v", txHashStr)
	return txHashStr, nil
}
282651
func Send(reqCtx reqContext.Context, tx *fab.Transaction, orderers []fab.Orderer) (*fab.TransactionResponse, error) {
	if len(orderers) == 0 {
		return nil, errors.New("orderers is nil")
	}
	if tx == nil {
		return nil, errors.New("transaction is nil")
	}
	if tx.Proposal == nil || tx.Proposal.Proposal == nil {
		return nil, errors.New("proposal is nil")
	}

	// the original header
	hdr, err := protoutil.GetHeader(tx.Proposal.Proposal.Header)
	if err != nil {
		return nil, errors.Wrap(err, "unmarshal proposal header failed")
	}
	// serialize the tx
	txBytes, err := protoutil.GetBytesTransaction(tx.Transaction)
	if err != nil {
		return nil, err
	}

	// create the payload
	payload := common.Payload{Header: hdr, Data: txBytes}

	transactionResponse, err := BroadcastPayload(reqCtx, &payload, orderers)
	if err != nil {
		return nil, err
	}

	return transactionResponse, nil
}
20561
func (c *Client) LookupSubscription(subID string) (*SubscriptionRequest, error) {
	s, err := c.subscriptions.lookupBySubscriptionID(subID)
	if err != nil {
		return nil, err
	}
	return s.Request, nil
}
278678
func (s *RegionRequestSender) SendReq(bo *Backoffer, req *tikvrpc.Request, regionID RegionVerID, timeout time.Duration) (*tikvrpc.Response, error) {
	resp, _, err := s.SendReqCtx(bo, req, regionID, timeout)
	return resp, err
}
384598
func sendRelayPayment(mctx libkb.MetaContext, walletState *WalletState,
	from stellar1.SecretKey, recipient stellarcommon.Recipient, amount string, displayBalance DisplayBalance,
	secretNote string, publicMemo string, quickReturn bool, senderEntryPrimary bool, baseFee uint64) (res SendPaymentResult, err error) {
	defer mctx.TraceTimed("Stellar.sendRelayPayment", func() error { return err })()
	appKey, teamID, err := relays.GetKey(mctx, recipient)
	if err != nil {
		return res, err
	}

	if isAmountLessThanMin(amount, minAmountRelayXLM) {
		return res, fmt.Errorf("you must send at least %s XLM to fund the account for %s", minAmountRelayXLM, recipient.Input)
	}

	sp, unlock := NewSeqnoProvider(mctx, walletState)
	defer unlock()
	tb, err := getTimeboundsForSending(mctx, walletState)
	if err != nil {
		return res, err
	}
	relay, err := relays.Create(relays.Input{
		From:          from,
		AmountXLM:     amount,
		Note:          secretNote,
		PublicMemo:    publicMemo,
		EncryptFor:    appKey,
		SeqnoProvider: sp,
		Timebounds:    tb,
		BaseFee:       baseFee,
	})
	if err != nil {
		return res, err
	}

	_, accountID, _, err := libkb.ParseStellarSecretKey(string(from))
	if err != nil {
		return res, err
	}
	if err := walletState.AddPendingTx(mctx.Ctx(), accountID, stellar1.TransactionID(relay.FundTx.TxHash), relay.FundTx.Seqno); err != nil {
		mctx.Debug("error calling AddPendingTx: %s", err)
	}

	post := stellar1.PaymentRelayPost{
		FromDeviceID:      mctx.ActiveDevice().DeviceID(),
		ToAssertion:       string(recipient.Input),
		RelayAccount:      relay.RelayAccountID,
		TeamID:            teamID,
		BoxB64:            relay.EncryptedB64,
		SignedTransaction: relay.FundTx.Signed,
		DisplayAmount:     displayBalance.Amount,
		DisplayCurrency:   displayBalance.Currency,
		QuickReturn:       quickReturn,
	}
	if recipient.User != nil {
		post.To = &recipient.User.UV
	}
	rres, err := walletState.SubmitRelayPayment(mctx.Ctx(), post)
	if err != nil {
		if rerr := walletState.RemovePendingTx(mctx.Ctx(), accountID, stellar1.TransactionID(relay.FundTx.TxHash)); rerr != nil {
			mctx.Debug("error calling RemovePendingTx: %s", rerr)
		}
		return res, err
	}
	mctx.Debug("sent payment (relay) kbTxID:%v txID:%v pending:%v", rres.KeybaseID, rres.StellarID, rres.Pending)

	if !rres.Pending {
		if err := walletState.RemovePendingTx(mctx.Ctx(), accountID, stellar1.TransactionID(relay.FundTx.TxHash)); err != nil {
			mctx.Debug("error calling RemovePendingTx: %s", err)
		}
	}

	if senderEntryPrimary {
		sendChat := func(mctx libkb.MetaContext) {
			if err := chatSendPaymentMessage(mctx, recipient, rres.StellarID); err != nil {
				// if the chat message fails to send, just log the error
				mctx.Debug("failed to send chat SendPayment message: %s", err)
			}
		}
		if post.QuickReturn {
			go sendChat(mctx.WithCtx(context.Background()))
		} else {
			sendChat(mctx)
		}
	} else {
		mctx.Debug("not sending chat message (relay): sending from non-primary account")
	}

	return SendPaymentResult{
		KbTxID:      rres.KeybaseID,
		TxID:        rres.StellarID,
		Pending:     rres.Pending,
		RelayTeamID: &teamID,
	}, nil
}
241419
func (s *StanServer) addSubscription(ss *subStore, sub *subState) error {
	// Store in client
	if !s.clients.addSub(sub.ClientID, sub) {
		return fmt.Errorf("can't find clientID: %v", sub.ClientID)
	}
	// Store this subscription in subStore
	if err := ss.Store(sub); err != nil {
		s.clients.removeSub(sub.ClientID, sub)
		return err
	}
	return nil
}
1180678
func (b *redisBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	var options broker.SubscribeOptions
	for _, o := range opts {
		o(&options)
	}

	s := subscriber{
		codec:  b.opts.Codec,
		conn:   &redis.PubSubConn{Conn: b.pool.Get()},
		topic:  topic,
		handle: handler,
		opts:   options,
	}

	// Run the receiver routine.
	go s.recv()

	if err := s.conn.Subscribe(s.topic); err != nil {
		return nil, err
	}

	return &s, nil
}
815891
func (t *Transactor) SendTransactionProposal(proposal *fab.TransactionProposal, targets []fab.ProposalProcessor) ([]*fab.TransactionProposalResponse, error) {
	ctx, ok := contextImpl.RequestClientContext(t.reqCtx)
	if !ok {
		return nil, errors.New("failed get client context from reqContext for SendTransactionProposal")
	}

	reqCtx, cancel := contextImpl.NewRequest(ctx, contextImpl.WithTimeoutType(fab.PeerResponse), contextImpl.WithParent(t.reqCtx))
	defer cancel()

	return txn.SendProposal(reqCtx, proposal, targets)
}
988350
func (sub *subscription) Delivered() (int64, error) {
	sub.Lock()
	defer sub.Unlock()
	if sub.inboxSub == nil {
		return -1, ErrBadSubscription
	}
	return sub.inboxSub.Delivered()
}
198554
func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
	var sp opentracing.Span
	ctx, sp := spancontext.StartSpan(
		context.TODO(),
		"send.offered.hashes",
	)
	defer sp.Finish()

	defer metrics.GetOrRegisterResettingTimer("send.offered.hashes", nil).UpdateSince(time.Now())

	hashes, from, to, proof, err := s.setNextBatch(f, t)
	if err != nil {
		return err
	}
	// true only when quitting
	if len(hashes) == 0 {
		return nil
	}
	if proof == nil {
		proof = &HandoverProof{
			Handover: &Handover{},
		}
	}
	s.currentBatch = hashes
	msg := &OfferedHashesMsg{
		HandoverProof: proof,
		Hashes:        hashes,
		From:          from,
		To:            to,
		Stream:        s.stream,
	}
	log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to)
	ctx = context.WithValue(ctx, "stream_send_tag", "send.offered.hashes")
	return p.SendPriority(ctx, msg, s.priority)
}
203217
func (s *sender) send(sm *sendable) {
	if err := s.Error(); err != nil {
		sm.unsent(err)
		return
	}
	bytes, err := s.session.connection.mc.Encode(sm.m, nil)
	close(sm.sent) // Safe to re-use sm.m now
	if err != nil {
		sm.unsent(err)
		return
	}
	d, err := s.pLink.SendMessageBytes(bytes)
	if err != nil {
		sm.unsent(err)
		return
	}
	if s.SndSettle() == SndSettled || (s.SndSettle() == SndMixed && sm.ack == nil) {
		d.Settle()                                // Pre-settled
		Outcome{Accepted, nil, sm.v}.send(sm.ack) // Assume accepted
	} else {
		// Register with handler to receive the remote outcome
		s.handler().sent[d] = sm
	}
}