// send sends a list of transactions to at most a given number of peers at
// once, never resending any particular transaction to the same peer twice
978465
func (self *LesTxRelay) send(txs types.Transactions, count int) {
	sendTo := make(map[*peer]types.Transactions)

	self.peerStartPos++ // rotate the starting position of the peer list
	if self.peerStartPos >= len(self.peerList) {
		self.peerStartPos = 0
	}

	for _, tx := range txs {
		hash := tx.Hash()
		ltr, ok := self.txSent[hash]
		if !ok {
			ltr = &ltrInfo{
				tx:     tx,
				sentTo: make(map[*peer]struct{}),
			}
			self.txSent[hash] = ltr
			self.txPending[hash] = struct{}{}
		}

		if len(self.peerList) > 0 {
			cnt := count
			pos := self.peerStartPos
			for {
				peer := self.peerList[pos]
				if _, ok := ltr.sentTo[peer]; !ok {
					sendTo[peer] = append(sendTo[peer], tx)
					ltr.sentTo[peer] = struct{}{}
					cnt--
				}
				if cnt == 0 {
					break // sent it to the desired number of peers
				}
				pos++
				if pos == len(self.peerList) {
					pos = 0
				}
				if pos == self.peerStartPos {
					break // tried all available peers
				}
			}
		}
	}

	for p, list := range sendTo {
		pp := p
		ll := list
		enc, _ := rlp.EncodeToBytes(ll)

		reqID := genReqID()
		rq := &distReq{
			getCost: func(dp distPeer) uint64 {
				peer := dp.(*peer)
				return peer.GetTxRelayCost(len(ll), len(enc))
			},
			canSend: func(dp distPeer) bool {
				return !dp.(*peer).isOnlyAnnounce && dp.(*peer) == pp
			},
			request: func(dp distPeer) func() {
				peer := dp.(*peer)
				cost := peer.GetTxRelayCost(len(ll), len(enc))
				peer.fcServer.QueuedRequest(reqID, cost)
				return func() { peer.SendTxs(reqID, cost, enc) }
			},
		}
		self.reqDist.queue(rq)
	}
}
1006791
func (s *Server) send() error {
	failures := 0
	for {
		err := s.sendBatch(s.batch)
		if err == nil {
			break
		}

		failures++
		log.Errorf("Send failed %d times: %v", failures, err)
		if failures >= s.config.MaxFailures {
			return fmt.Errorf("send: %d errors, giving up", failures)
		}

		time.Sleep(s.config.RetryDelay)
	}

	s.batch = s.batch[:0]
	return nil
}
527965
func (s *ServerStream) Send(message proto.Message, options ...yarpc.StreamOption) error {
	return writeToStream(context.Background(), s.stream, message)
}
1075690
func (c *Client) Send(packet Packet) error {
	addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", c.ip, c.port))
	if err != nil {
		return err
	}
	conn, err := net.DialUDP("udp", c.laddr, addr)
	if err != nil {
		return err
	}
	defer conn.Close()

	data, err := packet.MarshalBinary()
	if err != nil {
		return err
	}

	if _, err = conn.Write(data); err != nil {
		return err
	}
	return nil
}
764432
func (c *Conn) send(ctx context.Context, call *rpcCall) (err error) {
	defer func() {
		// Give precedence for user cancellation.
		select {
		case <-ctx.Done():
			err = ctx.Err()
		default:
		}
	}()

	c.mu.Lock()
	if c.closed {
		c.mu.Unlock()
		return ErrConnClosing
	}
	c.reqSeq++
	reqID := c.reqSeq
	c.pending[reqID] = call
	c.mu.Unlock()

	done := make(chan error, 1)
	go func() {
		c.reqMu.Lock()
		c.req.ID = reqID
		c.req.Method = call.Method
		c.req.Args = call.Args

		err := c.codec.WriteRequest(&c.req)

		c.req.Args = nil
		c.reqMu.Unlock()
		done <- err
	}()

	// Abort on user or connection cancellation.
	select {
	case <-ctx.Done():
		err = ctx.Err()
	case err = <-done:
	}

	if err != nil {
		// Remove reference on error, avoid
		// unnecessary work in recv.
		c.mu.Lock()
		delete(c.pending, reqID)
		c.mu.Unlock()
		return err
	}

	return nil
}
423514
func (s *Socket) Send(parts [][]byte) (err error) {
	for _, part := range parts[:len(parts)-1] {
		if err = s.SendPart(part, true); err != nil {
			return
		}
	}
	return s.SendPart(parts[len(parts)-1], false)
}
549435
func (conn *UDPConn) Send(p Packet) error {
	_, err := conn.Write(p.Bytes())
	return err
}
828545
func (t *SyncTransport) Send(body map[string]interface{}) error {
	return t.doSend(body, t.RetryAttempts)
}
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
}
1347512
func (c *DeliverConnection) Send(seekInfo *ab.SeekInfo) error {
	if c.Closed() {
		return errors.New("connection is closed")
	}

	logger.Debugf("Sending %#v", seekInfo)

	env, err := c.createSignedEnvelope(seekInfo)
	if err != nil {
		return err
	}

	return c.deliverStream().Send(env)
}
608114
func (c *Client) Send(data []byte) ([]byte, error) {
	if !c.running.Load() {
		panic(fmt.Sprintf("process.Client for %q has been closed", c.cmd.Path))
	}

	return c.client.Send(data)
}
950993
func (c *Transport) Send(ctx context.Context, messageParts ...[]byte) (<-chan Message, error) {
	conn, err := c.getConnection(ctx)
	if err != nil {
		return nil, driver.WithStack(err)
	}
	result, err := conn.Send(ctx, messageParts...)
	if err != nil {
		return nil, driver.WithStack(err)
	}
	return result, nil
}
377415
func (s *Socket) Send(t int, data []byte) (err error) {
	return s.Conn.WriteMessage(t, data)
}
765745
func (c *Client) Send(message string) error {
	_, err := c.conn.Write([]byte(message))
	return err
}
968516
func (ns *NatsRPCClient) Send(topic string, data []byte) error {
	if !ns.running {
		return constants.ErrRPCClientNotInitialized
	}
	return ns.conn.Publish(topic, data)
}
258430
func (list *TargetList) Send(event Event, targetIDs ...TargetID) <-chan TargetIDErr {
	errCh := make(chan TargetIDErr)

	go func() {
		defer close(errCh)

		var wg sync.WaitGroup
		for _, id := range targetIDs {
			list.RLock()
			target, ok := list.targets[id]
			list.RUnlock()
			if ok {
				wg.Add(1)
				go func(id TargetID, target Target) {
					defer wg.Done()
					if err := target.Save(event); err != nil {
						errCh <- TargetIDErr{
							ID:  id,
							Err: err,
						}
					}
				}(id, target)
			}
		}
		wg.Wait()
	}()

	return errCh
}
1170451
func (f *Flow) Send(pkts ...packet.Generic) *Flow {
	f.add(action{
		kind:    actionSend,
		packets: pkts,
	})

	return f
}