Mostly tests work

az-templates/stage-build.yml

@@ -13,20 +13,20 @@ stages:
         projects: 'src/*.sln'
         arguments: '-c $(BuildConfiguration) --no-incremental --nologo -p:TreatWarningsAsErrors=true -p:Version=$(SemVer) -p:InformationalVersion=$(InfoVer)'
 
-    - task: DotNetCoreCLI@2
+    - task: VSTest@2
       displayName: 'UnitTests .NetCoreApp2.2'
       inputs:
-        command: test
-        projects: 'src/Tests/**/UnitTests.csproj'
-        arguments: '-c $(BuildConfiguration) -f netcoreapp2.2 --no-build'
+        testSelector: 'testAssemblies'
+        testAssemblyVer2: 'src/Tests/UnitTests/bin/$(BuildConfiguration)/netcoreapp2.2/UnitTests.dll'
+        configuration: $(BuildConfiguration)
         testRunTitle: 'UnitTests .NetCoreApp2.2'
 
-    - task: DotNetCoreCLI@2
+    - task: VSTest@2
       displayName: 'UnitTests .Net4.5.2'
       inputs:
-        command: test
-        projects: 'src/Tests/**/UnitTests.csproj'
-        arguments: '-c $(BuildConfiguration) -f net452 --no-build'
+        testSelector: 'testAssemblies'
+        testAssemblyVer2: 'src/Tests/UnitTests/bin/$(BuildConfiguration)/net452/UnitTests.dll'
+        configuration: $(BuildConfiguration)
         testRunTitle: 'UnitTests .Net4.5.2'
 
     - task: PowerShell@2
@@ -62,29 +62,35 @@ stages:
           $natsServerDir = Get-ChildItem -Directory -Path "$(Agent.TempDirectory)\nats-server\nats-server-*" | Select -Expand FullName
           Write-Host "Found nats-server path: $natsServerDir"
           Write-Host "##vso[task.setvariable variable=PATH;]${env:PATH};$natsServerDir"
-  
-    - task: DotNetCoreCLI@2
+
+    - task: VSTest@2
       displayName: 'IntegrationTests .NetCoreApp2.2'
       inputs:
-        command: test
-        projects: 'src/Tests/**/IntegrationTests.csproj'
-        arguments: '-c $(BuildConfiguration) -f netcoreapp2.2 --no-build'
+        testSelector: 'testAssemblies'
+        testAssemblyVer2: 'src/Tests/IntegrationTests/bin/$(BuildConfiguration)/netcoreapp2.2/IntegrationTests.dll'
+        configuration: $(BuildConfiguration)
+        rerunFailedTests: True
+        rerunMaxAttempts: 2
         testRunTitle: 'IntegrationTests .NetCoreApp2.2'
 
-    - task: DotNetCoreCLI@2
+    - task: VSTest@2
       displayName: 'IntegrationTests .NetCoreApp3.0'
       inputs:
-        command: test
-        projects: 'src/Tests/**/IntegrationTests.csproj'
-        arguments: '-c $(BuildConfiguration) -f netcoreapp3.0 --no-build'
+        testSelector: 'testAssemblies'
+        testAssemblyVer2: 'src/Tests/IntegrationTests/bin/$(BuildConfiguration)/netcoreapp3.0/IntegrationTests.dll'
+        configuration: $(BuildConfiguration)
+        rerunFailedTests: True
+        rerunMaxAttempts: 2
         testRunTitle: 'IntegrationTests .NetCoreApp3.0'
 
-    - task: DotNetCoreCLI@2
+    - task: VSTest@2
       displayName: 'IntegrationTests .Net4.5.2'
       inputs:
-        command: test
-        projects: 'src/Tests/**/IntegrationTests.csproj'
-        arguments: '-c $(BuildConfiguration) -f net452 --no-build'
+        testSelector: 'testAssemblies'
+        testAssemblyVer2: 'src/Tests/IntegrationTests/bin/$(BuildConfiguration)/net452/IntegrationTests.dll'
+        configuration: $(BuildConfiguration)
+        rerunFailedTests: True
+        rerunMaxAttempts: 2
         testRunTitle: 'IntegrationTests .Net4.5.2'
 
     - task: DotNetCoreCLI@2

azure-pipelines-ci.yml

@@ -18,7 +18,6 @@ pr:
     - master
 
 pool:
-  name: Azure Pipelines
   vmImage: windows-2019
 
 stages:

src/Directory.Build.props

@@ -19,6 +19,5 @@
     <Copyright>Copyright © The NATS Authors 2016-2019</Copyright>
     <RepositoryUrl>https://github.com/nats-io/nats.net</RepositoryUrl>
     <RepositoryType>Git</RepositoryType>
-    <LangVersion>latest</LangVersion>
   </PropertyGroup>
 </Project>

src/NATS.Client/IEncodedConnection.cs

@@ -179,7 +179,7 @@ public interface IEncodedConnection : IDisposable
         /// <exception cref="NATSConnectionClosedException">The <see cref="Connection"/> is closed.</exception>
         /// <exception cref="NATSException">There was an unexpected exception performing an internal NATS call while executing the
         /// request. See <see cref="Exception.InnerException"/> for more details.</exception>
-        public void FlushBuffer();
+        void FlushBuffer();
 
         /// <summary>
         /// Closes the <see cref="IConnection"/> and all associated

src/NATS.Client/Rx/Ops/WhereObservable.cs

@@ -39,7 +39,7 @@ private sealed class WhereObserver : IObserver<T>
             public WhereObserver(IObserver<T> observer, Func<T, bool> predicate)
             {
                 this.observer = observer ?? throw new ArgumentNullException(nameof(observer));
-                this.predicate = predicate ?? throw new ArgumentNullException(nameof(observer));
+                this.predicate = predicate ?? throw new ArgumentNullException(nameof(predicate));
             }
 
             public void OnNext(T value)

