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

Use file scoped namespaces

Browse Source
This commit is contained in:
Peter Butzhammer
2024-02-07 09:54:18 +01:00
parent bacfbf05e2
commit e6cea1e04f
24 changed files with 1131 additions and 1191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
Sharp7.Rx.Tests/S7ValueConverterTests.cs
View File

@@ -1,5 +1,4 @@
using System; using NUnit.Framework;
using NUnit.Framework;
using Sharp7.Rx.Interfaces; using Sharp7.Rx.Interfaces;
using Shouldly; using Shouldly;
@@ -64,4 +63,4 @@ public class S7ValueConverterTests
//Act //Act
Should.Throw<ArgumentException>(() => S7ValueConverter.ConvertToType<T>(data, variableAddress)); Should.Throw<ArgumentException>(() => S7ValueConverter.ConvertToType<T>(data, variableAddress));
} }
} }

5
Sharp7.Rx.Tests/S7VariableNameParserTests.cs
View File

@@ -1,5 +1,4 @@
using System.Collections.Generic; using DeepEqual.Syntax;
using DeepEqual.Syntax;
using NUnit.Framework; using NUnit.Framework;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
@@ -39,4 +38,4 @@ internal class S7VariableNameParserTests
{ {
public override string ToString() => Input; public override string ToString() => Input;
} }
} }

229
Sharp7.Rx/Basics/ConcurrentSubjectDictionary.cs
View File

@@ -1,128 +1,125 @@
using System; using System.Collections.Concurrent;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Reactive; using System.Reactive;
using System.Reactive.Linq; using System.Reactive.Linq;
using System.Reactive.Subjects; using System.Reactive.Subjects;
namespace Sharp7.Rx.Basics namespace Sharp7.Rx.Basics;
internal class ConcurrentSubjectDictionary<TKey, TValue> : IDisposable
{ {
internal class ConcurrentSubjectDictionary<TKey, TValue> : IDisposable private readonly object dictionaryLock = new object();
private readonly Func<TKey, TValue> valueFactory;
private ConcurrentDictionary<TKey, SubjectWithRefCounter> dictionary;
public ConcurrentSubjectDictionary()
{ {
private readonly object dictionaryLock = new object(); dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>();
private readonly Func<TKey, TValue> valueFactory; }
private ConcurrentDictionary<TKey, SubjectWithRefCounter> dictionary;
public ConcurrentSubjectDictionary() public ConcurrentSubjectDictionary(IEqualityComparer<TKey> comparer)
{
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>(comparer);
}
public ConcurrentSubjectDictionary(TValue initialValue, IEqualityComparer<TKey> comparer)
{
valueFactory = _ => initialValue;
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>(comparer);
}
public ConcurrentSubjectDictionary(TValue initialValue)
{
valueFactory = _ => initialValue;
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>();
}
public ConcurrentSubjectDictionary(Func<TKey, TValue> valueFactory = null)
{
this.valueFactory = valueFactory;
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>();
}
public IEnumerable<TKey> ExistingKeys => dictionary.Keys;
public bool IsDisposed { get; private set; }
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
public DisposableItem<TValue> GetOrCreateObservable(TKey key)
{
lock (dictionaryLock)
{ {
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>(); var subject = dictionary.AddOrUpdate(key, k => new SubjectWithRefCounter {Counter = 1, Subject = CreateSubject(k)}, (key1, counter) =>
}
public ConcurrentSubjectDictionary(IEqualityComparer<TKey> comparer)
{
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>(comparer);
}
public ConcurrentSubjectDictionary(TValue initialValue, IEqualityComparer<TKey> comparer)
{
valueFactory = _ => initialValue;
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>(comparer);
}
public ConcurrentSubjectDictionary(TValue initialValue)
{
valueFactory = _ => initialValue;
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>();
}
public ConcurrentSubjectDictionary(Func<TKey, TValue> valueFactory = null)
{
this.valueFactory = valueFactory;
dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>();
}
public IEnumerable<TKey> ExistingKeys => dictionary.Keys;
public bool IsDisposed { get; private set; }
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
public DisposableItem<TValue> GetOrCreateObservable(TKey key)
{
lock (dictionaryLock)
{ {
var subject = dictionary.AddOrUpdate(key, k => new SubjectWithRefCounter {Counter = 1, Subject = CreateSubject(k)}, (key1, counter) => counter.Counter = counter.Counter + 1;
{ return counter;
counter.Counter = counter.Counter + 1; });
return counter;
});
return new DisposableItem<TValue>(subject.Subject.AsObservable(), () => RemoveIfNoLongerInUse(key)); return new DisposableItem<TValue>(subject.Subject.AsObservable(), () => RemoveIfNoLongerInUse(key));
}
}
public bool TryGetObserver(TKey key, out IObserver<TValue> subject)
{
SubjectWithRefCounter subjectWithRefCount;
if (dictionary.TryGetValue(key, out subjectWithRefCount))
{
subject = subjectWithRefCount.Subject.AsObserver();
return true;
}
subject = null;
return false;
}
protected virtual void Dispose(bool disposing)
{
if (IsDisposed)
return;
if (disposing && dictionary != null)
{
foreach (var subjectWithRefCounter in dictionary)
subjectWithRefCounter.Value.Subject.OnCompleted();
dictionary.Clear();
dictionary = null;
}
IsDisposed = true;
}
private ISubject<TValue> CreateSubject(TKey key)
{
if (valueFactory == null)
return new Subject<TValue>();
return new BehaviorSubject<TValue>(valueFactory(key));
}
private void RemoveIfNoLongerInUse(TKey variableName)
{
lock (dictionaryLock)
{
SubjectWithRefCounter subjectWithRefCount;
if (dictionary.TryGetValue(variableName, out subjectWithRefCount))
{
if (subjectWithRefCount.Counter == 1)
dictionary.TryRemove(variableName, out subjectWithRefCount);
else subjectWithRefCount.Counter--;
}
}
}
~ConcurrentSubjectDictionary()
{
Dispose(false);
}
class SubjectWithRefCounter
{
public int Counter { get; set; }
public ISubject<TValue> Subject { get; set; }
} }
} }
}
public bool TryGetObserver(TKey key, out IObserver<TValue> subject)
{
SubjectWithRefCounter subjectWithRefCount;
if (dictionary.TryGetValue(key, out subjectWithRefCount))
{
subject = subjectWithRefCount.Subject.AsObserver();
return true;
}
subject = null;
return false;
}
protected virtual void Dispose(bool disposing)
{
if (IsDisposed)
return;
if (disposing && dictionary != null)
{
foreach (var subjectWithRefCounter in dictionary)
subjectWithRefCounter.Value.Subject.OnCompleted();
dictionary.Clear();
dictionary = null;
}
IsDisposed = true;
}
private ISubject<TValue> CreateSubject(TKey key)
{
if (valueFactory == null)
return new Subject<TValue>();
return new BehaviorSubject<TValue>(valueFactory(key));
}
private void RemoveIfNoLongerInUse(TKey variableName)
{
lock (dictionaryLock)
{
SubjectWithRefCounter subjectWithRefCount;
if (dictionary.TryGetValue(variableName, out subjectWithRefCount))
{
if (subjectWithRefCount.Counter == 1)
dictionary.TryRemove(variableName, out subjectWithRefCount);
else subjectWithRefCount.Counter--;
}
}
}
~ConcurrentSubjectDictionary()
{
Dispose(false);
}
class SubjectWithRefCounter
{
public int Counter { get; set; }
public ISubject<TValue> Subject { get; set; }
}
}

63
Sharp7.Rx/Basics/DisposableItem.cs
View File

@@ -1,37 +1,34 @@
using System; namespace Sharp7.Rx.Basics;
namespace Sharp7.Rx.Basics internal class DisposableItem<TValue> : IDisposable
{ {
internal class DisposableItem<TValue> : IDisposable private readonly Action disposeAction;
bool disposed;
public DisposableItem(IObservable<TValue> observable, Action disposeAction)
{ {
private readonly Action disposeAction; this.disposeAction = disposeAction;
Observable = observable;
bool disposed;
public DisposableItem(IObservable<TValue> observable, Action disposeAction)
{
this.disposeAction = disposeAction;
Observable = observable;
}
public IObservable<TValue> Observable { get; }
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
protected virtual void Dispose(bool disposing)
{
if (disposed) return;
if (disposing)
{
disposeAction();
}
disposed = true;
}
} }
}
public IObservable<TValue> Observable { get; }
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
protected virtual void Dispose(bool disposing)
{
if (disposed) return;
if (disposing)
{
disposeAction();
}
disposed = true;
}
}

261
Sharp7.Rx/Basics/LimitedConcurrencyLevelTaskScheduler.cs
View File

