Observable backup and restore - SnapshotObservable and RestoreObservable unit tests

Author: robertlyson

src/Nest/Domain/Responses/RestoreObservable.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Text.RegularExpressions;
@@ -13,13 +14,22 @@ public class RestoreObservable : IDisposable, IObservable<IRecoveryStatusRespons
         private readonly TimeSpan _interval = TimeSpan.FromSeconds(2);
         private Timer _timer;
         private bool _disposed;
-        private string _renamePattern;
-        private string _renameReplacement;
+        private readonly RestoreStatusHumbleObject _restoreStatusHumbleObject;
+        private EventHandler<RestoreNextEventArgs> _nextEventHandlers;
+        private EventHandler<RestoreCompletedEventArgs> _completedEentHandlers;
+        private EventHandler<RestoreErrorEventArgs> _errorEventHandlers; 
 
         public RestoreObservable(IElasticClient elasticClient, IRestoreRequest restoreRequest)
         {
+            elasticClient.ThrowIfNull("elasticClient");
+            restoreRequest.ThrowIfNull("restoreRequest");
+
             _elasticClient = elasticClient;
             _restoreRequest = restoreRequest;
+
+            _restoreStatusHumbleObject = new RestoreStatusHumbleObject(elasticClient, restoreRequest);
+            _restoreStatusHumbleObject.Completed += StopTimer;
+            _restoreStatusHumbleObject.Error += StopTimer;
         }
 
         public RestoreObservable(IElasticClient elasticClient, IRestoreRequest restoreRequest, TimeSpan interval)
@@ -43,10 +53,19 @@ public IDisposable Subscribe(IObserver<IRecoveryStatusResponse> observer)
                 if (!restoreResponse.IsValid)
                     throw new RestoreException(restoreResponse.ConnectionStatus);
 
-                _renamePattern = _restoreRequest.RenamePattern;
-                _renameReplacement = _restoreRequest.RenameReplacement;
+                EventHandler<RestoreNextEventArgs> onNext = (sender, args) => observer.OnNext(args.RecoveryStatusResponse);
+                EventHandler<RestoreCompletedEventArgs> onCompleted = (sender, args) => observer.OnCompleted();
+                EventHandler<RestoreErrorEventArgs> onError = (sender, args) => observer.OnError(args.Exception);
+
+                _nextEventHandlers = onNext;
+                _completedEentHandlers = onCompleted;
+                _errorEventHandlers = onError;
 
-                _timer = new Timer(Restore, observer, _interval.Milliseconds, Timeout.Infinite);
+                _restoreStatusHumbleObject.Next += onNext;
+                _restoreStatusHumbleObject.Completed += onCompleted;
+                _restoreStatusHumbleObject.Error += onError;
+
+                _timer = new Timer(Restore, observer, (long)_interval.TotalMilliseconds, Timeout.Infinite);
             }
             catch (Exception exception)
             {
@@ -67,33 +86,20 @@ private void Restore(object state)
                 var watch = new Stopwatch();
                 watch.Start();
 
-                var recoveryStatus = _elasticClient.RecoveryStatus(
-                    descriptor =>
-                        descriptor.Indices(
-                            _restoreRequest.Indices.Select(
-                                x => Regex.Replace(x.Name, _renamePattern, _renameReplacement)).ToArray())
-                            .Detailed(true)
-                    );
-
-                if (!recoveryStatus.IsValid)
-                    throw new RestoreException(recoveryStatus.ConnectionStatus);
-
-                if (recoveryStatus.Indices.All(x => x.Value.Shards.All(s => s.Index.Files.Recovered == s.Index.Files.Total)))
-                {
-                    _timer.Change(Timeout.Infinite, Timeout.Infinite);
-                    observer.OnCompleted();
-                    return;
-                }
+                _restoreStatusHumbleObject.CheckStatus();
 
-                observer.OnNext(recoveryStatus);
-                _timer.Change(Math.Max(0, _interval.Milliseconds - watch.ElapsedMilliseconds), Timeout.Infinite);
+                _timer.Change(Math.Max(0, (long)_interval.TotalMilliseconds - watch.ElapsedMilliseconds), Timeout.Infinite);
             }
             catch (Exception exception)
             {
-                _timer.Change(Timeout.Infinite, Timeout.Infinite);
                 observer.OnError(exception);
             }
         }
