/*
Ack acknowledges a delivery by its delivery tag when having been consumed with
Channel.Consume or Channel.Get.

Ack acknowledges all message received prior to the delivery tag when multiple
is true.

See also Delivery.Ack
*/
552766
func (ch *Channel) Ack(tag uint64, multiple bool) error {
	ch.m.Lock()
	defer ch.m.Unlock()

	return ch.send(&basicAck{
		DeliveryTag: tag,
		Multiple:    multiple,
	})
}
158071
func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8, syncing bool) error {
	var msg interface{}

	spanName := "send.chunk.delivery"

	//we send different types of messages if delivery is for syncing or retrievals,
	//even if handling and content of the message are the same,
	//because swap accounting decides which messages need accounting based on the message type
	if syncing {
		msg = &ChunkDeliveryMsgSyncing{
			Addr:  chunk.Address(),
			SData: chunk.Data(),
		}
		spanName += ".syncing"
	} else {
		msg = &ChunkDeliveryMsgRetrieval{
			Addr:  chunk.Address(),
			SData: chunk.Data(),
		}
		spanName += ".retrieval"
	}

	ctx = context.WithValue(ctx, "stream_send_tag", nil)
	return p.SendPriority(ctx, msg, priority)
}
318372
func (c *TagCache) AddTag(ctx context.Context, pin common.Pin, tag string) error {
	if err := common.ValidatePin(pin, common.AnyHash); err != nil {
		return err
	}
	if err := common.ValidateInstanceTag(tag); err != nil {
		return err
	}

	c.lock.Lock()
	defer c.lock.Unlock()

	if c.addedTags == nil {
		c.addedTags = make(map[tagKey]*messages.TagCache_Entry, 1)
	}

	// 'Save' will merge this into 'c.cache' before dumping to disk.
	c.addedTags[makeTagKey(pin.PackageName, tag)] = &messages.TagCache_Entry{
		Service:    c.service,
		Package:    pin.PackageName,
		Tag:        tag,
		InstanceId: pin.InstanceID,
	}

	return nil
}
125532
func (s *StanServer) processAckMsg(m *nats.Msg) {
	ack := &pb.Ack{}
	if ack.Unmarshal(m.Data) != nil {
		if s.processCtrlMsg(m) {
			return
		}
	}
	c := s.channels.get(ack.Subject)
	if c == nil {
		s.log.Errorf("Unable to process ack seq=%d, channel %s not found", ack.Sequence, ack.Subject)
		return
	}
	sub := c.ss.LookupByAckInbox(m.Subject)
	if sub == nil {
		return
	}
	s.processAck(c, sub, ack.Sequence, true)
}
1152630
func (subs *consumers) add(tag string, consumer chan Delivery) {
	subs.Lock()
	defer subs.Unlock()

	if prev, found := subs.chans[tag]; found {
		close(prev)
	}

	in := make(chan *Delivery)
	subs.chans[tag] = in

	subs.Add(1)
	go subs.buffer(in, consumer)
}
1228235
func (b *AMQPCeleryBroker) GetTaskMessage() (*TaskMessage, error) {
	select {
	case delivery := <-b.consumingChannel:
		deliveryAck(delivery)
		var taskMessage TaskMessage
		if err := json.Unmarshal(delivery.Body, &taskMessage); err != nil {
			return nil, err
		}
		return &taskMessage, nil
	default:
		return nil, fmt.Errorf("consuming channel is empty")
	}
}
528549
func (s *Sentry) Tag(key, value string) {
	s.Lock()
	s.client.Tags[key] = value
	s.Unlock()
}
1004979
func (ch *channel) ackRequest(ok bool) error {
	if !ch.decided {
		return errUndecided
	}

	var msg interface{}
	if !ok {
		msg = channelRequestFailureMsg{
			PeersId: ch.remoteId,
		}
	} else {
		msg = channelRequestSuccessMsg{
			PeersId: ch.remoteId,
		}
	}
	return ch.sendMessage(msg)
}
89851
func (s *Session) ChannelMessageAck(channelID, messageID, lastToken string) (st *Ack, err error) {

	body, err := s.RequestWithBucketID("POST", EndpointChannelMessageAck(channelID, messageID), &Ack{Token: lastToken}, EndpointChannelMessageAck(channelID, ""))
	if err != nil {
		return
	}

	err = unmarshal(body, &st)
	return
}
67187
func (f *fetcher) deliver(ctx context.Context, ch Chunk) {
	f.deliverOnce.Do(func() {
		f.chunk = ch
		// closing the deliveredC channel will terminate ongoing requests
		close(f.deliveredC)
		log.Trace("n.getFetcher close deliveredC", "ref", ch.Address())
	})
}
372646
func (rm *retrieveManager) deliver(peer distPeer, msg *Msg) error {
	rm.lock.RLock()
	req, ok := rm.sentReqs[msg.ReqID]
	rm.lock.RUnlock()

	if ok {
		return req.deliver(peer, msg)
	}
	return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
}
1102870
func (s *Service) AddOnWebhookDeliveryInfo(ctx context.Context, addOnIdentity string, appWebhookDeliveryIdentity string) (*AddOnWebhookDeliveryInfoResult, error) {
	var addOnWebhookDelivery AddOnWebhookDeliveryInfoResult
	return &addOnWebhookDelivery, s.Get(ctx, &addOnWebhookDelivery, fmt.Sprintf("/addons/%v/webhook-deliveries/%v", addOnIdentity, appWebhookDeliveryIdentity), nil, nil)
}
1004785
func (b *EvalBroker) Nack(evalID, token string) error {
	b.l.Lock()
	defer b.l.Unlock()

	// Always delete the requeued evaluation since the Nack means the requeue is
	// invalid.
	delete(b.requeue, token)

	// Lookup the unack'd eval
	unack, ok := b.unack[evalID]
	if !ok {
		return fmt.Errorf("Evaluation ID not found")
	}
	if unack.Token != token {
		return fmt.Errorf("Token does not match for Evaluation ID")
	}

	// Stop the timer, doesn't matter if we've missed it
	unack.NackTimer.Stop()

	// Cleanup
	delete(b.unack, evalID)

	// Update the stats
	b.stats.TotalUnacked -= 1
	bySched := b.stats.ByScheduler[unack.Eval.Type]
	bySched.Unacked -= 1

	// Check if we've hit the delivery limit, and re-enqueue
	// in the failedQueue
	if dequeues := b.evals[evalID]; dequeues >= b.deliveryLimit {
		b.enqueueLocked(unack.Eval, failedQueue)
	} else {
		e := unack.Eval
		e.Wait = b.nackReenqueueDelay(e, dequeues)

		// See if there should be a delay before re-enqueuing
		if e.Wait > 0 {
			b.processWaitingEnqueue(e)
		} else {
			b.enqueueLocked(e, e.Type)
		}
	}

	return nil
}
126578
func (ch *Channel) dispatch(msg message) {
	switch m := msg.(type) {
	case *channelClose:
		// lock before sending connection.close-ok
		// to avoid unexpected interleaving with basic.publish frames if
		// publishing is happening concurrently
		ch.m.Lock()
		ch.send(&channelCloseOk{})
		ch.m.Unlock()
		ch.connection.closeChannel(ch, newError(m.ReplyCode, m.ReplyText))

	case *channelFlow:
		ch.notifyM.RLock()
		for _, c := range ch.flows {
			c <- m.Active
		}
		ch.notifyM.RUnlock()
		ch.send(&channelFlowOk{Active: m.Active})

	case *basicCancel:
		ch.notifyM.RLock()
		for _, c := range ch.cancels {
			c <- m.ConsumerTag
		}
		ch.notifyM.RUnlock()
		ch.consumers.cancel(m.ConsumerTag)

	case *basicReturn:
		ret := newReturn(*m)
		ch.notifyM.RLock()
		for _, c := range ch.returns {
			c <- *ret
		}
		ch.notifyM.RUnlock()

	case *basicAck:
		if ch.confirming {
			if m.Multiple {
				ch.confirms.Multiple(Confirmation{m.DeliveryTag, true})
			} else {
				ch.confirms.One(Confirmation{m.DeliveryTag, true})
			}
		}

	case *basicNack:
		if ch.confirming {
			if m.Multiple {
				ch.confirms.Multiple(Confirmation{m.DeliveryTag, false})
			} else {
				ch.confirms.One(Confirmation{m.DeliveryTag, false})
			}
		}

	case *basicDeliver:
		ch.consumers.send(m.ConsumerTag, newDelivery(ch, m))
		// TODO log failed consumer and close channel, this can happen when
		// deliveries are in flight and a no-wait cancel has happened

	default:
		ch.rpc <- msg
	}
}
745579
func consumeQueue(c *amqp.Channel, queue string, wg *sync.WaitGroup, done chan struct{}, out chan client.MessageSet) error {
	defer func() {
		log.With("queue", queue).Infoln("consuming complete")
		wg.Done()
	}()
	deliveries, err := c.Consume(queue, "transporter", false, false, false, false, nil)
	if err != nil {
		return err
	}
	for {
		select {
		case <-done:
			return nil
		case msg := <-deliveries:
			var result map[string]interface{}
			if jerr := json.NewDecoder(bytes.NewReader(msg.Body)).Decode(&result); jerr != nil {
				log.Errorf("unable to decode message to JSON, %s", jerr)
				continue
			}
			out <- client.MessageSet{
				Msg:  message.From(ops.Insert, queue, result),
				Mode: commitlog.Sync,
			}
			msg.Ack(false)
		}
	}
}
946373
func Ack() error {
	client, err := NewClientForPath(os.Getenv("EINHORN_SOCK_PATH"))
	if err != nil {
		return err
	}

	defer client.Close()

	return client.SendRequest(&ClientAckRequest{
		Command: "worker:ack",
		Pid:     os.Getpid(),
	})
}
283254
func (s *AuthService) ackMessages(c context.Context, subscription string, ackIDs []string) error {
	if len(ackIDs) == 0 {
		return nil
	}
	request := struct {
		AckIDs []string `json:"ackIds"`
	}{ackIDs}
	req := internal.Request{
		Method: "POST",
		URL:    s.pubSubURL(subscription + ":acknowledge"),
		Scopes: oauthScopes,
		Body:   &request,
	}
	return req.Do(c)
}