@@ -1,146 +1,139 @@
using System; namespace Sharp7.Rx.Basics;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace Sharp7.Rx.Basics /// <summary>
/// Provides a task scheduler that ensures a maximum concurrency level while
/// running on top of the ThreadPool.
/// from http://msdn.microsoft.com/en-us/library/ee789351.aspx
/// </summary>
internal class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
{ {
/// <summary>Whether the current thread is processing work items.</summary>
[ThreadStatic] private static bool currentThreadIsProcessingItems;
/// <summary>The maximum concurrency level allowed by this scheduler.</summary>
private readonly int maxDegreeOfParallelism;
/// <summary>The list of tasks to be executed.</summary>
private readonly LinkedList<Task> tasks = new LinkedList<Task>(); // protected by lock(_tasks)
/// <summary>Whether the scheduler is currently processing work items.</summary>
private int delegatesQueuedOrRunning; // protected by lock(_tasks)
/// <summary> /// <summary>
/// Provides a task scheduler that ensures a maximum concurrency level while /// Initializes an instance of the LimitedConcurrencyLevelTaskScheduler class with the
/// running on top of the ThreadPool. /// specified degree of parallelism.
/// from http://msdn.microsoft.com/en-us/library/ee789351.aspx
/// </summary> /// </summary>
internal class LimitedConcurrencyLevelTaskScheduler : TaskScheduler /// <param name="maxDegreeOfParallelism">The maximum degree of parallelism provided by this scheduler.</param>
public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
{ {
/// <summary>Whether the current thread is processing work items.</summary> if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
[ThreadStatic] private static bool currentThreadIsProcessingItems; this.maxDegreeOfParallelism = maxDegreeOfParallelism;
}
/// <summary>The maximum concurrency level allowed by this scheduler.</summary> /// <summary>Gets the maximum concurrency level supported by this scheduler.</summary>
private readonly int maxDegreeOfParallelism; public sealed override int MaximumConcurrencyLevel => maxDegreeOfParallelism;
/// <summary>The list of tasks to be executed.</summary> /// <summary>Gets an enumerable of the tasks currently scheduled on this scheduler.</summary>
private readonly LinkedList<Task> tasks = new LinkedList<Task>(); // protected by lock(_tasks) /// <returns>An enumerable of the tasks currently scheduled.</returns>
protected sealed override IEnumerable<Task> GetScheduledTasks()
/// <summary>Whether the scheduler is currently processing work items.</summary> {
private int delegatesQueuedOrRunning; // protected by lock(_tasks) var lockTaken = false;
try
/// <summary>
/// Initializes an instance of the LimitedConcurrencyLevelTaskScheduler class with the
/// specified degree of parallelism.
/// </summary>
/// <param name="maxDegreeOfParallelism">The maximum degree of parallelism provided by this scheduler.</param>
public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
{ {
if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism"); Monitor.TryEnter(tasks, ref lockTaken);
this.maxDegreeOfParallelism = maxDegreeOfParallelism; if (lockTaken) return tasks.ToArray();
else throw new NotSupportedException();
} }
finally
/// <summary>Gets the maximum concurrency level supported by this scheduler.</summary>
public sealed override int MaximumConcurrencyLevel => maxDegreeOfParallelism;
/// <summary>Gets an enumerable of the tasks currently scheduled on this scheduler.</summary>
/// <returns>An enumerable of the tasks currently scheduled.</returns>
protected sealed override IEnumerable<Task> GetScheduledTasks()
{ {
var lockTaken = false; if (lockTaken) Monitor.Exit(tasks);
try
{
Monitor.TryEnter(tasks, ref lockTaken);
if (lockTaken) return tasks.ToArray();
else throw new NotSupportedException();
}
finally
{
if (lockTaken) Monitor.Exit(tasks);
}
}
/// <summary>Queues a task to the scheduler.</summary>
/// <param name="task">The task to be queued.</param>
protected sealed override void QueueTask(Task task)
{
// Add the task to the list of tasks to be processed. If there aren't enough
// delegates currently queued or running to process tasks, schedule another.
lock (tasks)
{
tasks.AddLast(task);
if (delegatesQueuedOrRunning < maxDegreeOfParallelism)
{
++delegatesQueuedOrRunning;
NotifyThreadPoolOfPendingWork();
}
}
}
/// <summary>Attempts to remove a previously scheduled task from the scheduler.</summary>
/// <param name="task">The task to be removed.</param>
/// <returns>Whether the task could be found and removed.</returns>
protected sealed override bool TryDequeue(Task task)
{
lock (tasks)
{
return tasks.Remove(task);
}
}
/// <summary>Attempts to execute the specified task on the current thread.</summary>
/// <param name="task">The task to be executed.</param>
/// <param name="taskWasPreviouslyQueued"></param>
/// <returns>Whether the task could be executed on the current thread.</returns>
protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
{
// If this thread isn't already processing a task, we don't support inlining
if (!currentThreadIsProcessingItems) return false;
// If the task was previously queued, remove it from the queue
if (taskWasPreviouslyQueued) TryDequeue(task);
// Try to run the task.
return TryExecuteTask(task);
}
/// <summary>
/// Informs the ThreadPool that there's work to be executed for this scheduler.
/// </summary>
private void NotifyThreadPoolOfPendingWork()
{
ThreadPool.UnsafeQueueUserWorkItem(_ =>
{
// Note that the current thread is now processing work items.
// This is necessary to enable inlining of tasks into this thread.
currentThreadIsProcessingItems = true;
try
{
// Process all available items in the queue.
while (true)
{
Task item;
lock (tasks)
{
// When there are no more items to be processed,
// note that we're done processing, and get out.
if (tasks.Count == 0)
{
--delegatesQueuedOrRunning;
break;
}
// Get the next item from the queue
item = tasks.First.Value;
tasks.RemoveFirst();
}
// Execute the task we pulled out of the queue
TryExecuteTask(item);
}
}
// We're done processing items on the current thread
finally
{
currentThreadIsProcessingItems = false;
}
}, null);
} }
} }
}
/// <summary>Queues a task to the scheduler.</summary>
/// <param name="task">The task to be queued.</param>
protected sealed override void QueueTask(Task task)
{
// Add the task to the list of tasks to be processed. If there aren't enough
// delegates currently queued or running to process tasks, schedule another.
lock (tasks)
{
tasks.AddLast(task);
if (delegatesQueuedOrRunning < maxDegreeOfParallelism)
{
++delegatesQueuedOrRunning;
NotifyThreadPoolOfPendingWork();
}
}
}
/// <summary>Attempts to remove a previously scheduled task from the scheduler.</summary>
/// <param name="task">The task to be removed.</param>
/// <returns>Whether the task could be found and removed.</returns>
protected sealed override bool TryDequeue(Task task)
{
lock (tasks)
{
return tasks.Remove(task);
}
}
/// <summary>Attempts to execute the specified task on the current thread.</summary>
/// <param name="task">The task to be executed.</param>
/// <param name="taskWasPreviouslyQueued"></param>
/// <returns>Whether the task could be executed on the current thread.</returns>
protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
{
// If this thread isn't already processing a task, we don't support inlining
if (!currentThreadIsProcessingItems) return false;
// If the task was previously queued, remove it from the queue
if (taskWasPreviouslyQueued) TryDequeue(task);
// Try to run the task.
return TryExecuteTask(task);
}
/// <summary>
/// Informs the ThreadPool that there's work to be executed for this scheduler.
/// </summary>
private void NotifyThreadPoolOfPendingWork()
{
ThreadPool.UnsafeQueueUserWorkItem(_ =>
{
// Note that the current thread is now processing work items.
// This is necessary to enable inlining of tasks into this thread.
currentThreadIsProcessingItems = true;
try
{
// Process all available items in the queue.
while (true)
{
Task item;
lock (tasks)
{
// When there are no more items to be processed,
// note that we're done processing, and get out.
if (tasks.Count == 0)
{
--delegatesQueuedOrRunning;
break;
}
// Get the next item from the queue
item = tasks.First.Value;
tasks.RemoveFirst();
}
// Execute the task we pulled out of the queue
TryExecuteTask(item);
}
}
// We're done processing items on the current thread
finally
{
currentThreadIsProcessingItems = false;
}
}, null);
}
}

28
Sharp7.Rx/CacheVariableNameParser.cs
View File

@@ -1,20 +1,18 @@
using System; using System.Collections.Concurrent;
using System.Collections.Concurrent;
using Sharp7.Rx.Interfaces; using Sharp7.Rx.Interfaces;
namespace Sharp7.Rx namespace Sharp7.Rx;
internal class CacheVariableNameParser : IS7VariableNameParser
{ {
internal class CacheVariableNameParser : IS7VariableNameParser private static readonly ConcurrentDictionary<string, S7VariableAddress> addressCache = new ConcurrentDictionary<string, S7VariableAddress>(StringComparer.OrdinalIgnoreCase);
private readonly IS7VariableNameParser inner;
public CacheVariableNameParser(IS7VariableNameParser inner)
{ {
private static readonly ConcurrentDictionary<string, S7VariableAddress> addressCache = new ConcurrentDictionary<string, S7VariableAddress>(StringComparer.OrdinalIgnoreCase); this.inner = inner;
private readonly IS7VariableNameParser inner;
public CacheVariableNameParser(IS7VariableNameParser inner)
{
this.inner = inner;
}
public S7VariableAddress Parse(string input) => addressCache.GetOrAdd(input, inner.Parse);
} }
}
public S7VariableAddress Parse(string input) => addressCache.GetOrAdd(input, inner.Parse);
}

17
Sharp7.Rx/Enums/ConnectionState.cs
View File

@@ -1,10 +1,9 @@
namespace Sharp7.Rx.Enums namespace Sharp7.Rx.Enums;
public enum ConnectionState
{ {
public enum ConnectionState Initial,
{ Connected,
Initial, DisconnectedByUser,
Connected, ConnectionLost
DisconnectedByUser, }
ConnectionLost
}
}

17
Sharp7.Rx/Enums/CpuType.cs
View File

@@ -1,10 +1,9 @@
namespace Sharp7.Rx.Enums namespace Sharp7.Rx.Enums;
internal enum CpuType
{ {
internal enum CpuType S7_300,
{ S7_400,
S7_300, S7_1200,
S7_400, S7_1500
S7_1200, }
S7_1500
}
}

23
Sharp7.Rx/Enums/DbType.cs
View File

@@ -1,13 +1,12 @@
namespace Sharp7.Rx.Enums namespace Sharp7.Rx.Enums;
internal enum DbType
{ {
internal enum DbType Bit,
{ String,
Bit, Byte,
String, Double,
Byte, Integer,
Double, DInteger,
Integer, ULong
DInteger, }
ULong
}
}

17
Sharp7.Rx/Enums/Operand.cs
View File

