// Send the batch of messages to the next handler in the pipeline
975512
func (b *Batch) Send(cb func()) {
	if b.Deflate {
		b.Writer.(*zlib.Writer).Flush()
	} else {
		b.Writer.(*bufio.Writer).Flush()
	}

	b.Message.PushPayload(b.Buf.Bytes())
	cb()
}
245496
func (stream *Stream) sendMessage(request *orderer.StepRequest) {
	start := time.Now()
	var err error
	defer func() {
		if !stream.Logger.IsEnabledFor(zap.DebugLevel) {
			return
		}
		var result string
		if err != nil {
			result = fmt.Sprintf("but failed due to %s", err.Error())
		}
		stream.Logger.Debugf("Send of %s to %s(%s) took %v %s", requestAsString(request),
			stream.NodeName, stream.Endpoint, time.Since(start), result)
	}()

	f := func() (*orderer.StepResponse, error) {
		startSend := time.Now()
		stream.expCheck.checkExpiration(startSend, stream.Channel)
		err := stream.Cluster_StepClient.Send(request)
		stream.metrics.reportMsgSendTime(stream.Endpoint, stream.Channel, time.Since(startSend))
		return nil, err
	}

	_, err = stream.operateWithTimeout(f)
}
607008
func (ep *websocketPeer) Send(msg Message) error {
	b, err := ep.serializer.Serialize(msg)
	if err != nil {
		return err
	}

	ep.mu.Lock()
	writeError := ep.conn.WriteMessage(ep.payloadType, b)
	ep.mu.Unlock()

	return writeError
}
653199
func (s *trackingSender) Send(records []logfwd.Record) error {
	if err := s.SendCloser.Send(records); err != nil {
		return errors.Trace(err)
	}
	if err := s.tracker.setLastSent(records); err != nil {
		return errors.Trace(err)
	}
	return nil
}
133174
func (c *HoneycombClient) SendBatch(batch Batch) error {
	if len(batch) == 0 {
		// Nothing to send
		return nil
	}

	errs := []string{}
	for i := 0; i < len(batch); i += maxBatchSize {
		offset := i + maxBatchSize
		if offset > len(batch) {
			offset = len(batch)
		}
		if err := c.sendBatch(batch[i:offset]); err != nil {
			errs = append(errs, err.Error())
		}
	}

	if len(errs) > 0 {
		return errors.New(strings.Join(errs, "\n"))
	}

	return nil
}
43126
func (i *inotify) send(esch <-chan []*event) {
	for es := range esch {
		for _, e := range i.transform(es) {
			if e != nil {
				i.c <- e
			}
		}
	}
	i.wg.Done()
}
1038576
func (ch *Channel) send(msg message) (err error) {
	// If the channel is closed, use Channel.sendClosed()
	if atomic.LoadInt32(&ch.closed) == 1 {
		return ch.sendClosed(msg)
	}

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

	return f
}
337887
func (d *pcXTDriver) send(codes []string) {
	d.buffer = append(d.buffer, codes)
}
1366119
func (f *PeerEventsFilter) SentMessages() *PeerEventsFilter {
	f.eventType = simulations.EventTypeMsg
	b := false
	f.msgReceive = &b
	return f
}
566024
func (logger *StreamLogger) Send(message interface{}) {
	logger.mu.Lock()
	defer logger.mu.Unlock()

	for ch, name := range logger.subscribed {
		select {
		case ch <- message:
			deliveredCount.Add([]string{logger.name, name}, 1)
		default:
			deliveryDropCount.Add([]string{logger.name, name}, 1)
		}
	}
	sendCount.Add(logger.name, 1)
}
195674
func (JSONRPC2) Batch(arg BatchArg, replies *[]*json.RawMessage) (err error) {
	cli, srv := net.Pipe()
	defer cli.Close()
	go arg.srv.ServeCodec(NewServerCodecContext(arg.Context(), srv, arg.srv))

	replyc := make(chan *json.RawMessage, len(arg.reqs))
	donec := make(chan struct{}, 1)

	go func() {
		dec := json.NewDecoder(cli)
		*replies = make([]*json.RawMessage, 0, len(arg.reqs))
		for reply := range replyc {
			if reply != nil {
				*replies = append(*replies, reply)
			} else {
				*replies = append(*replies, new(json.RawMessage))
				if dec.Decode((*replies)[len(*replies)-1]) != nil {
					(*replies)[len(*replies)-1] = &jErrRequest
				}
			}
		}
		donec <- struct{}{}
	}()

	var testreq serverRequest
	for _, req := range arg.reqs {
		if req == nil || json.Unmarshal(*req, &testreq) != nil {
			replyc <- &jErrRequest
		} else {
			if testreq.ID != nil {
				replyc <- nil
			}
			if _, err = cli.Write(append(*req, '\n')); err != nil {
				break
			}
		}
	}

	close(replyc)
	<-donec
	return err
}
796605
func (client Client) Send(records []logfwd.Record) error {
	for _, rec := range records {
		msg, err := messageFromRecord(rec)
		if err != nil {
			return errors.Trace(err)
		}
		if err := client.Sender.Send(msg); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}
1263690
func (bot TgBot) SendMessage(cid int, text string, parsemode *ParseModeT, dwp *bool, rtmid *int, rm *ReplyMarkupInt) ResultWithMessage {
	var pm *string = nil
	if parsemode != nil {
		pmt := parsemode.String()
		pm = &pmt
	}
	payload := QuerySendMessage{cid, text, pm, dwp, rtmid, rm}
	return bot.SendMessageQuery(payload)
}
1273886
func SendMessage(serverID, to, method, message string) error {
	buffer := []byte(message)
	return SendtBytes(serverID, to, method, buffer)
}
605522
func (d *logDrain) Send(message string) error {
	if d.muted {
		return nil
	}
	d.mutex.Lock()
	defer d.mutex.Unlock()
	conn, err := d.getConnection(false)
	if err != nil {
		return err
	}
	_, err = fmt.Fprintln(conn, message)
	if err != nil {
		// Try again with a new connection in case the issue was a broken pipe
		conn, err = d.getConnection(true)
		if err != nil {
			return err
		}
		_, err = fmt.Fprintln(conn, message)
		if err != nil {
			return err
		}
	}
	return nil
}
250452
func (d *DockerMonitor) sendRequestToQueue(r *events.Message) {

	key0 := uint64(256203161)
	key1 := uint64(982451653)

	key := d.getHashKey(r)
	h := siphash.Hash(key0, key1, []byte(key))

	d.eventnotifications[int(h%uint64(d.numberOfQueues))] <- r
}