Method Dequeue

queue/sqlite/sqlite.go:390–483

(tenantId int64, queueName string, numToDequeue int, requeueIn int)

}

func (q *SQLiteQueue) Dequeue(tenantId int64, queueName string, numToDequeue int, requeueIn int) ([]*models.Message, error) {
	queue, err := q.getQueue(tenantId, queueName)
	if err != nil {
		return nil, err
	}

	// Queue is "paused"
	if queue.RateLimit == 0 {
		return nil, nil
	}

	visibilityTimeout := queue.VisibilityTimeout
	if requeueIn > -1 {
		visibilityTimeout = requeueIn
	}

	now := time.Now().UTC().Unix()

	q.Mu.Lock()
	defer q.Mu.Unlock()

	maxToDequeue, err := q.calculateRateLimit(queue, now, numToDequeue)
	if err != nil {
		return nil, err
	}

	var messages []Message

	res := q.DBG.Preload("KV").Where(
		"deliver_at <= ? AND delivered_at <= ? AND (tries < max_tries OR max_tries = -1) AND tenant_id = ? AND queue_id = ?",
		now, now, tenantId, queue.ID).
		Limit(maxToDequeue).
		Find(&messages)

	if res.Error != nil {
		return nil, err
	}

	if len(messages) == 0 {
		return nil, nil
	}

	rc := make([]*models.Message, len(messages))

	for i, message := range messages {
		rc[i] = message.ToModel()
	}

	messageIDs := make([]int64, len(rc))
	for i, message := range rc {
		messageIDs[i] = message.ID
	}

	err = q.DBG.Transaction(func(tx *gorm.DB) error {
		res = tx.Model(&Message{}).Where("tenant_id = ? AND queue_id = ? AND id in ?", tenantId, queue.ID, messageIDs).
			UpdateColumns(map[string]any{
				"tries":        gorm.Expr("tries+1"),
				"delivered_at": now,