@@ -1,10 +1,9 @@
namespace Sharp7.Rx.Enums namespace Sharp7.Rx.Enums;
internal enum Operand : byte
{ {
internal enum Operand : byte Input = 69,
{ Output = 65,
Input = 69, Marker = 77,
Output = 65, Db = 68,
Marker = 77, }
Db = 68,
}
}

13
Sharp7.Rx/Enums/TransmissionMode.cs
View File

@@ -1,8 +1,7 @@
namespace Sharp7.Rx.Enums namespace Sharp7.Rx.Enums;
public enum TransmissionMode
{ {
public enum TransmissionMode Cyclic = 3,
{ OnChange = 4,
Cyclic = 3, }
OnChange = 4,
}
}

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

@@ -1,13 +1,11 @@
using System; using System.Reactive.Disposables;
using System.Reactive.Disposables;
namespace Sharp7.Rx.Extensions namespace Sharp7.Rx.Extensions;
internal static class DisposableExtensions
{ {
internal static class DisposableExtensions public static void AddDisposableTo(this IDisposable disposable, CompositeDisposable compositeDisposable)
{ {
public static void AddDisposableTo(this IDisposable disposable, CompositeDisposable compositeDisposable) compositeDisposable.Add(disposable);
{
compositeDisposable.Add(disposable);
}
} }
} }

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

@@ -1,83 +1,81 @@
using System; using System.Reactive.Concurrency;
using System.Reactive.Concurrency;
using System.Reactive.Disposables; using System.Reactive.Disposables;
using System.Reactive.Linq; using System.Reactive.Linq;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
namespace Sharp7.Rx.Extensions namespace Sharp7.Rx.Extensions;
{
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> LogAndRetry<T>(this IObservable<T> source, ILogger logger, string message) internal static class ObservableExtensions
{
public static IObservable<T> DisposeMany<T>(this IObservable<T> source)
{
return Observable.Create<T>(obs =>
{ {
return source 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> LogAndRetry<T>(this IObservable<T> source, ILogger logger, string message)
{
return source
.Do(
_ => { },
ex => logger?.LogError(ex, message))
.Retry();
}
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( .Do(
_ => { }, _ => { },
ex => logger?.LogError(ex, message)) ex => logger?.LogError(ex, message));
.Retry();
}
public static IObservable<T> LogAndRetryAfterDelay<T>( return RetryAfterDelay(sourceLogged, retryDelay, retryCount, scheduler);
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);
}
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> RetryAfterDelay<T>(
this IObservable<T> source,
TimeSpan retryDelay,
int retryCount = -1,
IScheduler scheduler = null)
{
return RedoAfterDelay(source, retryDelay, retryCount, scheduler, Observable.Retry, Observable.Retry);
}
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;
var deferedObs = Observable.Defer(() => ((++attempt == 1) ? source : source.DelaySubscription(retryDelay, scheduler)));
return retryCount > 0 ? reDoCount(deferedObs, retryCount) : reDo(deferedObs);
}
} }
}
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> RetryAfterDelay<T>(
this IObservable<T> source,
TimeSpan retryDelay,
int retryCount = -1,
IScheduler scheduler = null)
{
return RedoAfterDelay(source, retryDelay, retryCount, scheduler, Observable.Retry, Observable.Retry);
}
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;
var deferedObs = Observable.Defer(() => ((++attempt == 1) ? source : source.DelaySubscription(retryDelay, scheduler)));
return retryCount > 0 ? reDoCount(deferedObs, retryCount) : reDo(deferedObs);
}
}

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

@@ -1,74 +1,71 @@
using System; using System.Reactive;
using System.Reactive;
using System.Reactive.Disposables; using System.Reactive.Disposables;
using System.Reactive.Linq; using System.Reactive.Linq;
using System.Reactive.Threading.Tasks; using System.Reactive.Threading.Tasks;
using System.Threading.Tasks;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
using Sharp7.Rx.Interfaces; using Sharp7.Rx.Interfaces;
namespace Sharp7.Rx.Extensions namespace Sharp7.Rx.Extensions;
public static class PlcExtensions
{ {
public static class PlcExtensions public static IObservable<TReturn> CreateDatatransferWithHandshake<TReturn>(this IPlc plc, string triggerAddress, string ackTriggerAddress, Func<IPlc, Task<TReturn>> readData, bool initialTransfer)
{ {
public static IObservable<TReturn> CreateDatatransferWithHandshake<TReturn>(this IPlc plc, string triggerAddress, string ackTriggerAddress, Func<IPlc, Task<TReturn>> readData, bool initialTransfer) return Observable.Create<TReturn>(async observer =>
{ {
return Observable.Create<TReturn>(async observer => var subscriptions = new CompositeDisposable();
var notification = plc
.CreateNotification<bool>(triggerAddress, TransmissionMode.OnChange)
.Publish()
.RefCount();
if (initialTransfer)
{ {
var subscriptions = new CompositeDisposable(); await plc.ConnectionState.FirstAsync(state => state == ConnectionState.Connected).ToTask();
var initialValue = await ReadData(plc, readData);
observer.OnNext(initialValue);
}
var notification = plc notification
.CreateNotification<bool>(triggerAddress, TransmissionMode.OnChange) .Where(trigger => trigger)
.Publish() .SelectMany(_ => ReadDataAndAcknowlodge(plc, readData, ackTriggerAddress))
.RefCount(); .Subscribe(observer)
.AddDisposableTo(subscriptions);
if (initialTransfer) notification
.Where(trigger => !trigger)
.SelectMany(async _ =>
{ {
await plc.ConnectionState.FirstAsync(state => state == ConnectionState.Connected).ToTask(); await plc.SetValue(ackTriggerAddress, false);
var initialValue = await ReadData(plc, readData); return Unit.Default;
observer.OnNext(initialValue); })
} .Subscribe()
.AddDisposableTo(subscriptions);
notification return subscriptions;
.Where(trigger => trigger) });
.SelectMany(_ => ReadDataAndAcknowlodge(plc, readData, ackTriggerAddress)) }
.Subscribe(observer)
.AddDisposableTo(subscriptions);
notification public static IObservable<TReturn> CreateDatatransferWithHandshake<TReturn>(this IPlc plc, string triggerAddress, string ackTriggerAddress, Func<IPlc, Task<TReturn>> readData)
.Where(trigger => !trigger) {
.SelectMany(async _ => return CreateDatatransferWithHandshake(plc, triggerAddress, ackTriggerAddress, readData, false);
{ }
await plc.SetValue(ackTriggerAddress, false);
return Unit.Default;
})
.Subscribe()
.AddDisposableTo(subscriptions);
return subscriptions; private static async Task<TReturn> ReadData<TReturn>(IPlc plc, Func<IPlc, Task<TReturn>> receiveData)
}); {
} return await receiveData(plc);
}
public static IObservable<TReturn> CreateDatatransferWithHandshake<TReturn>(this IPlc plc, string triggerAddress, string ackTriggerAddress, Func<IPlc, Task<TReturn>> readData) private static async Task<TReturn> ReadDataAndAcknowlodge<TReturn>(IPlc plc, Func<IPlc, Task<TReturn>> readData, string ackTriggerAddress)
{
try
{ {
return CreateDatatransferWithHandshake(plc, triggerAddress, ackTriggerAddress, readData, false); return await ReadData(plc, readData);
} }
finally
private static async Task<TReturn> ReadData<TReturn>(IPlc plc, Func<IPlc, Task<TReturn>> receiveData)
{ {
return await receiveData(plc); await plc.SetValue(ackTriggerAddress, true);
}
private static async Task<TReturn> ReadDataAndAcknowlodge<TReturn>(IPlc plc, Func<IPlc, Task<TReturn>> readData, string ackTriggerAddress)
{
try
{
return await ReadData(plc, readData);
}
finally
{
await plc.SetValue(ackTriggerAddress, true);
}
} }
} }
} }

23
Sharp7.Rx/Interfaces/IPlc.cs
View File

@@ -1,18 +1,15 @@
using System; using JetBrains.Annotations;
using System.Threading.Tasks;
using JetBrains.Annotations;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
namespace Sharp7.Rx.Interfaces namespace Sharp7.Rx.Interfaces;
[NoReorder]
public interface IPlc : IDisposable
{ {
[NoReorder] IObservable<TValue> CreateNotification<TValue>(string variableName, TransmissionMode transmissionMode);
public interface IPlc : IDisposable Task SetValue<TValue>(string variableName, TValue value);
{ Task<TValue> GetValue<TValue>(string variableName);
IObservable<TValue> CreateNotification<TValue>(string variableName, TransmissionMode transmissionMode); IObservable<ConnectionState> ConnectionState { get; }
Task SetValue<TValue>(string variableName, TValue value); ILogger Logger { get; }
Task<TValue> GetValue<TValue>(string variableName);
IObservable<ConnectionState> ConnectionState { get; }
ILogger Logger { get; }
}
} }

33
Sharp7.Rx/Interfaces/IS7Connector.cs
View File

@@ -1,26 +1,21 @@
using System; using JetBrains.Annotations;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using JetBrains.Annotations;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
namespace Sharp7.Rx.Interfaces namespace Sharp7.Rx.Interfaces;
[NoReorder]
internal interface IS7Connector : IDisposable
{ {
[NoReorder] IObservable<ConnectionState> ConnectionState { get; }
internal interface IS7Connector : IDisposable Task InitializeAsync();
{
IObservable<ConnectionState> ConnectionState { get; }
Task InitializeAsync();
Task<bool> Connect(); Task<bool> Connect();
Task Disconnect(); Task Disconnect();
Task<byte[]> ReadBytes(Operand operand, ushort startByteAddress, ushort bytesToRead, ushort dBNr, CancellationToken token); Task<byte[]> ReadBytes(Operand operand, ushort startByteAddress, ushort bytesToRead, ushort dBNr, CancellationToken token);
Task<bool> WriteBit(Operand operand, ushort startByteAddress, byte bitAdress, bool value, ushort dbNr, CancellationToken token); Task<bool> WriteBit(Operand operand, ushort startByteAddress, byte bitAdress, bool value, ushort dbNr, CancellationToken token);
Task<ushort> WriteBytes(Operand operand, ushort startByteAdress, byte[] data, ushort dBNr, CancellationToken token); Task<ushort> WriteBytes(Operand operand, ushort startByteAdress, byte[] data, ushort dBNr, CancellationToken token);
Task<Dictionary<string, byte[]>> ExecuteMultiVarRequest(IReadOnlyList<string> variableNames); Task<Dictionary<string, byte[]>> ExecuteMultiVarRequest(IReadOnlyList<string> variableNames);
} }
}

