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

Cleanup

Browse Source
This commit is contained in:
Peter Butzhammer
2024-02-07 09:35:21 +01:00
parent c79e07be33
commit babbb1a6bc
20 changed files with 372 additions and 374 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Sharp7.Rx/Extensions/DisposableExtensions.cs
View File

@@ -10,4 +10,4 @@ namespace Sharp7.Rx.Extensions
compositeDisposable.Add(disposable);
}
}
}
}

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

@@ -1,91 +1,83 @@
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Linq;
using System.Linq.Expressions;
using System.Reactive;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Sharp7.Rx.Resources;
namespace Sharp7.Rx.Extensions
{
internal static class ObservableExtensions
{
public static IObservable<T> LogAndRetry<T>(this IObservable<T> source, ILogger logger, string message)
{
return source
.Do(
_ => { },
ex => logger?.LogError(ex, message))
.Retry();
}
internal static class ObservableExtensions
{
public static IObservable<T> DisposeMany<T>(this IObservable<T> source)
{
return Observable.Create<T>(obs =>
{
var serialDisposable = new SerialDisposable();
var subscription =
source.Subscribe(
item =>
{
serialDisposable.Disposable = item as IDisposable;
obs.OnNext(item);
},
obs.OnError,
obs.OnCompleted);
return new CompositeDisposable(serialDisposable, subscription);
});
}
public static IObservable<T> RetryAfterDelay<T>(
this IObservable<T> source,
TimeSpan retryDelay,
int retryCount = -1,
IScheduler scheduler = null)
{
return RedoAfterDelay(source, retryDelay, retryCount, scheduler, Observable.Retry, Observable.Retry);
}
public static IObservable<T> LogAndRetry<T>(this IObservable<T> source, ILogger logger, string message)
{
return source
.Do(
_ => { },
ex => logger?.LogError(ex, message))
.Retry();
}
public static IObservable<T> RepeatAfterDelay<T>(
this IObservable<T> source,
TimeSpan retryDelay,
int repeatCount = -1,
IScheduler scheduler = null)
{
return RedoAfterDelay(source, retryDelay, repeatCount, scheduler, Observable.Repeat, Observable.Repeat);
}
public static IObservable<T> LogAndRetryAfterDelay<T>(
this IObservable<T> source,
ILogger logger,
TimeSpan retryDelay,
string message,
int retryCount = -1,
IScheduler scheduler = null)
{
var sourceLogged =
source
.Do(
_ => { },
ex => logger?.LogError(ex, message));
public static IObservable<T> LogAndRetryAfterDelay<T>(
this IObservable<T> source,
ILogger logger,
TimeSpan retryDelay,
string message,
int retryCount = -1,
IScheduler scheduler = null)
{
var sourceLogged =
source
.Do(
_ => { },
ex => logger?.LogError(ex, message));
return RetryAfterDelay(sourceLogged, retryDelay, retryCount, scheduler);
}
return RetryAfterDelay(sourceLogged, retryDelay, retryCount, scheduler);
}
public static IObservable<T> RepeatAfterDelay<T>(
this IObservable<T> source,
TimeSpan retryDelay,
int repeatCount = -1,
IScheduler scheduler = null)
{
return RedoAfterDelay(source, retryDelay, repeatCount, scheduler, Observable.Repeat, Observable.Repeat);
}
private static IObservable<T> RedoAfterDelay<T>(IObservable<T> source, TimeSpan retryDelay, int retryCount, IScheduler scheduler, Func<IObservable<T>, IObservable<T>> reDo,
Func<IObservable<T>, int, IObservable<T>> reDoCount)
{
scheduler = scheduler ?? TaskPoolScheduler.Default;
var attempt = 0;
public static IObservable<T> RetryAfterDelay<T>(
this IObservable<T> source,
TimeSpan retryDelay,
int retryCount = -1,
IScheduler scheduler = null)
{
return RedoAfterDelay(source, retryDelay, retryCount, scheduler, Observable.Retry, Observable.Retry);
}
var deferedObs = Observable.Defer(() => ((++attempt == 1) ? source : source.DelaySubscription(retryDelay, scheduler)));
return retryCount > 0 ? reDoCount(deferedObs, retryCount) : reDo(deferedObs);
}
private static IObservable<T> RedoAfterDelay<T>(IObservable<T> source, TimeSpan retryDelay, int retryCount, IScheduler scheduler, Func<IObservable<T>, IObservable<T>> reDo,
Func<IObservable<T>, int, IObservable<T>> reDoCount)
{
scheduler = scheduler ?? TaskPoolScheduler.Default;
var attempt = 0;
public static IObservable<T> DisposeMany<T>(this IObservable<T> source)
{
return Observable.Create<T>(obs =>
{
var serialDisposable = new SerialDisposable();
var subscription =
source.Subscribe(
item =>
{
serialDisposable.Disposable = item as IDisposable;
obs.OnNext(item);
},
obs.OnError,
obs.OnCompleted);
return new CompositeDisposable(serialDisposable, subscription);
});
}
}
var deferedObs = Observable.Defer(() => ((++attempt == 1) ? source : source.DelaySubscription(retryDelay, scheduler)));
return retryCount > 0 ? reDoCount(deferedObs, retryCount) : reDo(deferedObs);
}
}
}

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

@@ -38,10 +38,10 @@ namespace Sharp7.Rx.Extensions
notification
.Where(trigger => !trigger)
.SelectMany(async _ =>
{
await plc.SetValue(ackTriggerAddress, false);
return Unit.Default;
})
{
await plc.SetValue(ackTriggerAddress, false);
return Unit.Default;
})
.Subscribe()
.AddDisposableTo(subscriptions);
@@ -71,4 +71,4 @@ namespace Sharp7.Rx.Extensions
}
}
}
}
}