diff --git a/Knx.sln.DotSettings b/Knx.sln.DotSettings
index 4f9ff32..774accb 100644
--- a/Knx.sln.DotSettings
+++ b/Knx.sln.DotSettings
@@ -1,2 +1,3 @@
- IP
\ No newline at end of file
+ IP
+ True
\ No newline at end of file
diff --git a/Knx/ExtendedMessageInterface/KnxMessage.cs b/Knx/ExtendedMessageInterface/KnxMessage.cs
index 1ad4cab..4f9c480 100644
--- a/Knx/ExtendedMessageInterface/KnxMessage.cs
+++ b/Knx/ExtendedMessageInterface/KnxMessage.cs
@@ -78,13 +78,13 @@ public bool IsPositiveConfirmation
/// Gets or sets the source (device-) address.
///
/// The source address.
- public KnxDeviceAddress? SourceAddress { get; set; } = "0/0/0";
+ public KnxDeviceAddress SourceAddress { get; set; } = "0/0/0";
///
/// Gets or sets the destination address (device or individual/logical address).
///
/// The destination address.
- public KnxAddress? DestinationAddress { get; set; }
+ public KnxAddress DestinationAddress { get; set; } = (KnxLogicalAddress)"0/0/0";
///
/// Gets the hop count, a message did, till it was received.
@@ -255,4 +255,4 @@ public static KnxMessage Deserialize(byte[] bytes)
MessageType = msgType
};
}
-}
\ No newline at end of file
+}
diff --git a/Knx/KnxNetIp/IKnxNetIpClient.cs b/Knx/KnxNetIp/IKnxNetIpClient.cs
index ec6d848..4f33197 100644
--- a/Knx/KnxNetIp/IKnxNetIpClient.cs
+++ b/Knx/KnxNetIp/IKnxNetIpClient.cs
@@ -1,5 +1,4 @@
using System;
-using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Tasks;
@@ -7,8 +6,8 @@ namespace Knx.KnxNetIp;
public interface IKnxNetIpClient :
IAsyncDisposable,
- ISubject,
- ISubject
+ IObservable,
+ IObservable
{
Task ConnectAsync();
diff --git a/Knx/KnxNetIp/KnxNetIpRoutingClient.cs b/Knx/KnxNetIp/KnxNetIpRoutingClient.cs
index 3d192ad..b4fe661 100644
--- a/Knx/KnxNetIp/KnxNetIpRoutingClient.cs
+++ b/Knx/KnxNetIp/KnxNetIpRoutingClient.cs
@@ -14,7 +14,7 @@ namespace Knx.KnxNetIp;
///
/// Used to connect to the Knx Bus via KnxNetIpRouting protocol.
///
-public sealed class KnxNetIpRoutingClient : IKnxNetIpClient
+public sealed class KnxNetIpRoutingClient : IKnxNetIpClient, IObservable
{
private readonly KnxDeviceAddress _deviceAddress;
private readonly IPEndPoint _localEndpoint;
@@ -22,8 +22,10 @@ public sealed class KnxNetIpRoutingClient : IKnxNetIpClient
private readonly CancellationTokenSource _receivingMessagesCancellationTokenSource;
private readonly IPEndPoint _remoteEndPoint;
private UdpClient? _udpClient;
+
private readonly Subject _knxNetIpMessageSubject;
private readonly Subject _knxMessageSubject;
+ private readonly Subject _knxHpaiSubject;
///
/// Creates a new instance of the class.
@@ -48,6 +50,7 @@ public KnxNetIpRoutingClient(
_knxNetIpMessageSubject = new Subject();
_knxMessageSubject = new Subject();
+ _knxHpaiSubject = new Subject();
}
public async ValueTask DisposeAsync()
@@ -97,15 +100,14 @@ public async Task ConnectAsync()
await Task.CompletedTask;
}
- IDisposable IObservable.Subscribe(IObserver observer)
- {
- throw new NotImplementedException();
- }
+ IDisposable IObservable.Subscribe(IObserver observer) =>
+ _knxNetIpMessageSubject.Subscribe(observer);
- IDisposable IObservable.Subscribe(IObserver observer)
- {
- throw new NotImplementedException();
- }
+ IDisposable IObservable.Subscribe(IObserver observer) =>
+ _knxMessageSubject.Subscribe(observer);
+
+ IDisposable IObservable.Subscribe(IObserver observer) =>
+ _knxHpaiSubject.Subscribe(observer);
~KnxNetIpRoutingClient()
{
@@ -122,6 +124,7 @@ private ValueTask Dispose(bool disposing)
_knxNetIpMessageSubject.Dispose();
_knxMessageSubject.Dispose();
+ _knxHpaiSubject.Dispose();
}
return ValueTask.CompletedTask;
@@ -230,24 +233,9 @@ private void InvokeKnxMessageReceived(IKnxMessage knxMessage)
_knxMessageSubject.OnNext(knxMessage);
}
- private void InvokeKnxDeviceDiscovered(KnxHpai hostProtocolAddressInformation) =>
+ private void InvokeKnxDeviceDiscovered(KnxHpai hostProtocolAddressInformation)
+ {
KnxDeviceDiscovered?.Invoke(this, hostProtocolAddressInformation);
-
- void IObserver.OnCompleted() =>
- _knxNetIpMessageSubject.OnCompleted();
-
- void IObserver.OnError(Exception error) =>
- _knxMessageSubject.OnError(error);
-
- void IObserver.OnNext(IKnxMessage value) =>
- _knxMessageSubject.OnNext(value);
-
- void IObserver.OnCompleted() =>
- _knxMessageSubject.OnCompleted();
-
- void IObserver.OnError(Exception error) =>
- _knxNetIpMessageSubject.OnError(error);
-
- void IObserver.OnNext(KnxNetIpMessage value) =>
- _knxNetIpMessageSubject.OnNext(value);
+ _knxHpaiSubject.OnNext(hostProtocolAddressInformation);
+ }
}
diff --git a/Knx/KnxNetIp/KnxNetIpTunnelingClient.cs b/Knx/KnxNetIp/KnxNetIpTunnelingClient.cs
index 2b9707d..af30746 100644
--- a/Knx/KnxNetIp/KnxNetIpTunnelingClient.cs
+++ b/Knx/KnxNetIp/KnxNetIpTunnelingClient.cs
@@ -273,7 +273,8 @@ private void OnKnxNetIpMessageReceived(KnxNetIpMessage message)
}
KnxNetIpMessageReceived?.Invoke(this, message);
- ((ISubject)this).OnNext(message);
+
+ _knxNetIpMessageSubject.OnNext(message);
}
private async Task DisposeAsync(bool disposing)
@@ -512,22 +513,4 @@ private void SetSequenceCount(TunnelingMessageBody knxTunnelingMessageBody)
_sequenceCounter++;
}
}
-
- void IObserver.OnCompleted() =>
- _knxNetIpMessageSubject.OnCompleted();
-
- void IObserver.OnError(Exception error) =>
- _knxMessageSubject.OnError(error);
-
- void IObserver.OnNext(IKnxMessage value) =>
- _knxMessageSubject.OnNext(value);
-
- void IObserver.OnCompleted() =>
- _knxMessageSubject.OnCompleted();
-
- void IObserver.OnError(Exception error) =>
- _knxNetIpMessageSubject.OnError(error);
-
- void IObserver.OnNext(KnxNetIpMessage value) =>
- _knxNetIpMessageSubject.OnNext(value);
}