Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Routing driver #115

Merged
merged 11 commits into from
Nov 11, 2016
Merged

Routing driver #115

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
8181910
Refactoring on the connection pool
Oct 7, 2016
ec4ae51
Adding support for `bolt+routing` scheme.
Oct 11, 2016
ca36e6d
Added concurrent round robin set for accessing servers
Oct 19, 2016
5399603
Added basic routing:
Oct 20, 2016
3f52d99
Re-structed the classes related to routing in to one place
Oct 20, 2016
17765b2
Adding error handling hock into socketConnection send and receive method
Oct 21, 2016
d143b54
Added basic error handling callback
Oct 25, 2016
b7d5d46
Added test for `ClusterDiscoveryManager`
Oct 27, 2016
898dea3
Added tests for LoadBalancer
Oct 31, 2016
7fe2977
Added timeout of server routing results
Nov 2, 2016
1bd65e5
Fix after code review
Nov 10, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 3 additions & 6 deletions Neo4j.Driver/Neo4j.Driver.Tck.Tests/Tck/ErrorReportingSteps.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,12 +87,9 @@ public void WhenISetUpADriverToAnIncorrectPort()
[When(@"I set up a driver with wrong scheme")]
public void WhenISetUpADriverWithWrongScheme()
{
using (var driver = GraphDatabase.Driver("http://localhost"))
{
var ex = Xunit.Record.Exception(() => driver.Session());
ex.Should().BeOfType<NotSupportedException>();
ex.Message.Should().Be("Unsupported protocol: http");
}
var ex = Xunit.Record.Exception(() => GraphDatabase.Driver("http://localhost"));
ex.Should().BeOfType<NotSupportedException>();
ex.Message.Should().Be("Unsupported URI scheme: http");
}

[Then(@"it throws a `ClientException`")]
Expand Down
73 changes: 52 additions & 21 deletions Neo4j.Driver/Neo4j.Driver.Tests/ConnectionPoolTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ private IConnection MockedConnection
get
{
var mock = new Mock<IConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);
return mock.Object;
}
}
Expand All @@ -47,6 +47,37 @@ public AcquireMethod(ITestOutputHelper output)
_output = output;
}

[Fact]
public void ShouldAddExternalConnectionHandlerIfNotNull()
{
// Given
var mock = new Mock<IConnection>();
mock.Setup(x => x.IsOpen).Returns(true);

var mockedHandler = new Mock<IConnectionErrorHandler>();
var connectionPool = new ConnectionPool(mock.Object, exteralErrorHandler:mockedHandler.Object);
// When
connectionPool.Acquire();

//Then
mock.Verify(x=>x.AddConnectionErrorHander(mockedHandler.Object), Times.Once);
mock.Verify(x => x.Init(), Times.Once);
}

[Fact]
public void ShouldCallConnInit()
{
// Given
var mock = new Mock<IConnection>();
mock.Setup(x => x.IsOpen).Returns(true);
var connectionPool = new ConnectionPool(mock.Object);
// When
connectionPool.Acquire();

//Then
mock.Verify(x => x.Init(), Times.Once);
}

