Protocol: add support for BecomeMonitor. (#217)

.github/workflows/build.yaml

@@ -12,7 +12,7 @@ jobs:
       - name: Install .NET
         uses: actions/setup-dotnet@v3
         with:
-          dotnet-version: "7.0.x"
+          dotnet-version: "8.0.x"
 
       - name: Install reportgenerator tool
         run: dotnet tool install --global dotnet-reportgenerator-globaltool

samples/Monitor/Monitor.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Tmds.DBus.Protocol\Tmds.DBus.Protocol.csproj" />
+  </ItemGroup>
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+</Project>

samples/Monitor/Program.cs

@@ -0,0 +1,11 @@
+using Tmds.DBus.Protocol;
+
+string address = Address.Session ?? throw new ArgumentNullException(nameof(address));
+
+await foreach (DisposableMessage dmsg in Connection.MonitorBusAsync(address))
+{
+    using var _ = dmsg;
+    Message msg = dmsg.Message;
+
+    Console.WriteLine($"{msg.MessageType} {msg.SenderAsString} -> {msg.DestinationAsString}");
+}

src/Tmds.DBus.Protocol/ClientConnectionOptions.cs

@@ -18,6 +18,8 @@ protected ClientConnectionOptions()
 
     public bool AutoConnect { get; set; }
 
+    internal bool IsShared { get; set; }
+
     protected internal virtual ValueTask<ClientSetupResult> SetupAsync(CancellationToken cancellationToken)
     {
         return new ValueTask<ClientSetupResult>(

src/Tmds.DBus.Protocol/Connection.DBus.cs

@@ -1,3 +1,5 @@
+using System.Threading.Channels;
+
 namespace Tmds.DBus.Protocol;
 
 public partial class Connection
@@ -20,4 +22,76 @@ MessageBuffer CreateMessage()
             return writer.CreateMessage();
         }
     }
+
+    public async Task BecomeMonitorAsync(Action<Exception?, DisposableMessage> handler, IEnumerable<MatchRule>? rules = null)
+    {
+        if (_connectionOptions.IsShared)
+        {
+            throw new InvalidOperationException("Cannot become monitor on a shared connection.");
+        }
+
+        DBusConnection connection = await ConnectCoreAsync().ConfigureAwait(false);
+        await connection.BecomeMonitorAsync(handler, rules).ConfigureAwait(false);
+    }
+
+    public static async IAsyncEnumerable<DisposableMessage> MonitorBusAsync(string address, IEnumerable<MatchRule>? rules = null, [EnumeratorCancellation]CancellationToken ct = default)
+    {
+        ct.ThrowIfCancellationRequested();
+
+        var channel = Channel.CreateUnbounded<DisposableMessage>(
+            new UnboundedChannelOptions()
+            {
+                AllowSynchronousContinuations = true,
+                SingleReader = true,
+                SingleWriter = true,
+            }
+        );
+
+        using var connection = new Connection(address);
+        using CancellationTokenRegistration ctr =
+#if NETCOREAPP3_1_OR_GREATER
+                ct.UnsafeRegister(c => ((Connection)c!).Dispose(), connection);
+#else
+                ct.Register(c => ((Connection)c!).Dispose(), connection);
+#endif
+        try
+        {
+            await connection.ConnectAsync().ConfigureAwait(false);
+
+            await connection.BecomeMonitorAsync(
+                (Exception? ex, DisposableMessage message) =>
+                {
+                    if (ex is not null)
+                    {
+                        if (ct.IsCancellationRequested)
+                        {
+                            ex = new OperationCanceledException(ct);
+                        }
+                        channel.Writer.TryComplete(ex);
+                        return;
+                    }
+
+                    if (!channel.Writer.TryWrite(message))
+                    {
+                        message.Dispose();
+                    }
+                },
+                rules
+            ).ConfigureAwait(false);
+        }
+        catch
+        {
+            ct.ThrowIfCancellationRequested();
+
+            throw;
+        }
+
+        while (await channel.Reader.WaitToReadAsync().ConfigureAwait(false))
+        {
+            if (channel.Reader.TryRead(out DisposableMessage msg))
+            {
+                yield return msg;
+            }
+        }
+    }
 }

src/Tmds.DBus.Protocol/Connection.cs

@@ -249,7 +249,7 @@ private static Connection CreateConnection(ref Connection? field, string? addres
         {
             return connection;
         }
-        var newConnection = new Connection(new ClientConnectionOptions(address) { AutoConnect = true });
+        var newConnection = new Connection(new ClientConnectionOptions(address) { AutoConnect = true, IsShared = true });
         connection = Interlocked.CompareExchange(ref field, newConnection, null);
         if (connection != null)
         {