Skip to content

Remove sqlite internal message counter #16

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
reed-adams wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Remove sqlite internal message counter #16

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 63 additions & 63 deletions LegacyTestProject/packages.config
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,66 +1,66 @@
<?xml version="1.0" encoding="utf-8"?>
<packages>
<package id="Azure.Core" version="1.44.1" targetFramework="net472" />
<package id="Azure.Core.Amqp" version="1.3.1" targetFramework="net472" />
<package id="Azure.Messaging.ServiceBus" version="7.18.2" targetFramework="net472" />
<package id="Elastic.Clients.Elasticsearch" version="8.15.10" targetFramework="net472" />
<package id="Elastic.Transport" version="0.4.26" targetFramework="net472" />
<package id="KM.MessageQueue" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Azure.Topic" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Database.ElasticSearch" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Database.Sqlite" version="9.0.1" targetFramework="net472" />
<package id="KM.MessageQueue.FileSystem.Disk" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.ObjectToJsonObject" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.ObjectToJsonString" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.Specialized.Composition" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.StringToBytes" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Mqtt" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Specialized.Forwarder" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Azure.Amqp" version="2.6.9" targetFramework="net472" />
<package id="Microsoft.Bcl.AsyncInterfaces" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Bcl.HashCode" version="6.0.0" targetFramework="net472" />
<package id="Microsoft.CSharp" version="4.7.0" targetFramework="net472" />
<package id="Microsoft.Data.Sqlite.Core" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.DotNet.PlatformAbstractions" version="3.1.6" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Abstractions" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Analyzers" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Relational" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Sqlite" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Sqlite.Core" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.Extensions.Caching.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Caching.Memory" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Configuration" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Configuration.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Configuration.Binder" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.DependencyInjection" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.DependencyInjection.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.DependencyModel" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Logging" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Logging.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Options" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Primitives" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.NETFramework.ReferenceAssemblies" version="1.0.3" targetFramework="net472" developmentDependency="true" />
<package id="Microsoft.NETFramework.ReferenceAssemblies.net472" version="1.0.3" targetFramework="net472" developmentDependency="true" />
<package id="MQTTnet" version="4.3.7.1207" targetFramework="net472" />
<package id="Newtonsoft.Json" version="13.0.3" targetFramework="net472" />
<package id="SQLitePCLRaw.bundle_e_sqlite3" version="2.1.10" targetFramework="net472" />
<package id="SQLitePCLRaw.core" version="2.1.10" targetFramework="net472" />
<package id="SQLitePCLRaw.lib.e_sqlite3" version="2.1.10" targetFramework="net472" />
<package id="SQLitePCLRaw.provider.dynamic_cdecl" version="2.1.10" targetFramework="net472" />
<package id="System.Buffers" version="4.6.0" targetFramework="net472" />
<package id="System.ClientModel" version="1.2.1" targetFramework="net472" />
<package id="System.Collections.Immutable" version="9.0.0" targetFramework="net472" />
<package id="System.ComponentModel.Annotations" version="5.0.0" targetFramework="net472" />
<package id="System.Diagnostics.DiagnosticSource" version="9.0.0" targetFramework="net472" />
<package id="System.IO.Pipelines" version="9.0.0" targetFramework="net472" />
<package id="System.Memory" version="4.6.0" targetFramework="net472" />
<package id="System.Memory.Data" version="9.0.0" targetFramework="net472" />
<package id="System.Numerics.Vectors" version="4.6.0" targetFramework="net472" />
<package id="System.Runtime.CompilerServices.Unsafe" version="6.1.0" targetFramework="net472" />
<package id="System.Runtime.InteropServices.RuntimeInformation" version="4.3.0" targetFramework="net472" />
<package id="System.Text.Encodings.Web" version="9.0.0" targetFramework="net472" />
<package id="System.Text.Json" version="9.0.0" targetFramework="net472" />
<package id="System.Threading.Tasks.Extensions" version="4.6.0" targetFramework="net472" />
<package id="System.ValueTuple" version="4.5.0" targetFramework="net472" />
<package id="Azure.Core" version="1.44.1" targetFramework="net472" />
<package id="Azure.Core.Amqp" version="1.3.1" targetFramework="net472" />
<package id="Azure.Messaging.ServiceBus" version="7.18.2" targetFramework="net472" />
<package id="Elastic.Clients.Elasticsearch" version="8.15.10" targetFramework="net472" />
<package id="Elastic.Transport" version="0.4.26" targetFramework="net472" />
<package id="KM.MessageQueue" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Azure.Topic" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Database.ElasticSearch" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Database.Sqlite" version="9.0.1" targetFramework="net472" />
<package id="KM.MessageQueue.FileSystem.Disk" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.ObjectToJsonObject" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.ObjectToJsonString" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.Specialized.Composition" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Formatters.StringToBytes" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Mqtt" version="9.0.0" targetFramework="net472" />
<package id="KM.MessageQueue.Specialized.Forwarder" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Azure.Amqp" version="2.6.9" targetFramework="net472" />
<package id="Microsoft.Bcl.AsyncInterfaces" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Bcl.HashCode" version="6.0.0" targetFramework="net472" />
<package id="Microsoft.CSharp" version="4.7.0" targetFramework="net472" />
<package id="Microsoft.Data.Sqlite.Core" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.DotNet.PlatformAbstractions" version="3.1.6" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Abstractions" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Analyzers" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Relational" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Sqlite" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.EntityFrameworkCore.Sqlite.Core" version="3.1.32" targetFramework="net472" />
<package id="Microsoft.Extensions.Caching.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Caching.Memory" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Configuration" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Configuration.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Configuration.Binder" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.DependencyInjection" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.DependencyInjection.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.DependencyModel" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Logging" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Logging.Abstractions" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Options" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.Extensions.Primitives" version="9.0.0" targetFramework="net472" />
<package id="Microsoft.NETFramework.ReferenceAssemblies" version="1.0.3" targetFramework="net472" developmentDependency="true" />
<package id="Microsoft.NETFramework.ReferenceAssemblies.net472" version="1.0.3" targetFramework="net472" developmentDependency="true" />
<package id="MQTTnet" version="4.3.7.1207" targetFramework="net472" />
<package id="Newtonsoft.Json" version="13.0.3" targetFramework="net472" />
<package id="SQLitePCLRaw.bundle_e_sqlite3" version="2.1.10" targetFramework="net472" />
<package id="SQLitePCLRaw.core" version="2.1.10" targetFramework="net472" />
<package id="SQLitePCLRaw.lib.e_sqlite3" version="2.1.12" targetFramework="net472" />
<package id="SQLitePCLRaw.provider.dynamic_cdecl" version="2.1.10" targetFramework="net472" />
<package id="System.Buffers" version="4.6.0" targetFramework="net472" />
<package id="System.ClientModel" version="1.2.1" targetFramework="net472" />
<package id="System.Collections.Immutable" version="9.0.0" targetFramework="net472" />
<package id="System.ComponentModel.Annotations" version="5.0.0" targetFramework="net472" />
<package id="System.Diagnostics.DiagnosticSource" version="9.0.0" targetFramework="net472" />
<package id="System.IO.Pipelines" version="9.0.0" targetFramework="net472" />
<package id="System.Memory" version="4.6.0" targetFramework="net472" />
<package id="System.Memory.Data" version="9.0.0" targetFramework="net472" />
<package id="System.Numerics.Vectors" version="4.6.0" targetFramework="net472" />
<package id="System.Runtime.CompilerServices.Unsafe" version="6.1.0" targetFramework="net472" />
<package id="System.Runtime.InteropServices.RuntimeInformation" version="4.3.0" targetFramework="net472" />
<package id="System.Text.Encodings.Web" version="9.0.0" targetFramework="net472" />
<package id="System.Text.Json" version="9.0.0" targetFramework="net472" />
<package id="System.Threading.Tasks.Extensions" version="4.6.0" targetFramework="net472" />
<package id="System.ValueTuple" version="4.5.0" targetFramework="net472" />
</packages>
39 changes: 17 additions & 22 deletions MessageQueue.Database.Sqlite/SqliteMessageQueue.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,11 +42,12 @@ public SqliteMessageQueue(ILogger<SqliteMessageQueue<TMessage>> logger, IOptions
}

