maier_st/Sharp7Reactive
1
0
Fork 0
mirror of https://github.com/evopro-ag/Sharp7Reactive.git synced 2025-12-16 19:52:53 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Changed from select to select many implementation in transfer with handshake

Browse Source
This commit is contained in:
Stefan Werner
2019-09-20 15:29:24 +02:00
parent 999ad76825
commit e3600e30d1
2 changed files with 7 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
Sharp7.Rx/Extensions/ObservableExtensions.cs
View File

@@ -16,20 +16,6 @@ namespace Sharp7.Rx.Extensions
{ {
internal static class ObservableExtensions internal static class ObservableExtensions
{ {
public static IObservable<Unit> Select<TSource>(this IObservable<TSource> source, Func<TSource, Task> selector)
{
return source
.Select(x => Observable.FromAsync(async () => await selector(x)))
.Concat();
}
public static IObservable<TResult> Select<TSource, TResult>(this IObservable<TSource> source, Func<TSource, Task<TResult>> selector)
{
return source
.Select(x => Observable.FromAsync(async () => await selector(x)))
.Concat();
}
public static IObservable<T> LogAndRetry<T>(this IObservable<T> source, ILogger logger, string message) public static IObservable<T> LogAndRetry<T>(this IObservable<T> source, ILogger logger, string message)
{ {
return source return source
@@ -101,14 +87,5 @@ namespace Sharp7.Rx.Extensions
return new CompositeDisposable(serialDisposable, subscription); return new CompositeDisposable(serialDisposable, subscription);
}); });
} }
public static IObservable<Unit> SelectMany<T>(this IObservable<T> source, Func<T, Task> selector)
{
return source.SelectMany(async item =>
{
await selector(item);
return Unit.Default;
});
}
} }
} }

9
Sharp7.Rx/Extensions/PlcExtensions.cs
View File

@@ -1,4 +1,5 @@
using System; using System;
using System.Reactive;
using System.Reactive.Disposables; using System.Reactive.Disposables;
using System.Reactive.Linq; using System.Reactive.Linq;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -28,13 +29,17 @@ namespace Sharp7.Rx.Extensions
notification notification
.Where(trigger => trigger) .Where(trigger => trigger)
.Select(_ => ReadDataAndAcknowlodge(plc, readData, ackTriggerAddress)) .SelectMany(_ => ReadDataAndAcknowlodge(plc, readData, ackTriggerAddress))
.Subscribe(observer) .Subscribe(observer)
.AddDisposableTo(subscriptions); .AddDisposableTo(subscriptions);
notification notification
.Where(trigger => !trigger) .Where(trigger => !trigger)
.Select(_ => plc.SetValue(ackTriggerAddress, false)) .SelectMany(async _ =>
{
await plc.SetValue(ackTriggerAddress, false);
return Unit.Default;
})
.Subscribe() .Subscribe()
.AddDisposableTo(subscriptions); .AddDisposableTo(subscriptions);