From 956f39cc6658fb3e3b4264b6a2cf20c14c7406c1 Mon Sep 17 00:00:00 2001
From: Peter Butzhammer <peter.butzhammer@evopro-ag.de>
Date: Wed, 7 Feb 2024 08:51:40 +0100
Subject: [PATCH] Complete Notifications when Plc is disposed

---
 .../Basics/ConcurrentSubjectDictionary.cs     |  3 +-
 Sharp7.Rx/Extensions/DisposableExtensions.cs  |  8 ----
 Sharp7.Rx/Sharp7Plc.cs                        | 44 +++++++------------
 3 files changed, 17 insertions(+), 38 deletions(-)

diff --git a/Sharp7.Rx/Basics/ConcurrentSubjectDictionary.cs b/Sharp7.Rx/Basics/ConcurrentSubjectDictionary.cs
index fa5184e..041991c 100644
--- a/Sharp7.Rx/Basics/ConcurrentSubjectDictionary.cs
+++ b/Sharp7.Rx/Basics/ConcurrentSubjectDictionary.cs
@@ -85,7 +85,8 @@ namespace Sharp7.Rx.Basics
                 return;
             if (disposing && dictionary != null)
             {
-                dictionary.Values.DisposeItems();
+                foreach (var subjectWithRefCounter in dictionary) 
+                    subjectWithRefCounter.Value.Subject.OnCompleted();
                 dictionary.Clear();
                 dictionary = null;
             }
diff --git a/Sharp7.Rx/Extensions/DisposableExtensions.cs b/Sharp7.Rx/Extensions/DisposableExtensions.cs
index 0ad5c06..89f3c75 100644
--- a/Sharp7.Rx/Extensions/DisposableExtensions.cs
+++ b/Sharp7.Rx/Extensions/DisposableExtensions.cs
@@ -1,6 +1,4 @@
 using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Reactive.Disposables;
 
 namespace Sharp7.Rx.Extensions
@@ -11,11 +9,5 @@ namespace Sharp7.Rx.Extensions
         {
             compositeDisposable.Add(disposable);
         }
-        
-        public static void DisposeItems(this IEnumerable<object> disposables)
-        {
-            foreach (IDisposable disposable in disposables.OfType<IDisposable>())
-                disposable?.Dispose();
-        }
     }
 }
diff --git a/Sharp7.Rx/Sharp7Plc.cs b/Sharp7.Rx/Sharp7Plc.cs
index ca46309..a1a7664 100644
--- a/Sharp7.Rx/Sharp7Plc.cs
+++ b/Sharp7.Rx/Sharp7Plc.cs
@@ -5,7 +5,6 @@ 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;
@@ -22,7 +21,6 @@ namespace Sharp7.Rx
     {
         private readonly IS7VariableNameParser varaibleNameParser = new CacheVariableNameParser(new S7VariableNameParser());
         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);
@@ -182,33 +180,26 @@ namespace Sharp7.Rx
         public void Dispose()
         {
             Dispose(true);
+            GC.SuppressFinalize(this);
         }
 
         protected virtual void Dispose(bool disposing)
         {
-            if (!disposed)
+            if (disposed) return;
+            disposed = true;
+
+            if (disposing)
             {
-                if (disposing)
-                {
-                    Disposables.Dispose();
+                Disposables.Dispose();
                     
-                    if (disposingSubject != null)
-                    {
-                        disposingSubject.OnNext(Unit.Default);
-                        disposingSubject.OnCompleted();
-                        var disposable = (disposingSubject as IDisposable);
-                        if (disposable != null) disposable.Dispose();
-                        disposingSubject = null;
-                    }
-                    if (s7Connector != null)
-                    {
-                        s7Connector.Disconnect().Wait();
-                        s7Connector.Dispose();
-                        s7Connector = null;
-                    }
+                if (s7Connector != null)
+                {
+                    s7Connector.Disconnect().Wait();
+                    s7Connector.Dispose();
+                    s7Connector = null;
                 }
 
-                disposed = true;
+                multiVariableSubscriptions.Dispose();
             }
         }
 
@@ -224,7 +215,6 @@ namespace Sharp7.Rx
                 .SelectMany(connected => GetAllValues(connected, connector))
                 .RepeatAfterDelay(cycle)
                 .LogAndRetryAfterDelay(Logger, cycle, "Error while getting batch notifications from plc")
-                .TakeUntil(disposingSubject)
                 .Subscribe();
         }
         
@@ -239,15 +229,11 @@ namespace Sharp7.Rx
             var stopWatch = Stopwatch.StartNew();
             foreach (var partsOfMultiVarRequest in multiVariableSubscriptions.ExistingKeys.Buffer(MultiVarRequestMaxItems))
             {
-                var multiVarRequest = await connector.ExecuteMultiVarRequest(partsOfMultiVarRequest as IReadOnlyList<string>??partsOfMultiVarRequest.ToList());
-
+                var multiVarRequest = await connector.ExecuteMultiVarRequest(partsOfMultiVarRequest as IReadOnlyList<string>);
+                
                 foreach (var pair in multiVarRequest)
-                {
-                    if (multiVariableSubscriptions.TryGetObserver(pair.Key, out var subject))
-                    {
+                    if (multiVariableSubscriptions.TryGetObserver(pair.Key, out var subject)) 
                         subject.OnNext(pair.Value);
-                    }
-                }
             }
 
             stopWatch.Stop();