11
Sharp7.Rx/Interfaces/IS7VariableNameParser.cs
View File

@@ -1,7 +1,6 @@
namespace Sharp7.Rx.Interfaces namespace Sharp7.Rx.Interfaces;
internal interface IS7VariableNameParser
{ {
internal interface IS7VariableNameParser S7VariableAddress Parse(string input);
{ }
S7VariableAddress Parse(string input);
}
}

41
Sharp7.Rx/S7ErrorCodes.cs
View File

@@ -1,27 +1,24 @@
using System.Collections.Generic; namespace Sharp7.Rx;
namespace Sharp7.Rx public static class S7ErrorCodes
{ {
public static class S7ErrorCodes /// <summary>
/// This list is not exhaustive and should be considered work in progress.
/// </summary>
private static readonly HashSet<int> notDisconnectedErrorCodes = new HashSet<int>
{ {
/// <summary> 0x000000, // OK
/// This list is not exhaustive and should be considered work in progress. 0xC00000, // CPU: Item not available
/// </summary> 0x900000, // CPU: Address out of range
private static readonly HashSet<int> notDisconnectedErrorCodes = new HashSet<int> };
{
0x000000, // OK
0xC00000, // CPU: Item not available
0x900000, // CPU: Address out of range
};
/// <summary> /// <summary>
/// Some error codes indicate connection lost, in which case, the driver tries to reestablish connection. /// Some error codes indicate connection lost, in which case, the driver tries to reestablish connection.
/// Other error codes indicate a user error, like reading from an unavailable DB or exceeding /// Other error codes indicate a user error, like reading from an unavailable DB or exceeding
/// the DBs range. In this case the driver should not consider the connection to be lost. /// the DBs range. In this case the driver should not consider the connection to be lost.
/// </summary> /// </summary>
public static bool AssumeConnectionLost(int errorCode) public static bool AssumeConnectionLost(int errorCode)
{ {
return !notDisconnectedErrorCodes.Contains(errorCode); return !notDisconnectedErrorCodes.Contains(errorCode);
}
} }
} }

142
Sharp7.Rx/S7ValueConverter.cs
View File

@@ -1,85 +1,83 @@
using System; using System.Buffers.Binary;
using System.Buffers.Binary;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
using System.Text; using System.Text;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
namespace Sharp7.Rx namespace Sharp7.Rx;
internal static class S7ValueConverter
{ {
internal static class S7ValueConverter public static TValue ConvertToType<TValue>(byte[] buffer, S7VariableAddress address)
{ {
public static TValue ConvertToType<TValue>(byte[] buffer, S7VariableAddress address) if (typeof(TValue) == typeof(bool))
return (TValue) (object) (((buffer[0] >> address.Bit) & 1) > 0);
if (typeof(TValue) == typeof(int))
{ {
if (typeof(TValue) == typeof(bool)) if (address.Length == 2)
return (TValue) (object) (((buffer[0] >> address.Bit) & 1) > 0); return (TValue) (object) (int) BinaryPrimitives.ReadInt16BigEndian(buffer);
if (address.Length == 4)
return (TValue) (object) BinaryPrimitives.ReadInt32BigEndian(buffer);
if (typeof(TValue) == typeof(int)) throw new InvalidOperationException($"length must be 2 or 4 but is {address.Length}");
{
if (address.Length == 2)
return (TValue) (object) (int) BinaryPrimitives.ReadInt16BigEndian(buffer);
if (address.Length == 4)
return (TValue) (object) BinaryPrimitives.ReadInt32BigEndian(buffer);
throw new InvalidOperationException($"length must be 2 or 4 but is {address.Length}");
}
if (typeof(TValue) == typeof(long))
return (TValue) (object) BinaryPrimitives.ReadInt64BigEndian(buffer);
if (typeof(TValue) == typeof(ulong))
return (TValue) (object) BinaryPrimitives.ReadUInt64BigEndian(buffer);
if (typeof(TValue) == typeof(short))
return (TValue) (object) BinaryPrimitives.ReadInt16BigEndian(buffer);
if (typeof(TValue) == typeof(byte))
return (TValue) (object) buffer[0];
if (typeof(TValue) == typeof(char))
return (TValue) (object) (char) buffer[0];
if (typeof(TValue) == typeof(byte[]))
return (TValue) (object) buffer;
if (typeof(TValue) == typeof(double))
{
var d = new UInt32SingleMap
{
UInt32 = BinaryPrimitives.ReadUInt32BigEndian(buffer)
};
return (TValue) (object) (double) d.Single;
}
if (typeof(TValue) == typeof(float))
{
var d = new UInt32SingleMap
{
UInt32 = BinaryPrimitives.ReadUInt32BigEndian(buffer)
};
return (TValue) (object) d.Single;
}
if (typeof(TValue) == typeof(string))
if (address.Type == DbType.String)
{
// First byte is maximal length
// Second byte is actual length
// https://cache.industry.siemens.com/dl/files/480/22506480/att_105176/v1/s7_scl_string_parameterzuweisung_e.pdf
var length = Math.Min(address.Length, buffer[1]);
return (TValue) (object) Encoding.ASCII.GetString(buffer, 2, length);
}
else
return (TValue) (object) Encoding.ASCII.GetString(buffer).Trim();
throw new InvalidOperationException(string.Format("type '{0}' not supported.", typeof(TValue)));
} }
[StructLayout(LayoutKind.Explicit)] if (typeof(TValue) == typeof(long))
private struct UInt32SingleMap return (TValue) (object) BinaryPrimitives.ReadInt64BigEndian(buffer);
if (typeof(TValue) == typeof(ulong))
return (TValue) (object) BinaryPrimitives.ReadUInt64BigEndian(buffer);
if (typeof(TValue) == typeof(short))
return (TValue) (object) BinaryPrimitives.ReadInt16BigEndian(buffer);
if (typeof(TValue) == typeof(byte))
return (TValue) (object) buffer[0];
if (typeof(TValue) == typeof(char))
return (TValue) (object) (char) buffer[0];
if (typeof(TValue) == typeof(byte[]))
return (TValue) (object) buffer;
if (typeof(TValue) == typeof(double))
{ {
[FieldOffset(0)] public uint UInt32; var d = new UInt32SingleMap
[FieldOffset(0)] public float Single; {
UInt32 = BinaryPrimitives.ReadUInt32BigEndian(buffer)
};
return (TValue) (object) (double) d.Single;
} }
if (typeof(TValue) == typeof(float))
{
var d = new UInt32SingleMap
{
UInt32 = BinaryPrimitives.ReadUInt32BigEndian(buffer)
};
return (TValue) (object) d.Single;
}
if (typeof(TValue) == typeof(string))
if (address.Type == DbType.String)
{
// First byte is maximal length
// Second byte is actual length
// https://cache.industry.siemens.com/dl/files/480/22506480/att_105176/v1/s7_scl_string_parameterzuweisung_e.pdf
var length = Math.Min(address.Length, buffer[1]);
return (TValue) (object) Encoding.ASCII.GetString(buffer, 2, length);
}
else
return (TValue) (object) Encoding.ASCII.GetString(buffer).Trim();
throw new InvalidOperationException(string.Format("type '{0}' not supported.", typeof(TValue)));
} }
}
[StructLayout(LayoutKind.Explicit)]
private struct UInt32SingleMap
{
[FieldOffset(0)] public uint UInt32;
[FieldOffset(0)] public float Single;
}
}

23
Sharp7.Rx/S7VariableAddress.cs
View File

@@ -1,16 +1,15 @@
using JetBrains.Annotations; using JetBrains.Annotations;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
namespace Sharp7.Rx namespace Sharp7.Rx;
[NoReorder]
internal class S7VariableAddress
{ {
[NoReorder] public Operand Operand { get; set; }
internal class S7VariableAddress public ushort DbNr { get; set; }
{ public ushort Start { get; set; }
public Operand Operand { get; set; } public ushort Length { get; set; }
public ushort DbNr { get; set; } public byte Bit { get; set; }
public ushort Start { get; set; } public DbType Type { get; set; }
public ushort Length { get; set; } }
public byte Bit { get; set; }
public DbType Type { get; set; }
}
}

135
Sharp7.Rx/S7VariableNameParser.cs
View File

