// sender pops messages from a queue and sends them.
359281
func (c *connection) sender(sender electron.Sender) {
	q := c.broker.queues.Get(sender.Source())
	for {
		if sender.Error() != nil {
			debugf("%v closed: %v", sender, sender.Error())
			return
		}
		select {

		case m := <-q:
			debugf("%v: sent %v", sender, m.Body())
			sm := sentMessage{m, q}
			c.broker.sent <- sm                    // Record sent message
			sender.SendAsync(m, c.broker.acks, sm) // Receive outcome on c.broker.acks with Value sm

		case <-sender.Done(): // break if sender is closed
			break
		}
	}
}
1299796
func (p *pullMediatorImpl) SendRes(items []string, context interface{}, nonce uint64) {
	items2return := []*proto.Envelope{}
	p.RLock()
	defer p.RUnlock()
	for _, item := range items {
		if msg, exists := p.itemID2Msg[item]; exists {
			items2return = append(items2return, msg.Envelope)
		}
	}
	returnedUpdate := &proto.GossipMessage{
		Channel: p.config.Channel,
		Tag:     p.config.Tag,
		Nonce:   0,
		Content: &proto.GossipMessage_DataUpdate{
			DataUpdate: &proto.DataUpdate{
				MsgType: p.config.MsgType,
				Nonce:   nonce,
				Data:    items2return,
			},
		},
	}
	remotePeer := context.(proto.ReceivedMessage).GetConnectionInfo()
	p.logger.Debug("Sending", len(returnedUpdate.GetDataUpdate().Data), p.config.MsgType, "items to", remotePeer)
	context.(proto.ReceivedMessage).Respond(returnedUpdate)
}
103875
func (s *BounceAction) SetSender(v string) *BounceAction {
	s.Sender = &v
	return s
}
769731
func (c *Client) SendMessages(m []models.Message, handler func(ack string) error) error {
	_, err := c.conn.SendRequest("msg.send", m, func(ctx *neptulon.ResCtx) error {
		var ack string
		if err := ctx.Result(&ack); err != nil {
			return fmt.Errorf("client: msg.send: error reading response: %v", err)
		}
		return handler(ack)
	})

	if err != nil {
		return fmt.Errorf("client: msg.send: error sending request: %v", err)
	}

	return nil
}
913212
func (c *Conn) Send(req Message) (*Message, error) {
	err := Transmit(c.conn, nil, req)
	if err != nil {
		return nil, err
	}

	if !req.IsConfirmable() {
		return nil, nil
	}

	rv, err := Receive(c.conn, c.buf)
	if err != nil {
		return nil, err
	}

	return &rv, nil
}
889424
func DefaultSender(logger std.Logger) Sender {
	return func(w http.ResponseWriter, r *http.Request, sendable jsh.Sendable) {
		sendableError, isType := sendable.(jsh.ErrorType)
		if isType && sendableError.StatusCode() >= 500 {
			logger.Printf("Returning ISE: %s\n", sendableError.Error())
		}

		sendError := jsh.Send(w, r, sendable)
		if sendError != nil && sendError.Status >= 500 {
			logger.Printf("Error sending response: %s\n", sendError.Error())
		}
	}
}
771096
func (c *conn) SendMessages(messages []Message) error {
	var buf []byte
	for _, m := range messages {
		b, err := m.MarshalBinary()
		if err != nil {
			return err
		}

		buf = append(buf, b...)
	}

	addr := &unix.SockaddrNetlink{
		Family: unix.AF_NETLINK,
	}

	return os.NewSyscallError("sendmsg", c.s.Sendmsg(buf, nil, addr, 0))
}
388707
func (rtm *RTM) SendMessage(msg *OutgoingMessage) {
	if msg == nil {
		rtm.Debugln("Error: Attempted to SendMessage(nil)")
		return
	}

	rtm.outgoingMessages <- *msg
}
267396
func (c *Client) queue(msg message) {
	c.once.Do(c.startLoop)
	msg.setMessageId(c.uid())
	msg.setTimestamp(timestamp(c.now()))
	c.msgs <- msg
}
958087
func pinger(conn *websocket.Conn, done chan struct{}) {
	ticker := time.NewTicker(pingPeriod)
	defer ticker.Stop()

	for { // blocking loop with select to wait for stimulation.
		select {
		case <-ticker.C:
			if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait)); err != nil {
				err := conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseGoingAway, err.Error()), time.Time{})
				if err != nil {
					log.Println("[ERROR] WriteControl failed: ", err)
				}
				return
			}
		case <-done:
			return // clean up this routine.
		}
	}
}
124663
func (s *MessageService) SendSMS(from, to, body string) (*Message, *Response, error) {
	return s.Send(from, to, MessageParams{Body: body})
}
238668
func NewStringAppender(buffer *bytes.Buffer) send.Sender {
	s, _ := send.NewStreamLogger("", stringAppender{buffer}, send.LevelInfo{Default: level.Debug, Threshold: level.Debug})
	return s
}
1309997
func (s *Server) SendSync(m Message) error {
	var eg errgroup.Group
	msg, err := s.serializer.Marshal(m)
	if err != nil {
		return fmt.Errorf("marshaling message: %v", err)
	}
	msg = append([]byte{getMessageType(m)}, msg...)

	for _, node := range s.cluster.Nodes() {
		node := node
		// Don't forward the message to ourselves.
		if s.uri == node.URI {
			continue
		}

		eg.Go(func() error {
			return s.defaultClient.SendMessage(context.Background(), &node.URI, msg)
		})
	}

	return eg.Wait()
}
966246
func SendMessage(tel, msg string) (uint64, error) {
	omsg := &outgoingMessage{
		tel: tel,
		msg: msg,
	}
	return sendMessage(omsg)
}
735861
func (c *Connection) sendTicker() {
	c.hbd.sc = 0
	ticker := time.NewTicker(time.Duration(c.hbd.sti))
	defer ticker.Stop()
hbSend:
	for {
		select {
		case <-ticker.C:
			c.log("HeartBeat Send data")
			// Send a heartbeat
			f := Frame{"\n", Headers{}, NULLBUFF} // Heartbeat frame
			r := make(chan error)
			if e := c.writeWireData(wiredata{f, r}); e != nil {
				c.Hbsf = true
				break hbSend
			}
			e := <-r
			//
			c.hbd.sdl.Lock()
			if e != nil {
				fmt.Printf("Heartbeat Send Failure: %v\n", e)
				c.Hbsf = true
			} else {
				c.Hbsf = false
				c.hbd.sc++
			}
			c.hbd.sdl.Unlock()
			//
		case _ = <-c.hbd.ssd:
			break hbSend
		case _ = <-c.ssdc:
			break hbSend
		} // End of select
	} // End of for
	c.log("Heartbeat Send Ends", time.Now())
	return
}
565360
func (s *server) Transporter() Transporter {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return s.transporter
}
1205454
func (m serverSender) Send(msg *irc.Message) error {
	level.Debug(m.logger()).Log("action", "send", "message", msg.String())
	return m.writer.Encode(msg)
}