// RequeueMessage requeues a message based on `time.Duration`, ie:
//
// `timeoutMs` == 0 - requeue a message immediately
// `timeoutMs`  > 0 - asynchronously wait for the specified timeout
//     and requeue a message (aka "deferred requeue")
//
190765
func (c *Channel) RequeueMessage(clientID int64, id MessageID, timeout time.Duration) error {
	// remove from inflight first
	msg, err := c.popInFlightMessage(clientID, id)
	if err != nil {
		return err
	}
	c.removeFromInFlightPQ(msg)
	atomic.AddUint64(&c.requeueCount, 1)

	if timeout == 0 {
		c.exitMutex.RLock()
		if c.Exiting() {
			c.exitMutex.RUnlock()
			return errors.New("exiting")
		}
		err := c.put(msg)
		c.exitMutex.RUnlock()
		return err
	}

	// deferred requeue
	return c.StartDeferredTimeout(msg, timeout)
}
630487
func rebaseMessage(cls error, err error) string {
	desc := errDesc(err)
	clss := cls.Error()
	if desc == clss {
		return ""
	}

	return strings.TrimSuffix(desc, ": "+clss)
}
1158851
func WriteMessage(w io.Writer, m *Message) error {
	_, err := w.Write(m.raw)
	return err
}
607022
func (cb *RedisCeleryBroker) GetCeleryMessage() (*CeleryMessage, error) {
	conn := cb.Get()
	defer conn.Close()
	messageJSON, err := conn.Do("BLPOP", cb.queueName, "1")
	if err != nil {
		return nil, err
	}
	if messageJSON == nil {
		return nil, fmt.Errorf("null message received from redis")
	}
	messageList := messageJSON.([]interface{})
	if string(messageList[0].([]byte)) != "celery" {
		return nil, fmt.Errorf("not a celery message: %v", messageList[0])
	}
	var message CeleryMessage
	if err := json.Unmarshal(messageList[1].([]byte), &message); err != nil {
		return nil, err
	}
	return &message, nil
}
1236076
func (router *Metadata) Enqueue(msg *core.Message) error {
	metadata := msg.TryGetMetadata()
	if metadata == nil {
		return router.Broadcast.Enqueue(msg)
	}

	streamName := metadata.GetValueString(router.key)
	targetID := core.GetStreamID(streamName)
	if targetID == core.InvalidStreamID {
		return router.Broadcast.Enqueue(msg)
	}

	targetRouter := core.StreamRegistry.GetRouterOrFallback(targetID)

	msg.SetStreamID(targetID)
	return core.Route(msg, targetRouter)
}
1242174
func (r *messageRing) Dequeue() (*Message, error) {
	r.mu.Lock()
	for len(r.queue) == 0 && !r.closed {
		r.wait.Wait()
	}

	if r.closed {
		r.mu.Unlock()
		return nil, errClosed
	}

	msg := r.queue[0]
	r.queue = r.queue[1:]
	r.sizeBytes -= int64(len(msg.Line))
	r.mu.Unlock()
	return msg, nil
}
752837
func (l *Limiter) GetMessage() string {
	l.RLock()
	defer l.RUnlock()
	return l.message
}
204131
func (s *Subscription) NextMsg(timeout time.Duration) (*Msg, error) {
	if s == nil {
		return nil, ErrBadSubscription
	}

	s.mu.Lock()
	err := s.validateNextMsgState()
	if err != nil {
		s.mu.Unlock()
		return nil, err
	}

	// snapshot
	mch := s.mch
	s.mu.Unlock()

	var ok bool
	var msg *Msg

	// If something is available right away, let's optimize that case.
	select {
	case msg, ok = <-mch:
		if !ok {
			return nil, ErrConnectionClosed
		}
		if err := s.processNextMsgDelivered(msg); err != nil {
			return nil, err
		} else {
			return msg, nil
		}
	default:
	}

	// If we are here a message was not immediately available, so lets loop
	// with a timeout.

	t := globalTimerPool.Get(timeout)
	defer globalTimerPool.Put(t)

	select {
	case msg, ok = <-mch:
		if !ok {
			return nil, ErrConnectionClosed
		}
		if err := s.processNextMsgDelivered(msg); err != nil {
			return nil, err
		}
	case <-t.C:
		return nil, ErrTimeout
	}

	return msg, nil
}
763941
func (c *Connection) writeMessage(msg []byte) (err error) {
	// Avoid the retry loop in os.File.Write.
	n, err := syscall.Write(int(c.dev.Fd()), msg)
	if err != nil {
		return
	}

	if n != len(msg) {
		err = fmt.Errorf("Wrote %d bytes; expected %d", n, len(msg))
		return
	}

	return
}
761878
func (p *MockPeer) ReadNextMessage() ([]byte, error) {
	select {
	case b := <-p.IncomingMsgs:
		return b, nil
	case <-p.readDeadline:
		return nil, fmt.Errorf("read timeout expired")
	case <-p.RemoteQuit:
		return nil, fmt.Errorf("remote closed connected")
	case <-p.Quit:
		return nil, fmt.Errorf("connection closed")
	}
}
1024941
func (w *Writer) WriteMessage(m *Message) error {
	return w.Write(m.String())
}
468206
func (b *Bot) SendMessage(target string, message string, sender *User) {
	message = b.executeFilterCommands(&FilterCmd{
		Target:  target,
		Message: message,
		User:    sender})
	if message == "" {
		return
	}

	select {
	case b.msgsToSend <- responseMessage{target, message, sender}:
	default:
		b.errored("Failed to queue message to send.", errors.New("Too busy"))
	}
}
582641
func (c *Client) QueueMsg(msg *Message) error {
	if msg.Timestamp == nil {
		curTime := time.Now()
		msg.Timestamp = &curTime
	}

	c.msgChan <- msg
	return nil
}
1298847
func (l *Logtap) WriteMessage(msg Message) {
	group := sync.WaitGroup{}
	for _, drain := range l.drains {
		group.Add(1)
		go func(myDrain drainChannels) {
			select {
			case <-myDrain.done:
			case myDrain.send <- msg:
			}
			group.Done()
		}(drain)
	}
	group.Wait()
}
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)
}
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
}
138253
func PublishMessage(config *Config, msg *packet.Message, timeout time.Duration) error {
	// create client
	client := New()

	// connect to broker
	future, err := client.Connect(config)
	if err != nil {
		return err
	}

	// wait on future
	err = future.Wait(timeout)
	if err != nil {
		return err
	}

	// publish message
	publishFuture, err := client.PublishMessage(msg)
	if err != nil {
		return err
	}

	// wait on future
	err = publishFuture.Wait(timeout)
	if err != nil {
		return err
	}

	// disconnect
	err = client.Disconnect()
	if err != nil {
		return err
	}

	return nil
}