[Fact]
public void ShouldNotThrowExceptionWhenIdlePoolSizeReached()
{
Expand Down Expand Up @@ -116,7 +147,7 @@ public void ShouldCreateNewWhenQueueOnlyContainsUnhealthyConnections()
var conns = new Queue<IPooledConnection>();
var unhealthyId = Guid.NewGuid();
var unhealthyMock = new Mock<IPooledConnection>();
unhealthyMock.Setup(x => x.IsHealthy).Returns(false);
unhealthyMock.Setup(x => x.IsOpen).Returns(false);
unhealthyMock.Setup(x => x.Id).Returns(unhealthyId);

conns.Enqueue(unhealthyMock.Object);
Expand All @@ -129,7 +160,7 @@ public void ShouldCreateNewWhenQueueOnlyContainsUnhealthyConnections()

pool.NumberOfAvailableConnections.Should().Be(0);
pool.NumberOfInUseConnections.Should().Be(1);
unhealthyMock.Verify(x => x.IsHealthy, Times.Once);
unhealthyMock.Verify(x => x.IsOpen, Times.Once);
unhealthyMock.Verify(x => x.Close(), Times.Once);

conn.Should().NotBeNull();
Expand All @@ -141,7 +172,7 @@ public void ShouldReuseOldWhenReusableConnectionInQueue()
{
var conns = new Queue<IPooledConnection>();
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);

conns.Enqueue(mock.Object);
var pool = new ConnectionPool(MockedConnection, conns);
Expand All @@ -153,7 +184,7 @@ public void ShouldReuseOldWhenReusableConnectionInQueue()

pool.NumberOfAvailableConnections.Should().Be(0);
pool.NumberOfInUseConnections.Should().Be(1);
mock.Verify(x => x.IsHealthy, Times.Once);
mock.Verify(x => x.IsOpen, Times.Once);
conn.Should().Be(mock.Object);
}

Expand All @@ -162,9 +193,9 @@ public void ShouldReuseReusableWhenReusableConnectionInQueue()
{
var conns = new Queue<IPooledConnection>();
var healthyMock = new Mock<IPooledConnection>();
healthyMock.Setup(x => x.IsHealthy).Returns(true);
healthyMock.Setup(x => x.IsOpen).Returns(true);
var unhealthyMock = new Mock<IPooledConnection>();
unhealthyMock.Setup(x => x.IsHealthy).Returns(false);
unhealthyMock.Setup(x => x.IsOpen).Returns(false);

conns.Enqueue(unhealthyMock.Object);
conns.Enqueue(healthyMock.Object);
Expand Down Expand Up @@ -201,7 +232,7 @@ public void ShouldAcquireNewWhenBeingUsedConcurrentlyBy(int numberOfThreads)
for (var i = 0; i < numberOfThreads; i++)
{
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);
mock.Setup(x => x.Id).Returns(ids[i]);
conns.Enqueue(mock.Object);
mockConns.Enqueue(mock);
Expand Down Expand Up @@ -246,7 +277,7 @@ public void ShouldAcquireNewWhenBeingUsedConcurrentlyBy(int numberOfThreads)

foreach (var mock in mockConns)
{
mock.Verify(x => x.IsHealthy, Times.Once);
mock.Verify(x => x.IsOpen, Times.Once);
}
}

Expand Down Expand Up @@ -277,7 +308,7 @@ public void ShouldCloseAcquiredConnectionIfPoolDisposeStarted()
// This is to simulate Acquire called first,
// but before Acquire put a new conn into inUseConn, Dispose get called.
// Note: Once dispose get called, it is forbiden to put anything into queue.
healthyMock.Setup(x => x.IsHealthy).Returns(true)
healthyMock.Setup(x => x.IsOpen).Returns(true)
.Callback(() => pool.DisposeCalled = true); // Simulte Dispose get called at this time
conns.Enqueue(healthyMock.Object);
pool.NumberOfAvailableConnections.Should().Be(1);
Expand All @@ -286,7 +317,7 @@ public void ShouldCloseAcquiredConnectionIfPoolDisposeStarted()

pool.NumberOfAvailableConnections.Should().Be(0);
pool.NumberOfInUseConnections.Should().Be(0);
healthyMock.Verify(x => x.IsHealthy, Times.Once);
healthyMock.Verify(x => x.IsOpen, Times.Once);
healthyMock.Verify(x => x.Close(), Times.Once);
exception.Should().BeOfType<InvalidOperationException>();
exception.Message.Should().Contain("the driver has already started to dispose");
Expand All @@ -299,7 +330,7 @@ public class ReleaseMethod
public void ShouldReturnToPoolWhenConnectionIsReusableAndPoolIsNotFull()
{
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);
var id = new Guid();