using var dbContext = GetDatabaseContext();
(_currentQueueSize, _sequenceNumber) = GetCurrentStats(dbContext);
var currentQueueSize = GetCurrentQueueSize(dbContext);
_sequenceNumber = GetCurrentSequenceNumber(dbContext);

Name = opts.Name ?? nameof(SqliteMessageQueue<TMessage>);

_logger.LogTrace($"{Name} initialized with {_currentQueueSize} stored messages");
_logger.LogTrace($"{Name} initialized with {currentQueueSize} stored messages");
}

private readonly SqliteMessageQueueOptions<TMessage> _options;
Expand All @@ -59,7 +60,6 @@ public SqliteMessageQueue(ILogger<SqliteMessageQueue<TMessage>> logger, IOptions
private readonly IMessageFormatter<TMessage, string> _messageFormatter;
private readonly int? _maxQueueSize;

private int _currentQueueSize;
private long _sequenceNumber;

private readonly CancellationTokenSource _cancellationSource = new();
Expand All @@ -73,16 +73,20 @@ public SqliteMessageQueue(ILogger<SqliteMessageQueue<TMessage>> logger, IOptions
public int MaxReadCount { get; }


private static (int QueueSize, long SequenceNumber) GetCurrentStats(SqliteDatabaseContext dbContext)
private static int GetCurrentQueueSize(SqliteDatabaseContext dbContext)
{
var queueSize = dbContext.SqliteQueueMessages.Count();
return queueSize;
}

private static long GetCurrentSequenceNumber(SqliteDatabaseContext dbContext)
{
var sequenceNumber = dbContext.SqliteQueueMessages
.Select(x => x.SequenceNumber)
.Max()
?? 0L;

return (queueSize, sequenceNumber);
return sequenceNumber;
}


Expand Down Expand Up @@ -164,9 +168,11 @@ public async Task PostManyMessagesAsync(IEnumerable<(TMessage message, MessageAt
throw new InvalidOperationException($"Message count exceeds max write count of {MaxWriteCount}");
}

using var dbContext = GetDatabaseContext();
if (_maxQueueSize is { } maxQueueSize)
{
if (_currentQueueSize >= maxQueueSize)

if (GetCurrentQueueSize(dbContext) >= maxQueueSize)
{
_logger.LogError($"{Name} {nameof(PostManyMessagesAsync)} exceeded maximum queue size of {{MaxQueueSize}}", maxQueueSize);
throw new InvalidOperationException($"{Name} {nameof(PostManyMessagesAsync)} exceeded maximum queue size of {maxQueueSize}");
Expand Down Expand Up @@ -199,13 +205,8 @@ public async Task PostManyMessagesAsync(IEnumerable<(TMessage message, MessageAt
var messageString = messageCount == 1 ? sqlMessages[0].Body : $"{messageCount} messages";
_logger.LogTrace($"{Name} {nameof(PostManyMessagesAsync)} posting to store, Message: {{Message}}", messageString);

using (var dbContext = GetDatabaseContext())
{
dbContext.SqliteQueueMessages.AddRange(sqlMessages);
_ = await dbContext.SaveChangesAsync(linkedCancellation.Token).ConfigureAwait(false);
}

_ = Interlocked.Add(ref _currentQueueSize, sqlMessages.Count);
dbContext.SqliteQueueMessages.AddRange(sqlMessages);
_ = await dbContext.SaveChangesAsync(linkedCancellation.Token).ConfigureAwait(false);
}

public Task<IMessageQueueReader<TMessage>> GetReaderAsync(MessageQueueReaderOptions<TMessage> options, CancellationToken cancellationToken)
Expand Down Expand Up @@ -249,12 +250,6 @@ public Task<IMessageQueueReader<TMessage>> GetReaderAsync(MessageQueueReaderOpti
{
linkedCancellation.Token.ThrowIfCancellationRequested();

if (_currentQueueSize == 0)
{
shouldWait = true;
continue;
}

using var dbContext = GetDatabaseContext();

var dbMessages = await dbContext.SqliteQueueMessages
Expand Down Expand Up @@ -310,7 +305,9 @@ public Task<IMessageQueueReader<TMessage>> GetReaderAsync(MessageQueueReaderOpti
{
try
{
(_currentQueueSize, _sequenceNumber) = GetCurrentStats(dbContext);
using var recoveryDbContext = GetDatabaseContext();
var recoverySequenceNumber = GetCurrentSequenceNumber(recoveryDbContext);
_ = Interlocked.Exchange(ref _sequenceNumber, recoverySequenceNumber);
}
catch (Exception inner)
{
Expand All @@ -319,8 +316,6 @@ public Task<IMessageQueueReader<TMessage>> GetReaderAsync(MessageQueueReaderOpti

throw;
}

_ = Interlocked.Add(ref _currentQueueSize, -itemsToRemove.Count);
}

return (completionResult, result);
Expand Down
2 changes: 1 addition & 1 deletion MessageQueue.Http/MessageQueue.Http.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@
</ItemGroup>

<ItemGroup Condition="'$(TargetFramework)' != 'net8' and '$(TargetFramework)' != 'net9'">
<PackageReference Include="Microsoft.AspNetCore.WebUtilities" Version="2.2.0" />
<PackageReference Include="Microsoft.AspNetCore.WebUtilities" Version="2.3.11" />
<PackageReference Include="System.Text.Encodings.Web" Version="9.0.0" />
</ItemGroup>

Expand Down