Added multivar create notification

Sharp7.Rx/Basics/ConcurrentSubjectDictionary.cs

@@ -0,0 +1,128 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Reactive;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using Sharp7.Rx.Extensions;
+
+namespace Sharp7.Rx.Basics
+{
+    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()
+        {
+            dictionary = new ConcurrentDictionary<TKey, SubjectWithRefCounter>();
+        }
+
+        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;
+                });
+
+                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)
+            {
+                dictionary.Values.DisposeItems();
+                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; }
+        }
+    }
+}

Sharp7.Rx/Basics/DisposableItem.cs

@@ -0,0 +1,38 @@
+using System;
+using System.Linq;
+
+namespace Sharp7.Rx.Basics
+{
+    internal class DisposableItem<TValue> : IDisposable
+    {
+        private readonly Action disposeAction;
+
+        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;
+        }
+    }
+}

Sharp7.Rx/Extensions/DisposableExtensions.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Reactive.Disposables;
 
 namespace Sharp7.Rx.Extensions
@@ -9,5 +11,11 @@ namespace Sharp7.Rx.Extensions
         {
             compositeDisposable.Add(disposable);
         }
+        
+        public static void DisposeItems(this IEnumerable<object> disposables)
+        {
+            foreach (IDisposable disposable in disposables.OfType<IDisposable>())
+                disposable?.Dispose();
+        }
     }
 }

Sharp7.Rx/Interfaces/IS7Connector.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
@@ -20,5 +21,6 @@ namespace Sharp7.Rx.Interfaces
         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);
 	    ILogger Logger { get; }
+        Task<Dictionary<string, byte[]>> ExecuteMultiVarRequest(IEnumerable<string> variableNames);
     }
 }

Sharp7.Rx/Interfaces/IS7VariableNameParser.cs

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

Sharp7.Rx/S7VariableNameParser.cs

@@ -4,10 +4,11 @@ using System.Globalization;
 using System.Linq;
 using System.Text.RegularExpressions;
 using Sharp7.Rx.Enums;