var inUseconns = new Dictionary<Guid, IPooledConnection>();
Expand All @@ -319,7 +350,7 @@ public void ShouldReturnToPoolWhenConnectionIsReusableAndPoolIsNotFull()
public void ShouldCloseConnectionWhenConnectionIsUnhealthy()
{
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(false);
mock.Setup(x => x.IsOpen).Returns(false);
var id = new Guid();

var inUseConns = new Dictionary<Guid, IPooledConnection>();
Expand All @@ -337,10 +368,10 @@ public void ShouldCloseConnectionWhenConnectionIsUnhealthy()
}

[Fact]
public void ShouldCloseConnectionWhenConnectionIsHealthyButNotResetable()
public void ShouldCloseConnectionWhenConnectionIsOpenButNotResetable()
{
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);
mock.Setup(x => x.ClearConnection()).Throws<ClientException>();
var id = new Guid();

Expand All @@ -362,7 +393,7 @@ public void ShouldCloseConnectionWhenConnectionIsHealthyButNotResetable()
public void ShouldCloseTheConnectionIfSessionIsReusableButThePoolIsFull()
{
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);
var id = new Guid();

var inUseConns = new Dictionary<Guid, IPooledConnection>();
Expand Down Expand Up @@ -408,7 +439,7 @@ public void ShouldCloseConnectionIfPoolDisposeStarted()
// this is to simulate Release called first,
// but before Release put a new conn into availConns, Dispose get called.
// Note: Once dispose get called, it is forbiden to put anything into queue.
mock.Setup(x => x.IsHealthy).Returns(true)
mock.Setup(x => x.IsOpen).Returns(true)
.Callback(() => pool.DisposeCalled = true); // Simulte Dispose get called at this time
pool.Release(id);

Expand All @@ -425,14 +456,14 @@ public class DisposeMethod
public void ShouldReleaseAll()
{
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);
var id = Guid.NewGuid();
var inUseConns = new Dictionary<Guid, IPooledConnection>();
inUseConns.Add(id, mock.Object);

var availableConns = new Queue<IPooledConnection>();
var mock1 = new Mock<IPooledConnection>();
mock1.Setup(x => x.IsHealthy).Returns(true);
mock1.Setup(x => x.IsOpen).Returns(true);

availableConns.Enqueue(mock1.Object);

