Add option for running continuations asynchronously.

docs/howto.md

@@ -1,5 +1,29 @@
 # How-to
 
+## Threading
+
+D-Bus APIs assume peers will consume the messages from a single logical thread.
+
+For example, this preserves the ordering of values from `PropertyChanged` signals, and method returns of `Properties.Get`.
+
+By design .NET Tasks do not preserve the ordering because (a.) a Task can complete
+synchronously, and (b.) a Task can complete on the threadpool when using `TaskCreationOptions.RunContinuationsAsynchronously`.
+
+`Tmds.DBus` supports two modes of operations that preserve the ordering.
+
+When the application has a single-threaded `SynchronizationContext`, it can be set on `ConnectionOptions`.
+All signals will be emitted on that context. The user is assumed to be using that `SynchronizationContext` while making method calls,
+causing the completions to be executed on that context. This casues the `SynchronizationContext` to be used as the single logical thread.
+
+Otherwise, the `SynchronizationContext` should be kept at the default value of `null`.
+All signals will be emitted directly from the read loop, and all method continuations will be completed synchronously from that loop.
+This causes the read loop to be used as the single logical thread.
+The user must ensure the loop is not blocked by not making synchronous calls on the continuation.
+
+If you are writing some re-usable code (like a library), you can either let the user provide a `SynchronizationContext` and set it on `ConnectionOptions`. Or, you can choose to use a `null` `SynchronizationContext` and ensure all method are awaited `ConfigureAwait(false)`.
+
+If your API usage doesn't require ordering to be preserved, you can set the `RunContinuationsAsynchronously` `ConnectionOption` to `true`.
+
 ## Local Server
 
 Tmds.DBus supports running an in-process server that accepts connections. This allows other clients to connect

src/Tmds.DBus.Protocol/ClientConnectionOptions.cs

@@ -30,4 +30,8 @@ protected internal virtual ValueTask<ClientSetupResult> SetupAsync(CancellationT
 
     protected internal virtual void Teardown(object? token)
     { }
+
+    public SynchronizationContext? SynchronizationContext { get; set; } = null;
+
+    public bool RunContinuationsAsynchronously {get; set; } = false;
 }

src/Tmds.DBus.Protocol/ConnectionOptions.cs

@@ -4,8 +4,4 @@ public abstract class ConnectionOptions
 {
     internal ConnectionOptions()
     { }
-
-    public SynchronizationContext? SynchronizationContext { get; set; } = null;
-
-    public bool RunContinuationsAsynchronously {get; set; } = false;
 }

src/Tmds.DBus.Protocol/DBusConnection.cs

@@ -124,7 +124,7 @@ public void Invoke(Exception? exception, in Message message)
     private readonly List<Observer> _matchedObservers;
     private readonly Dictionary<string, IMethodHandler> _pathHandlers;
     private readonly SynchronizationContext? _synchronizationContext;
-    private readonly bool _runContinuationsAsynchronosly;
+    private readonly bool _runContinuationsAsynchronously;
 
     private IMessageStream? _messageStream;
     private ConnectionState _state;
