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

Merge branch 'master' of https://github.com/evopro-ag/Sharp7Reactive

Browse Source
This commit is contained in:
Federico Barresi
2019-11-21 14:43:03 +01:00
parent 0f3125fe20 e3600e30d1
commit fd7dd01a5e
3 changed files with 9 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
{
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)
{
return source
@@ -101,14 +87,5 @@ namespace Sharp7.Rx.Extensions
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.Reactive;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Threading.Tasks;
@@ -28,13 +29,17 @@ namespace Sharp7.Rx.Extensions
notification
.Where(trigger => trigger)
.Select(_ => ReadDataAndAcknowlodge(plc, readData, ackTriggerAddress))
.SelectMany(_ => ReadDataAndAcknowlodge(plc, readData, ackTriggerAddress))
.Subscribe(observer)
.AddDisposableTo(subscriptions);
notification
.Where(trigger => !trigger)
.Select(_ => plc.SetValue(ackTriggerAddress, false))
.SelectMany(async _ =>
{
await plc.SetValue(ackTriggerAddress, false);
return Unit.Default;
})
.Subscribe()
.AddDisposableTo(subscriptions);

2
Sharp7.Rx/Sharp7.Rx.csproj
View File

@@ -10,6 +10,8 @@
<Description>Reactive framework for Sharp7, the Ethernet S7 PLC communication suite</Description>
<PackageProjectUrl>https://github.com/evopro-ag/Sharp7Reactive</PackageProjectUrl>
<PackageLicenseUrl>https://raw.githubusercontent.com/evopro-ag/Sharp7Reactive/master/LICENSE</PackageLicenseUrl>
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
</PropertyGroup>
<ItemGroup>