public Task<IFetchedMessage> FetchNextMessageAsync()
public Task<IFetchedMessage> FetchNextMessageAsync()
{
{
//Last execute statement(FOR UPDATE to fix dirty read) :
//SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
//START TRANSACTION;
//SELECT MessageId,MessageType FROM `{_prefix}.queue` LIMIT 1 FOR UPDATE;
//DELETE FROM `{_prefix}.queue` LIMIT 1;
//COMMIT;
var sql = $@"
var sql = $@"
SELECT `MessageId`,`MessageType` FROM `{_prefix}.queue` LIMIT 1 FOR UPDATE;
DELETE FROM `{_prefix}.queue` LIMIT 1;";
SELECT @MessageId:=`MessageId` as MessageId, @MessageType:=`MessageType` as MessageType FROM `{_prefix}.queue` LIMIT 1;
DELETE FROM `{_prefix}.queue` where `MessageId` = @MessageId AND `MessageType`=@MessageType;";
return FetchNextMessageCoreAsync(sql);
return FetchNextMessageCoreAsync(sql);
}
}
public async Task<CapPublishedMessage> GetNextPublishedMessageToBeEnqueuedAsync()
public async Task<CapPublishedMessage> GetNextPublishedMessageToBeEnqueuedAsync()
{
{
var sql = $"SELECT * FROM `{_prefix}.published` WHERE `StatusName` = '{StatusName.Scheduled}' LIMIT 1;";
var sql = $@"
UPDATE `{_prefix}.published` SET Id=LAST_INSERT_ID(Id),ExpiresAt='{DateTimeMaxValue}' WHERE ExpiresAt IS NULL AND `StatusName` = '{StatusName.Scheduled}' LIMIT 1;
SELECT * FROM `{_prefix}.published` WHERE Id=LAST_INSERT_ID();";
using (var connection = new MySqlConnection(Options.ConnectionString))
using (var connection = new MySqlConnection(Options.ConnectionString))
public async Task<CapReceivedMessage> GetNextReceivedMessageToBeEnqueuedAsync()
public async Task<CapReceivedMessage> GetNextReceivedMessageToBeEnqueuedAsync()
{
{
var sql = $"SELECT * FROM `{_prefix}.received` WHERE `StatusName` = '{StatusName.Scheduled}' LIMIT 1;";
var sql = $@"
UPDATE `{_prefix}.received` SET Id=LAST_INSERT_ID(Id),ExpiresAt='{DateTimeMaxValue}' WHERE ExpiresAt IS NULL AND `StatusName` = '{StatusName.Scheduled}' LIMIT 1;
SELECT * FROM `{_prefix}.received` WHERE Id=LAST_INSERT_ID();";
using (var connection = new MySqlConnection(Options.ConnectionString))
using (var connection = new MySqlConnection(Options.ConnectionString))