+using Sharp7.Rx.Interfaces;
 
 namespace Sharp7.Rx
 {
-    internal class S7VariableNameParser
+    internal class S7VariableNameParser : IS7VariableNameParser
     {
         private 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);
 

Sharp7.Rx/Settings/PlcConnectionSettings.cs

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

Sharp7.Rx/Sharp7.Rx.csproj

@@ -17,6 +17,7 @@
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.Logging" Version="2.1.1" />
     <PackageReference Include="Sharp7" Version="1.0.18" />
+    <PackageReference Include="System.Interactive" Version="4.0.0" />
     <PackageReference Include="System.Reactive" Version="4.1.0" />
   </ItemGroup>
 

Sharp7.Rx/Sharp7Connector.cs

@@ -1,4 +1,7 @@
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
@@ -9,12 +12,16 @@ using Sharp7.Rx.Enums;
 using Sharp7.Rx.Extensions;
 using Sharp7.Rx.Interfaces;
 using Sharp7.Rx.Resources;
+using Sharp7.Rx.Settings;
 
 namespace Sharp7.Rx
 {
     internal class Sharp7Connector : IS7Connector
     {
+        private readonly IS7VariableNameParser variableNameParser;
         private readonly BehaviorSubject<ConnectionState> connectionStateSubject = new BehaviorSubject<ConnectionState>(Enums.ConnectionState.Initial);
+        private ConcurrentDictionary<string, S7VariableAddress> s7VariableAddresses = new ConcurrentDictionary<string, S7VariableAddress>();
+
         private readonly CompositeDisposable disposables = new CompositeDisposable();
         private readonly TaskScheduler scheduler = TaskScheduler.Current;
         private readonly string ipAddress;
@@ -26,13 +33,43 @@ namespace Sharp7.Rx
         private bool disposed;
 
 		public ILogger Logger { get; set; }
+        public async Task<Dictionary<string, byte[]>> ExecuteMultiVarRequest(IEnumerable<string> variableNames)
+        {
+            var enumerable = variableNames as string[] ?? variableNames.ToArray();
 
-		public Sharp7Connector(string ipAddress, int rackNr = 0, int cpuSlotNr = 2, int port = 102)
+            if (enumerable.IsEmpty())
+                return new Dictionary<string, byte[]>();
+
+            var s7MultiVar = new S7MultiVar(sharp7);
+
+            var buffers = enumerable.Select(key => new {VariableName = key, Address = s7VariableAddresses.GetOrAdd(key, s => variableNameParser.Parse(s))})
+                .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)
+            {
+                await EvaluateErrorCode(result);
+                throw new InvalidOperationException($"Error in MultiVar request for variables: {string.Join(",", enumerable)}");
+            }
+
+            return buffers.ToDictionary(arg => arg.VariableName, arg => arg.Buffer);
+        }
+
+        
+        
+        public Sharp7Connector(PlcConnectionSettings settings, IS7VariableNameParser variableNameParser)
 		{
-            this.ipAddress = ipAddress;
+            this.variableNameParser = variableNameParser;
-            this.cpuSlotNr = cpuSlotNr;
+            this.ipAddress = settings.IpAddress;
-			this.port = port;
+            this.cpuSlotNr = settings.CpuMpiAddress;
-			this.rackNr = rackNr;
+			this.port = settings.Port;
+			this.rackNr = settings.Port;
 
 			ReconnectDelay = TimeSpan.FromSeconds(5);
         }

Sharp7.Rx/Sharp7Plc.cs

@@ -1,16 +1,21 @@
 using System;
+using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Reactive;
+using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
+using Sharp7.Rx.Basics;
 using Sharp7.Rx.Enums;
 using Sharp7.Rx.Extensions;
 using Sharp7.Rx.Interfaces;
 using Sharp7.Rx.Resources;
+using Sharp7.Rx.Settings;
 
 namespace Sharp7.Rx
 {
@@ -19,16 +24,26 @@ namespace Sharp7.Rx
         private readonly string ipAddress;
         private readonly int rackNumber;
         private readonly int cpuMpiAddress;
-        private readonly S7VariableNameParser varaibleNameParser;
+        private readonly int port;
+        private readonly IS7VariableNameParser varaibleNameParser;
         private bool disposed;
         private ISubject<Unit> disposingSubject = new Subject<Unit>();
         private IS7Connector s7Connector;
+        private readonly PlcConnectionSettings plcConnectionSettings;
+        private readonly ConcurrentSubjectDictionary<string, byte[]> multiVariableSubscriptions = new ConcurrentSubjectDictionary<string, byte[]>(StringComparer.InvariantCultureIgnoreCase);
+        protected readonly CompositeDisposable Disposables = new CompositeDisposable();        
+        private readonly List<long> performanceCoutner = new List<long>(1000);
 
-        public Sharp7Plc(string ipAddress, int rackNumber, int cpuMpiAddress)
+
+
+        public Sharp7Plc(string ipAddress, int rackNumber, int cpuMpiAddress, int port = 102)
         {
             this.ipAddress = ipAddress;
             this.rackNumber = rackNumber;
             this.cpuMpiAddress = cpuMpiAddress;
+            this.port = port;
+            
+            plcConnectionSettings = new PlcConnectionSettings(){IpAddress = ipAddress, RackNumber = rackNumber, CpuMpiAddress = cpuMpiAddress, Port = port};
 
             varaibleNameParser = new S7VariableNameParser();
         }
@@ -37,10 +52,13 @@ namespace Sharp7.Rx
 
         public async Task<bool> InitializeAsync()
         {
-            s7Connector = new Sharp7Connector(ipAddress, rackNumber, cpuMpiAddress);
+            s7Connector = new Sharp7Connector(plcConnectionSettings, varaibleNameParser);
             ConnectionState = s7Connector.ConnectionState;
 
             await s7Connector.InitializeAsync();
+            
+            RunNotifications(s7Connector, TimeSpan.FromMilliseconds(100))
+                .AddDisposableTo(Disposables);
 
 #pragma warning disable 4014
             Task.Run(async () =>
@@ -64,6 +82,72 @@ namespace Sharp7.Rx
             return GetValue<TValue>(variableName, CancellationToken.None);
         }
 
+        private TValue ConvertToType<TValue>(byte[] buffer, S7VariableAddress address)
+        {
+            if (typeof(TValue) == typeof(bool))
+            {
+                return (TValue) (object) Convert.ToBoolean(buffer[0] & (1 << address.Bit));
+            }
+
+            if (typeof(TValue) == typeof(int))
+            {
+                if (address.Length == 2)
+                    return (TValue)(object)((buffer[0] << 8) + buffer[1]);
+                if (address.Length == 4)
+                {
+                    Array.Reverse(buffer);
+                    return (TValue)(object)BitConverter.ToInt32(buffer,0);
+                }
+
+                throw new InvalidOperationException($"length must be 2 or 4 but is {address.Length}");
+            }
+
+            if (typeof(TValue) == typeof(long))
+            {
+                Array.Reverse(buffer);
+                return (TValue)(object)BitConverter.ToInt64(buffer,0);
+            }
+
+            if (typeof(TValue) == typeof(ulong))
+            {
+                Array.Reverse(buffer);
+                return (TValue)(object)BitConverter.ToUInt64(buffer, 0);
+            }
+
+            if (typeof(TValue) == typeof(short))
+            {
+                return (TValue)(object)(short)((buffer[0] << 8) + buffer[1]);
+            }
+
+            if (typeof(TValue) == typeof(byte) || typeof(TValue) == typeof(char))
+            {
+                return (TValue)(object)buffer[0];
+            }
+
+            if (typeof(TValue) == typeof(byte[]))
+            {
+                return (TValue)(object)buffer;
+            }
+
+            if (typeof(TValue) == typeof(double) || typeof(TValue) == typeof(float))
+            {
+                var d = BitConverter.ToSingle(buffer.Reverse().ToArray(),0);
+                return (TValue)(object)d;
+            }
+
+            if (typeof(TValue) == typeof(string))
+                if (address.Type == DbType.String)
+                {
+                    return (TValue) (object) Encoding.ASCII.GetString(buffer);
+                }
+                else
+                {
+                    return (TValue) (object) Encoding.ASCII.GetString(buffer).Trim();
+                }
+
+            throw new InvalidOperationException(string.Format("type '{0}' not supported.", typeof(TValue)));
+        }
+        
         public async Task<TValue> GetValue<TValue>(string variableName, CancellationToken token)
         {
             var address = varaibleNameParser.Parse(variableName);
@@ -235,44 +319,30 @@ namespace Sharp7.Rx
             }
         }
 
-        public IObservable<TValue> CreateNotification<TValue>(string variableName, TransmissionMode transmissionMode, TimeSpan cycle)
+        public IObservable<TValue> CreateNotification<TValue>(string variableName, TransmissionMode transmissionMode, TimeSpan cycleTime)
         {
-            var address = varaibleNameParser.Parse(variableName);
+            return Observable.Create<TValue>(observer =>
-            if (address == null) throw new ArgumentException("Input variable name is not valid", nameof(variableName));
+            {
+                var address = varaibleNameParser.Parse(variableName);
+                if (address == null) throw new ArgumentException("Input variable name is not valid", nameof(variableName));
 
-            if (cycle < TimeSpan.FromMilliseconds(100))
+                var disposables = new CompositeDisposable();
-                cycle = TimeSpan.FromMilliseconds(100);
+                var disposeableContainer = multiVariableSubscriptions.GetOrCreateObservable(variableName);
+                disposeableContainer.AddDisposableTo(disposables);
 
-            var notification = ConnectionState.FirstAsync().Select(states => states == Enums.ConnectionState.Connected)
+                var observable = disposeableContainer.Observable
-                .SelectMany(async connected =>
+                    .Select(bytes => ConvertToType<TValue>(bytes, address));
-                {
-                    var value = default(TValue);
-                    if (connected)
-                    {
-                        value = await GetValue<TValue>(variableName, CancellationToken.None);
-                    }
 
-                    return new
+                if (transmissionMode == TransmissionMode.OnChange)
-                    {
+                    observable = observable.DistinctUntilChanged();
-                        HasValue = connected,
-                        Value = value
-                    };
-                })
-                .RepeatAfterDelay(cycle)
-                .LogAndRetryAfterDelay(s7Connector.Logger, cycle, StringResources.StrLogErrorReadingDataFromPlc)
-                .TakeUntil(disposingSubject)
-                .Where(union => union.HasValue)
-                .Select(union => union.Value);
 
-            if (transmissionMode == TransmissionMode.Cyclic)
+                observable.Subscribe(observer)
-                return notification;
+                    .AddDisposableTo(disposables);
 
-            if (transmissionMode == TransmissionMode.OnChange)
+                return disposables;
-                return notification.DistinctUntilChanged();
+            });
-
-            throw new ArgumentException("Transmission mode can either be Cyclic or OnChange", nameof(transmissionMode));
         }
-
+        
         public void Dispose()
         {
             Dispose(true);
@@ -284,6 +354,8 @@ namespace Sharp7.Rx
             {
                 if (disposing)
                 {
+                    Disposables.Dispose();
+                    
                     if (disposingSubject != null)
                     {
                         disposingSubject.OnNext(Unit.Default);
@@ -308,5 +380,62 @@ namespace Sharp7.Rx
         {
             Dispose(false);
         }
+        
+        private IDisposable RunNotifications(IS7Connector connector, TimeSpan cycle)
+        {
+            return ConnectionState.FirstAsync()
+                .Select(states => states == Enums.ConnectionState.Connected)
+                .SelectMany(connected => GetAllValues(connected, connector))
+                .RepeatAfterDelay(cycle)
+                .LogAndRetryAfterDelay(s7Connector.Logger, cycle, "Error while getting batch notifications from plc")
+                .TakeUntil(disposingSubject)
+                .Subscribe();
+        }
+        
+        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);
+
+                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();
+
+                s7Connector.Logger.LogInformation("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();
+            }
+        }
+
+        public int MultiVarRequestMaxItems { get; set; } = 16;
     }
 }