Minor async changes #15 looking to fix change feed bug in cursor

bchavez · Nov 22, 2015 · 34c25b4 · 34c25b4
1 parent 1dc1699
commit 34c25b4
Show file tree

Hide file tree

Showing 5 changed files with 49 additions and 31 deletions.
diff --git a/Source/RethinkDb.Driver.Tests/ReQL/RxReactiveExtensionTests.cs b/Source/RethinkDb.Driver.Tests/ReQL/RxReactiveExtensionTests.cs
@@ -26,7 +26,8 @@ public void basic_change_feed_with_reactive_extensions()
             result.ChangesAs<JObject>().Dump();
 
             var changes = r.db(DbName).table(TableName)
-                .changes()[new {include_states = true, include_initial = true}]
+                //.changes()[new {include_states = true, include_initial = true}]
+                .changes()[new { include_states = true }]
                 //.runCursor<Change<JObject>>(conn);
                 .runChanges<JObject>(conn);
 

diff --git a/Source/RethinkDb.Driver.Tests/TestingCommon.cs b/Source/RethinkDb.Driver.Tests/TestingCommon.cs
@@ -242,6 +242,10 @@ public Err(string errorType, string errorMessage, object obj)
 
             public override bool Equals(object obj)
             {
+                if( obj is AggregateException )
+                {
+                    obj = (obj as AggregateException).InnerException;
+                }
                 if( obj.GetType() != this.clazz )
                 {
                     obj.GetType().Should().Be(this.clazz);
@@ -652,6 +656,10 @@ public ErrRegex(String classname, String message_rgx)
 
             public override bool Equals(Object other)
             {
+                if( other is AggregateException )
+                {
+                    other = (other as AggregateException).InnerException;
+                }
                 var otherClass = other.GetType().Name;
 
                 if( !otherClass.EndsWith(this.clazz) )

diff --git a/Source/RethinkDb.Driver/Net/ConnectionInstance.cs b/Source/RethinkDb.Driver/Net/ConnectionInstance.cs
@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using RethinkDb.Driver.Ast;

diff --git a/Source/RethinkDb.Driver/Net/Cursor.cs b/Source/RethinkDb.Driver/Net/Cursor.cs
@@ -2,6 +2,7 @@
 using System.Collections;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using Newtonsoft.Json.Linq;
 using RethinkDb.Driver.Ast;
 using RethinkDb.Driver.Proto;
@@ -32,6 +33,7 @@ public abstract class Cursor<T> : IEnumerable<T>, IEnumerator<T>, ICursor
         protected internal Exception error = null;
         public bool IsFeed { get; }
 
+
         protected Cursor(Connection connection, Query query, Response firstResponse)
         {
             this.connection = connection;
@@ -158,6 +160,10 @@ public override bool MoveNext()
                     //finished awaiting.
                     var result = connection.AwaitResponseAsync(query, NetUtil.Deadline(timeout))
                         .RunSync();
+
+                    //there's a problem here....... with change feed types, 
+                    //of cursors, data can arrive without having to "send";
+
                     this.Extend(result);
                 }
 

diff --git a/Source/RethinkDb.Driver/Net/SocketWrapper.cs b/Source/RethinkDb.Driver/Net/SocketWrapper.cs
@@ -58,7 +58,7 @@ public virtual void Connect(byte[] handshake)
                     throw new ReqlDriverError($"Server dropped connection with message: '{msg}'");
                 }
 
-                Task.Factory.StartNew(ResponseLoop, TaskCreationOptions.LongRunning);
+                Task.Factory.StartNew(ResponsePump, TaskCreationOptions.LongRunning);
             }
             catch when( !taskComplete )
             {
@@ -101,44 +101,48 @@ private string ReadNullTerminatedString(TimeSpan? deadline)
         /// <summary>
         /// Started just after connect.
         /// </summary>
-        private void ResponseLoop()
+        private void ResponsePump()
         {
             pump = new CancellationTokenSource();
 
-            while( true )
+            using( pump )
             {
-                if( pump.Token.IsCancellationRequested )
+                while( true )
                 {
-                    Log.Trace("Response Loop: shutting down. Cancel is requested.");
-                    break;
-                }
-                if( this.Closed )
-                {
-                    Log.Trace("Response Loop: The connected socket is not open. Response Loop exiting.");
-                    break;
-                }
+                    if( pump.Token.IsCancellationRequested )
+                    {
+                        Log.Trace("Response Loop: shutting down. Cancel is requested.");
+                        break;
+                    }
+                    if( this.Closed )
+                    {
+                        Log.Trace("Response Loop: The connected socket is not open. Response Loop exiting.");
+                        break;
+                    }
 
-                try
-                {
-                    var response = this.Read();
-                    TaskCompletionSource<Response> awaitingTask;
-                    if( awaiters.TryRemove(response.Token, out awaitingTask) )
+                    try
                     {
-                        //Push, don't block.
-                        Task.Run(() => awaitingTask.SetResult(response));
-                        //See ya...
+                        var response = this.Read();
+                        Log.Trace($"Message Pump: Read {response.Token}");
+                        TaskCompletionSource<Response> awaitingTask;
+                        if( awaiters.TryRemove(response.Token, out awaitingTask) )
+                        {
+                            //Push, don't block.
+                            Task.Run(() => awaitingTask.SetResult(response));
+                            //See ya later alligator
+                        }
+                        else
+                        {
+                            //Wow, there's nobody waiting for this response.
+                            Log.Debug($"Response Loop: There are no awaiters waiting for {response.Token} token.");
+                            //I guess we'll ignore for now, perhaps a cursor was killed
+                        }
                     }
-                    else
+                    catch( Exception e ) when( !pump.Token.IsCancellationRequested )
                     {
-                        //Wow, there's nobody waiting for this response.
-                        Log.Debug($"Response Loop: There are no awaiters waiting for {response.Token} token.");
-                        //I guess we'll ignore for now, perhaps a cursor was killed
+                        Log.Debug($"Response Loop: Exception - {e.Message}");
                     }
                 }
-                catch( Exception e ) when( !pump.Token.IsCancellationRequested )
-                {
-                    Log.Debug($"Response Loop: Exception - {e.Message}");
-                }
             }
 
             //clean up.
@@ -191,7 +195,7 @@ public virtual void WriteQuery(long token, string json, bool assignAwaiter = tru
 
         public virtual void Close()
         {
-            this.pump.Cancel();
+            this.pump?.Cancel();
             try
             {
 #if DNXCORE50