Method EnqueueMessageToDLQ

common/persistence/sql/queue.go:163–190

(
	ctx context.Context,
	blob *commonpb.DataBlob,
)

}

func (q *sqlQueue) EnqueueMessageToDLQ(
	ctx context.Context,
	blob *commonpb.DataBlob,
) (int64, error) {
	var lastMessageID int64
	err := q.txExecute(ctx, "EnqueueMessageToDLQ", func(tx sqlplugin.Tx) error {
		var err error
		lastMessageID, err = tx.GetLastEnqueuedMessageIDForUpdate(ctx, q.getDLQTypeFromQueueType())
		switch err {
		case nil:
			_, err = tx.InsertIntoMessages(ctx, []sqlplugin.QueueMessageRow{
				newQueueRow(q.getDLQTypeFromQueueType(), lastMessageID+1, blob),
			})
			return err
		case sql.ErrNoRows:
			_, err = tx.InsertIntoMessages(ctx, []sqlplugin.QueueMessageRow{
				newQueueRow(q.getDLQTypeFromQueueType(), persistence.EmptyQueueMessageID+1, blob),
			})
			return err
		default:
			return fmt.Errorf("failed to get last enqueued message id from DLQ: %v", err)
		}
	})
	if err != nil {
		return persistence.EmptyQueueMessageID, serviceerror.NewUnavailable(err.Error())
	}
	return lastMessageID + 1, nil
}

func (q *sqlQueue) ReadMessagesFromDLQ(
	ctx context.Context,