@@ -146,7 +146,7 @@ public DBusConnection(Connection parent, SynchronizationContext? synchronization
     {
         _parentConnection = parent;
         _synchronizationContext = synchronizationContext;
-        _runContinuationsAsynchronosly = runContinuationsAsynchronously;
+        _runContinuationsAsynchronously = runContinuationsAsynchronously;
         _connectCts = new();
         _pendingCalls = new();
         _matchMakers = new();
@@ -238,7 +238,7 @@ public async ValueTask ConnectAsync(string address, string? userId, bool support
 
     private async Task<string?> GetLocalNameAsync()
     {
-        MyValueTaskSource<string?> vts = new(_runContinuationsAsynchronosly);
+        MyValueTaskSource<string?> vts = new(_runContinuationsAsynchronously);
 
         await CallMethodAsync(
             message: CreateHelloMessage(),
@@ -528,7 +528,7 @@ public async Task<T> CallMethodAsync<T>(MessageBuffer message, MessageValueReade
             }
         };
 
-        MyValueTaskSource<T> vts = new(_runContinuationsAsynchronosly);
+        MyValueTaskSource<T> vts = new(_runContinuationsAsynchronously);
         MessageHandler handler = new(fn, valueReader, vts, state);
 
         await CallMethodAsync(message, handler).ConfigureAwait(false);
@@ -538,7 +538,7 @@ public async Task<T> CallMethodAsync<T>(MessageBuffer message, MessageValueReade
 
     public async Task CallMethodAsync(MessageBuffer message)
     {
-        MyValueTaskSource<object?> vts = new(_runContinuationsAsynchronosly);
+        MyValueTaskSource<object?> vts = new(_runContinuationsAsynchronously);
 
         await CallMethodAsync(message,
             static (Exception? exception, in Message message, object? state) => CompleteCallValueTaskSource(exception, in message, state), vts).ConfigureAwait(false);
@@ -629,7 +629,7 @@ private async ValueTask<IDisposable> AddMatchAsync(MatchRule rule, MessageHandle
             if (sendMessage)
             {
                 addMatchMessage = CreateAddMatchMessage(matchMaker.RuleString);
-                matchMaker.AddMatchTcs = new(_runContinuationsAsynchronosly);
+                matchMaker.AddMatchTcs = new(_runContinuationsAsynchronously);
 
                 MessageHandlerDelegate fn = static (Exception? exception, in Message message, object? state1, object? state2, object? state3) =>
                 {

src/Tmds.DBus/ClientConnectionOptions.cs

@@ -56,5 +56,10 @@ protected internal virtual Task<ClientSetupResult> SetupAsync()
         /// </summary>
         protected internal virtual void Teardown(object token)
         {}
+
+        /// <summary>
+        /// Run Task continuations asynchronously.
+        /// </summary>
+        public bool RunContinuationsAsynchronously { get; set; }
     }
 }

src/Tmds.DBus/Connection.cs

@@ -64,6 +64,7 @@ public T CreateProxy<T>(string serviceName, ObjectPath path)
         private readonly Func<Task<ClientSetupResult>> _connectFunction;
         private readonly Action<object> _disposeAction;
         private readonly SynchronizationContext _synchronizationContext;
+        private readonly bool _runContinuationsAsynchronously;
         private readonly ConnectionType _connectionType;
 
         private ConnectionState _state = ConnectionState.Created;
@@ -147,12 +148,13 @@ public Connection(ConnectionOptions connectionOptions)
                 _connectionType = clientConnectionOptions.AutoConnect ? ConnectionType.ClientAutoConnect : ConnectionType.ClientManual ;
                 _connectFunction = clientConnectionOptions.SetupAsync;
                 _disposeAction = clientConnectionOptions.Teardown;
+                _runContinuationsAsynchronously = clientConnectionOptions.RunContinuationsAsynchronously;
             }
             else if (connectionOptions is ServerConnectionOptions serverConnectionOptions)
             {
                 _connectionType = ConnectionType.Server;
                 _state = ConnectionState.Connected;
-                _dbusConnection = new DBusConnection(localServer: true);
+                _dbusConnection = new DBusConnection(localServer: true, runContinuationsAsynchronously: false);
                 _dbusConnectionTask = Task.FromResult(_dbusConnection);
                 serverConnectionOptions.Connection = this;
             }
@@ -224,7 +226,7 @@ private async Task<DBusConnection> DoConnectAsync()
             {
                 ClientSetupResult connectionContext = await _connectFunction().ConfigureAwait(false);
                 disposeUserToken = connectionContext.TeardownToken;
-                connection = await DBusConnection.ConnectAsync(connectionContext, OnDisconnect, _connectCts.Token).ConfigureAwait(false);
+                connection = await DBusConnection.ConnectAsync(connectionContext, _runContinuationsAsynchronously, OnDisconnect, _connectCts.Token).ConfigureAwait(false);
             }
             catch (ConnectException ce)
             {