@@ -1,84 +1,81 @@
using System; using System.Globalization;
using System.Collections.Generic;
using System.Globalization;
using System.Text.RegularExpressions; using System.Text.RegularExpressions;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
using Sharp7.Rx.Interfaces; using Sharp7.Rx.Interfaces;
namespace Sharp7.Rx namespace Sharp7.Rx;
internal class S7VariableNameParser : IS7VariableNameParser
{ {
internal class S7VariableNameParser : IS7VariableNameParser private static readonly Regex regex = new Regex(@"^(?<operand>db{1})(?<dbNr>\d{1,4})\.?(?<type>dbx|x|s|string|b|dbb|d|int|dbw|w|dint|dul|dulint|dulong|){1}(?<start>\d+)(\.(?<bitOrLength>\d+))?$", RegexOptions.IgnoreCase | RegexOptions.Compiled | RegexOptions.CultureInvariant);
private static readonly IReadOnlyDictionary<string, DbType> types = new Dictionary<string, DbType>(StringComparer.OrdinalIgnoreCase)
{ {
private static readonly Regex regex = new Regex(@"^(?<operand>db{1})(?<dbNr>\d{1,4})\.?(?<type>dbx|x|s|string|b|dbb|d|int|dbw|w|dint|dul|dulint|dulong|){1}(?<start>\d+)(\.(?<bitOrLength>\d+))?$", RegexOptions.IgnoreCase | RegexOptions.Compiled | RegexOptions.CultureInvariant); {"x", DbType.Bit},
{"dbx", DbType.Bit},
{"s", DbType.String},
{"string", DbType.String},
{"b", DbType.Byte},
{"dbb", DbType.Byte},
{"d", DbType.Double},
{"int", DbType.Integer},
{"dint", DbType.DInteger},
{"w", DbType.Integer},
{"dbw", DbType.Integer},
{"dul", DbType.ULong},
{"dulint", DbType.ULong},
{"dulong", DbType.ULong}
};
private static readonly IReadOnlyDictionary<string, DbType> types = new Dictionary<string, DbType>(StringComparer.OrdinalIgnoreCase) public S7VariableAddress Parse(string input)
{
var match = regex.Match(input);
if (match.Success)
{ {
{"x", DbType.Bit}, var operand = (Operand) Enum.Parse(typeof(Operand), match.Groups["operand"].Value, true);
{"dbx", DbType.Bit}, var dbNr = ushort.Parse(match.Groups["dbNr"].Value, NumberStyles.Integer, CultureInfo.InvariantCulture);
{"s", DbType.String}, var start = ushort.Parse(match.Groups["start"].Value, NumberStyles.Integer, CultureInfo.InvariantCulture);
{"string", DbType.String}, if (!types.TryGetValue(match.Groups["type"].Value, out var type))
{"b", DbType.Byte}, return null;
{"dbb", DbType.Byte},
{"d", DbType.Double},
{"int", DbType.Integer},
{"dint", DbType.DInteger},
{"w", DbType.Integer},
{"dbw", DbType.Integer},
{"dul", DbType.ULong},
{"dulint", DbType.ULong},
{"dulong", DbType.ULong}
};
public S7VariableAddress Parse(string input)
{ var s7VariableAddress = new S7VariableAddress
var match = regex.Match(input);
if (match.Success)
{ {
var operand = (Operand) Enum.Parse(typeof(Operand), match.Groups["operand"].Value, true); Operand = operand,
var dbNr = ushort.Parse(match.Groups["dbNr"].Value, NumberStyles.Integer, CultureInfo.InvariantCulture); DbNr = dbNr,
var start = ushort.Parse(match.Groups["start"].Value, NumberStyles.Integer, CultureInfo.InvariantCulture); Start = start,
if (!types.TryGetValue(match.Groups["type"].Value, out var type)) Type = type,
return null; };
switch (type)
var s7VariableAddress = new S7VariableAddress {
{ case DbType.Bit:
Operand = operand, s7VariableAddress.Length = 1;
DbNr = dbNr, s7VariableAddress.Bit = byte.Parse(match.Groups["bitOrLength"].Value);
Start = start, break;
Type = type, case DbType.Byte:
}; s7VariableAddress.Length = match.Groups["bitOrLength"].Success ? ushort.Parse(match.Groups["bitOrLength"].Value) : (ushort) 1;
break;
switch (type) case DbType.String:
{ s7VariableAddress.Length = match.Groups["bitOrLength"].Success ? ushort.Parse(match.Groups["bitOrLength"].Value) : (ushort) 0;
case DbType.Bit: break;
s7VariableAddress.Length = 1; case DbType.Integer:
s7VariableAddress.Bit = byte.Parse(match.Groups["bitOrLength"].Value); s7VariableAddress.Length = 2;
break; break;
case DbType.Byte: case DbType.DInteger:
s7VariableAddress.Length = match.Groups["bitOrLength"].Success ? ushort.Parse(match.Groups["bitOrLength"].Value) : (ushort) 1; s7VariableAddress.Length = 4;
break; break;
case DbType.String: case DbType.ULong:
s7VariableAddress.Length = match.Groups["bitOrLength"].Success ? ushort.Parse(match.Groups["bitOrLength"].Value) : (ushort) 0; s7VariableAddress.Length = 8;
break; break;
case DbType.Integer: case DbType.Double:
s7VariableAddress.Length = 2; s7VariableAddress.Length = 4;
break; break;
case DbType.DInteger:
s7VariableAddress.Length = 4;
break;
case DbType.ULong:
s7VariableAddress.Length = 8;
break;
case DbType.Double:
s7VariableAddress.Length = 4;
break;
}
return s7VariableAddress;
} }
return null; return s7VariableAddress;
} }
return null;
} }
} }

17
Sharp7.Rx/Settings/PlcConnectionSettings.cs
View File

@@ -1,10 +1,9 @@
namespace Sharp7.Rx.Settings namespace Sharp7.Rx.Settings;
internal class PlcConnectionSettings
{ {
internal class PlcConnectionSettings public int CpuMpiAddress { get; set; }
{ public string IpAddress { get; set; }
public int CpuMpiAddress { get; set; } public int Port { get; set; }
public string IpAddress { get; set; } public int RackNumber { get; set; }
public int Port { get; set; } }
public int RackNumber { get; set; }
}
}

490
Sharp7.Rx/Sharp7Connector.cs
View File

