CSHARP-2267: ChangeStream cursor raises resumable exception when there is no event since last replica set election.

Author: rstam

src/MongoDB.Driver.Core/Core/Operations/ChangeStreamCursor.cs

@@ -81,20 +81,18 @@ public void Dispose()
         public bool MoveNext(CancellationToken cancellationToken = default(CancellationToken))
         {
             bool hasMore;
-            try
+            while (true)
             {
-                hasMore = _cursor.MoveNext(cancellationToken);
-            }
-            catch (Exception ex)
-            {
-                if (RetryabilityHelper.IsResumableChangeStreamException(ex))
+                try
                 {
-                    _cursor = _changeStreamOperation.Resume(_binding, cancellationToken);
                     hasMore = _cursor.MoveNext(cancellationToken);
+                    break;
                 }
-                else
+                catch (Exception ex) when (RetryabilityHelper.IsResumableChangeStreamException(ex))
                 {
-                    throw;
+                    var newCursor = _changeStreamOperation.Resume(_binding, cancellationToken);
+                    _cursor.Dispose();
+                    _cursor = newCursor;
                 }
             }
 
@@ -104,22 +102,20 @@ public void Dispose()
 
         /// <inheritdoc/>
         public async Task<bool> MoveNextAsync(CancellationToken cancellationToken = default(CancellationToken))
-        {
+        {          
             bool hasMore;
-            try
+            while (true)
             {
-                hasMore = await _cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false);
-            }
-            catch (Exception ex)
-            {
-                if (RetryabilityHelper.IsResumableChangeStreamException(ex))
+                try
                 {
-                    _cursor = await _changeStreamOperation.ResumeAsync(_binding, cancellationToken).ConfigureAwait(false);
                     hasMore = await _cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false);
+                    break;
                 }
-                else
+                catch (Exception ex) when (RetryabilityHelper.IsResumableChangeStreamException(ex))
                 {
-                    throw;
+                    var newCursor = await _changeStreamOperation.ResumeAsync(_binding, cancellationToken).ConfigureAwait(false);
+                    _cursor.Dispose();
+                    _cursor = newCursor;
                 }
             }