diff --git a/src/Namotion.Interceptor.Mqtt.SampleClient/Namotion.Interceptor.Mqtt.SampleClient.csproj b/src/Namotion.Interceptor.Mqtt.SampleClient/Namotion.Interceptor.Mqtt.SampleClient.csproj
new file mode 100644
index 00000000..fd7c5108
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.SampleClient/Namotion.Interceptor.Mqtt.SampleClient.csproj
@@ -0,0 +1,22 @@
+
+
+
+ Exe
+ net9.0
+ enable
+ enable
+ false
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/Namotion.Interceptor.Mqtt.SampleClient/Program.cs b/src/Namotion.Interceptor.Mqtt.SampleClient/Program.cs
new file mode 100644
index 00000000..7c52fa43
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.SampleClient/Program.cs
@@ -0,0 +1,41 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Namotion.Interceptor;
+using Namotion.Interceptor.Hosting;
+using Namotion.Interceptor.Mqtt;
+using Namotion.Interceptor.Mqtt.Client;
+using Namotion.Interceptor.Registry;
+using Namotion.Interceptor.SamplesModel;
+using Namotion.Interceptor.SamplesModel.Workers;
+using Namotion.Interceptor.Sources.Paths;
+using Namotion.Interceptor.Tracking;
+using Namotion.Interceptor.Validation;
+
+var builder = Host.CreateApplicationBuilder(args);
+
+var context = InterceptorSubjectContext
+ .Create()
+ .WithFullPropertyTracking()
+ .WithRegistry()
+ .WithParents()
+ .WithLifecycle()
+ .WithDataAnnotationValidation()
+ .WithHostedServices(builder.Services);
+
+var root = Root.CreateWithPersons(context);
+context.AddService(root);
+
+builder.Services.AddSingleton(root);
+builder.Services.AddHostedService();
+builder.Services.AddMqttSubjectClient(
+ _ => root,
+ _ => new MqttClientConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("mqtt", "/", null)
+ });
+
+using var performanceProfiler = new PerformanceProfiler(context, "Client");
+var host = builder.Build();
+host.Run();
diff --git a/src/Namotion.Interceptor.Mqtt.SampleServer/Namotion.Interceptor.Mqtt.SampleServer.csproj b/src/Namotion.Interceptor.Mqtt.SampleServer/Namotion.Interceptor.Mqtt.SampleServer.csproj
new file mode 100644
index 00000000..fd7c5108
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.SampleServer/Namotion.Interceptor.Mqtt.SampleServer.csproj
@@ -0,0 +1,22 @@
+
+
+
+ Exe
+ net9.0
+ enable
+ enable
+ false
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/Namotion.Interceptor.Mqtt.SampleServer/Program.cs b/src/Namotion.Interceptor.Mqtt.SampleServer/Program.cs
new file mode 100644
index 00000000..c291f8c0
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.SampleServer/Program.cs
@@ -0,0 +1,40 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Namotion.Interceptor;
+using Namotion.Interceptor.Hosting;
+using Namotion.Interceptor.Mqtt.Server;
+using Namotion.Interceptor.Registry;
+using Namotion.Interceptor.SamplesModel;
+using Namotion.Interceptor.SamplesModel.Workers;
+using Namotion.Interceptor.Sources.Paths;
+using Namotion.Interceptor.Tracking;
+using Namotion.Interceptor.Validation;
+
+var builder = Host.CreateApplicationBuilder(args);
+
+var context = InterceptorSubjectContext
+ .Create()
+ .WithFullPropertyTracking()
+ .WithRegistry()
+ .WithParents()
+ .WithLifecycle()
+ .WithDataAnnotationValidation()
+ .WithHostedServices(builder.Services);
+
+var root = Root.CreateWithPersons(context);
+context.AddService(root);
+
+builder.Services.AddSingleton(root);
+builder.Services.AddHostedService();
+builder.Services.AddMqttSubjectServer(
+ _ => root,
+ _ => new MqttServerConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("mqtt", "/")
+ });
+
+using var performanceProfiler = new PerformanceProfiler(context, "Server");
+var host = builder.Build();
+host.Run();
diff --git a/src/Namotion.Interceptor.Mqtt.Tests/JsonMqttValueConverterTests.cs b/src/Namotion.Interceptor.Mqtt.Tests/JsonMqttValueConverterTests.cs
new file mode 100644
index 00000000..ce0e8540
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.Tests/JsonMqttValueConverterTests.cs
@@ -0,0 +1,144 @@
+using System.Buffers;
+using System.Text;
+using System.Text.Json;
+using Xunit;
+
+namespace Namotion.Interceptor.Mqtt.Tests;
+
+public class JsonMqttValueConverterTests
+{
+ private readonly JsonMqttValueConverter _converter = new();
+
+ [Fact]
+ public void Serialize_NullValue_ReturnsJsonNull()
+ {
+ // Act
+ var result = _converter.Serialize(null, typeof(object));
+
+ // Assert
+ Assert.Equal("null", Encoding.UTF8.GetString(result));
+ }
+
+ [Theory]
+ [InlineData(42, typeof(int), "42")]
+ [InlineData(3.14, typeof(double), "3.14")]
+ [InlineData(true, typeof(bool), "true")]
+ [InlineData(false, typeof(bool), "false")]
+ [InlineData("test", typeof(string), "\"test\"")]
+ public void Serialize_PrimitiveTypes_ReturnsCorrectJson(object value, Type type, string expected)
+ {
+ // Act
+ var result = _converter.Serialize(value, type);
+
+ // Assert
+ Assert.Equal(expected, Encoding.UTF8.GetString(result));
+ }
+
+ [Fact]
+ public void Serialize_ComplexObject_ReturnsCorrectJson()
+ {
+ // Arrange
+ var obj = new TestObject { Name = "Test", Value = 123 };
+
+ // Act
+ var result = _converter.Serialize(obj, typeof(TestObject));
+
+ // Assert
+ var json = Encoding.UTF8.GetString(result);
+ Assert.Contains("\"name\":", json); // camelCase
+ Assert.Contains("\"Test\"", json);
+ Assert.Contains("\"value\":", json);
+ Assert.Contains("123", json);
+ }
+
+ [Fact]
+ public void Deserialize_EmptyPayload_ReturnsNull()
+ {
+ // Arrange
+ var payload = new ReadOnlySequence(Array.Empty());
+
+ // Act
+ var result = _converter.Deserialize(payload, typeof(string));
+
+ // Assert
+ Assert.Null(result);
+ }
+
+ [Fact]
+ public void Deserialize_JsonNull_ReturnsNull()
+ {
+ // Arrange
+ var payload = new ReadOnlySequence(Encoding.UTF8.GetBytes("null"));
+
+ // Act
+ var result = _converter.Deserialize(payload, typeof(string));
+
+ // Assert
+ Assert.Null(result);
+ }
+
+ [Theory]
+ [InlineData("42", typeof(int), 42)]
+ [InlineData("3.14", typeof(double), 3.14)]
+ [InlineData("true", typeof(bool), true)]
+ [InlineData("\"test\"", typeof(string), "test")]
+ public void Deserialize_PrimitiveTypes_ReturnsCorrectValue(string json, Type type, object expected)
+ {
+ // Arrange
+ var payload = new ReadOnlySequence(Encoding.UTF8.GetBytes(json));
+
+ // Act
+ var result = _converter.Deserialize(payload, type);
+
+ // Assert
+ Assert.Equal(expected, result);
+ }
+
+ [Fact]
+ public void Deserialize_ComplexObject_ReturnsCorrectObject()
+ {
+ // Arrange
+ var json = "{\"name\":\"Test\",\"value\":123}";
+ var payload = new ReadOnlySequence(Encoding.UTF8.GetBytes(json));
+
+ // Act
+ var result = _converter.Deserialize(payload, typeof(TestObject)) as TestObject;
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal("Test", result.Name);
+ Assert.Equal(123, result.Value);
+ }
+
+ [Fact]
+ public void Deserialize_InvalidJson_ThrowsJsonException()
+ {
+ // Arrange
+ var payload = new ReadOnlySequence(Encoding.UTF8.GetBytes("invalid json"));
+
+ // Act & Assert
+ Assert.Throws(() => _converter.Deserialize(payload, typeof(int)));
+ }
+
+ [Fact]
+ public void RoundTrip_PreservesValue()
+ {
+ // Arrange
+ var original = new TestObject { Name = "RoundTrip", Value = 456 };
+
+ // Act
+ var serialized = _converter.Serialize(original, typeof(TestObject));
+ var deserialized = _converter.Deserialize(new ReadOnlySequence(serialized), typeof(TestObject)) as TestObject;
+
+ // Assert
+ Assert.NotNull(deserialized);
+ Assert.Equal(original.Name, deserialized.Name);
+ Assert.Equal(original.Value, deserialized.Value);
+ }
+
+ private class TestObject
+ {
+ public string Name { get; set; } = string.Empty;
+ public int Value { get; set; }
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt.Tests/MqttClientConfigurationTests.cs b/src/Namotion.Interceptor.Mqtt.Tests/MqttClientConfigurationTests.cs
new file mode 100644
index 00000000..fee7f3ef
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.Tests/MqttClientConfigurationTests.cs
@@ -0,0 +1,140 @@
+using Namotion.Interceptor.Mqtt.Client;
+using Namotion.Interceptor.Sources.Paths;
+using Xunit;
+
+namespace Namotion.Interceptor.Mqtt.Tests;
+
+public class MqttClientConfigurationTests
+{
+ [Fact]
+ public void Validate_ValidConfiguration_DoesNotThrow()
+ {
+ // Arrange
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Act & Assert
+ config.Validate(); // Should not throw
+ }
+
+ [Fact]
+ public void Validate_NullBrokerHost_ThrowsArgumentException()
+ {
+ // Arrange
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = null!,
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Fact]
+ public void Validate_EmptyBrokerHost_ThrowsArgumentException()
+ {
+ // Arrange
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = "",
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Theory]
+ [InlineData(0)]
+ [InlineData(-1)]
+ [InlineData(65536)]
+ public void Validate_InvalidBrokerPort_ThrowsArgumentException(int port)
+ {
+ // Arrange
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = port,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Fact]
+ public void Validate_NullPathProvider_ThrowsArgumentException()
+ {
+ // Arrange
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = null!
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Fact]
+ public void Validate_NegativeWriteRetryQueueSize_ThrowsArgumentException()
+ {
+ // Arrange
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null),
+ WriteRetryQueueSize = -1
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Fact]
+ public void Validate_MaxReconnectDelayLessThanReconnectDelay_ThrowsArgumentException()
+ {
+ // Arrange
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null),
+ ReconnectDelay = TimeSpan.FromSeconds(10),
+ MaximumReconnectDelay = TimeSpan.FromSeconds(5)
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Fact]
+ public void DefaultValues_AreCorrect()
+ {
+ // Arrange & Act
+ var config = new MqttClientConfiguration
+ {
+ BrokerHost = "localhost",
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Assert
+ Assert.Equal(1883, config.BrokerPort);
+ Assert.False(config.UseTls);
+ Assert.True(config.CleanSession);
+ Assert.Equal(TimeSpan.FromSeconds(15), config.KeepAliveInterval);
+ Assert.Equal(TimeSpan.FromSeconds(10), config.ConnectTimeout);
+ Assert.Equal(TimeSpan.FromSeconds(2), config.ReconnectDelay);
+ Assert.Equal(TimeSpan.FromMinutes(1), config.MaximumReconnectDelay);
+ Assert.NotNull(config.ValueConverter);
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt.Tests/MqttServerConfigurationTests.cs b/src/Namotion.Interceptor.Mqtt.Tests/MqttServerConfigurationTests.cs
new file mode 100644
index 00000000..70b33be5
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.Tests/MqttServerConfigurationTests.cs
@@ -0,0 +1,69 @@
+using Namotion.Interceptor.Mqtt.Server;
+using Namotion.Interceptor.Sources.Paths;
+using Xunit;
+
+namespace Namotion.Interceptor.Mqtt.Tests;
+
+public class MqttServerConfigurationTests
+{
+ [Fact]
+ public void Validate_ValidConfiguration_DoesNotThrow()
+ {
+ // Arrange
+ var config = new MqttServerConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Act & Assert
+ config.Validate(); // Should not throw
+ }
+
+ [Fact]
+ public void Validate_NullBrokerHost_ThrowsArgumentException()
+ {
+ // Arrange
+ var config = new MqttServerConfiguration
+ {
+ BrokerHost = null!,
+ BrokerPort = 1883,
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Fact]
+ public void Validate_NullPathProvider_ThrowsArgumentException()
+ {
+ // Arrange
+ var config = new MqttServerConfiguration
+ {
+ BrokerHost = "localhost",
+ BrokerPort = 1883,
+ PathProvider = null!
+ };
+
+ // Act & Assert
+ Assert.Throws(() => config.Validate());
+ }
+
+ [Fact]
+ public void DefaultValues_AreCorrect()
+ {
+ // Arrange & Act
+ var config = new MqttServerConfiguration
+ {
+ BrokerHost = "localhost",
+ PathProvider = new AttributeBasedSourcePathProvider("test", "/", null)
+ };
+
+ // Assert
+ Assert.Equal(1883, config.BrokerPort);
+ Assert.Equal(10000, config.MaxPendingMessagesPerClient);
+ Assert.NotNull(config.ValueConverter);
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt.Tests/Namotion.Interceptor.Mqtt.Tests.csproj b/src/Namotion.Interceptor.Mqtt.Tests/Namotion.Interceptor.Mqtt.Tests.csproj
new file mode 100644
index 00000000..fab37b05
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt.Tests/Namotion.Interceptor.Mqtt.Tests.csproj
@@ -0,0 +1,31 @@
+
+
+
+ net9.0
+ enable
+ enable
+ false
+ true
+
+
+
+
+
+
+
+ runtime; build; native; contentfiles; analyzers; buildtransitive
+ all
+
+
+ runtime; build; native; contentfiles; analyzers; buildtransitive
+ all
+
+
+
+
+
+
+
+
+
+
diff --git a/src/Namotion.Interceptor.Mqtt/Client/MqttClientConfiguration.cs b/src/Namotion.Interceptor.Mqtt/Client/MqttClientConfiguration.cs
new file mode 100644
index 00000000..1bf572ca
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/Client/MqttClientConfiguration.cs
@@ -0,0 +1,213 @@
+using System;
+using MQTTnet.Protocol;
+using Namotion.Interceptor.Sources.Paths;
+
+namespace Namotion.Interceptor.Mqtt.Client;
+
+///
+/// Configuration for MQTT client source.
+///
+public class MqttClientConfiguration
+{
+ ///
+ /// Gets or sets the MQTT broker hostname or IP address.
+ ///
+ public required string BrokerHost { get; init; }
+
+ ///
+ /// Gets or sets the MQTT broker port. Default is 1883.
+ ///
+ public int BrokerPort { get; init; } = 1883;
+
+ ///
+ /// Gets or sets the username for broker authentication.
+ ///
+ public string? Username { get; init; }
+
+ ///
+ /// Gets or sets the password for broker authentication.
+ ///
+ public string? Password { get; init; }
+
+ ///
+ /// Gets or sets whether to use TLS/SSL for the connection. Default is false.
+ ///
+ public bool UseTls { get; init; }
+
+ ///
+ /// Gets or sets the client identifier. Default is a unique GUID-based identifier.
+ ///
+ public string ClientId { get; init; } = $"Namotion_{Guid.NewGuid():N}";
+
+ ///
+ /// Gets or sets whether to use a clean session. Default is true.
+ /// When false, the broker preserves subscriptions and queued messages across reconnects.
+ ///
+ public bool CleanSession { get; init; } = true;
+
+ ///
+ /// Gets or sets the optional topic prefix. When set, all topics are prefixed with this value.
+ ///
+ public string? TopicPrefix { get; init; }
+
+ ///
+ /// Gets or sets the source path provider for property-to-topic mapping.
+ ///
+ public required ISourcePathProvider PathProvider { get; init; }
+
+ // QoS settings
+
+ ///
+ /// Gets or sets the default QoS level for publish/subscribe operations. Default is AtMostOnce (0) for high throughput.
+ ///
+ public MqttQualityOfServiceLevel DefaultQualityOfService { get; init; } = MqttQualityOfServiceLevel.AtMostOnce;
+
+ ///
+ /// Gets or sets whether to use retained messages. Default is true.
+ /// Retained messages enable initial state loading.
+ ///
+ public bool UseRetainedMessages { get; init; } = true;
+
+ ///
+ /// Gets or sets the connection timeout. Default is 10 seconds.
+ ///
+ public TimeSpan ConnectTimeout { get; init; } = TimeSpan.FromSeconds(10);
+
+ ///
+ /// Gets or sets the initial delay before attempting to reconnect. Default is 2 seconds.
+ ///
+ public TimeSpan ReconnectDelay { get; init; } = TimeSpan.FromSeconds(2);
+
+ ///
+ /// Gets or sets the maximum delay between reconnection attempts (for exponential backoff). Default is 60 seconds.
+ ///
+ public TimeSpan MaximumReconnectDelay { get; init; } = TimeSpan.FromSeconds(60);
+
+ ///
+ /// Gets or sets the keep-alive interval. Default is 15 seconds.
+ ///
+ public TimeSpan KeepAliveInterval { get; init; } = TimeSpan.FromSeconds(15);
+
+ ///
+ /// Gets or sets the interval for connection health checks. Default is 30 seconds.
+ /// Health checks use TryPingAsync to verify the connection is still alive.
+ ///
+ public TimeSpan HealthCheckInterval { get; init; } = TimeSpan.FromSeconds(30);
+
+ ///
+ /// Gets or sets the maximum number of health check iterations while reconnecting before forcing a reset.
+ /// Default is 10 iterations. Combined with HealthCheckInterval, this provides a stall timeout.
+ /// Example: 10 iterations × 30s = 5 minutes timeout for hung reconnection attempts.
+ /// Set to 0 to disable stall detection.
+ ///
+ public int ReconnectStallThreshold { get; init; } = 10;
+
+ ///
+ /// Gets or sets the number of consecutive connection failures before the circuit breaker opens.
+ /// Default is 5 failures. When open, reconnection attempts are paused for the cooldown period.
+ /// Set to 0 to disable circuit breaker.
+ ///
+ public int CircuitBreakerFailureThreshold { get; init; } = 5;
+
+ ///
+ /// Gets or sets the cooldown period after the circuit breaker opens. Default is 60 seconds.
+ /// During cooldown, no reconnection attempts are made. After cooldown, one retry is allowed.
+ ///
+ public TimeSpan CircuitBreakerCooldown { get; init; } = TimeSpan.FromSeconds(60);
+
+ ///
+ /// Gets or sets the time to buffer property changes before sending. Default is 8ms.
+ ///
+ public TimeSpan BufferTime { get; init; } = TimeSpan.FromMilliseconds(8);
+
+ ///
+ /// Gets or sets the time between retry attempts for failed writes. Default is 10 seconds.
+ ///
+ public TimeSpan RetryTime { get; init; } = TimeSpan.FromSeconds(10);
+
+ ///
+ /// Gets or sets the maximum number of writes to queue during disconnection. Default is 1000.
+ /// Set to 0 to disable write buffering.
+ ///
+ public int WriteRetryQueueSize { get; init; } = 1000;
+
+ ///
+ /// Gets or sets the value converter for serialization/deserialization. Default is JSON.
+ ///
+ public IMqttValueConverter ValueConverter { get; init; } = new JsonMqttValueConverter();
+
+ ///
+ /// Gets or sets the MQTT user property name for the source timestamp. Default is "ts".
+ /// Set to null to disable timestamp extraction.
+ ///
+ public string? SourceTimestampPropertyName { get; init; } = "ts";
+
+ ///
+ /// Gets or sets the converter function for serializing timestamps to strings.
+ /// Default converts to Unix milliseconds.
+ ///
+ public Func SourceTimestampConverter { get; init; } =
+ static timestamp => timestamp.ToUnixTimeMilliseconds().ToString();
+
+ ///
+ /// Validates the configuration and throws if invalid.
+ ///
+ /// Thrown when configuration is invalid.
+ public void Validate()
+ {
+ if (string.IsNullOrWhiteSpace(BrokerHost))
+ {
+ throw new ArgumentException("BrokerHost must be specified.", nameof(BrokerHost));
+ }
+
+ if (BrokerPort is < 1 or > 65535)
+ {
+ throw new ArgumentException($"BrokerPort must be between 1 and 65535, got: {BrokerPort}", nameof(BrokerPort));
+ }
+
+ if (PathProvider is null)
+ {
+ throw new ArgumentException("PathProvider must be specified.", nameof(PathProvider));
+ }
+
+ if (ConnectTimeout <= TimeSpan.Zero)
+ {
+ throw new ArgumentException($"ConnectTimeout must be positive, got: {ConnectTimeout}", nameof(ConnectTimeout));
+ }
+
+ if (ReconnectDelay <= TimeSpan.Zero)
+ {
+ throw new ArgumentException($"ReconnectDelay must be positive, got: {ReconnectDelay}", nameof(ReconnectDelay));
+ }
+
+ if (MaximumReconnectDelay < ReconnectDelay)
+ {
+ throw new ArgumentException($"MaxReconnectDelay must be >= ReconnectDelay, got: {MaximumReconnectDelay}", nameof(MaximumReconnectDelay));
+ }
+
+ if (WriteRetryQueueSize < 0)
+ {
+ throw new ArgumentException($"WriteRetryQueueSize must be non-negative, got: {WriteRetryQueueSize}", nameof(WriteRetryQueueSize));
+ }
+
+ if (ValueConverter is null)
+ {
+ throw new ArgumentException("ValueConverter must be specified.", nameof(ValueConverter));
+ }
+
+ if (ReconnectStallThreshold < 0)
+ {
+ throw new ArgumentException($"ReconnectStallThreshold must be non-negative, got: {ReconnectStallThreshold}", nameof(ReconnectStallThreshold));
+ }
+
+ if (CircuitBreakerFailureThreshold < 0)
+ {
+ throw new ArgumentException($"CircuitBreakerFailureThreshold must be non-negative, got: {CircuitBreakerFailureThreshold}", nameof(CircuitBreakerFailureThreshold));
+ }
+
+ if (CircuitBreakerCooldown <= TimeSpan.Zero && CircuitBreakerFailureThreshold > 0)
+ {
+ throw new ArgumentException($"CircuitBreakerCooldown must be positive when circuit breaker is enabled, got: {CircuitBreakerCooldown}", nameof(CircuitBreakerCooldown));
+ }
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt/Client/MqttConnectionLifetime.cs b/src/Namotion.Interceptor.Mqtt/Client/MqttConnectionLifetime.cs
new file mode 100644
index 00000000..96a10997
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/Client/MqttConnectionLifetime.cs
@@ -0,0 +1,32 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Namotion.Interceptor.Mqtt.Client;
+
+internal sealed class MqttConnectionLifetime : IDisposable, IAsyncDisposable
+{
+ private readonly Func _disposeAsync;
+ private int _disposed;
+
+ public MqttConnectionLifetime(Func disposeAsync)
+ {
+ _disposeAsync = disposeAsync;
+ }
+
+ public void Dispose()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 0)
+ {
+ _disposeAsync().GetAwaiter().GetResult();
+ }
+ }
+
+ public async ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 0)
+ {
+ await _disposeAsync().ConfigureAwait(false);
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/Namotion.Interceptor.Mqtt/Client/MqttConnectionMonitor.cs b/src/Namotion.Interceptor.Mqtt/Client/MqttConnectionMonitor.cs
new file mode 100644
index 00000000..9845e229
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/Client/MqttConnectionMonitor.cs
@@ -0,0 +1,244 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using MQTTnet;
+using Namotion.Interceptor.Sources.Resilience;
+
+namespace Namotion.Interceptor.Mqtt.Client;
+
+///
+/// Monitors MQTT client connection health and handles reconnection with exponential backoff.
+/// Uses a hybrid approach: events trigger immediate action, but actual reconnection happens in monitoring task.
+///
+internal sealed class MqttConnectionMonitor : IAsyncDisposable
+{
+ private readonly IMqttClient _client;
+ private readonly MqttClientConfiguration _configuration;
+ private readonly ILogger _logger;
+
+ private readonly Func _optionsBuilder;
+ private readonly Func _onReconnected;
+ private readonly Func _onDisconnected;
+
+ private readonly SemaphoreSlim _reconnectSignal = new(0, 1);
+ private readonly CircuitBreaker? _circuitBreaker;
+
+ private int _isReconnecting;
+ private int _reconnectingIterations; // Tracks health check iterations while reconnecting (for stall detection)
+ private int _disposed;
+
+ public MqttConnectionMonitor(
+ IMqttClient client,
+ MqttClientConfiguration configuration,
+ Func optionsBuilder,
+ Func onReconnected,
+ Func onDisconnected,
+ ILogger logger)
+ {
+ _client = client ?? throw new ArgumentNullException(nameof(client));
+ _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
+ _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+ _optionsBuilder = optionsBuilder ?? throw new ArgumentNullException(nameof(optionsBuilder));
+ _onReconnected = onReconnected ?? throw new ArgumentNullException(nameof(onReconnected));
+ _onDisconnected = onDisconnected ?? throw new ArgumentNullException(nameof(onDisconnected));
+
+ // Initialize circuit breaker if enabled
+ if (configuration.CircuitBreakerFailureThreshold > 0)
+ {
+ _circuitBreaker = new CircuitBreaker(
+ configuration.CircuitBreakerFailureThreshold,
+ configuration.CircuitBreakerCooldown);
+ }
+ }
+
+ ///
+ /// Signals that a reconnection is needed (called by DisconnectedAsync event handler in MqttSubjectClientSource).
+ ///
+ public void SignalReconnectNeeded()
+ {
+ try
+ {
+ // Only signal if not already signaled (semaphore maxCount is 1)
+ _reconnectSignal.Release();
+ }
+ catch (SemaphoreFullException)
+ {
+ // Already signaled, ignore
+ }
+ catch (ObjectDisposedException)
+ {
+ // Monitor is being disposed, ignore (race between disconnect event and disposal)
+ }
+ }
+
+ ///
+ /// Monitors connection health and performs reconnection with exponential backoff, circuit breaker, and stall detection.
+ /// This is a blocking method that runs until a cancellation is requested.
+ /// Uses hybrid approach: Waits for disconnect event OR periodic health check.
+ ///
+ /// Cancellation token to stop monitoring.
+ public async Task MonitorConnectionAsync(CancellationToken cancellationToken)
+ {
+ var healthCheckInterval = _configuration.HealthCheckInterval;
+ var maxDelay = _configuration.MaximumReconnectDelay;
+ var stallThreshold = _configuration.ReconnectStallThreshold;
+
+ while (!cancellationToken.IsCancellationRequested)
+ {
+ try
+ {
+ // Wait for either: Disconnect event signal OR periodic health check timeout
+ var signaled = await _reconnectSignal.WaitAsync(healthCheckInterval, cancellationToken).ConfigureAwait(false);
+ if (signaled)
+ {
+ _logger.LogWarning("MQTT disconnect event received.");
+ await _onDisconnected().ConfigureAwait(false);
+ }
+ else
+ {
+ // Periodic health check: Use TryPingAsync (recommended by MQTTnet)
+ var isHealthy = _client.IsConnected && await _client.TryPingAsync(cancellationToken).ConfigureAwait(false);
+ if (isHealthy)
+ {
+ // Connection healthy - reset stall detection counter
+ Interlocked.Exchange(ref _reconnectingIterations, 0);
+ continue;
+ }
+
+ _logger.LogWarning("MQTT health check failed.");
+ }
+
+ // Stall detection: Check if reconnection is hung
+ if (Volatile.Read(ref _isReconnecting) == 1 && stallThreshold > 0)
+ {
+ var iterations = Interlocked.Increment(ref _reconnectingIterations);
+ if (iterations > stallThreshold)
+ {
+ // Timeout: iterations × health check interval (e.g., 10 × 30s = 5 minutes)
+ _logger.LogError(
+ "Reconnection stalled after {Iterations} iterations (~{Timeout}s). Forcing reset.",
+ iterations,
+ (int)(iterations * healthCheckInterval.TotalSeconds));
+
+ // Force reset reconnection flag to allow recovery
+ Interlocked.Exchange(ref _isReconnecting, 0);
+ Interlocked.Exchange(ref _reconnectingIterations, 0);
+
+ // Reset circuit breaker to allow immediate retry
+ _circuitBreaker?.Reset();
+ }
+ }
+
+ if (!_client.IsConnected)
+ {
+ if (Interlocked.Exchange(ref _isReconnecting, 1) == 1)
+ {
+ continue; // Already reconnecting
+ }
+
+ try
+ {
+ var reconnectDelay = _configuration.ReconnectDelay;
+ while (!cancellationToken.IsCancellationRequested)
+ {
+ // Check circuit breaker
+ if (_circuitBreaker is not null && !_circuitBreaker.ShouldAttempt())
+ {
+ var cooldownRemaining = _circuitBreaker.GetCooldownRemaining();
+ _logger.LogWarning(
+ "Circuit breaker open after {TripCount} trips. Pausing reconnection attempts for {Cooldown}s.",
+ _circuitBreaker.TripCount,
+ (int)cooldownRemaining.TotalSeconds);
+
+ // Wait for cooldown period (or until cancellation)
+ await Task.Delay(cooldownRemaining, cancellationToken).ConfigureAwait(false);
+ continue;
+ }
+
+ try
+ {
+ _logger.LogInformation("Attempting to reconnect to MQTT broker in {Delay}...", reconnectDelay);
+ await Task.Delay(reconnectDelay, cancellationToken).ConfigureAwait(false);
+
+ var options = _optionsBuilder();
+ await _client.ConnectAsync(options, cancellationToken).ConfigureAwait(false);
+
+ _logger.LogInformation("Reconnected to MQTT broker successfully.");
+ await _onReconnected(cancellationToken).ConfigureAwait(false);
+
+ // Success - close circuit breaker and reset counters
+ _circuitBreaker?.RecordSuccess();
+ Interlocked.Exchange(ref _reconnectingIterations, 0);
+ break;
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ _logger.LogInformation("Reconnection cancelled due to shutdown.");
+ break;
+ }
+ catch (Exception ex)
+ {
+ var isTransient = MqttExceptionClassifier.IsTransient(ex);
+ var description = MqttExceptionClassifier.GetFailureDescription(ex);
+
+ if (isTransient)
+ {
+ _logger.LogError(ex,
+ "Failed to reconnect to MQTT broker: {Description}.",
+ description);
+ }
+ else
+ {
+ _logger.LogError(ex,
+ "Permanent connection failure detected: {Description}. " +
+ "Reconnection will be retried, but this likely requires configuration changes.",
+ description);
+ }
+
+ // Record failure in circuit breaker
+ if (_circuitBreaker is not null && _circuitBreaker.RecordFailure())
+ {
+ _logger.LogWarning(
+ "Circuit breaker tripped after {Threshold} consecutive failures. " +
+ "Pausing reconnection attempts for {Cooldown}s.",
+ _configuration.CircuitBreakerFailureThreshold,
+ (int)_configuration.CircuitBreakerCooldown.TotalSeconds);
+ }
+
+ // Exponential backoff with jitter
+ var jitter = Random.Shared.NextDouble() * 0.1 + 0.95; // 0.95 to 1.05
+ reconnectDelay = TimeSpan.FromMilliseconds(
+ Math.Min(reconnectDelay.TotalMilliseconds * 2 * jitter, maxDelay.TotalMilliseconds));
+ }
+ }
+ }
+ finally
+ {
+ Interlocked.Exchange(ref _isReconnecting, 0);
+ }
+ }
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ _logger.LogInformation("Connection monitoring cancelled due to shutdown.");
+ break;
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Error in connection monitoring.");
+ }
+ }
+ }
+
+ public ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 1)
+ {
+ return ValueTask.CompletedTask;
+ }
+
+ _reconnectSignal.Dispose();
+ return ValueTask.CompletedTask;
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt/Client/MqttExceptionClassifier.cs b/src/Namotion.Interceptor.Mqtt/Client/MqttExceptionClassifier.cs
new file mode 100644
index 00000000..29f65450
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/Client/MqttExceptionClassifier.cs
@@ -0,0 +1,80 @@
+using System;
+using System.Net.Sockets;
+using MQTTnet.Exceptions;
+
+namespace Namotion.Interceptor.Mqtt.Client;
+
+///
+/// Classifies MQTT exceptions as transient (retryable) or permanent (configuration/design-time errors).
+///
+internal static class MqttExceptionClassifier
+{
+ ///
+ /// Determines if an exception represents a transient failure that can be retried.
+ /// Returns true for transient errors (network issues, broker temporarily unavailable),
+ /// false for permanent errors (bad credentials, invalid configuration).
+ ///
+ public static bool IsTransient(Exception exception)
+ {
+ return exception switch
+ {
+ // Authentication failures - permanent
+ MqttCommunicationException { InnerException: SocketException { SocketErrorCode: SocketError.ConnectionRefused } } => false,
+
+ // Invalid configuration - permanent
+ ArgumentNullException => false,
+ ArgumentException => false,
+ InvalidOperationException ex when ex.Message.Contains("not allowed to connect", StringComparison.OrdinalIgnoreCase) => false,
+
+ // DNS resolution failures - permanent (invalid hostname)
+ SocketException { SocketErrorCode: SocketError.HostNotFound } => false,
+ SocketException { SocketErrorCode: SocketError.NoData } => false,
+
+ // TLS/Certificate failures - potentially permanent (depends on configuration)
+ MqttCommunicationException ex when ex.Message.Contains("TLS", StringComparison.OrdinalIgnoreCase) => false,
+ MqttCommunicationException ex when ex.Message.Contains("certificate", StringComparison.OrdinalIgnoreCase) => false,
+ MqttCommunicationException ex when ex.Message.Contains("authentication", StringComparison.OrdinalIgnoreCase) => false,
+
+ // All other exceptions are considered transient (network issues, timeout, etc.)
+ _ => true
+ };
+ }
+
+ ///
+ /// Gets a user-friendly description of the failure type for logging.
+ ///
+ public static string GetFailureDescription(Exception exception)
+ {
+ return exception switch
+ {
+ MqttCommunicationException { InnerException: SocketException { SocketErrorCode: SocketError.ConnectionRefused } }
+ => "Connection refused - broker may be down or authentication failed",
+
+ SocketException { SocketErrorCode: SocketError.HostNotFound }
+ => "Host not found - invalid broker hostname",
+
+ SocketException { SocketErrorCode: SocketError.NoData }
+ => "No DNS data - invalid broker hostname",
+
+ MqttCommunicationException ex when ex.Message.Contains("TLS", StringComparison.OrdinalIgnoreCase)
+ => "TLS connection failed - check certificate configuration",
+
+ MqttCommunicationException ex when ex.Message.Contains("certificate", StringComparison.OrdinalIgnoreCase)
+ => "Certificate validation failed - check TLS configuration",
+
+ MqttCommunicationException ex when ex.Message.Contains("authentication", StringComparison.OrdinalIgnoreCase)
+ => "Authentication failed - check username/password",
+
+ InvalidOperationException ex when ex.Message.Contains("not allowed to connect", StringComparison.OrdinalIgnoreCase)
+ => "Connection state error - client already connected or connecting",
+
+ SocketException socketEx
+ => $"Network error: {socketEx.SocketErrorCode}",
+
+ OperationCanceledException
+ => "Operation cancelled",
+
+ _ => exception.GetType().Name
+ };
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt/Client/MqttSubjectClientSource.cs b/src/Namotion.Interceptor.Mqtt/Client/MqttSubjectClientSource.cs
new file mode 100644
index 00000000..6c6b4e9d
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/Client/MqttSubjectClientSource.cs
@@ -0,0 +1,390 @@
+using System;
+using System.Buffers;
+using System.Collections.Concurrent;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using MQTTnet;
+using MQTTnet.Packets;
+using Namotion.Interceptor.Registry;
+using Namotion.Interceptor.Registry.Abstractions;
+using Namotion.Interceptor.Sources;
+using Namotion.Interceptor.Sources.Paths;
+using Namotion.Interceptor.Tracking.Change;
+
+namespace Namotion.Interceptor.Mqtt.Client;
+
+///
+/// MQTT client source that subscribes to an MQTT broker and synchronizes properties.
+///
+internal sealed class MqttSubjectClientSource : BackgroundService, ISubjectSource, IAsyncDisposable
+{
+ private readonly IInterceptorSubject _subject;
+ private readonly MqttClientConfiguration _configuration;
+ private readonly ILogger _logger;
+
+ private readonly MqttClientFactory _factory;
+
+ // TODO(memory): Might lead to memory leaks
+ private readonly ConcurrentDictionary _topicToProperty = new();
+ private readonly ConcurrentDictionary _propertyToTopic = new();
+
+ private IMqttClient? _client;
+ private SubjectPropertyWriter? _propertyWriter;
+ private MqttConnectionMonitor? _connectionMonitor;
+
+ private int _disposed;
+ private volatile bool _isStarted;
+
+ public MqttSubjectClientSource(
+ IInterceptorSubject subject,
+ MqttClientConfiguration configuration,
+ ILogger logger)
+ {
+ _subject = subject ?? throw new ArgumentNullException(nameof(subject));
+ _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
+ _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+
+ _factory = new MqttClientFactory();
+
+ configuration.Validate();
+ }
+
+ ///
+ public bool IsPropertyIncluded(RegisteredSubjectProperty property) =>
+ _configuration.PathProvider.IsPropertyIncluded(property);
+
+ ///
+ public int WriteBatchSize => 0; // No server-imposed limit for MQTT
+
+ ///
+ public async Task StartListeningAsync(SubjectPropertyWriter propertyWriter, CancellationToken cancellationToken)
+ {
+ _propertyWriter = propertyWriter;
+ _logger.LogInformation("Connecting to MQTT broker at {Host}:{Port}.", _configuration.BrokerHost, _configuration.BrokerPort);
+
+ _client = _factory.CreateMqttClient();
+ _client.ApplicationMessageReceivedAsync += OnMessageReceivedAsync;
+ _client.DisconnectedAsync += OnDisconnectedAsync;
+
+ await _client.ConnectAsync(GetClientOptions(), cancellationToken).ConfigureAwait(false);
+ _logger.LogInformation("Connected to MQTT broker successfully.");
+
+ await SubscribeToPropertiesAsync(cancellationToken).ConfigureAwait(false);
+
+ _connectionMonitor = new MqttConnectionMonitor(
+ _client,
+ _configuration,
+ GetClientOptions,
+ async ct => await OnReconnectedAsync(ct).ConfigureAwait(false),
+ async () =>
+ {
+ _propertyWriter?.StartBuffering();
+ await Task.CompletedTask;
+ }, _logger);
+
+ _isStarted = true;
+
+ return new MqttConnectionLifetime(async () =>
+ {
+ if (_client?.IsConnected == true)
+ {
+ await _client.DisconnectAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+ }
+ });
+ }
+
+ ///
+ public Task LoadInitialStateAsync(CancellationToken cancellationToken)
+ {
+ // Retained messages are received through the normal message handler: No separate initial load needed/possible
+ return Task.FromResult(null);
+ }
+
+ ///
+ public async ValueTask WriteChangesAsync(ReadOnlyMemory changes, CancellationToken cancellationToken)
+ {
+ var client = _client;
+ if (client is null || !client.IsConnected)
+ {
+ throw new InvalidOperationException("MQTT client is not connected.");
+ }
+
+ var length = changes.Length;
+ if (length == 0) return;
+
+ // Rent array from pool for messages
+ var messagesPool = ArrayPool.Shared;
+ var messages = messagesPool.Rent(length);
+ var messageCount = 0;
+ try
+ {
+ var changesSpan = changes.Span;
+
+ // Build all messages first
+ for (var i = 0; i < length; i++)
+ {
+ var change = changesSpan[i];
+ var property = change.Property.TryGetRegisteredProperty();
+ if (property is null || property.HasChildSubjects)
+ {
+ continue;
+ }
+
+ var topic = TryGetTopicForProperty(change.Property, property);
+ if (topic is null) continue;
+
+ byte[] payload;
+ try
+ {
+ payload = _configuration.ValueConverter.Serialize(
+ change.GetNewValue(),
+ property.Type);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Failed to serialize value for property {PropertyName}.", property.Name);
+ continue;
+ }
+
+ var message = new MqttApplicationMessage
+ {
+ Topic = topic,
+ PayloadSegment = new ArraySegment(payload),
+ QualityOfServiceLevel = _configuration.DefaultQualityOfService,
+ Retain = _configuration.UseRetainedMessages
+ };
+
+ if (_configuration.SourceTimestampPropertyName is not null)
+ {
+ message.UserProperties =
+ [
+ new MqttUserProperty(
+ _configuration.SourceTimestampPropertyName,
+ _configuration.SourceTimestampConverter(change.ChangedTimestamp))
+ ];
+ }
+
+ messages[messageCount++] = message;
+ }
+
+ // TODO(perf): Add batch API?
+ for (var i = 0; i < messageCount; i++)
+ {
+ await client.PublishAsync(messages[i], cancellationToken).ConfigureAwait(false);
+ }
+ }
+ finally
+ {
+ messagesPool.Return(messages);
+ }
+ }
+
+ private async Task SubscribeToPropertiesAsync(CancellationToken cancellationToken)
+ {
+ var registeredSubject = _subject.TryGetRegisteredSubject();
+ if (registeredSubject is null)
+ {
+ _logger.LogWarning("Subject is not registered. No MQTT subscriptions will be created.");
+ return;
+ }
+
+ var properties = registeredSubject
+ .GetAllProperties()
+ .Where(p => !p.HasChildSubjects && IsPropertyIncluded(p))
+ .ToList();
+
+ if (properties.Count == 0)
+ {
+ _logger.LogWarning("No MQTT properties found to subscribe.");
+ return;
+ }
+
+ var subscribeOptionsBuilder = _factory!.CreateSubscribeOptionsBuilder();
+
+ foreach (var property in properties)
+ {
+ var topic = TryGetTopicForProperty(property.Reference, property);
+ if (topic is null) continue;
+
+ _topicToProperty[topic] = property.Reference;
+ subscribeOptionsBuilder.WithTopicFilter(f => f
+ .WithTopic(topic)
+ .WithQualityOfServiceLevel(_configuration.DefaultQualityOfService));
+ }
+
+ await _client!.SubscribeAsync(subscribeOptionsBuilder.Build(), cancellationToken).ConfigureAwait(false);
+
+ _logger.LogInformation("Subscribed to {Count} MQTT topics.", properties.Count);
+ }
+
+ private string? TryGetTopicForProperty(PropertyReference propertyReference, RegisteredSubjectProperty property)
+ {
+ return _propertyToTopic.GetOrAdd(propertyReference, static (_, state) =>
+ {
+ var (p, pathProvider, subject, topicPrefix) = state;
+ var path = p.TryGetSourcePath(pathProvider, subject);
+ return path is null ? null : MqttHelper.BuildTopic(path, topicPrefix);
+ }, (property, _configuration.PathProvider, _subject, _configuration.TopicPrefix));
+ }
+
+ private PropertyReference? TryGetPropertyForTopic(string topic)
+ {
+ return _topicToProperty.GetOrAdd(topic, static (t, state) =>
+ {
+ var (subject, pathProvider, topicPrefix) = state;
+ var path = MqttHelper.StripTopicPrefix(t, topicPrefix);
+ var (property, _) = subject.TryGetPropertyFromSourcePath(path, pathProvider);
+ return property?.Reference;
+ }, (_subject, _configuration.PathProvider, _configuration.TopicPrefix));
+ }
+
+ private Task OnMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs e)
+ {
+ var topic = e.ApplicationMessage.Topic;
+ if (TryGetPropertyForTopic(topic) is not { } propertyReference)
+ {
+ return Task.CompletedTask;
+ }
+
+ var registeredProperty = propertyReference.TryGetRegisteredProperty();
+ if (registeredProperty is null)
+ {
+ return Task.CompletedTask;
+ }
+
+ object? value;
+ try
+ {
+ var payload = e.ApplicationMessage.Payload;
+ value = _configuration.ValueConverter.Deserialize(payload, registeredProperty.Type);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Failed to deserialize MQTT message for topic {Topic}.", topic);
+ return Task.CompletedTask;
+ }
+
+ var propertyWriter = _propertyWriter;
+ if (propertyWriter is null)
+ {
+ return Task.CompletedTask;
+ }
+
+ // Extract timestamps
+ var receivedTimestamp = DateTimeOffset.UtcNow;
+ var sourceTimestamp = MqttHelper.ExtractSourceTimestamp(
+ e.ApplicationMessage.UserProperties,
+ _configuration.SourceTimestampPropertyName) ?? receivedTimestamp;
+
+ // Use static delegate to avoid allocations on hot path
+ propertyWriter.Write(
+ (propertyReference, value, this, sourceTimestamp, receivedTimestamp),
+ static state => state.propertyReference.SetValueFromSource(state.Item3, state.sourceTimestamp, state.receivedTimestamp, state.value));
+
+ return Task.CompletedTask;
+ }
+
+ ///
+ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+ {
+ // Wait until StartListeningAsync has been called
+ while (!_isStarted && !stoppingToken.IsCancellationRequested)
+ {
+ await Task.Delay(100, stoppingToken).ConfigureAwait(false);
+ }
+
+ if (_connectionMonitor is not null)
+ {
+ await _connectionMonitor.MonitorConnectionAsync(stoppingToken).ConfigureAwait(false);
+ }
+ }
+
+ private Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs e)
+ {
+ if (Interlocked.CompareExchange(ref _disposed, 0, 0) == 1)
+ {
+ return Task.CompletedTask;
+ }
+
+ _logger.LogWarning(e.Exception, "MQTT client disconnected. Reason: {Reason}.", e.Reason);
+ _connectionMonitor?.SignalReconnectNeeded();
+
+ return Task.CompletedTask;
+ }
+
+ private async Task OnReconnectedAsync(CancellationToken cancellationToken)
+ {
+ await SubscribeToPropertiesAsync(cancellationToken).ConfigureAwait(false);
+ if (_propertyWriter is not null)
+ {
+ await _propertyWriter.CompleteInitializationAsync(cancellationToken).ConfigureAwait(false);
+ }
+ }
+
+ private MqttClientOptions GetClientOptions()
+ {
+ var options = new MqttClientOptionsBuilder()
+ .WithTcpServer(_configuration.BrokerHost, _configuration.BrokerPort)
+ .WithClientId(_configuration.ClientId)
+ .WithCleanSession(_configuration.CleanSession)
+ .WithKeepAlivePeriod(_configuration.KeepAliveInterval)
+ .WithTimeout(_configuration.ConnectTimeout);
+
+ if (_configuration.UseTls)
+ {
+ options.WithTlsOptions(o => o.UseTls());
+ }
+
+ if (_configuration.Username is not null)
+ {
+ options.WithCredentials(_configuration.Username, _configuration.Password);
+ }
+
+ return options.Build();
+ }
+
+ ///
+ public async ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 1)
+ {
+ return;
+ }
+
+ if (_connectionMonitor is not null)
+ {
+ await _connectionMonitor.DisposeAsync().ConfigureAwait(false);
+ }
+
+ var client = _client;
+ if (client is not null)
+ {
+ client.ApplicationMessageReceivedAsync -= OnMessageReceivedAsync;
+ client.DisconnectedAsync -= OnDisconnectedAsync;
+
+ if (client.IsConnected)
+ {
+ try
+ {
+ await client.DisconnectAsync().ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(ex, "Error disconnecting MQTT client.");
+ }
+ }
+
+ client.Dispose();
+ _client = null;
+ }
+
+ // Clear caches to allow GC of subject references
+ _topicToProperty.Clear();
+ _propertyToTopic.Clear();
+
+ Dispose();
+ }
+}
\ No newline at end of file
diff --git a/src/Namotion.Interceptor.Mqtt/IMqttValueConverter.cs b/src/Namotion.Interceptor.Mqtt/IMqttValueConverter.cs
new file mode 100644
index 00000000..6e7bb223
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/IMqttValueConverter.cs
@@ -0,0 +1,26 @@
+using System;
+using System.Buffers;
+
+namespace Namotion.Interceptor.Mqtt;
+
+///
+/// Converts values between CLR types and MQTT message payloads.
+///
+public interface IMqttValueConverter
+{
+ ///
+ /// Serializes a value to an MQTT message payload.
+ ///
+ /// The value to serialize.
+ /// The type of the value.
+ /// The serialized payload.
+ byte[] Serialize(object? value, Type type);
+
+ ///
+ /// Deserializes an MQTT message payload to a value.
+ ///
+ /// The payload to deserialize.
+ /// The target type.
+ /// The deserialized value.
+ object? Deserialize(ReadOnlySequence payload, Type type);
+}
diff --git a/src/Namotion.Interceptor.Mqtt/JsonMqttValueConverter.cs b/src/Namotion.Interceptor.Mqtt/JsonMqttValueConverter.cs
new file mode 100644
index 00000000..3b881f5d
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/JsonMqttValueConverter.cs
@@ -0,0 +1,86 @@
+using System;
+using System.Buffers;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace Namotion.Interceptor.Mqtt;
+
+///
+/// JSON-based MQTT value converter using System.Text.Json for high performance.
+///
+public sealed class JsonMqttValueConverter : IMqttValueConverter
+{
+ [ThreadStatic]
+ private static ArrayBufferWriter? _bufferWriter;
+
+ private readonly JsonSerializerOptions _options;
+
+ ///
+ /// Initializes a new instance with default options.
+ ///
+ public JsonMqttValueConverter()
+ : this(CreateDefaultOptions())
+ {
+ }
+
+ ///
+ /// Initializes a new instance with custom options.
+ ///
+ /// The JSON serializer options.
+ public JsonMqttValueConverter(JsonSerializerOptions options)
+ {
+ _options = options ?? throw new ArgumentNullException(nameof(options));
+ }
+
+ ///
+ public byte[] Serialize(object? value, Type type)
+ {
+ // Reuse thread-local buffer writer to avoid allocations
+ var bufferWriter = _bufferWriter ??= new ArrayBufferWriter(256);
+ bufferWriter.Clear();
+
+ using var writer = new Utf8JsonWriter(bufferWriter);
+ JsonSerializer.Serialize(writer, value, type, _options);
+
+ return bufferWriter.WrittenSpan.ToArray();
+ }
+
+ ///
+ public object? Deserialize(ReadOnlySequence payload, Type type)
+ {
+ if (payload.IsEmpty)
+ {
+ return type.IsValueType ? Activator.CreateInstance(type) : null;
+ }
+
+ // Single segment - no copy needed (most common case)
+ if (payload.IsSingleSegment)
+ {
+ return JsonSerializer.Deserialize(payload.FirstSpan, type, _options);
+ }
+
+ // Multi-segment - need to copy to contiguous buffer
+ var length = (int)payload.Length;
+ var buffer = ArrayPool.Shared.Rent(length);
+ try
+ {
+ payload.CopyTo(buffer);
+ return JsonSerializer.Deserialize(buffer.AsSpan(0, length), type, _options);
+ }
+ finally
+ {
+ ArrayPool.Shared.Return(buffer);
+ }
+ }
+
+ private static JsonSerializerOptions CreateDefaultOptions()
+ {
+ return new JsonSerializerOptions
+ {
+ PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
+ DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
+ NumberHandling = JsonNumberHandling.AllowReadingFromString,
+ Converters = { new JsonStringEnumConverter() }
+ };
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt/MqttHelper.cs b/src/Namotion.Interceptor.Mqtt/MqttHelper.cs
new file mode 100644
index 00000000..36b816fa
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/MqttHelper.cs
@@ -0,0 +1,48 @@
+using System;
+using System.Collections.Generic;
+using MQTTnet.Packets;
+
+namespace Namotion.Interceptor.Mqtt;
+
+internal static class MqttHelper
+{
+ public static DateTimeOffset? ExtractSourceTimestamp(
+ IReadOnlyCollection? userProperties,
+ string? timestampPropertyName)
+ {
+ if (timestampPropertyName is null || userProperties is null)
+ {
+ return null;
+ }
+
+ foreach (var prop in userProperties)
+ {
+ if (prop.Name == timestampPropertyName && long.TryParse(prop.Value, out var unixMs))
+ {
+ return DateTimeOffset.FromUnixTimeMilliseconds(unixMs);
+ }
+ }
+
+ return null;
+ }
+
+ public static string BuildTopic(string path, string? topicPrefix)
+ {
+ return topicPrefix is null
+ ? path
+ : string.Concat(topicPrefix, "/", path);
+ }
+
+ public static string StripTopicPrefix(string topic, string? topicPrefix)
+ {
+ if (topicPrefix is not null &&
+ topic.StartsWith(topicPrefix, StringComparison.Ordinal) &&
+ topic.Length > topicPrefix.Length &&
+ topic[topicPrefix.Length] == '/')
+ {
+ return topic.Substring(topicPrefix.Length + 1);
+ }
+
+ return topic;
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt/MqttSubjectExtensions.cs b/src/Namotion.Interceptor.Mqtt/MqttSubjectExtensions.cs
index caf50f56..256ca0f6 100644
--- a/src/Namotion.Interceptor.Mqtt/MqttSubjectExtensions.cs
+++ b/src/Namotion.Interceptor.Mqtt/MqttSubjectExtensions.cs
@@ -2,7 +2,9 @@
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Namotion.Interceptor;
-using Namotion.Interceptor.Mqtt;
+using Namotion.Interceptor.Mqtt.Client;
+using Namotion.Interceptor.Mqtt.Server;
+using Namotion.Interceptor.Sources;
using Namotion.Interceptor.Sources.Paths;
// ReSharper disable once CheckNamespace
@@ -10,25 +12,104 @@ namespace Microsoft.Extensions.DependencyInjection;
public static class MqttSubjectExtensions
{
+ ///
+ /// Adds an MQTT client source that subscribes to an MQTT broker and synchronizes properties.
+ ///
+ public static IServiceCollection AddMqttSubjectClient(
+ this IServiceCollection serviceCollection,
+ string brokerHost,
+ string sourceName,
+ int brokerPort = 1883,
+ string? topicPrefix = null)
+ where TSubject : IInterceptorSubject
+ {
+ return serviceCollection.AddMqttSubjectClient(
+ sp => sp.GetRequiredService(),
+ _ => new MqttClientConfiguration
+ {
+ BrokerHost = brokerHost,
+ BrokerPort = brokerPort,
+ TopicPrefix = topicPrefix,
+ PathProvider = new AttributeBasedSourcePathProvider(sourceName, "/")
+ });
+ }
+
+ ///
+ /// Adds an MQTT client source with custom configuration.
+ ///
+ public static IServiceCollection AddMqttSubjectClient(
+ this IServiceCollection serviceCollection,
+ Func subjectSelector,
+ Func configurationProvider)
+ {
+ var key = Guid.NewGuid().ToString();
+ return serviceCollection
+ .AddKeyedSingleton(key, (sp, _) => configurationProvider(sp))
+ .AddKeyedSingleton(key, (sp, _) => subjectSelector(sp))
+ .AddKeyedSingleton(key, (sp, _) =>
+ {
+ var subject = sp.GetRequiredKeyedService(key);
+ return new MqttSubjectClientSource(
+ subject,
+ sp.GetRequiredKeyedService(key),
+ sp.GetRequiredService>());
+ })
+ .AddSingleton(sp => sp.GetRequiredKeyedService(key))
+ .AddSingleton(sp =>
+ {
+ var configuration = sp.GetRequiredKeyedService(key);
+ var subject = sp.GetRequiredKeyedService(key);
+ return new SubjectSourceBackgroundService(
+ sp.GetRequiredKeyedService(key),
+ subject.Context,
+ sp.GetRequiredService>(),
+ configuration.BufferTime,
+ configuration.RetryTime,
+ configuration.WriteRetryQueueSize);
+ });
+ }
+
+ ///
+ /// Adds an MQTT server that publishes property changes to an MQTT broker.
+ ///
public static IServiceCollection AddMqttSubjectServer(
- this IServiceCollection serviceCollection, string sourceName, string? pathPrefix = null)
+ this IServiceCollection serviceCollection,
+ string brokerHost,
+ string sourceName,
+ int brokerPort = 1883,
+ string? topicPrefix = null)
where TSubject : IInterceptorSubject
{
- return serviceCollection.AddMqttSubjectServer(sp => sp.GetRequiredService(), sourceName, pathPrefix);
+ return serviceCollection.AddMqttSubjectServer(
+ sp => sp.GetRequiredService(),
+ _ => new MqttServerConfiguration
+ {
+ BrokerHost = brokerHost,
+ BrokerPort = brokerPort,
+ TopicPrefix = topicPrefix,
+ PathProvider = new AttributeBasedSourcePathProvider(sourceName, "/")
+ });
}
- public static IServiceCollection AddMqttSubjectServer(this IServiceCollection serviceCollection,
- Func subjectSelector, string sourceName, string? pathPrefix = null)
+ ///
+ /// Adds an MQTT server with custom configuration.
+ ///
+ public static IServiceCollection AddMqttSubjectServer(
+ this IServiceCollection serviceCollection,
+ Func subjectSelector,
+ Func configurationProvider)
{
var key = Guid.NewGuid().ToString();
return serviceCollection
+ .AddKeyedSingleton(key, (sp, _) => configurationProvider(sp))
.AddKeyedSingleton(key, (sp, _) => subjectSelector(sp))
.AddKeyedSingleton(key, (sp, _) =>
{
var subject = sp.GetRequiredKeyedService(key);
- var pathProvider = new AttributeBasedSourcePathProvider(sourceName, "/", pathPrefix);
return new MqttSubjectServerBackgroundService(
- subject, pathProvider, sp.GetRequiredService>());
+ subject,
+ sp.GetRequiredKeyedService(key),
+ sp.GetRequiredService>());
})
.AddSingleton(sp => sp.GetRequiredKeyedService(key));
}
diff --git a/src/Namotion.Interceptor.Mqtt/MqttSubjectServerBackgroundService.cs b/src/Namotion.Interceptor.Mqtt/MqttSubjectServerBackgroundService.cs
deleted file mode 100644
index 3cac4118..00000000
--- a/src/Namotion.Interceptor.Mqtt/MqttSubjectServerBackgroundService.cs
+++ /dev/null
@@ -1,191 +0,0 @@
-using System;
-using System.Linq;
-using System.Text;
-using System.Text.Json;
-using System.Threading;
-using System.Threading.Tasks;
-using Microsoft.Extensions.Hosting;
-using Microsoft.Extensions.Logging;
-using MQTTnet;
-using MQTTnet.Server;
-using Namotion.Interceptor.Registry;
-using Namotion.Interceptor.Registry.Abstractions;
-using Namotion.Interceptor.Sources;
-using Namotion.Interceptor.Sources.Paths;
-using Namotion.Interceptor.Tracking.Change;
-
-namespace Namotion.Interceptor.Mqtt
-{
- public class MqttSubjectServerBackgroundService : BackgroundService
- {
- private readonly string _serverClientId = "Server_" + Guid.NewGuid().ToString("N");
-
- private readonly IInterceptorSubject _subject;
- private readonly ISourcePathProvider _pathProvider;
- private readonly ILogger _logger;
- private readonly TimeSpan? _bufferTime;
-
- private int _numberOfClients;
- private MqttServer? _mqttServer;
-
- public int Port { get; set; } = 1883;
-
- public bool IsListening { get; private set; }
-
- public int? NumberOfClients => _numberOfClients;
-
- public MqttSubjectServerBackgroundService(IInterceptorSubject subject,
- ISourcePathProvider pathProvider,
- ILogger logger,
- TimeSpan? bufferTime = null)
- {
- _subject = subject;
- _pathProvider = pathProvider;
- _logger = logger;
- _bufferTime = bufferTime;
- }
-
- private bool IsPropertyIncluded(RegisteredSubjectProperty property)
- {
- return _pathProvider.IsPropertyIncluded(property);
- }
-
- protected override async Task ExecuteAsync(CancellationToken stoppingToken)
- {
- _mqttServer = new MqttServerFactory()
- .CreateMqttServer(new MqttServerOptions
- {
- DefaultEndpointOptions =
- {
- IsEnabled = true,
- Port = Port
- }
- });
-
- _mqttServer.ClientConnectedAsync += ClientConnectedAsync;
- _mqttServer.ClientDisconnectedAsync += ClientDisconnectedAsync;
- _mqttServer.InterceptingPublishAsync += InterceptingPublishAsync;
-
- while (!stoppingToken.IsCancellationRequested)
- {
- try
- {
- await _mqttServer.StartAsync();
- IsListening = true;
-
- try
- {
- using var changeQueueProcessor = new ChangeQueueProcessor(
- source: this, _subject.Context,
- propertyFilter: IsPropertyIncluded, writeHandler: WriteChangesAsync,
- _bufferTime, _logger);
-
- await changeQueueProcessor.ProcessAsync(stoppingToken);
- }
- finally
- {
- await _mqttServer.StopAsync();
- IsListening = false;
- }
- }
- catch (Exception ex)
- {
- IsListening = false;
-
- if (ex is TaskCanceledException or OperationCanceledException) return;
-
- _logger.LogError(ex, "Error in MQTT server.");
- await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
- }
- }
- }
-
- private async ValueTask WriteChangesAsync(ReadOnlyMemory changes, CancellationToken cancellationToken)
- {
- for (var i = 0; i < changes.Length; i++)
- {
- var change = changes.Span[i];
- var registeredProperty = change.Property.TryGetRegisteredProperty();
- if (registeredProperty is not { HasChildSubjects: false })
- {
- continue;
- }
-
- var path = registeredProperty.TryGetSourcePath(_pathProvider, _subject);
- if (path is not null)
- {
- await PublishPropertyValueAsync(path, change.GetNewValue(), cancellationToken);
- }
- }
- }
-
- private Task ClientConnectedAsync(ClientConnectedEventArgs arg)
- {
- Interlocked.Increment(ref _numberOfClients);
-
- Task.Run(async () =>
- {
- await Task.Delay(1000);
- foreach (var (path, property) in _subject
- .TryGetRegisteredSubject()?
- .GetAllProperties()
- .Where(p => !p.HasChildSubjects)
- .GetSourcePaths(_pathProvider, _subject) ?? [])
- {
- // TODO: Send only to new client
- await PublishPropertyValueAsync(path, property.GetValue(), CancellationToken.None);
- }
- });
-
- return Task.CompletedTask;
- }
-
- private async ValueTask PublishPropertyValueAsync(string path, object? value, CancellationToken cancellationToken)
- {
- await _mqttServer!.InjectApplicationMessage(
- new InjectedMqttApplicationMessage(
- new MqttApplicationMessage
- {
- Topic = path,
- ContentType = "application/json",
- PayloadSegment = new ArraySegment(
- JsonSerializer.SerializeToUtf8Bytes(value)),
- })
- {
- SenderClientId = _serverClientId
- }, cancellationToken);
- }
-
- private Task InterceptingPublishAsync(InterceptingPublishEventArgs args)
- {
- if (args.ClientId == _serverClientId)
- {
- return Task.CompletedTask;
- }
-
- var path = args.ApplicationMessage.Topic;
- var payload = Encoding.UTF8.GetString(args.ApplicationMessage.Payload);
-
- try
- {
- var document = JsonDocument.Parse(payload);
- _subject.UpdatePropertyValueFromSourcePath(path,
- DateTimeOffset.UtcNow, // TODO: What timestamp to use here?
- (property, _) => document.Deserialize(property.Type),
- _pathProvider, this);
- }
- catch (Exception ex)
- {
- _logger.LogError(ex, "Failed to deserialize MQTT payload.");
- }
-
- return Task.CompletedTask;
- }
-
- private Task ClientDisconnectedAsync(ClientDisconnectedEventArgs arg)
- {
- Interlocked.Decrement(ref _numberOfClients);
- return Task.CompletedTask;
- }
- }
-}
diff --git a/src/Namotion.Interceptor.Mqtt/Namotion.Interceptor.Mqtt.csproj b/src/Namotion.Interceptor.Mqtt/Namotion.Interceptor.Mqtt.csproj
index 84749016..752c04df 100644
--- a/src/Namotion.Interceptor.Mqtt/Namotion.Interceptor.Mqtt.csproj
+++ b/src/Namotion.Interceptor.Mqtt/Namotion.Interceptor.Mqtt.csproj
@@ -5,6 +5,9 @@
true
+
+
+
diff --git a/src/Namotion.Interceptor.Mqtt/Server/MqttServerConfiguration.cs b/src/Namotion.Interceptor.Mqtt/Server/MqttServerConfiguration.cs
new file mode 100644
index 00000000..e22c57b1
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/Server/MqttServerConfiguration.cs
@@ -0,0 +1,99 @@
+using System;
+using MQTTnet.Protocol;
+using Namotion.Interceptor.Sources.Paths;
+
+namespace Namotion.Interceptor.Mqtt.Server;
+
+///
+/// Configuration for MQTT server (publisher) background service.
+///
+public class MqttServerConfiguration
+{
+ ///
+ /// Gets or sets the MQTT broker hostname or IP address.
+ ///
+ public required string BrokerHost { get; init; }
+
+ ///
+ /// Gets or sets the MQTT broker port. Default is 1883.
+ ///
+ public int BrokerPort { get; init; } = 1883;
+
+ ///
+ /// Gets or sets the client identifier.
+ ///
+ public string ClientId { get; init; } = $"Namotion_{Guid.NewGuid():N}";
+
+ ///
+ /// Gets or sets the optional topic prefix.
+ ///
+ public string? TopicPrefix { get; init; }
+
+ ///
+ /// Gets or sets the source path provider for property-to-topic mapping.
+ ///
+ public required ISourcePathProvider PathProvider { get; init; }
+
+ ///
+ /// Gets or sets the default QoS level. Default is AtMostOnce (0) for high throughput.
+ ///
+ public MqttQualityOfServiceLevel DefaultQualityOfService { get; init; } = MqttQualityOfServiceLevel.AtMostOnce;
+
+ ///
+ /// Gets or sets whether to use retained messages. Default is true.
+ ///
+ public bool UseRetainedMessages { get; init; } = true;
+
+ ///
+ /// Gets or sets the maximum pending messages per client. Default is 10000.
+ /// Messages are dropped when the queue exceeds this limit.
+ ///
+ public int MaxPendingMessagesPerClient { get; init; } = 10000;
+
+ ///
+ /// Gets or sets the time to buffer property changes before sending. Default is 8ms.
+ ///
+ public TimeSpan BufferTime { get; init; } = TimeSpan.FromMilliseconds(8);
+
+ ///
+ /// Gets or sets the delay before publishing the initial state to a newly connected client.
+ /// This allows time for the client to complete its subscription setup.
+ /// Set to zero to disable initial state publishing (relies on retained messages only).
+ /// Default is 500ms.
+ ///
+ public TimeSpan InitialStateDelay { get; init; } = TimeSpan.FromMilliseconds(500);
+
+ ///
+ /// Gets or sets the value converter. Default is JSON.
+ ///
+ public IMqttValueConverter ValueConverter { get; init; } = new JsonMqttValueConverter();
+
+ ///
+ /// Gets or sets the MQTT user property name for the source timestamp. Default is "ts".
+ /// Set to null to disable timestamp inclusion.
+ ///
+ public string? SourceTimestampPropertyName { get; init; } = "ts";
+
+ ///
+ /// Gets or sets the converter function for serializing timestamps to strings.
+ /// Default converts to Unix milliseconds.
+ ///
+ public Func SourceTimestampConverter { get; init; } =
+ static timestamp => timestamp.ToUnixTimeMilliseconds().ToString();
+
+ ///
+ /// Validates the configuration.
+ ///
+ public void Validate()
+ {
+ if (string.IsNullOrWhiteSpace(BrokerHost))
+ {
+ throw new ArgumentException("BrokerHost must be specified.", nameof(BrokerHost));
+ }
+
+ if (PathProvider is null)
+ {
+ throw new ArgumentException("PathProvider must be specified.", nameof(PathProvider));
+ }
+ }
+}
diff --git a/src/Namotion.Interceptor.Mqtt/Server/MqttSubjectServerBackgroundService.cs b/src/Namotion.Interceptor.Mqtt/Server/MqttSubjectServerBackgroundService.cs
new file mode 100644
index 00000000..c66ca741
--- /dev/null
+++ b/src/Namotion.Interceptor.Mqtt/Server/MqttSubjectServerBackgroundService.cs
@@ -0,0 +1,445 @@
+using System;
+using System.Buffers;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using MQTTnet;
+using MQTTnet.Packets;
+using MQTTnet.Server;
+using Namotion.Interceptor.Registry;
+using Namotion.Interceptor.Registry.Abstractions;
+using Namotion.Interceptor.Registry.Performance;
+using Namotion.Interceptor.Sources;
+using Namotion.Interceptor.Sources.Paths;
+using Namotion.Interceptor.Tracking.Change;
+
+namespace Namotion.Interceptor.Mqtt.Server;
+
+///
+/// Background service that hosts an MQTT broker and publishes property changes.
+///
+public class MqttSubjectServerBackgroundService : BackgroundService, IAsyncDisposable
+{
+ private static readonly ObjectPool> UserPropertiesPool
+ = new(() => new List(1));
+
+ private readonly string _serverClientId;
+ private readonly IInterceptorSubject _subject;
+ private readonly IInterceptorSubjectContext _context;
+ private readonly MqttServerConfiguration _configuration;
+ private readonly ILogger _logger;
+
+ // TODO(memory): Might lead to memory leaks
+ private readonly ConcurrentDictionary _propertyToTopic = new();
+ private readonly ConcurrentDictionary _pathToProperty = new();
+
+ private readonly List _runningInitialStateTasks = [];
+ private readonly Lock _initialStateTasksLock = new();
+
+ private int _numberOfClients;
+ private int _disposed;
+ private int _isListening;
+ private MqttServer? _mqttServer;
+
+ ///
+ /// Gets whether the MQTT server is listening.
+ ///
+ public bool IsListening => Volatile.Read(ref _isListening) == 1;
+
+ ///
+ /// Gets the number of connected clients.
+ ///
+ public int NumberOfClients => Volatile.Read(ref _numberOfClients);
+
+ public MqttSubjectServerBackgroundService(
+ IInterceptorSubject subject,
+ MqttServerConfiguration configuration,
+ ILogger logger)
+ {
+ _subject = subject ?? throw new ArgumentNullException(nameof(subject));
+ _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
+ _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+ _context = subject.Context;
+ _serverClientId = _configuration.ClientId;
+
+ configuration.Validate();
+ }
+
+ private bool IsPropertyIncluded(RegisteredSubjectProperty property) =>
+ _configuration.PathProvider.IsPropertyIncluded(property);
+
+ ///
+ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+ {
+ var options = new MqttServerOptionsBuilder()
+ .WithDefaultEndpoint()
+ .WithDefaultEndpointPort(_configuration.BrokerPort)
+ .WithMaxPendingMessagesPerClient(_configuration.MaxPendingMessagesPerClient)
+ .Build();
+
+ _mqttServer = new MqttServerFactory().CreateMqttServer(options);
+
+ _mqttServer.ClientConnectedAsync += ClientConnectedAsync;
+ _mqttServer.ClientDisconnectedAsync += ClientDisconnectedAsync;
+ _mqttServer.InterceptingPublishAsync += InterceptingPublishAsync;
+
+ while (!stoppingToken.IsCancellationRequested)
+ {
+ try
+ {
+ await _mqttServer.StartAsync().ConfigureAwait(false);
+ Volatile.Write(ref _isListening, 1);
+
+ _logger.LogInformation("MQTT server started on port {Port}.", _configuration.BrokerPort);
+
+ try
+ {
+ using var changeQueueProcessor = new ChangeQueueProcessor(
+ source: this,
+ _context,
+ propertyFilter: IsPropertyIncluded,
+ writeHandler: WriteChangesAsync,
+ _configuration.BufferTime,
+ _logger);
+
+ await changeQueueProcessor.ProcessAsync(stoppingToken).ConfigureAwait(false);
+ }
+ finally
+ {
+ await _mqttServer.StopAsync().ConfigureAwait(false);
+ Volatile.Write(ref _isListening, 0);
+ }
+ }
+ catch (Exception ex)
+ {
+ Volatile.Write(ref _isListening, 0);
+
+ if (ex is TaskCanceledException or OperationCanceledException)
+ {
+ return;
+ }
+
+ _logger.LogError(ex, "Error in MQTT server.");
+ await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken).ConfigureAwait(false);
+ }
+ }
+ }
+
+ private async ValueTask WriteChangesAsync(ReadOnlyMemory changes, CancellationToken cancellationToken)
+ {
+ var length = changes.Length;
+ if (length == 0) return;
+
+ var server = _mqttServer;
+ if (server is null) return;
+
+ // Rent arrays from pool to avoid allocations
+ var messagesPool = ArrayPool.Shared;
+ var userPropsArrayPool = ArrayPool?>.Shared;
+
+ var messages = messagesPool.Rent(length);
+ var userPropertiesArray = _configuration.SourceTimestampPropertyName is not null
+ ? userPropsArrayPool.Rent(length)
+ : null;
+ var messageCount = 0;
+
+ try
+ {
+ var changesSpan = changes.Span;
+
+ // Build all messages first
+ for (var i = 0; i < length; i++)
+ {
+ var change = changesSpan[i];
+ var registeredProperty = change.Property.TryGetRegisteredProperty();
+ if (registeredProperty is not { HasChildSubjects: false })
+ {
+ continue;
+ }
+
+ var topic = TryGetTopicForProperty(change.Property, registeredProperty);
+ if (topic is null) continue;
+
+ byte[] payload;
+ try
+ {
+ payload = _configuration.ValueConverter.Serialize(
+ change.GetNewValue(),
+ registeredProperty.Type);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Failed to serialize value for topic {Topic}.", topic);
+ continue;
+ }
+
+ // Create message directly without builder to reduce allocations
+ var message = new MqttApplicationMessage
+ {
+ Topic = topic,
+ PayloadSegment = new ArraySegment(payload),
+ QualityOfServiceLevel = _configuration.DefaultQualityOfService,
+ Retain = _configuration.UseRetainedMessages
+ };
+
+ if (userPropertiesArray is not null)
+ {
+ var userProps = UserPropertiesPool.Rent();
+ userProps.Clear();
+ userProps.Add(new MqttUserProperty(
+ _configuration.SourceTimestampPropertyName!,
+ _configuration.SourceTimestampConverter(change.ChangedTimestamp)));
+ message.UserProperties = userProps;
+ userPropertiesArray[messageCount] = userProps;
+ }
+
+ messages[messageCount++] = new InjectedMqttApplicationMessage(message)
+ {
+ SenderClientId = _serverClientId
+ };
+ }
+
+ // TODO(nuget): Use batch API for better performance
+ // if (messageCount > 0)
+ // {
+ // await server.InjectApplicationMessages(
+ // new ArraySegment(messages, 0, messageCount),
+ // cancellationToken).ConfigureAwait(false);
+ // }
+
+ // TODO(nuget): Keep this as fallback for older NuGet versions
+ for (var i = 0; i < messageCount; i++)
+ {
+ await server.InjectApplicationMessage(messages[i], cancellationToken).ConfigureAwait(false);
+ }
+ }
+ finally
+ {
+ // Return user property lists to pool
+ if (userPropertiesArray is not null)
+ {
+ for (var i = 0; i < messageCount; i++)
+ {
+ if (userPropertiesArray[i] is { } list)
+ {
+ UserPropertiesPool.Return(list);
+ }
+ }
+ userPropsArrayPool.Return(userPropertiesArray);
+ }
+
+ messagesPool.Return(messages);
+ }
+ }
+
+ private string? TryGetTopicForProperty(PropertyReference propertyReference, RegisteredSubjectProperty property)
+ {
+ return _propertyToTopic.GetOrAdd(propertyReference, static (_, state) =>
+ {
+ var (p, pathProvider, subject, topicPrefix) = state;
+ var path = p.TryGetSourcePath(pathProvider, subject);
+ return path is null ? null : MqttHelper.BuildTopic(path, topicPrefix);
+ }, (property, _configuration.PathProvider, _subject, _configuration.TopicPrefix));
+ }
+
+ private PropertyReference? TryGetPropertyForTopic(string path)
+ {
+ return _pathToProperty.GetOrAdd(path, static (p, state) =>
+ {
+ var (subject, pathProvider) = state;
+ var (property, _) = subject.TryGetPropertyFromSourcePath(p, pathProvider);
+ return property?.Reference;
+ }, (_subject, _configuration.PathProvider));
+ }
+
+ private Task ClientConnectedAsync(ClientConnectedEventArgs arg)
+ {
+ var count = Interlocked.Increment(ref _numberOfClients);
+ _logger.LogInformation("Client {ClientId} connected. Total clients: {Count}.", arg.ClientId, count);
+
+ // Publish all current property values to new client
+ var task = Task.Run(async () =>
+ {
+ try
+ {
+ if (Interlocked.CompareExchange(ref _disposed, 0, 0) == 0)
+ {
+ await PublishInitialStateAsync().ConfigureAwait(false);
+ }
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Failed to publish initial state to client {ClientId}.", arg.ClientId);
+ }
+ });
+
+ lock (_initialStateTasksLock)
+ {
+ // Clean up completed tasks to prevent memory leak
+ _runningInitialStateTasks.RemoveAll(t => t.IsCompleted);
+ _runningInitialStateTasks.Add(task);
+ }
+
+ return Task.CompletedTask;
+ }
+
+ private async Task PublishInitialStateAsync()
+ {
+ try
+ {
+ // Wait for the client to complete subscription setup before sending initial values.
+ // This delay is configurable; set to zero to rely on retained messages only.
+ var delay = _configuration.InitialStateDelay;
+ if (delay <= TimeSpan.Zero)
+ {
+ return;
+ }
+
+ await Task.Delay(delay).ConfigureAwait(false);
+
+ var properties = _subject
+ .TryGetRegisteredSubject()?
+ .GetAllProperties()
+ .Where(p => !p.HasChildSubjects)
+ .GetSourcePaths(_configuration.PathProvider, _subject);
+
+ if (properties is null) return;
+
+ var server = _mqttServer;
+ if (server is null) return;
+
+ foreach (var (path, property) in properties)
+ {
+ var topic = MqttHelper.BuildTopic(path, _configuration.TopicPrefix);
+
+ var payload = _configuration.ValueConverter.Serialize(
+ property.GetValue(),
+ property.Type);
+
+ var message = new MqttApplicationMessage
+ {
+ Topic = topic,
+ PayloadSegment = new ArraySegment(payload),
+ ContentType = "application/json",
+ QualityOfServiceLevel = _configuration.DefaultQualityOfService,
+ Retain = _configuration.UseRetainedMessages
+ };
+
+ await server.InjectApplicationMessage(
+ new InjectedMqttApplicationMessage(message) { SenderClientId = _serverClientId },
+ CancellationToken.None).ConfigureAwait(false);
+ }
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Failed to publish initial state to client.");
+ }
+ }
+
+ private Task InterceptingPublishAsync(InterceptingPublishEventArgs args)
+ {
+ // Skip messages published by this server (injected messages may have null/empty ClientId)
+ if (string.IsNullOrEmpty(args.ClientId) || args.ClientId == _serverClientId)
+ {
+ return Task.CompletedTask;
+ }
+
+ var topic = args.ApplicationMessage.Topic;
+ var path = MqttHelper.StripTopicPrefix(topic, _configuration.TopicPrefix);
+
+ if (TryGetPropertyForTopic(path) is not { } propertyReference)
+ {
+ return Task.CompletedTask;
+ }
+
+ var registeredProperty = propertyReference.TryGetRegisteredProperty();
+ if (registeredProperty is null)
+ {
+ return Task.CompletedTask;
+ }
+
+ try
+ {
+ var payload = args.ApplicationMessage.Payload;
+ var value = _configuration.ValueConverter.Deserialize(payload, registeredProperty.Type);
+
+ var receivedTimestamp = DateTimeOffset.UtcNow;
+ var sourceTimestamp = MqttHelper.ExtractSourceTimestamp(
+ args.ApplicationMessage.UserProperties,
+ _configuration.SourceTimestampPropertyName) ?? receivedTimestamp;
+
+ propertyReference.SetValueFromSource(this, sourceTimestamp, receivedTimestamp, value);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Failed to deserialize MQTT payload for topic {Topic}.", topic);
+ }
+
+ return Task.CompletedTask;
+ }
+
+ private Task ClientDisconnectedAsync(ClientDisconnectedEventArgs arg)
+ {
+ var count = Interlocked.Decrement(ref _numberOfClients);
+ _logger.LogInformation("Client {ClientId} disconnected. Total clients: {Count}.", arg.ClientId, count);
+ return Task.CompletedTask;
+ }
+
+ ///
+ public async ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 1)
+ {
+ return;
+ }
+
+ var server = _mqttServer;
+ if (server is not null)
+ {
+ server.ClientConnectedAsync -= ClientConnectedAsync;
+ server.ClientDisconnectedAsync -= ClientDisconnectedAsync;
+ server.InterceptingPublishAsync -= InterceptingPublishAsync;
+
+ // Wait for all running initial state tasks to complete
+ try
+ {
+ Task[] tasksSnapshot;
+ lock (_initialStateTasksLock)
+ {
+ tasksSnapshot = _runningInitialStateTasks.ToArray();
+ }
+ await Task.WhenAll(tasksSnapshot).ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(ex, "Error waiting for initial state tasks to complete.");
+ }
+
+ if (IsListening)
+ {
+ try
+ {
+ await server.StopAsync().ConfigureAwait(false);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogWarning(ex, "Error stopping MQTT server.");
+ }
+ }
+
+ server.Dispose();
+ _mqttServer = null;
+ }
+
+ // Clear caches to allow GC of subject references
+ _propertyToTopic.Clear();
+ _pathToProperty.Clear();
+
+ Volatile.Write(ref _isListening, 0);
+ Dispose();
+ }
+}
diff --git a/src/Namotion.Interceptor.OpcUa.SampleClient/Namotion.Interceptor.OpcUa.SampleClient.csproj b/src/Namotion.Interceptor.OpcUa.SampleClient/Namotion.Interceptor.OpcUa.SampleClient.csproj
index 7bb04193..b461b023 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleClient/Namotion.Interceptor.OpcUa.SampleClient.csproj
+++ b/src/Namotion.Interceptor.OpcUa.SampleClient/Namotion.Interceptor.OpcUa.SampleClient.csproj
@@ -16,7 +16,7 @@
-
+
diff --git a/src/Namotion.Interceptor.OpcUa.SampleClient/Program.cs b/src/Namotion.Interceptor.OpcUa.SampleClient/Program.cs
index fe5effc3..18ae1bf1 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleClient/Program.cs
+++ b/src/Namotion.Interceptor.OpcUa.SampleClient/Program.cs
@@ -2,9 +2,9 @@
using Microsoft.Extensions.Hosting;
using Namotion.Interceptor;
using Namotion.Interceptor.Hosting;
-using Namotion.Interceptor.OpcUa.SampleClient;
-using Namotion.Interceptor.OpcUa.SampleModel;
using Namotion.Interceptor.Registry;
+using Namotion.Interceptor.SamplesModel;
+using Namotion.Interceptor.SamplesModel.Workers;
using Namotion.Interceptor.Tracking;
using Namotion.Interceptor.Validation;
using Opc.Ua;
@@ -22,12 +22,13 @@
Utils.SetTraceMask(Utils.TraceMasks.All);
+// OPC UA client creates just the root - persons array will be loaded from server
var root = new Root(context);
context.AddService(root);
builder.Services.AddSingleton(root);
+builder.Services.AddHostedService();
builder.Services.AddOpcUaSubjectClient("opc.tcp://localhost:4840", "opc", rootName: "Root");
-builder.Services.AddHostedService();
using var performanceProfiler = new PerformanceProfiler(context, "Client");
var host = builder.Build();
diff --git a/src/Namotion.Interceptor.OpcUa.SampleClient/Worker.cs b/src/Namotion.Interceptor.OpcUa.SampleClient/Worker.cs
deleted file mode 100644
index 5ea48858..00000000
--- a/src/Namotion.Interceptor.OpcUa.SampleClient/Worker.cs
+++ /dev/null
@@ -1,23 +0,0 @@
-using Microsoft.Extensions.Hosting;
-using Namotion.Interceptor.OpcUa.SampleModel;
-
-namespace Namotion.Interceptor.OpcUa.SampleClient;
-
-public class Worker : BackgroundService
-{
- private readonly Root _root;
-
- public Worker(Root root)
- {
- _root = root;
- }
-
- protected override async Task ExecuteAsync(CancellationToken stoppingToken)
- {
- while (!stoppingToken.IsCancellationRequested)
- {
- //_root.Number++;
- await Task.Delay(1000, stoppingToken);
- }
- }
-}
\ No newline at end of file
diff --git a/src/Namotion.Interceptor.OpcUa.SampleModel/Root.cs b/src/Namotion.Interceptor.OpcUa.SampleModel/Root.cs
deleted file mode 100644
index ea38ec0a..00000000
--- a/src/Namotion.Interceptor.OpcUa.SampleModel/Root.cs
+++ /dev/null
@@ -1,23 +0,0 @@
-using Namotion.Interceptor.Attributes;
-using Namotion.Interceptor.Sources.Paths.Attributes;
-
-namespace Namotion.Interceptor.OpcUa.SampleModel;
-
-[InterceptorSubject]
-public partial class Root
-{
- [SourcePath("opc", "Name")]
- public partial string Name { get; set; }
-
- [SourcePath("opc", "Number")]
- public partial decimal Number { get; set; }
-
- [SourcePath("opc", "Persons")]
- public partial Person[] Persons { get; set; }
-
- public Root()
- {
- Name = "My root name";
- Persons = [];
- }
-}
\ No newline at end of file
diff --git a/src/Namotion.Interceptor.OpcUa.SampleServer/Namotion.Interceptor.OpcUa.SampleServer.csproj b/src/Namotion.Interceptor.OpcUa.SampleServer/Namotion.Interceptor.OpcUa.SampleServer.csproj
index 59396a74..8b3ca306 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleServer/Namotion.Interceptor.OpcUa.SampleServer.csproj
+++ b/src/Namotion.Interceptor.OpcUa.SampleServer/Namotion.Interceptor.OpcUa.SampleServer.csproj
@@ -16,7 +16,7 @@
-
+
diff --git a/src/Namotion.Interceptor.OpcUa.SampleServer/Program.cs b/src/Namotion.Interceptor.OpcUa.SampleServer/Program.cs
index 5cd7559c..7b413f93 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleServer/Program.cs
+++ b/src/Namotion.Interceptor.OpcUa.SampleServer/Program.cs
@@ -2,9 +2,9 @@
using Microsoft.Extensions.Hosting;
using Namotion.Interceptor;
using Namotion.Interceptor.Hosting;
-using Namotion.Interceptor.OpcUa.SampleModel;
-using Namotion.Interceptor.OpcUa.SampleServer;
using Namotion.Interceptor.Registry;
+using Namotion.Interceptor.SamplesModel;
+using Namotion.Interceptor.SamplesModel.Workers;
using Namotion.Interceptor.Tracking;
using Namotion.Interceptor.Validation;
@@ -19,21 +19,12 @@
.WithDataAnnotationValidation()
.WithHostedServices(builder.Services);
-var root = new Root(context);
+var root = Root.CreateWithPersons(context);
context.AddService(root);
-root.Persons = Enumerable
- .Range(0, 10000)
- .Select(i => new Person
- {
- FirstName = "John " + i,
- LastName = "Doe" + i
- })
- .ToArray();
-
builder.Services.AddSingleton(root);
+builder.Services.AddHostedService();
builder.Services.AddOpcUaSubjectServer("opc", rootName: "Root");
-builder.Services.AddHostedService();
using var performanceProfiler = new PerformanceProfiler(context, "Server");
var host = builder.Build();
diff --git a/src/Namotion.Interceptor.OpcUa/Client/Polling/PollingManager.cs b/src/Namotion.Interceptor.OpcUa/Client/Polling/PollingManager.cs
index 1d4fcec5..feddf40f 100644
--- a/src/Namotion.Interceptor.OpcUa/Client/Polling/PollingManager.cs
+++ b/src/Namotion.Interceptor.OpcUa/Client/Polling/PollingManager.cs
@@ -4,6 +4,7 @@
using Namotion.Interceptor.OpcUa.Client.Connection;
using Namotion.Interceptor.Registry.Abstractions;
using Namotion.Interceptor.Sources;
+using Namotion.Interceptor.Sources.Resilience;
using Namotion.Interceptor.Tracking.Change;
using Opc.Ua;
using Opc.Ua.Client;
@@ -21,7 +22,7 @@ internal sealed class PollingManager : IDisposable
private readonly SessionManager _sessionManager;
private readonly SubjectPropertyWriter _propertyWriter;
private readonly OpcUaClientConfiguration _configuration;
- private readonly PollingCircuitBreaker _circuitBreaker;
+ private readonly CircuitBreaker _circuitBreaker;
private readonly PollingMetrics _metrics = new();
private readonly ConcurrentDictionary _pollingItems = new();
@@ -49,7 +50,7 @@ public PollingManager(OpcUaSubjectClientSource source,
_propertyWriter = propertyWriter;
_configuration = configuration;
- _circuitBreaker = new PollingCircuitBreaker(configuration.PollingCircuitBreakerThreshold, configuration.PollingCircuitBreakerCooldown);
+ _circuitBreaker = new CircuitBreaker(configuration.PollingCircuitBreakerThreshold, configuration.PollingCircuitBreakerCooldown);
_timer = new PeriodicTimer(configuration.PollingInterval);
}
diff --git a/src/Namotion.Interceptor.OpcUa/OpcUaSourceExtensions.cs b/src/Namotion.Interceptor.OpcUa/OpcUaSourceExtensions.cs
new file mode 100644
index 00000000..e69de29b
diff --git a/src/Namotion.Interceptor.SampleWeb/Program.cs b/src/Namotion.Interceptor.SampleWeb/Program.cs
index bfb299d9..46c12682 100644
--- a/src/Namotion.Interceptor.SampleWeb/Program.cs
+++ b/src/Namotion.Interceptor.SampleWeb/Program.cs
@@ -114,7 +114,7 @@ public static void Main(string[] args)
// builder.Services.AddOpcUaSubjectClient("opc.tcp://localhost:4840", "opc", rootName: "Root");
// expose subject via MQTT
- builder.Services.AddMqttSubjectServer("mqtt");
+ builder.Services.AddMqttSubjectServer("localhost", "mqtt");
//builder.Services.AddMqttSubjectServer(sp => sp.GetRequiredService().Tires[2], "mqtt");
// expose subject via GraphQL
diff --git a/src/Namotion.Interceptor.OpcUa.SampleModel/Namotion.Interceptor.OpcUa.SampleModel.csproj b/src/Namotion.Interceptor.SamplesModel/Namotion.Interceptor.SamplesModel.csproj
similarity index 71%
rename from src/Namotion.Interceptor.OpcUa.SampleModel/Namotion.Interceptor.OpcUa.SampleModel.csproj
rename to src/Namotion.Interceptor.SamplesModel/Namotion.Interceptor.SamplesModel.csproj
index 0e84c8cf..02ea70f2 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleModel/Namotion.Interceptor.OpcUa.SampleModel.csproj
+++ b/src/Namotion.Interceptor.SamplesModel/Namotion.Interceptor.SamplesModel.csproj
@@ -1,4 +1,4 @@
-
+
net9.0
@@ -6,10 +6,15 @@
enable
false
-
+
+
+
+
+
+
diff --git a/src/Namotion.Interceptor.OpcUa.SampleModel/PerformanceProfiler.cs b/src/Namotion.Interceptor.SamplesModel/PerformanceProfiler.cs
similarity index 94%
rename from src/Namotion.Interceptor.OpcUa.SampleModel/PerformanceProfiler.cs
rename to src/Namotion.Interceptor.SamplesModel/PerformanceProfiler.cs
index cbb77051..fdeaa4e2 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleModel/PerformanceProfiler.cs
+++ b/src/Namotion.Interceptor.SamplesModel/PerformanceProfiler.cs
@@ -3,7 +3,7 @@
using System.Reactive.Linq;
using Namotion.Interceptor.Tracking;
-namespace Namotion.Interceptor.OpcUa.SampleModel;
+namespace Namotion.Interceptor.SamplesModel;
public class PerformanceProfiler : IDisposable
{
@@ -21,7 +21,7 @@ public PerformanceProfiler(IInterceptorSubjectContext context, string roleTitle)
var windowStartTime = startTime;
var lastAllThroughputTime = startTime;
- long windowStartTotalAllocatedBytes = GC.GetTotalAllocatedBytes(precise: true); // track allocation baseline per window
+ long windowStartTotalAllocatedBytes = GC.GetTotalAllocatedBytes(precise: true);
static double Percentile(IReadOnlyList sortedAsc, double p)
{
@@ -46,7 +46,6 @@ static double StdDev(IReadOnlyList values, double mean)
void PrintStats(string title, DateTimeOffset windowStartTimeCopy, List changedLatencyData, List receivedLatencyData, List throughputData)
{
- // Memory metrics
var proc = Process.GetCurrentProcess();
var workingSetMb = proc.WorkingSet64 / (1024.0 * 1024.0);
var totalMemoryMb = GC.GetTotalMemory(forceFullCollection: false) / (1024.0 * 1024.0);
@@ -66,7 +65,6 @@ void PrintStats(string title, DateTimeOffset windowStartTimeCopy, List c
Console.WriteLine($"Avg allocations over last {elapsedSec}s: {Math.Round(allocRateMbPerSec, 2)} MB/s");
Console.WriteLine();
- // Single compact table for all metrics
Console.WriteLine($"{"Metric",-29} {"Avg",10} {"P50",10} {"P90",10} {"P95",10} {"P99",10} {"P99.9",10} {"Max",10} {"Min",10} {"StdDev",10} {"Count",10}");
Console.WriteLine(new string('-', 139));
@@ -82,13 +80,11 @@ void PrintStats(string title, DateTimeOffset windowStartTimeCopy, List c
var p99Throughput = Percentile(sortedTp, 0.99);
var p999Throughput = Percentile(sortedTp, 0.999);
var stdThroughput = StdDev(sortedTp, avgThroughput);
-
+
Console.WriteLine($"{"Modifications (changes/s)",-29} {avgThroughput,10:F2} {p50Throughput,10:F2} {p90Throughput,10:F2} {p95Throughput,10:F2} {p99Throughput,10:F2} {p999Throughput,10:F2} {maxThroughput,10:F2} {minThroughput,10:F2} {stdThroughput,10:F2} {"-",10}");
}
- // Processing latency: Time from receiving change to applying it locally
PrintLatency("Processing latency (ms)", receivedLatencyData);
- // End-to-end latency: Time from source change to local application
PrintLatency("End-to-end latency (ms)", changedLatencyData);
}
@@ -119,6 +115,9 @@ void PrintLatency(string label, IEnumerable doubles)
.GetPropertyChangeObservable(ImmediateScheduler.Instance)
.Subscribe(change =>
{
+ if (change.Source == null)
+ return;
+
var now = DateTimeOffset.UtcNow;
lock (syncLock)
{
@@ -134,7 +133,7 @@ void PrintLatency(string label, IEnumerable doubles)
var receivedLatencyMs = (now - receivedTimestamp).TotalMilliseconds;
allReceivedLatencies.Add(receivedLatencyMs);
}
-
+
var timeSinceLastAllSample = (now - lastAllThroughputTime).TotalSeconds;
if (timeSinceLastAllSample >= 1.0)
{
@@ -171,7 +170,7 @@ void PrintLatency(string label, IEnumerable doubles)
windowStartTime = startTime + TimeSpan.FromSeconds(10) + index * TimeSpan.FromSeconds(60);
lastAllThroughputTime = windowStartTime;
}
-
+
if (index == 0)
{
PrintStats("Benchmark - Intermediate (10 seconds)", windowStartTimeCopy, allChangedLatenciesCopy, allReceivedLatenciesCopy, allThroughputSamplesCopy);
@@ -181,7 +180,7 @@ void PrintLatency(string label, IEnumerable doubles)
PrintStats("Benchmark - 1 minute", windowStartTimeCopy, allChangedLatenciesCopy, allReceivedLatenciesCopy, allThroughputSamplesCopy);
}
- windowStartTotalAllocatedBytes = GC.GetTotalAllocatedBytes(precise: true); // reset baseline for next window
+ windowStartTotalAllocatedBytes = GC.GetTotalAllocatedBytes(precise: true);
});
}
@@ -190,4 +189,4 @@ public void Dispose()
_timer.Dispose();
_change.Dispose();
}
-}
\ No newline at end of file
+}
diff --git a/src/Namotion.Interceptor.OpcUa.SampleModel/Person.cs b/src/Namotion.Interceptor.SamplesModel/Person.cs
similarity index 83%
rename from src/Namotion.Interceptor.OpcUa.SampleModel/Person.cs
rename to src/Namotion.Interceptor.SamplesModel/Person.cs
index a9b4174b..e8d945ba 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleModel/Person.cs
+++ b/src/Namotion.Interceptor.SamplesModel/Person.cs
@@ -1,8 +1,7 @@
-
using Namotion.Interceptor.Attributes;
using Namotion.Interceptor.Sources.Paths.Attributes;
-namespace Namotion.Interceptor.OpcUa.SampleModel;
+namespace Namotion.Interceptor.SamplesModel;
[InterceptorSubject]
public partial class Person
@@ -13,13 +12,14 @@ public Person()
}
[SourcePath("opc", "FirstName")]
+ [SourcePath("mqtt", "FirstName")]
public partial string? FirstName { get; set; }
[SourcePath("opc", "LastName")]
+ [SourcePath("mqtt", "LastName")]
public partial string? LastName { get; set; }
-
+
[Derived]
- [SourcePath("opc", "FullName")]
public string FullName => $"{FirstName} {LastName}";
public partial Person? Father { get; set; }
@@ -27,4 +27,4 @@ public Person()
public partial Person? Mother { get; set; }
public partial IReadOnlyCollection Children { get; set; }
-}
\ No newline at end of file
+}
diff --git a/src/Namotion.Interceptor.SamplesModel/Root.cs b/src/Namotion.Interceptor.SamplesModel/Root.cs
new file mode 100644
index 00000000..ab8406cf
--- /dev/null
+++ b/src/Namotion.Interceptor.SamplesModel/Root.cs
@@ -0,0 +1,47 @@
+using Namotion.Interceptor.Attributes;
+using Namotion.Interceptor.Sources.Paths.Attributes;
+
+namespace Namotion.Interceptor.SamplesModel;
+
+[InterceptorSubject]
+public partial class Root
+{
+ [SourcePath("opc", "Name")]
+ [SourcePath("mqtt", "Name")]
+ public partial string Name { get; set; }
+
+ [SourcePath("opc", "Number")]
+ [SourcePath("mqtt", "Number")]
+ public partial decimal Number { get; set; }
+
+ [SourcePath("opc", "Persons")]
+ [SourcePath("mqtt", "Persons")]
+ public partial Person[] Persons { get; set; }
+
+ public Root()
+ {
+ Name = "Sample Root";
+ Persons = [];
+ }
+
+ ///
+ /// Creates a Root with pre-instantiated persons for testing.
+ ///
+ public static Root CreateWithPersons(IInterceptorSubjectContext context, int count = 20_000)
+ {
+ var root = new Root(context);
+ var persons = new Person[count];
+
+ for (var i = 0; i < count; i++)
+ {
+ persons[i] = new Person(context)
+ {
+ FirstName = $"Person{i}",
+ LastName = "Test"
+ };
+ }
+
+ root.Persons = persons;
+ return root;
+ }
+}
diff --git a/src/Namotion.Interceptor.SamplesModel/Workers/ClientWorker.cs b/src/Namotion.Interceptor.SamplesModel/Workers/ClientWorker.cs
new file mode 100644
index 00000000..4f6944b0
--- /dev/null
+++ b/src/Namotion.Interceptor.SamplesModel/Workers/ClientWorker.cs
@@ -0,0 +1,44 @@
+using System.Diagnostics;
+using Microsoft.Extensions.Hosting;
+
+namespace Namotion.Interceptor.SamplesModel.Workers;
+
+public class ClientWorker : BackgroundService
+{
+ private readonly Root _root;
+
+ public ClientWorker(Root root)
+ {
+ _root = root;
+ }
+
+ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+ {
+ // Expected updates/second = number of persons * 2 / delay
+
+ var delay = TimeSpan.FromSeconds(1);
+ var lastChange = DateTimeOffset.UtcNow;
+ while (!stoppingToken.IsCancellationRequested)
+ {
+ var mod = _root.Persons.Length / 50;
+ var now = DateTimeOffset.UtcNow;
+ if (now - lastChange > delay)
+ {
+ lastChange = lastChange.AddSeconds(1);
+
+ for (var index = 0; index < _root.Persons.Length; index++)
+ {
+ var person = _root.Persons[index];
+ person.LastName = Stopwatch.GetTimestamp().ToString();
+
+ if (index % mod == 0) // distribute updates over approx. 0.5s
+ {
+ await Task.Delay(10, stoppingToken);
+ }
+ }
+ }
+
+ await Task.Delay(10, stoppingToken);
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/Namotion.Interceptor.OpcUa.SampleServer/Worker.cs b/src/Namotion.Interceptor.SamplesModel/Workers/ServerWorker.cs
similarity index 82%
rename from src/Namotion.Interceptor.OpcUa.SampleServer/Worker.cs
rename to src/Namotion.Interceptor.SamplesModel/Workers/ServerWorker.cs
index ec558985..91747795 100644
--- a/src/Namotion.Interceptor.OpcUa.SampleServer/Worker.cs
+++ b/src/Namotion.Interceptor.SamplesModel/Workers/ServerWorker.cs
@@ -1,14 +1,13 @@
using System.Diagnostics;
using Microsoft.Extensions.Hosting;
-using Namotion.Interceptor.OpcUa.SampleModel;
-namespace Namotion.Interceptor.OpcUa.SampleServer;
+namespace Namotion.Interceptor.SamplesModel.Workers;
-public class Worker : BackgroundService
+public class ServerWorker : BackgroundService
{
private readonly Root _root;
- public Worker(Root root)
+ public ServerWorker(Root root)
{
_root = root;
}
@@ -30,8 +29,6 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
for (var index = 0; index < _root.Persons.Length; index++)
{
var person = _root.Persons[index];
-
- // Triggers 2 changes: FirstName and FullName
person.FirstName = Stopwatch.GetTimestamp().ToString();
if (index % mod == 0) // distribute updates over approx. 0.5s
diff --git a/src/Namotion.Interceptor.OpcUa.Tests/Client/Polling/PollingCircuitBreakerTests.cs b/src/Namotion.Interceptor.Sources.Tests/CircuitBreakerTests.cs
similarity index 76%
rename from src/Namotion.Interceptor.OpcUa.Tests/Client/Polling/PollingCircuitBreakerTests.cs
rename to src/Namotion.Interceptor.Sources.Tests/CircuitBreakerTests.cs
index c85654b3..77cd2437 100644
--- a/src/Namotion.Interceptor.OpcUa.Tests/Client/Polling/PollingCircuitBreakerTests.cs
+++ b/src/Namotion.Interceptor.Sources.Tests/CircuitBreakerTests.cs
@@ -1,21 +1,21 @@
-using Namotion.Interceptor.OpcUa.Client.Polling;
+using Namotion.Interceptor.Sources.Resilience;
-namespace Namotion.Interceptor.OpcUa.Tests.Client.Polling;
+namespace Namotion.Interceptor.Sources.Tests;
///
-/// Tests for PollingCircuitBreaker focusing on thread-safety and correctness.
+/// Tests for CircuitBreaker focusing on thread-safety and correctness.
/// Verifies that the circuit breaker handles concurrent access correctly and prevents race conditions.
///
-public class PollingCircuitBreakerTests
+public class CircuitBreakerTests
{
[Fact]
public void Constructor_WithInvalidThreshold_ThrowsArgumentOutOfRangeException()
{
// Act & Assert
Assert.Throws(() =>
- new PollingCircuitBreaker(failureThreshold: 0, cooldownPeriod: TimeSpan.FromSeconds(5)));
+ new CircuitBreaker(failureThreshold: 0, cooldownPeriod: TimeSpan.FromSeconds(5)));
Assert.Throws(() =>
- new PollingCircuitBreaker(failureThreshold: -1, cooldownPeriod: TimeSpan.FromSeconds(5)));
+ new CircuitBreaker(failureThreshold: -1, cooldownPeriod: TimeSpan.FromSeconds(5)));
}
[Fact]
@@ -23,16 +23,16 @@ public void Constructor_WithInvalidCooldown_ThrowsArgumentOutOfRangeException()
{
// Act & Assert
Assert.Throws(() =>
- new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.Zero));
+ new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.Zero));
Assert.Throws(() =>
- new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(-1)));
+ new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(-1)));
}
[Fact]
public void ShouldAttempt_InitialState_ReturnsTrue()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
// Act
var result = breaker.ShouldAttempt();
@@ -46,7 +46,7 @@ public void ShouldAttempt_InitialState_ReturnsTrue()
public void RecordFailure_BelowThreshold_DoesNotOpenCircuit()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
// Act
var tripped1 = breaker.RecordFailure();
@@ -63,7 +63,7 @@ public void RecordFailure_BelowThreshold_DoesNotOpenCircuit()
public void RecordFailure_AtThreshold_OpensCircuit()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
// Act
breaker.RecordFailure();
@@ -81,7 +81,7 @@ public void RecordFailure_AtThreshold_OpensCircuit()
public void RecordSuccess_ResetsFailureCount()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
// Act - Record some failures, then success
breaker.RecordFailure();
@@ -99,7 +99,7 @@ public void RecordSuccess_ResetsFailureCount()
public void RecordSuccess_ClosesOpenCircuit()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 2, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 2, cooldownPeriod: TimeSpan.FromSeconds(5));
breaker.RecordFailure();
breaker.RecordFailure(); // Trip circuit
Assert.True(breaker.IsOpen);
@@ -116,7 +116,7 @@ public void RecordSuccess_ClosesOpenCircuit()
public void ShouldAttempt_DuringCooldown_ReturnsFalse()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(10));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(10));
breaker.RecordFailure(); // Trip circuit
// Act & Assert
@@ -128,7 +128,7 @@ public void ShouldAttempt_DuringCooldown_ReturnsFalse()
public async Task ShouldAttempt_AfterCooldown_ReturnsTrue()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(100));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(100));
breaker.RecordFailure(); // Trip circuit
Assert.True(breaker.IsOpen);
@@ -144,7 +144,7 @@ public async Task ShouldAttempt_AfterCooldown_ReturnsTrue()
public async Task ShouldAttempt_AfterCooldown_ClosesOnSuccess()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(100));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(100));
breaker.RecordFailure(); // Trip circuit
await Task.Delay(150); // Wait for cooldown
@@ -160,7 +160,7 @@ public async Task ShouldAttempt_AfterCooldown_ClosesOnSuccess()
public void GetCooldownRemaining_WhenClosed_ReturnsZero()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
// Act
var remaining = breaker.GetCooldownRemaining();
@@ -173,7 +173,7 @@ public void GetCooldownRemaining_WhenClosed_ReturnsZero()
public void GetCooldownRemaining_WhenOpen_ReturnsPositiveValue()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(10));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(10));
breaker.RecordFailure(); // Trip circuit
// Act
@@ -188,7 +188,7 @@ public void GetCooldownRemaining_WhenOpen_ReturnsPositiveValue()
public void Reset_ClearsStateAndClosesCircuit()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(10));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(10));
breaker.RecordFailure(); // Trip circuit
Assert.True(breaker.IsOpen);
@@ -205,7 +205,7 @@ public void Reset_ClearsStateAndClosesCircuit()
public void RecordFailure_MultipleTrips_IncrementsTripCount()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(1));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromSeconds(1));
// Act - Trip multiple times
breaker.RecordFailure(); // Trip 1
@@ -226,7 +226,7 @@ public void RecordFailure_MultipleTrips_IncrementsTripCount()
public async Task ConcurrentShouldAttempt_AfterCooldown_IsThreadSafe()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(100));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(100));
breaker.RecordFailure(); // Trip circuit
await Task.Delay(150); // Wait for cooldown
@@ -248,7 +248,7 @@ public async Task ConcurrentShouldAttempt_AfterCooldown_IsThreadSafe()
public async Task ConcurrentRecordFailure_OnlyTripsOnce()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 3, cooldownPeriod: TimeSpan.FromSeconds(5));
// Act - Multiple threads record failures concurrently
var tasks = Enumerable.Range(0, 10)
@@ -266,7 +266,7 @@ public async Task ConcurrentRecordFailure_OnlyTripsOnce()
public async Task ConcurrentRecordSuccessAndFailure_IsThreadSafe()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 100, cooldownPeriod: TimeSpan.FromSeconds(5));
+ var breaker = new CircuitBreaker(failureThreshold: 100, cooldownPeriod: TimeSpan.FromSeconds(5));
var random = new Random();
// Act - Mix of concurrent successes and failures
@@ -298,7 +298,7 @@ public async Task ConcurrentRecordSuccessAndFailure_IsThreadSafe()
public async Task RecordFailure_WhenAlreadyOpen_DoesNotIncrementTripCount()
{
// Arrange
- var breaker = new PollingCircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(50));
+ var breaker = new CircuitBreaker(failureThreshold: 1, cooldownPeriod: TimeSpan.FromMilliseconds(50));
// Act
var tripped1 = breaker.RecordFailure(); // First trip
diff --git a/src/Namotion.Interceptor.Sources/RegisteredSubjectPropertyExtensions.cs b/src/Namotion.Interceptor.Sources/RegisteredSubjectPropertyExtensions.cs
index 054b06fc..82fe9e2f 100644
--- a/src/Namotion.Interceptor.Sources/RegisteredSubjectPropertyExtensions.cs
+++ b/src/Namotion.Interceptor.Sources/RegisteredSubjectPropertyExtensions.cs
@@ -14,6 +14,7 @@ public static class RegisteredSubjectPropertyExtensions
/// The new value.
public static void SetValueFromSource(this RegisteredSubjectProperty property, object source, DateTimeOffset? timestamp, object? value)
{
+ // TODO: Remove this method and only provide the one on PropertyReference? (also missing received timestamp parameter)
property.Reference.SetValueFromSource(source, timestamp, null, value);
}
}
diff --git a/src/Namotion.Interceptor.OpcUa/Client/Polling/PollingCircuitBreaker.cs b/src/Namotion.Interceptor.Sources/Resilience/CircuitBreaker.cs
similarity index 81%
rename from src/Namotion.Interceptor.OpcUa/Client/Polling/PollingCircuitBreaker.cs
rename to src/Namotion.Interceptor.Sources/Resilience/CircuitBreaker.cs
index 99686b50..8f118385 100644
--- a/src/Namotion.Interceptor.OpcUa/Client/Polling/PollingCircuitBreaker.cs
+++ b/src/Namotion.Interceptor.Sources/Resilience/CircuitBreaker.cs
@@ -1,10 +1,14 @@
-namespace Namotion.Interceptor.OpcUa.Client.Polling;
+using System;
+using System.Threading;
+
+namespace Namotion.Interceptor.Sources.Resilience;
///
-/// Circuit breaker for polling operations that prevents resource exhaustion during persistent failures.
+/// Circuit breaker pattern implementation that prevents resource exhaustion during persistent failures.
/// Thread-safe implementation using volatile fields and interlocked operations.
+/// Can be used by any source implementation (MQTT, OPC UA, HTTP, etc.) to handle reconnection failures.
///
-internal sealed class PollingCircuitBreaker
+public sealed class CircuitBreaker
{
private readonly int _failureThreshold;
private readonly TimeSpan _cooldownPeriod;
@@ -14,7 +18,12 @@ internal sealed class PollingCircuitBreaker
private long _circuitOpenedAtTicks; // Using long (ticks) for atomic operations instead of DateTimeOffset struct
private long _tripCount;
- public PollingCircuitBreaker(int failureThreshold, TimeSpan cooldownPeriod)
+ ///
+ /// Creates a new circuit breaker.
+ ///
+ /// Number of consecutive failures before opening the circuit.
+ /// Time to wait before allowing retry after circuit opens.
+ public CircuitBreaker(int failureThreshold, TimeSpan cooldownPeriod)
{
if (failureThreshold <= 0)
throw new ArgumentOutOfRangeException(nameof(failureThreshold), "Failure threshold must be greater than zero");
@@ -33,6 +42,7 @@ public PollingCircuitBreaker(int failureThreshold, TimeSpan cooldownPeriod)
///
/// Gets the total number of times the circuit breaker has tripped.
+ /// Useful for monitoring and alerting.
///
public long TripCount => Interlocked.Read(ref _tripCount);
@@ -62,6 +72,7 @@ public bool ShouldAttempt()
///
/// Gets the time remaining until the circuit breaker cooldown completes.
/// Returns TimeSpan.Zero if the circuit is closed.
+ /// Useful for logging and waiting.
///
public TimeSpan GetCooldownRemaining()
{
@@ -79,6 +90,7 @@ public TimeSpan GetCooldownRemaining()
///
/// Records a successful operation, closing the circuit and resetting the failure count atomically.
+ /// Call this after a successful connection/operation attempt.
///
public void RecordSuccess()
{
@@ -91,6 +103,7 @@ public void RecordSuccess()
///
/// Records a failed operation, potentially opening the circuit if threshold is reached.
/// Returns true if the circuit was opened as a result of this failure.
+ /// Call this after each failed connection/operation attempt.
///
public bool RecordFailure()
{
@@ -110,7 +123,8 @@ public bool RecordFailure()
}
///
- /// Resets the circuit breaker to closed state with zero failures.
+ /// Manually resets the circuit breaker to closed state with zero failures.
+ /// Useful for manual recovery or testing.
///
public void Reset()
{
diff --git a/src/Namotion.Interceptor.Sources/SubjectPropertyWriter.cs b/src/Namotion.Interceptor.Sources/SubjectPropertyWriter.cs
index b10852e4..b549e90c 100644
--- a/src/Namotion.Interceptor.Sources/SubjectPropertyWriter.cs
+++ b/src/Namotion.Interceptor.Sources/SubjectPropertyWriter.cs
@@ -105,6 +105,9 @@ public async Task CompleteInitializationAsync(CancellationToken cancellationToke
/// The update action to apply to the subject.
public void Write(TState state, Action update)
{
+ // Hot path optimization: plain read (no volatile read) is fastest.
+ // Changes to _updates are rare (only during initialization/reconnection).
+ // If we see stale non-null during transition, we take lock and re-check - still correct.
var updates = _updates;
if (updates is not null)
{
diff --git a/src/Namotion.Interceptor.sln b/src/Namotion.Interceptor.sln
index b60d9425..b3909215 100644
--- a/src/Namotion.Interceptor.sln
+++ b/src/Namotion.Interceptor.sln
@@ -77,12 +77,18 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.OpcUa.
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.OpcUa.SampleServer", "Namotion.Interceptor.OpcUa.SampleServer\Namotion.Interceptor.OpcUa.SampleServer.csproj", "{9AAB5E5D-B32D-440C-87DC-5BCB72F25E08}"
EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.OpcUa.SampleModel", "Namotion.Interceptor.OpcUa.SampleModel\Namotion.Interceptor.OpcUa.SampleModel.csproj", "{D99E51C9-2F57-448C-B33C-3FCEA68058E9}"
-EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.Dynamic", "Namotion.Interceptor.Dynamic\Namotion.Interceptor.Dynamic.csproj", "{DD4E4986-56F1-4A07-A857-B71CBF0C453C}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.Dynamic.Tests", "Namotion.Interceptor.Dynamic.Tests\Namotion.Interceptor.Dynamic.Tests.csproj", "{288013AC-398A-4ECC-B793-E223C2EDBAC4}"
EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.Mqtt.SampleClient", "Namotion.Interceptor.Mqtt.SampleClient\Namotion.Interceptor.Mqtt.SampleClient.csproj", "{BE44817A-BD7B-4427-A5FF-1DC6B9D634A2}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.Mqtt.SampleServer", "Namotion.Interceptor.Mqtt.SampleServer\Namotion.Interceptor.Mqtt.SampleServer.csproj", "{2F54E37B-8A8D-4DB0-A28D-B1E2CC635296}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.SamplesModel", "Namotion.Interceptor.SamplesModel\Namotion.Interceptor.SamplesModel.csproj", "{1BF9824C-D211-493A-9091-782765FCC4F0}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Namotion.Interceptor.Mqtt.Tests", "Namotion.Interceptor.Mqtt.Tests\Namotion.Interceptor.Mqtt.Tests.csproj", "{7F88375F-0C6A-4D00-BB3D-4CC10EFD06D2}"
+EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@@ -201,10 +207,6 @@ Global
{9AAB5E5D-B32D-440C-87DC-5BCB72F25E08}.Debug|Any CPU.Build.0 = Debug|Any CPU
{9AAB5E5D-B32D-440C-87DC-5BCB72F25E08}.Release|Any CPU.ActiveCfg = Release|Any CPU
{9AAB5E5D-B32D-440C-87DC-5BCB72F25E08}.Release|Any CPU.Build.0 = Release|Any CPU
- {D99E51C9-2F57-448C-B33C-3FCEA68058E9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
- {D99E51C9-2F57-448C-B33C-3FCEA68058E9}.Debug|Any CPU.Build.0 = Debug|Any CPU
- {D99E51C9-2F57-448C-B33C-3FCEA68058E9}.Release|Any CPU.ActiveCfg = Release|Any CPU
- {D99E51C9-2F57-448C-B33C-3FCEA68058E9}.Release|Any CPU.Build.0 = Release|Any CPU
{DD4E4986-56F1-4A07-A857-B71CBF0C453C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{DD4E4986-56F1-4A07-A857-B71CBF0C453C}.Debug|Any CPU.Build.0 = Debug|Any CPU
{DD4E4986-56F1-4A07-A857-B71CBF0C453C}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -213,6 +215,22 @@ Global
{288013AC-398A-4ECC-B793-E223C2EDBAC4}.Debug|Any CPU.Build.0 = Debug|Any CPU
{288013AC-398A-4ECC-B793-E223C2EDBAC4}.Release|Any CPU.ActiveCfg = Release|Any CPU
{288013AC-398A-4ECC-B793-E223C2EDBAC4}.Release|Any CPU.Build.0 = Release|Any CPU
+ {BE44817A-BD7B-4427-A5FF-1DC6B9D634A2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {BE44817A-BD7B-4427-A5FF-1DC6B9D634A2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {BE44817A-BD7B-4427-A5FF-1DC6B9D634A2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {BE44817A-BD7B-4427-A5FF-1DC6B9D634A2}.Release|Any CPU.Build.0 = Release|Any CPU
+ {2F54E37B-8A8D-4DB0-A28D-B1E2CC635296}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {2F54E37B-8A8D-4DB0-A28D-B1E2CC635296}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {2F54E37B-8A8D-4DB0-A28D-B1E2CC635296}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {2F54E37B-8A8D-4DB0-A28D-B1E2CC635296}.Release|Any CPU.Build.0 = Release|Any CPU
+ {1BF9824C-D211-493A-9091-782765FCC4F0}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {1BF9824C-D211-493A-9091-782765FCC4F0}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {1BF9824C-D211-493A-9091-782765FCC4F0}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {1BF9824C-D211-493A-9091-782765FCC4F0}.Release|Any CPU.Build.0 = Release|Any CPU
+ {7F88375F-0C6A-4D00-BB3D-4CC10EFD06D2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {7F88375F-0C6A-4D00-BB3D-4CC10EFD06D2}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {7F88375F-0C6A-4D00-BB3D-4CC10EFD06D2}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {7F88375F-0C6A-4D00-BB3D-4CC10EFD06D2}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@@ -237,9 +255,12 @@ Global
{D266C5C9-D82A-4E4E-9ABB-8C1941295A08} = {31C2A126-61B9-402D-B2F4-9772341EBF09}
{993D95D7-2BB6-4414-8F8A-C097B40173D4} = {D977E1EA-C167-4BEB-B7C5-4214DDD470D3}
{9AAB5E5D-B32D-440C-87DC-5BCB72F25E08} = {D977E1EA-C167-4BEB-B7C5-4214DDD470D3}
- {D99E51C9-2F57-448C-B33C-3FCEA68058E9} = {D977E1EA-C167-4BEB-B7C5-4214DDD470D3}
{288013AC-398A-4ECC-B793-E223C2EDBAC4} = {31C2A126-61B9-402D-B2F4-9772341EBF09}
{8B5D2A7F-3C9E-4F8A-9D5B-1E7F8C9A6B2D} = {31C2A126-61B9-402D-B2F4-9772341EBF09}
+ {BE44817A-BD7B-4427-A5FF-1DC6B9D634A2} = {D977E1EA-C167-4BEB-B7C5-4214DDD470D3}
+ {2F54E37B-8A8D-4DB0-A28D-B1E2CC635296} = {D977E1EA-C167-4BEB-B7C5-4214DDD470D3}
+ {1BF9824C-D211-493A-9091-782765FCC4F0} = {D977E1EA-C167-4BEB-B7C5-4214DDD470D3}
+ {7F88375F-0C6A-4D00-BB3D-4CC10EFD06D2} = {31C2A126-61B9-402D-B2F4-9772341EBF09}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {D7E90F8C-B0AE-4ACF-9226-306D8EA3D35B}