src/Tests/IntegrationTests/IntegrationTests.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.3.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.4.0" />
     <PackageReference Include="xunit" Version="2.4.1" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.4.1">
       <PrivateAssets>all</PrivateAssets>

src/Tests/IntegrationTests/TestAsyncAwaitDeadlocks.cs

@@ -1,5 +1,19 @@
-#if !NET452
+// Copyright 2015-2019 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
 
+
+#if !NET452
+using System;
 using System.Collections.Concurrent;
 using System.Threading.Tasks;
 using NATS.Client;
@@ -272,9 +286,8 @@ public void EnsurePubSubWithAsyncHandler()
                     {
                         using (var cn = Context.ConnectionFactory.CreateConnection(Context.Server1.Url))
                         {
-                            using (cn.SubscribeAsync(subject, async (_, m) =>
+                            using (cn.SubscribeAsync(subject, (_, m) =>
                             {
-                                await Task.Delay(100);
                                 recieved.Enqueue(m.Message);
                                 if(recieved.Count == 5)
                                     sync.SignalComplete();
@@ -292,6 +305,143 @@ public void EnsurePubSubWithAsyncHandler()
                 }
             });
         }
+
+        [Fact]
+        public void EnsureAutoUnsubscribeForSyncSub()
+        {
+            var subject = "de82267b22454dd7afc37c9e34a8e0ab";
+            var recieved = new ConcurrentQueue<Msg>();
+
+            AsyncContext.Run(() =>
+            {
+                using (NATSServer.CreateFast(Context.Server1.Port))
+                {
+                    using (var cn = Context.ConnectionFactory.CreateConnection(Context.Server1.Url))
+                    {
+                        using (var sub = cn.SubscribeSync(subject))
+                        {
+                            sub.AutoUnsubscribe(1);
+
+                            cn.Publish(subject, new byte[0]);
+
+                            recieved.Enqueue(sub.NextMessage());
+
+                            cn.Publish(subject, new byte[0]);
+
+                            Assert.Equal(1, sub.Delivered);
+                        }
+                    }
+
+                    Assert.Single(recieved);
+                }
+            });
+        }
+
+        [Fact]
+        public void EnsureAutoUnsubscribeForAsyncSub()
+        {
+            var subject = "0be903f6c9c14c10973e78ce03ad47e1";
+            var recieved = new ConcurrentQueue<Msg>();
+
+            AsyncContext.Run(async () =>
+            {
+                using (NATSServer.CreateFast(Context.Server1.Port))
+                {
+                    using (var sync = TestSync.SingleActor())
+                    {
+                        using (var cn = Context.ConnectionFactory.CreateConnection(Context.Server1.Url))
+                        {
+                            using (var sub = cn.SubscribeAsync(subject, (_, m) =>
+                            {
+                                recieved.Enqueue(m.Message);
+                                sync.SignalComplete();
+                            }))
+                            {
+                                sub.AutoUnsubscribe(1);
+
+                                cn.Publish(subject, new byte[0]);
+                                cn.Publish(subject, new byte[0]);
+
+                                sync.WaitForAll();
+
+                                await Task.Delay(100);
+                                Assert.Equal(1, sub.Delivered);
+                            }
+                        }
+
+                        Assert.Single(recieved);
+                    }
+                }
+            });
+        }
+
+        [Fact]
+        public void EnsureUnsubscribeForSyncSub()
+        {
+            var subject = "b2dcc4f56fd041cb985300d4966bd1c1";
+            var recieved = new ConcurrentQueue<Msg>();
+
+            AsyncContext.Run(() =>
+            {
+                using (NATSServer.CreateFast(Context.Server1.Port))
+                {
+                    using (var cn = Context.ConnectionFactory.CreateConnection(Context.Server1.Url))
+                    {
+                        using (var sub = cn.SubscribeSync(subject))
+                        {
+                            cn.Publish(subject, new byte[0]);
+
+                            recieved.Enqueue(sub.NextMessage());
+
+                            sub.Unsubscribe();
+
+                            cn.Publish(subject, new byte[0]);
+                            Assert.Throws<NATSBadSubscriptionException>(sub.NextMessage);
+                            Assert.Equal(1, sub.Delivered);
+                        }
+                    }
+
+                    Assert.Single(recieved);
+                }
+            });
+        }
+
+        [Fact]
+        public void EnsureUnsubscribeForAsyncSub()
+        {
+            var subject = "d37e3729c5c84702b836a4bb4edf7241";
+            var recieved = new ConcurrentQueue<Msg>();
+
+            AsyncContext.Run(async () =>
+            {
+                using (NATSServer.CreateFast(Context.Server1.Port))
+                {
+                    using (var sync = TestSync.SingleActor())
+                    {
+                        using (var cn = Context.ConnectionFactory.CreateConnection(Context.Server1.Url))
+                        {
+                            using (var sub = cn.SubscribeAsync(subject, (_, m) =>
+                            {
+                                recieved.Enqueue(m.Message);
+                                sync.SignalComplete();
+                            }))
+                            {
+                                cn.Publish(subject, new byte[0]);
+                                sync.WaitForAll();
+
+                                sub.Unsubscribe();
+
+                                cn.Publish(subject, new byte[0]);
+                                await Task.Delay(100);
+                                Assert.Equal(1, sub.Delivered);
+                            }
+                        }
+
+                        Assert.Single(recieved);
+                    }
+                }
+            });
+        }
     }
 }
 #endif