+
+        private void StopTimer(object sender, EventArgs restoreCompletedEventArgs)
+        {
+            _timer.Change(Timeout.Infinite, Timeout.Infinite);
+        }
 
         public void Dispose()
         {
@@ -105,6 +111,12 @@ protected virtual void Dispose(bool disposing)
             if (_disposed) return;
 
             _timer.Dispose();
+            _restoreStatusHumbleObject.Next -= _nextEventHandlers;
+            _restoreStatusHumbleObject.Completed -= _completedEentHandlers;
+            _restoreStatusHumbleObject.Error -= _errorEventHandlers;
+
+            _restoreStatusHumbleObject.Completed -= StopTimer;
+            _restoreStatusHumbleObject.Error -= StopTimer;
 
             _disposed = true;
         }
@@ -114,4 +126,112 @@ protected virtual void Dispose(bool disposing)
             Dispose(false);
         }
     }
+
+    public class RestoreNextEventArgs : EventArgs
+    {
+        public IRecoveryStatusResponse RecoveryStatusResponse { get; private set; }
+
+        public RestoreNextEventArgs(IRecoveryStatusResponse recoveryStatusResponse)
+        {
+            RecoveryStatusResponse = recoveryStatusResponse;
+        }
+    }
+
+    public class RestoreCompletedEventArgs : EventArgs
+    {
+        public IRecoveryStatusResponse RecoveryStatusResponse { get; private set; }
+
+        public RestoreCompletedEventArgs(IRecoveryStatusResponse recoveryStatusResponse)
+        {
+            RecoveryStatusResponse = recoveryStatusResponse;
+        }
+    }
+
+    public class RestoreErrorEventArgs : EventArgs
+    {
+        public Exception Exception { get; private set; }
+
+        public RestoreErrorEventArgs(Exception exception)
+        {
+            Exception = exception;
+        }
+    }
+
+    public class RestoreStatusHumbleObject
+    {
+        private readonly IElasticClient _elasticClient;
+        private readonly IRestoreRequest _restoreRequest;
+        private string _renamePattern;
+        private string _renameReplacement;
+
+        public event EventHandler<RestoreCompletedEventArgs> Completed;
+        public event EventHandler<RestoreErrorEventArgs> Error;
+        public event EventHandler<RestoreNextEventArgs> Next;
+
+        public RestoreStatusHumbleObject(IElasticClient elasticClient, IRestoreRequest restoreRequest)
+        {
+            elasticClient.ThrowIfNull("elasticClient");
+            restoreRequest.ThrowIfNull("restoreRequest");
+
+            _elasticClient = elasticClient;
+            _restoreRequest = restoreRequest;
+
+            _renamePattern = string.IsNullOrEmpty(_restoreRequest.RenamePattern) ? string.Empty : _restoreRequest.RenamePattern;
+            _renameReplacement = string.IsNullOrEmpty(_restoreRequest.RenameReplacement) ? string.Empty : _restoreRequest.RenameReplacement;
+        }
+
+        public void CheckStatus()
+        {
+            try
+            {
+                var indices =
+                    _restoreRequest.Indices.Select(
+                        x => new IndexNameMarker
+                        {
+                            Name = Regex.Replace(x.Name, _renamePattern, _renameReplacement),
+                            Type = x.Type
+                        })
+                        .ToArray();
+
+                var recoveryStatus = _elasticClient.RecoveryStatus(new RecoveryStatusRequest
+                {
+                    Detailed = true,
+                    Indices = indices
+                });
+
+                if (!recoveryStatus.IsValid)
+                    throw new RestoreException(recoveryStatus.ConnectionStatus);
+
+                if (recoveryStatus.Indices.All(x => x.Value.Shards.All(s => s.Index.Files.Recovered == s.Index.Files.Total)))
+                {
+                    OnCompleted(new RestoreCompletedEventArgs(recoveryStatus));
+                    return;
+                }
+
+                OnNext(new RestoreNextEventArgs(recoveryStatus));
+            }
+            catch (Exception exception)
+            {
+                OnError(new RestoreErrorEventArgs(exception));
+            }
+        }
+
+        protected virtual void OnNext(RestoreNextEventArgs nextEventArgs)
+        {
+            var handler = Next;
+            if (handler != null) handler(this, nextEventArgs);
+        }
+
+        protected virtual void OnCompleted(RestoreCompletedEventArgs completedEventArgs)
+        {
+            var handler = Completed;
+            if (handler != null) handler(this, completedEventArgs);
+        }
+
+        protected virtual void OnError(RestoreErrorEventArgs errorEventArgs)
+        {
+            var handler = Error;
+            if (handler != null) handler(this, errorEventArgs);
+        }
+    }
 }