Expand All @@ -451,14 +482,14 @@ public void ShouldLogInUseAndAvailableConnectionIds()
{
var mockLogger = new Mock<ILogger>();
var mock = new Mock<IPooledConnection>();
mock.Setup(x => x.IsHealthy).Returns(true);
mock.Setup(x => x.IsOpen).Returns(true);
var id = Guid.NewGuid();
var inUseConns = new Dictionary<Guid, IPooledConnection>();
inUseConns.Add(id, mock.Object);

var availableConns = new Queue<IPooledConnection>();
var mock1 = new Mock<IPooledConnection>();
mock1.Setup(x => x.IsHealthy).Returns(true);
mock1.Setup(x => x.IsOpen).Returns(true);

availableConns.Enqueue(mock1.Object);

Expand Down
127 changes: 127 additions & 0 deletions Neo4j.Driver/Neo4j.Driver.Tests/Connector/PooledConnectionTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
// Copyright (c) 2002-2016 "Neo Technology,"
// Network Engine for Objects in Lund AB [http://neotechnology.com]
//
// This file is part of Neo4j.
//
// 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.

using FluentAssertions;
using Moq;
using Neo4j.Driver.Internal.Connector;
using Neo4j.Driver.Internal.Messaging;
using Neo4j.Driver.V1;
using Xunit;

namespace Neo4j.Driver.Tests.Connector
{
public class PooledConnectionTests
{
private static ILogger Logger => new Mock<ILogger>().Object;

private static Mock<ISocketClient> MockSocketClient => new Mock<ISocketClient>();

public class Constructor
{
[Fact]
public void ShouldAddPooledConnectionErrorHandler()
{
var mockedSocketConn = new Mock<IConnection>();
var conn = new PooledConnection(mockedSocketConn.Object);

mockedSocketConn.Verify(x=>x.AddConnectionErrorHander(It.IsAny<PooledConnection.PooledConnectionErrorHandler>()), Times.Once);
}
}

public class HasUnrecoverableError
{
[Fact]
public void ShouldReportErrorIfIsTransientException()
{
var mock = MockSocketClient;
var mockResponseHandler = new Mock<IMessageResponseHandler>();
var con = new PooledConnection(new SocketConnection(mock.Object, AuthTokens.None, Logger, mockResponseHandler.Object));

mockResponseHandler.Setup(x => x.Error).Returns(new TransientException("BLAH", "lalala"));
con.HasUnrecoverableError.Should().BeFalse();
}

[Fact]
public void ShouldReportErrorIfIsDatabaseException()
{
var mock = MockSocketClient;
var mockResponseHandler = new Mock<IMessageResponseHandler>();
var con = new PooledConnection(new SocketConnection(mock.Object, AuthTokens.None, Logger, mockResponseHandler.Object));

mockResponseHandler.Setup(x => x.HasError).Returns(true);
mockResponseHandler.Setup(x => x.Error).Returns(new DatabaseException("BLAH", "lalala"));

var exception = Record.Exception(() => con.ReceiveOne());
exception.Should().BeOfType<DatabaseException>();
exception.Message.Should().Be("lalala");

con.HasUnrecoverableError.Should().BeTrue();
mockResponseHandler.VerifySet(x => x.Error = null, Times.Once);
}

[Fact]
public void ShouldNotReportErrorIfIsOtherExceptions()
{
var mock = MockSocketClient;
var mockResponseHandler = new Mock<IMessageResponseHandler>();
var con = new PooledConnection(new SocketConnection(mock.Object, AuthTokens.None, Logger, mockResponseHandler.Object));

mockResponseHandler.Setup(x => x.Error).Returns(new ClientException("BLAH", "lalala"));
con.HasUnrecoverableError.Should().BeFalse();
}
}

public class IsOpenMethod
{
[Fact]
public void ShouldBeFalseWhenConectionIsNotOpen()
{
var mockClient = new Mock<ISocketClient>();
mockClient.Setup(x => x.IsOpen).Returns(false);
var mockResponseHandler = new Mock<IMessageResponseHandler>();
mockResponseHandler.Setup(x => x.Error).Returns(new ClientException()); // has no unrecoverable error

var conn = new PooledConnection(new SocketConnection(mockClient.Object, AuthTokens.None, Logger, mockResponseHandler.Object));
conn.IsOpen.Should().BeFalse();
}

[Fact]
public void ShouldBeFalseWhenConnectionHasUnrecoverableError()
{
var mockClient = new Mock<ISocketClient>();
mockClient.Setup(x => x.IsOpen).Returns(false);
var mockResponseHandler = new Mock<IMessageResponseHandler>();
mockResponseHandler.Setup(x => x.Error).Returns(new DatabaseException()); // unrecoverable error

var conn = new PooledConnection(new SocketConnection(mockClient.Object, AuthTokens.None, Logger, mockResponseHandler.Object));
conn.IsOpen.Should().BeFalse();
}

[Fact]
public void ShouldReturnTrueWhenIsHealthy()
{
var mockClient = new Mock<ISocketClient>();
mockClient.Setup(x => x.IsOpen).Returns(true);
var mockResponseHandler = new Mock<IMessageResponseHandler>();
mockResponseHandler.Setup(x => x.Error).Returns(new ClientException()); // has no unrecoverable error

var conn = new PooledConnection(new SocketConnection(mockClient.Object, AuthTokens.None, Logger, mockResponseHandler.Object));
conn.IsOpen.Should().BeTrue();
}
}
}
}
Loading