@@ -1,11 +1,6 @@
using System; using System.Reactive.Disposables;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Disposables;
using System.Reactive.Linq; using System.Reactive.Linq;
using System.Reactive.Subjects; using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Sharp7.Rx.Basics; using Sharp7.Rx.Basics;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
@@ -14,266 +9,265 @@ using Sharp7.Rx.Interfaces;
using Sharp7.Rx.Resources; using Sharp7.Rx.Resources;
using Sharp7.Rx.Settings; using Sharp7.Rx.Settings;
namespace Sharp7.Rx namespace Sharp7.Rx;
internal class Sharp7Connector : IS7Connector
{ {
internal class Sharp7Connector : IS7Connector private readonly BehaviorSubject<ConnectionState> connectionStateSubject = new BehaviorSubject<ConnectionState>(Enums.ConnectionState.Initial);
private readonly int cpuSlotNr;
private readonly CompositeDisposable disposables = new CompositeDisposable();
private readonly string ipAddress;
private readonly int port;
private readonly int rackNr;
private readonly LimitedConcurrencyLevelTaskScheduler scheduler = new LimitedConcurrencyLevelTaskScheduler(maxDegreeOfParallelism: 1);
private readonly IS7VariableNameParser variableNameParser;
private bool disposed;
private S7Client sharp7;
public Sharp7Connector(PlcConnectionSettings settings, IS7VariableNameParser variableNameParser)
{ {
private readonly BehaviorSubject<ConnectionState> connectionStateSubject = new BehaviorSubject<ConnectionState>(Enums.ConnectionState.Initial); this.variableNameParser = variableNameParser;
private readonly int cpuSlotNr; ipAddress = settings.IpAddress;
cpuSlotNr = settings.CpuMpiAddress;
port = settings.Port;
rackNr = settings.RackNumber;
private readonly CompositeDisposable disposables = new CompositeDisposable(); ReconnectDelay = TimeSpan.FromSeconds(5);
private readonly string ipAddress; }
private readonly int port;
private readonly int rackNr;
private readonly LimitedConcurrencyLevelTaskScheduler scheduler = new LimitedConcurrencyLevelTaskScheduler(maxDegreeOfParallelism: 1);
private readonly IS7VariableNameParser variableNameParser;
private bool disposed;
private S7Client sharp7; public IObservable<ConnectionState> ConnectionState => connectionStateSubject.DistinctUntilChanged().AsObservable();
public ILogger Logger { get; set; }
public Sharp7Connector(PlcConnectionSettings settings, IS7VariableNameParser variableNameParser) public TimeSpan ReconnectDelay { get; set; }
private bool IsConnected => connectionStateSubject.Value == Enums.ConnectionState.Connected;
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
public async Task<bool> Connect()
{
if (sharp7 == null)
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized);
try
{ {
this.variableNameParser = variableNameParser; var errorCode = await Task.Factory.StartNew(() => sharp7.ConnectTo(ipAddress, rackNr, cpuSlotNr), CancellationToken.None, TaskCreationOptions.None, scheduler);
ipAddress = settings.IpAddress; var success = EvaluateErrorCode(errorCode);
cpuSlotNr = settings.CpuMpiAddress; if (success)
port = settings.Port;
rackNr = settings.RackNumber;
ReconnectDelay = TimeSpan.FromSeconds(5);
}
public IObservable<ConnectionState> ConnectionState => connectionStateSubject.DistinctUntilChanged().AsObservable();
public ILogger Logger { get; set; }
public TimeSpan ReconnectDelay { get; set; }
private bool IsConnected => connectionStateSubject.Value == Enums.ConnectionState.Connected;
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
public async Task<bool> Connect()
{
if (sharp7 == null)
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized);
try
{ {
var errorCode = await Task.Factory.StartNew(() => sharp7.ConnectTo(ipAddress, rackNr, cpuSlotNr), CancellationToken.None, TaskCreationOptions.None, scheduler); connectionStateSubject.OnNext(Enums.ConnectionState.Connected);
var success = EvaluateErrorCode(errorCode);
if (success)
{
connectionStateSubject.OnNext(Enums.ConnectionState.Connected);
return true;
}
}
catch (Exception ex)
{
// TODO:
}
return false;
}
public async Task Disconnect()
{
connectionStateSubject.OnNext(Enums.ConnectionState.DisconnectedByUser);
await CloseConnection();
}
public async Task<Dictionary<string, byte[]>> ExecuteMultiVarRequest(IReadOnlyList<string> variableNames)
{
if (variableNames.IsEmpty())
return new Dictionary<string, byte[]>();
var s7MultiVar = new S7MultiVar(sharp7);
var buffers = variableNames
.Select(key => new {VariableName = key, Address = variableNameParser.Parse(key)})
.Select(x =>
{
var buffer = new byte[x.Address.Length];
s7MultiVar.Add(S7Consts.S7AreaDB, S7Consts.S7WLByte, x.Address.DbNr, x.Address.Start, x.Address.Length, ref buffer);
return new {x.VariableName, Buffer = buffer};
})
.ToArray();
var result = await Task.Factory.StartNew(() => s7MultiVar.Read(), CancellationToken.None, TaskCreationOptions.None, scheduler);
if (result != 0)
{
EvaluateErrorCode(result);
throw new InvalidOperationException($"Error in MultiVar request for variables: {string.Join(",", variableNames)}");
}
return buffers.ToDictionary(arg => arg.VariableName, arg => arg.Buffer);
}
public Task InitializeAsync()
{
try
{
sharp7 = new S7Client();
sharp7.PLCPort = port;
var subscription =
ConnectionState
.Where(state => state == Enums.ConnectionState.ConnectionLost)
.Take(1)
.SelectMany(_ => Reconnect())
.RepeatAfterDelay(ReconnectDelay)
.LogAndRetry(Logger, "Error while reconnecting to S7.")
.Subscribe();
disposables.Add(subscription);
}
catch (Exception ex)
{
Logger?.LogError(ex, StringResources.StrErrorS7DriverCouldNotBeInitialized);
}
return Task.FromResult(true);
}
public async Task<byte[]> ReadBytes(Operand operand, ushort startByteAddress, ushort bytesToRead, ushort dBNr, CancellationToken token)
{
EnsureConnectionValid();
var buffer = new byte[bytesToRead];
var result =
await Task.Factory.StartNew(() => sharp7.ReadArea(operand.ToArea(), dBNr, startByteAddress, bytesToRead, S7WordLength.Byte, buffer), token, TaskCreationOptions.None, scheduler);
token.ThrowIfCancellationRequested();
if (result != 0)
{
EvaluateErrorCode(result);
var errorText = sharp7.ErrorText(result);
throw new InvalidOperationException($"Error reading {operand}{dBNr}:{startByteAddress}->{bytesToRead} ({errorText})");
}
return buffer;
}
public async Task<bool> WriteBit(Operand operand, ushort startByteAddress, byte bitAdress, bool value, ushort dbNr, CancellationToken token)
{
EnsureConnectionValid();
var buffer = new[] {value ? (byte) 0xff : (byte) 0};
var offsetStart = (startByteAddress * 8) + bitAdress;
var result = await Task.Factory.StartNew(() => sharp7.WriteArea(operand.ToArea(), dbNr, offsetStart, 1, S7WordLength.Bit, buffer), token, TaskCreationOptions.None, scheduler);
token.ThrowIfCancellationRequested();
if (result != 0)
{
EvaluateErrorCode(result);
return (false);
}
return (true);
}
public async Task<ushort> WriteBytes(Operand operand, ushort startByteAdress, byte[] data, ushort dBNr, CancellationToken token)
{
EnsureConnectionValid();
var result = await Task.Factory.StartNew(() => sharp7.WriteArea(operand.ToArea(), dBNr, startByteAdress, data.Length, S7WordLength.Byte, data), token, TaskCreationOptions.None, scheduler);
token.ThrowIfCancellationRequested();
if (result != 0)
{
EvaluateErrorCode(result);
return 0;
}
return (ushort) (data.Length);
}
protected virtual void Dispose(bool disposing)
{
if (!disposed)
{
if (disposing)
{
disposables.Dispose();
if (sharp7 != null)
{
sharp7.Disconnect();
sharp7 = null;
}
connectionStateSubject?.OnCompleted();
connectionStateSubject?.Dispose();
}
disposed = true;
}
}
private async Task CloseConnection()
{
if (sharp7 == null)
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized);
await Task.Factory.StartNew(() => sharp7.Disconnect(), CancellationToken.None, TaskCreationOptions.None, scheduler);
}
private void EnsureConnectionValid()
{
if (disposed)
throw new ObjectDisposedException("S7Connector");
if (sharp7 == null)
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized);
if (!IsConnected)
throw new InvalidOperationException("Plc is not connected");
}
private bool EvaluateErrorCode(int errorCode)
{
if (errorCode == 0)
return true; return true;
}
if (sharp7 == null) }
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized); catch (Exception ex)
{
var errorText = sharp7.ErrorText(errorCode); // TODO:
Logger?.LogError($"Error Code {errorCode} {errorText}");
if (S7ErrorCodes.AssumeConnectionLost(errorCode))
SetConnectionLostState();
return false;
} }
private async Task<bool> Reconnect() return false;
{ }
await CloseConnection();
return await Connect();
public async Task Disconnect()
{
connectionStateSubject.OnNext(Enums.ConnectionState.DisconnectedByUser);
await CloseConnection();
}
public async Task<Dictionary<string, byte[]>> ExecuteMultiVarRequest(IReadOnlyList<string> variableNames)
{
if (variableNames.IsEmpty())
return new Dictionary<string, byte[]>();
var s7MultiVar = new S7MultiVar(sharp7);
var buffers = variableNames
.Select(key => new {VariableName = key, Address = variableNameParser.Parse(key)})
.Select(x =>
{
var buffer = new byte[x.Address.Length];
s7MultiVar.Add(S7Consts.S7AreaDB, S7Consts.S7WLByte, x.Address.DbNr, x.Address.Start, x.Address.Length, ref buffer);
return new {x.VariableName, Buffer = buffer};
})
.ToArray();
var result = await Task.Factory.StartNew(() => s7MultiVar.Read(), CancellationToken.None, TaskCreationOptions.None, scheduler);
if (result != 0)
{
EvaluateErrorCode(result);
throw new InvalidOperationException($"Error in MultiVar request for variables: {string.Join(",", variableNames)}");
} }
private void SetConnectionLostState() return buffers.ToDictionary(arg => arg.VariableName, arg => arg.Buffer);
{ }
if (connectionStateSubject.Value == Enums.ConnectionState.ConnectionLost) return;
connectionStateSubject.OnNext(Enums.ConnectionState.ConnectionLost); public Task InitializeAsync()
{
try
{
sharp7 = new S7Client();
sharp7.PLCPort = port;
var subscription =
ConnectionState
.Where(state => state == Enums.ConnectionState.ConnectionLost)
.Take(1)
.SelectMany(_ => Reconnect())
.RepeatAfterDelay(ReconnectDelay)
.LogAndRetry(Logger, "Error while reconnecting to S7.")
.Subscribe();
disposables.Add(subscription);
}
catch (Exception ex)
{
Logger?.LogError(ex, StringResources.StrErrorS7DriverCouldNotBeInitialized);
} }
~Sharp7Connector() return Task.FromResult(true);
}
public async Task<byte[]> ReadBytes(Operand operand, ushort startByteAddress, ushort bytesToRead, ushort dBNr, CancellationToken token)
{
EnsureConnectionValid();
var buffer = new byte[bytesToRead];
var result =
await Task.Factory.StartNew(() => sharp7.ReadArea(operand.ToArea(), dBNr, startByteAddress, bytesToRead, S7WordLength.Byte, buffer), token, TaskCreationOptions.None, scheduler);
token.ThrowIfCancellationRequested();
if (result != 0)
{ {
Dispose(false); EvaluateErrorCode(result);
var errorText = sharp7.ErrorText(result);
throw new InvalidOperationException($"Error reading {operand}{dBNr}:{startByteAddress}->{bytesToRead} ({errorText})");
}
return buffer;
}
public async Task<bool> WriteBit(Operand operand, ushort startByteAddress, byte bitAdress, bool value, ushort dbNr, CancellationToken token)
{
EnsureConnectionValid();
var buffer = new[] {value ? (byte) 0xff : (byte) 0};
var offsetStart = (startByteAddress * 8) + bitAdress;
var result = await Task.Factory.StartNew(() => sharp7.WriteArea(operand.ToArea(), dbNr, offsetStart, 1, S7WordLength.Bit, buffer), token, TaskCreationOptions.None, scheduler);
token.ThrowIfCancellationRequested();
if (result != 0)
{
EvaluateErrorCode(result);
return (false);
}
return (true);
}
public async Task<ushort> WriteBytes(Operand operand, ushort startByteAdress, byte[] data, ushort dBNr, CancellationToken token)
{
EnsureConnectionValid();
var result = await Task.Factory.StartNew(() => sharp7.WriteArea(operand.ToArea(), dBNr, startByteAdress, data.Length, S7WordLength.Byte, data), token, TaskCreationOptions.None, scheduler);
token.ThrowIfCancellationRequested();
if (result != 0)
{
EvaluateErrorCode(result);
return 0;
}
return (ushort) (data.Length);
}
protected virtual void Dispose(bool disposing)
{
if (!disposed)
{
if (disposing)
{
disposables.Dispose();
if (sharp7 != null)
{
sharp7.Disconnect();
sharp7 = null;
}
connectionStateSubject?.OnCompleted();
connectionStateSubject?.Dispose();
}
disposed = true;
} }
} }
private async Task CloseConnection()
{
if (sharp7 == null)
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized);
await Task.Factory.StartNew(() => sharp7.Disconnect(), CancellationToken.None, TaskCreationOptions.None, scheduler);
}
private void EnsureConnectionValid()
{
if (disposed)
throw new ObjectDisposedException("S7Connector");
if (sharp7 == null)
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized);
if (!IsConnected)
throw new InvalidOperationException("Plc is not connected");
}
private bool EvaluateErrorCode(int errorCode)
{
if (errorCode == 0)
return true;
if (sharp7 == null)
throw new InvalidOperationException(StringResources.StrErrorS7DriverNotInitialized);
var errorText = sharp7.ErrorText(errorCode);
Logger?.LogError($"Error Code {errorCode} {errorText}");
if (S7ErrorCodes.AssumeConnectionLost(errorCode))
SetConnectionLostState();
return false;
}
private async Task<bool> Reconnect()
{
await CloseConnection();
return await Connect();
}
private void SetConnectionLostState()
{
if (connectionStateSubject.Value == Enums.ConnectionState.ConnectionLost) return;
connectionStateSubject.OnNext(Enums.ConnectionState.ConnectionLost);
}
~Sharp7Connector()
{
Dispose(false);
}
} }

