This commit is contained in:
ig 2023-10-02 15:36:50 +02:00
parent b7cda31b58
commit 7b7a17ca49
5 changed files with 36 additions and 22 deletions

View File

@ -73,7 +73,13 @@ public static class AsyncEnumerableEx
}
}
#pragma warning disable 1998
public static async void ForEach<T>(this IAsyncEnumerable<T> ts, Action<T> action)
{
await foreach (var t in ts)
action(t);
}
#pragma warning disable 1998
public static async IAsyncEnumerable<T> Repeat<T>(this T t, [EnumeratorCancellation] CancellationToken ct = default)
{
while(!ct.IsCancellationRequested)

View File

@ -59,17 +59,18 @@ public static class GraphTraversal
}
[SuppressMessage("ReSharper", "PossibleMultipleEnumeration")]
private static IEnumerable<T> Traverse<T>(IEnumerable<T> sources,
private static IEnumerable<T> Traverse<T>(
IEnumerable<T> sources,
Func<T , Func<T, IEnumerable<T>>,IEnumerable<T>> traversor,
Func<T, IEnumerable<T>> getChildren,
IEqualityComparer<T>? comparer = null)
{
var set = new HashSet<T>(sources, comparer ?? EqualityComparer<T>.Default);
IEnumerable<T> GetUniqueChildren(T n) => getChildren(n).Where(set.Add);
IEnumerable<T> GetUniqueChildren(T n) => getChildren(n).Where(set!.Add);
return from s in sources
from e in traversor(s, GetUniqueChildren)
select e;
from e in traversor(s, GetUniqueChildren)
select e;
}
// TODO: IEqualityComparer

View File

@ -1,5 +1,5 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
using System.Runtime.CompilerServices;
using static System.Runtime.CompilerServices.MethodImplOptions;
@ -171,4 +171,12 @@ public static class Utils
}
public static String ExecutingProcessName => Process.GetCurrentProcess().ProcessName;
public static Func<A, R> Memoize<A, R>(Func<A, R> func) where A : notnull
{
var cache = new ConcurrentDictionary<A, R>();
return arg => cache.GetOrAdd(arg, func);
}
}

View File

@ -43,22 +43,22 @@ public static class VePropertiesDbus
public static async Task<Exception> PublishOnDBus(this VeProperties veProperties, DBusConnection con)
{
var calls = con.IncomingMessages
.Where(m => m.Type == MessageType.MethodCall)
.Select(msg => AnswerMethodCall(veProperties, msg))
.Do(con.OutgoingMessages.OnNext);
.Where(m => m.Type == MessageType.MethodCall)
.Select(msg => AnswerMethodCall(veProperties, msg))
.Do(con.OutgoingMessages.OnNext);
var changes = veProperties
.PropertyChanged
.Do(con.BroadcastPropertiesChanged);
.PropertyChanged
.Do(con.BroadcastPropertiesChanged);
var initialPropertiesChanged = veProperties
.ToObservable(TaskPoolScheduler.Default)
.Do(con.BroadcastPropertiesChanged);
.ToObservable(TaskPoolScheduler.Default)
.Do(con.BroadcastPropertiesChanged);
return await calls
.MergeErrors(changes)
.MergeErrors(initialPropertiesChanged)
.Finally(con.Dispose);
.MergeErrors(changes)
.MergeErrors(initialPropertiesChanged)
.Finally(con.Dispose);
}
private static async Task<T> GetName<T>(this DBusDaemonConnection dbusConnection, String busName)
@ -107,11 +107,11 @@ public static class VePropertiesDbus
return msg.Member switch
{
VeDBusApi.GetValue => msg.Ok(p.Value),
VeDBusApi.GetText => msg.Ok(p.Text),
VeDBusApi.GetText => msg.Ok(p.Text),
VeDBusApi.GetItems => msg.Ok(p.GetItem()),
VeDBusApi.SetValue => (p.Writeable && msg.ObjectPath != null)
? msg.Ok((p.Value==msg.Payload! || props.Set(path: msg.ObjectPath.Value, value: msg.Payload!, writable: true)) ? 0 : -1)
: msg.ObjectNotWritable(msg.Member),
VeDBusApi.SetValue => p.Writeable && msg.ObjectPath != null
? msg.Ok(p.Value==msg.Payload! || props.Set(path: msg.ObjectPath.Value, value: msg.Payload!, writable: true) ? 0 : -1)
: msg.ObjectNotWritable(msg.Member),
_ => msg.UnknownMember(msg.Member)
};
}
@ -150,7 +150,6 @@ public static class VePropertiesDbus
return CreateErrorReply(msg, $"Not Writable Property: {objectPath}");
}
private static Message CreateErrorReply(Message msg, String error)
{
Debug.WriteLine("ERROR: " + error);