468
Sharp7.Rx/Sharp7Plc.cs
View File

@@ -1,13 +1,8 @@
using System; using System.Diagnostics;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Reactive; using System.Reactive;
using System.Reactive.Disposables; using System.Reactive.Disposables;
using System.Reactive.Linq; using System.Reactive.Linq;
using System.Text; using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Sharp7.Rx.Basics; using Sharp7.Rx.Basics;
using Sharp7.Rx.Enums; using Sharp7.Rx.Enums;
@@ -15,270 +10,269 @@ using Sharp7.Rx.Extensions;
using Sharp7.Rx.Interfaces; using Sharp7.Rx.Interfaces;
using Sharp7.Rx.Settings; using Sharp7.Rx.Settings;
namespace Sharp7.Rx namespace Sharp7.Rx;
public class Sharp7Plc : IPlc
{ {
public class Sharp7Plc : IPlc protected readonly CompositeDisposable Disposables = new CompositeDisposable();
private readonly ConcurrentSubjectDictionary<string, byte[]> multiVariableSubscriptions = new ConcurrentSubjectDictionary<string, byte[]>(StringComparer.InvariantCultureIgnoreCase);
private readonly List<long> performanceCoutner = new List<long>(1000);
private readonly PlcConnectionSettings plcConnectionSettings;
private readonly IS7VariableNameParser varaibleNameParser = new CacheVariableNameParser(new S7VariableNameParser());
private bool disposed;
private IS7Connector s7Connector;
/// <summary>
/// </summary>
/// <param name="ipAddress"></param>
/// <param name="rackNumber"></param>
/// <param name="cpuMpiAddress"></param>
/// <param name="port"></param>
/// <param name="multiVarRequestCycleTime">
/// <para>
/// Polling interval used to read multi variable requests from PLC.
/// </para>
/// <para>
/// This is the wait time between two successive reads from PLC and determines the
/// time resolution for all variable reads reated with CreateNotification.
/// </para>
/// <para>
/// Default is 100 ms. The minimum supported time is 5 ms.
/// </para>
/// </param>
public Sharp7Plc(string ipAddress, int rackNumber, int cpuMpiAddress, int port = 102, TimeSpan? multiVarRequestCycleTime = null)
{ {
protected readonly CompositeDisposable Disposables = new CompositeDisposable(); plcConnectionSettings = new PlcConnectionSettings {IpAddress = ipAddress, RackNumber = rackNumber, CpuMpiAddress = cpuMpiAddress, Port = port};
private readonly ConcurrentSubjectDictionary<string, byte[]> multiVariableSubscriptions = new ConcurrentSubjectDictionary<string, byte[]>(StringComparer.InvariantCultureIgnoreCase);
private readonly List<long> performanceCoutner = new List<long>(1000);
private readonly PlcConnectionSettings plcConnectionSettings;
private readonly IS7VariableNameParser varaibleNameParser = new CacheVariableNameParser(new S7VariableNameParser());
private bool disposed;
private IS7Connector s7Connector;
if (multiVarRequestCycleTime != null && multiVarRequestCycleTime > TimeSpan.FromMilliseconds(5))
MultiVarRequestCycleTime = multiVarRequestCycleTime.Value;
}
/// <summary> public IObservable<ConnectionState> ConnectionState { get; private set; }
/// </summary> public ILogger Logger { get; set; }
/// <param name="ipAddress"></param> public TimeSpan MultiVarRequestCycleTime { get; } = TimeSpan.FromSeconds(0.1);
/// <param name="rackNumber"></param>
/// <param name="cpuMpiAddress"></param>
/// <param name="port"></param>
/// <param name="multiVarRequestCycleTime">
/// <para>
/// Polling interval used to read multi variable requests from PLC.
/// </para>
/// <para>
/// This is the wait time between two successive reads from PLC and determines the
/// time resolution for all variable reads reated with CreateNotification.
/// </para>
/// <para>
/// Default is 100 ms. The minimum supported time is 5 ms.
/// </para>
/// </param>
public Sharp7Plc(string ipAddress, int rackNumber, int cpuMpiAddress, int port = 102, TimeSpan? multiVarRequestCycleTime = null)
{
plcConnectionSettings = new PlcConnectionSettings {IpAddress = ipAddress, RackNumber = rackNumber, CpuMpiAddress = cpuMpiAddress, Port = port};
if (multiVarRequestCycleTime != null && multiVarRequestCycleTime > TimeSpan.FromMilliseconds(5)) public int MultiVarRequestMaxItems { get; set; } = 16;
MultiVarRequestCycleTime = multiVarRequestCycleTime.Value;
}
public IObservable<ConnectionState> ConnectionState { get; private set; } public void Dispose()
public ILogger Logger { get; set; } {
public TimeSpan MultiVarRequestCycleTime { get; } = TimeSpan.FromSeconds(0.1); Dispose(true);
GC.SuppressFinalize(this);
}
public int MultiVarRequestMaxItems { get; set; } = 16; public IObservable<TValue> CreateNotification<TValue>(string variableName, TransmissionMode transmissionMode)
{
public void Dispose() return Observable.Create<TValue>(observer =>
{
Dispose(true);
GC.SuppressFinalize(this);
}
public IObservable<TValue> CreateNotification<TValue>(string variableName, TransmissionMode transmissionMode)
{
return Observable.Create<TValue>(observer =>
{
var address = varaibleNameParser.Parse(variableName);
if (address == null) throw new ArgumentException("Input variable name is not valid", nameof(variableName));
var disposables = new CompositeDisposable();
var disposeableContainer = multiVariableSubscriptions.GetOrCreateObservable(variableName);
disposeableContainer.AddDisposableTo(disposables);
var observable = disposeableContainer.Observable
.Select(bytes => S7ValueConverter.ConvertToType<TValue>(bytes, address));
if (transmissionMode == TransmissionMode.OnChange)
observable = observable.DistinctUntilChanged();
observable.Subscribe(observer)
.AddDisposableTo(disposables);
return disposables;
});
}
public Task<TValue> GetValue<TValue>(string variableName)
{
return GetValue<TValue>(variableName, CancellationToken.None);
}
public Task SetValue<TValue>(string variableName, TValue value)
{
return SetValue(variableName, value, CancellationToken.None);
}
public async Task<TValue> GetValue<TValue>(string variableName, CancellationToken token)
{ {
var address = varaibleNameParser.Parse(variableName); var address = varaibleNameParser.Parse(variableName);
if (address == null) throw new ArgumentException("Input variable name is not valid", nameof(variableName)); if (address == null) throw new ArgumentException("Input variable name is not valid", nameof(variableName));
var data = await s7Connector.ReadBytes(address.Operand, address.Start, address.Length, address.DbNr, token); var disposables = new CompositeDisposable();
return S7ValueConverter.ConvertToType<TValue>(data, address); var disposeableContainer = multiVariableSubscriptions.GetOrCreateObservable(variableName);
} disposeableContainer.AddDisposableTo(disposables);
public async Task<bool> InitializeAsync() var observable = disposeableContainer.Observable
{ .Select(bytes => S7ValueConverter.ConvertToType<TValue>(bytes, address));
s7Connector = new Sharp7Connector(plcConnectionSettings, varaibleNameParser) {Logger = Logger};
ConnectionState = s7Connector.ConnectionState;
await s7Connector.InitializeAsync(); if (transmissionMode == TransmissionMode.OnChange)
observable = observable.DistinctUntilChanged();
observable.Subscribe(observer)
.AddDisposableTo(disposables);
return disposables;
});
}
public Task<TValue> GetValue<TValue>(string variableName)
{
return GetValue<TValue>(variableName, CancellationToken.None);
}
public Task SetValue<TValue>(string variableName, TValue value)
{
return SetValue(variableName, value, CancellationToken.None);
}
public async Task<TValue> GetValue<TValue>(string variableName, CancellationToken token)
{
var address = varaibleNameParser.Parse(variableName);
if (address == null) throw new ArgumentException("Input variable name is not valid", nameof(variableName));
var data = await s7Connector.ReadBytes(address.Operand, address.Start, address.Length, address.DbNr, token);
return S7ValueConverter.ConvertToType<TValue>(data, address);
}
public async Task<bool> InitializeAsync()
{
s7Connector = new Sharp7Connector(plcConnectionSettings, varaibleNameParser) {Logger = Logger};
ConnectionState = s7Connector.ConnectionState;
await s7Connector.InitializeAsync();
#pragma warning disable 4014 #pragma warning disable 4014
Task.Run(async () => Task.Run(async () =>
{
try
{ {
try await s7Connector.Connect();
{ }
await s7Connector.Connect(); catch (Exception e)
} {
catch (Exception e) Logger?.LogError(e, "Error while connecting to PLC");
{ }
Logger?.LogError(e, "Error while connecting to PLC"); });
}
});
#pragma warning restore 4014 #pragma warning restore 4014
RunNotifications(s7Connector, MultiVarRequestCycleTime) RunNotifications(s7Connector, MultiVarRequestCycleTime)
.AddDisposableTo(Disposables); .AddDisposableTo(Disposables);
return true; return true;
} }
public async Task SetValue<TValue>(string variableName, TValue value, CancellationToken token) public async Task SetValue<TValue>(string variableName, TValue value, CancellationToken token)
{
var address = varaibleNameParser.Parse(variableName);
if (address == null) throw new ArgumentException("Input variable name is not valid", "variableName");
if (typeof(TValue) == typeof(bool))
{ {
var address = varaibleNameParser.Parse(variableName); await s7Connector.WriteBit(address.Operand, address.Start, address.Bit, (bool) (object) value, address.DbNr, token);
if (address == null) throw new ArgumentException("Input variable name is not valid", "variableName"); }
else if (typeof(TValue) == typeof(int) || typeof(TValue) == typeof(short))
{
byte[] bytes;
if (address.Length == 4)
bytes = BitConverter.GetBytes((int) (object) value);
else
bytes = BitConverter.GetBytes((short) (object) value);
if (typeof(TValue) == typeof(bool)) Array.Reverse(bytes);
{
await s7Connector.WriteBit(address.Operand, address.Start, address.Bit, (bool) (object) value, address.DbNr, token);
}
else if (typeof(TValue) == typeof(int) || typeof(TValue) == typeof(short))
{
byte[] bytes;
if (address.Length == 4)
bytes = BitConverter.GetBytes((int) (object) value);
else
bytes = BitConverter.GetBytes((short) (object) value);
Array.Reverse(bytes); await s7Connector.WriteBytes(address.Operand, address.Start, bytes, address.DbNr, token);
}
else if (typeof(TValue) == typeof(byte) || typeof(TValue) == typeof(char))
{
var bytes = new[] {Convert.ToByte(value)};
await s7Connector.WriteBytes(address.Operand, address.Start, bytes, address.DbNr, token);
}
else if (typeof(TValue) == typeof(byte[]))
{
await s7Connector.WriteBytes(address.Operand, address.Start, (byte[]) (object) value, address.DbNr, token);
}
else if (typeof(TValue) == typeof(float))
{
var buffer = new byte[sizeof(float)];
buffer.SetRealAt(0, (float) (object) value);
await s7Connector.WriteBytes(address.Operand, address.Start, buffer, address.DbNr, token);
}
else if (typeof(TValue) == typeof(string))
{
var stringValue = value as string;
if (stringValue == null) throw new ArgumentException("Value must be of type string", "value");
await s7Connector.WriteBytes(address.Operand, address.Start, bytes, address.DbNr, token); var bytes = Encoding.ASCII.GetBytes(stringValue);
} Array.Resize(ref bytes, address.Length);
else if (typeof(TValue) == typeof(byte) || typeof(TValue) == typeof(char))
{
var bytes = new[] {Convert.ToByte(value)};
await s7Connector.WriteBytes(address.Operand, address.Start, bytes, address.DbNr, token);
}
else if (typeof(TValue) == typeof(byte[]))
{
await s7Connector.WriteBytes(address.Operand, address.Start, (byte[]) (object) value, address.DbNr, token);
}
else if (typeof(TValue) == typeof(float))
{
var buffer = new byte[sizeof(float)];
buffer.SetRealAt(0, (float) (object) value);
await s7Connector.WriteBytes(address.Operand, address.Start, buffer, address.DbNr, token);
}
else if (typeof(TValue) == typeof(string))
{
var stringValue = value as string;
if (stringValue == null) throw new ArgumentException("Value must be of type string", "value");
var bytes = Encoding.ASCII.GetBytes(stringValue); if (address.Type == DbType.String)
Array.Resize(ref bytes, address.Length); {
var bytesWritten = await s7Connector.WriteBytes(address.Operand, address.Start, new[] {(byte) address.Length, (byte) bytes.Length}, address.DbNr, token);
if (address.Type == DbType.String) token.ThrowIfCancellationRequested();
if (bytesWritten == 2)
{ {
var bytesWritten = await s7Connector.WriteBytes(address.Operand, address.Start, new[] {(byte) address.Length, (byte) bytes.Length}, address.DbNr, token); var stringStartAddress = (ushort) (address.Start + 2);
token.ThrowIfCancellationRequested();
if (bytesWritten == 2)
{
var stringStartAddress = (ushort) (address.Start + 2);
token.ThrowIfCancellationRequested();
await s7Connector.WriteBytes(address.Operand, stringStartAddress, bytes, address.DbNr, token);
}
}
else
{
await s7Connector.WriteBytes(address.Operand, address.Start, bytes, address.DbNr, token);
token.ThrowIfCancellationRequested(); token.ThrowIfCancellationRequested();
await s7Connector.WriteBytes(address.Operand, stringStartAddress, bytes, address.DbNr, token);
} }
} }
else else
{ {
throw new InvalidOperationException($"type '{typeof(TValue)}' not supported."); await s7Connector.WriteBytes(address.Operand, address.Start, bytes, address.DbNr, token);
token.ThrowIfCancellationRequested();
} }
} }
else
protected virtual void Dispose(bool disposing)
{ {
if (disposed) return; throw new InvalidOperationException($"type '{typeof(TValue)}' not supported.");
disposed = true;
if (disposing)
{
Disposables.Dispose();
if (s7Connector != null)
{
s7Connector.Disconnect().Wait();
s7Connector.Dispose();
s7Connector = null;
}
multiVariableSubscriptions.Dispose();
}
}
private async Task<Unit> GetAllValues(bool connected, IS7Connector connector)
{
if (!connected)
return Unit.Default;
if (multiVariableSubscriptions.ExistingKeys.IsEmpty())
return Unit.Default;
var stopWatch = Stopwatch.StartNew();
foreach (var partsOfMultiVarRequest in multiVariableSubscriptions.ExistingKeys.Buffer(MultiVarRequestMaxItems))
{
var multiVarRequest = await connector.ExecuteMultiVarRequest(partsOfMultiVarRequest as IReadOnlyList<string>);
foreach (var pair in multiVarRequest)
if (multiVariableSubscriptions.TryGetObserver(pair.Key, out var subject))
subject.OnNext(pair.Value);
}
stopWatch.Stop();
performanceCoutner.Add(stopWatch.ElapsedMilliseconds);
PrintAndResetPerformanceStatistik();
return Unit.Default;
}
private void PrintAndResetPerformanceStatistik()
{
if (performanceCoutner.Count == performanceCoutner.Capacity)
{
var average = performanceCoutner.Average();
var min = performanceCoutner.Min();
var max = performanceCoutner.Max();
Logger?.LogTrace("Performance statistic during {0} elements of plc notification. Min: {1}, Max: {2}, Average: {3}, Plc: '{4}', Number of variables: {5}, Batch size: {6}", performanceCoutner.Capacity, min, max, average, plcConnectionSettings.IpAddress,
multiVariableSubscriptions.ExistingKeys.Count(),
MultiVarRequestMaxItems);
performanceCoutner.Clear();
}
}
private IDisposable RunNotifications(IS7Connector connector, TimeSpan cycle)
{
return ConnectionState.FirstAsync()
.Select(states => states == Enums.ConnectionState.Connected)
.SelectMany(connected => GetAllValues(connected, connector))
.RepeatAfterDelay(MultiVarRequestCycleTime)
.LogAndRetryAfterDelay(Logger, cycle, "Error while getting batch notifications from plc")
.Subscribe();
}
~Sharp7Plc()
{
Dispose(false);
} }
} }
}
protected virtual void Dispose(bool disposing)
{
if (disposed) return;
disposed = true;
if (disposing)
{
Disposables.Dispose();
if (s7Connector != null)
{
s7Connector.Disconnect().Wait();
s7Connector.Dispose();
s7Connector = null;
}
multiVariableSubscriptions.Dispose();
}
}
private async Task<Unit> GetAllValues(bool connected, IS7Connector connector)
{
if (!connected)
return Unit.Default;
if (multiVariableSubscriptions.ExistingKeys.IsEmpty())
return Unit.Default;
var stopWatch = Stopwatch.StartNew();
foreach (var partsOfMultiVarRequest in multiVariableSubscriptions.ExistingKeys.Buffer(MultiVarRequestMaxItems))
{
var multiVarRequest = await connector.ExecuteMultiVarRequest(partsOfMultiVarRequest as IReadOnlyList<string>);
foreach (var pair in multiVarRequest)
if (multiVariableSubscriptions.TryGetObserver(pair.Key, out var subject))
subject.OnNext(pair.Value);
}
stopWatch.Stop();
performanceCoutner.Add(stopWatch.ElapsedMilliseconds);
PrintAndResetPerformanceStatistik();
return Unit.Default;
}
private void PrintAndResetPerformanceStatistik()
{
if (performanceCoutner.Count == performanceCoutner.Capacity)
{
var average = performanceCoutner.Average();
var min = performanceCoutner.Min();
var max = performanceCoutner.Max();
Logger?.LogTrace("Performance statistic during {0} elements of plc notification. Min: {1}, Max: {2}, Average: {3}, Plc: '{4}', Number of variables: {5}, Batch size: {6}", performanceCoutner.Capacity, min, max, average, plcConnectionSettings.IpAddress,
multiVariableSubscriptions.ExistingKeys.Count(),
MultiVarRequestMaxItems);
performanceCoutner.Clear();
}
}
private IDisposable RunNotifications(IS7Connector connector, TimeSpan cycle)
{
return ConnectionState.FirstAsync()
.Select(states => states == Enums.ConnectionState.Connected)
.SelectMany(connected => GetAllValues(connected, connector))
.RepeatAfterDelay(MultiVarRequestCycleTime)
.LogAndRetryAfterDelay(Logger, cycle, "Error while getting batch notifications from plc")
.Subscribe();
}
~Sharp7Plc()
{
Dispose(false);
}
}