RoutingProcedureClusterCompositionProvider implements ClusterCompositionProvider { @Override public CompletionStage<ClusterComposition> getClusterComposition( Connection connection, DatabaseName databaseName, Bookmark bookmark ) { RoutingProcedureRunner runner; if ( supportsMultiDatabase( connection ) ) { runner = multiDatabaseRoutingProcedureRunner; } else { runner = routingProcedureRunner; } return runner.run( connection, databaseName, bookmark ) .thenApply( this::processRoutingResponse ); } RoutingProcedureClusterCompositionProvider( Clock clock, RoutingContext routingContext );   RoutingProcedureClusterCompositionProvider( Clock clock, RoutingProcedureRunner routingProcedureRunner,
            MultiDatabasesRoutingProcedureRunner multiDatabaseRoutingProcedureRunner ); @Override CompletionStage<ClusterComposition> getClusterComposition( Connection connection, DatabaseName databaseName, Bookmark bookmark );  }
@Test void shouldProtocolErrorWhenNoRecord() { RoutingProcedureRunner mockedRunner = newProcedureRunnerMock(); Connection connection = mock( Connection.class ); ClusterCompositionProvider provider = newClusterCompositionProvider( mockedRunner, connection ); RoutingProcedureResponse noRecordsResponse = newRoutingResponse(); when( mockedRunner.run( eq( connection ), any( DatabaseName.class ), any( InternalBookmark.class ) ) ) .thenReturn( completedFuture( noRecordsResponse ) ); ProtocolException error = assertThrows( ProtocolException.class, () -> await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ) ); assertThat( error.getMessage(), containsString( "records received '0' is too few or too many." ) ); }
@Test void shouldProtocolErrorWhenMoreThanOneRecord() { RoutingProcedureRunner mockedRunner = newProcedureRunnerMock(); Connection connection = mock( Connection.class ); ClusterCompositionProvider provider = newClusterCompositionProvider( mockedRunner, connection ); Record aRecord = new InternalRecord( asList( "key1", "key2" ), new Value[]{ new StringValue( "a value" ) } ); RoutingProcedureResponse routingResponse = newRoutingResponse( aRecord, aRecord ); when( mockedRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ).thenReturn( completedFuture( routingResponse ) ); ProtocolException error = assertThrows( ProtocolException.class, () -> await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ) ); assertThat( error.getMessage(), containsString( "records received '2' is too few or too many." ) ); }
@Test void shouldProtocolErrorWhenUnparsableRecord() { RoutingProcedureRunner mockedRunner = newProcedureRunnerMock(); Connection connection = mock( Connection.class ); ClusterCompositionProvider provider = newClusterCompositionProvider( mockedRunner, connection ); Record aRecord = new InternalRecord( asList( "key1", "key2" ), new Value[]{ new StringValue( "a value" ) } ); RoutingProcedureResponse routingResponse = newRoutingResponse( aRecord ); when( mockedRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ).thenReturn( completedFuture( routingResponse ) ); ProtocolException error = assertThrows( ProtocolException.class, () -> await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ) ); assertThat( error.getMessage(), containsString( "unparsable record received." ) ); }
@Test void shouldProtocolErrorWhenNoRouters() { MultiDatabasesRoutingProcedureRunner mockedRunner = newMultiDBProcedureRunnerMock(); Connection connection = mock( Connection.class ); Clock mockedClock = mock( Clock.class ); ClusterCompositionProvider provider = newClusterCompositionProvider( mockedRunner, connection, mockedClock ); Record record = new InternalRecord( asList( "ttl", "servers" ), new Value[]{ value( 100 ), value( asList( serverInfo( "READ", "one:1337", "two:1337" ), serverInfo( "WRITE", "one:1337" ) ) ) } ); RoutingProcedureResponse routingResponse = newRoutingResponse( record ); when( mockedRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ).thenReturn( completedFuture( routingResponse ) ); when( mockedClock.millis() ).thenReturn( 12345L ); ProtocolException error = assertThrows( ProtocolException.class, () -> await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ) ); assertThat( error.getMessage(), containsString( "no router or reader found in response." ) ); }
@Test void shouldProtocolErrorWhenNoReaders() { MultiDatabasesRoutingProcedureRunner mockedRunner = newMultiDBProcedureRunnerMock(); Connection connection = mock( Connection.class ); Clock mockedClock = mock( Clock.class ); ClusterCompositionProvider provider = newClusterCompositionProvider( mockedRunner, connection, mockedClock ); Record record = new InternalRecord( asList( "ttl", "servers" ), new Value[]{ value( 100 ), value( asList( serverInfo( "WRITE", "one:1337" ), serverInfo( "ROUTE", "one:1337", "two:1337" ) ) ) } ); RoutingProcedureResponse routingResponse = newRoutingResponse( record ); when( mockedRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ).thenReturn( completedFuture( routingResponse ) ); when( mockedClock.millis() ).thenReturn( 12345L ); ProtocolException error = assertThrows( ProtocolException.class, () -> await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ) ); assertThat( error.getMessage(), containsString( "no router or reader found in response." ) ); }
@Test void shouldPropagateConnectionFailureExceptions() { RoutingProcedureRunner mockedRunner = newProcedureRunnerMock(); Connection connection = mock( Connection.class ); ClusterCompositionProvider provider = newClusterCompositionProvider( mockedRunner, connection ); when( mockedRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ).thenReturn( failedFuture( new ServiceUnavailableException( "Connection breaks during cypher execution" ) ) ); ServiceUnavailableException e = assertThrows( ServiceUnavailableException.class, () -> await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ) ); assertThat( e.getMessage(), containsString( "Connection breaks during cypher execution" ) ); }
@Test void shouldReturnSuccessResultWhenNoError() { Clock mockedClock = mock( Clock.class ); Connection connection = mock( Connection.class ); MultiDatabasesRoutingProcedureRunner mockedRunner = newMultiDBProcedureRunnerMock(); ClusterCompositionProvider provider = newClusterCompositionProvider( mockedRunner, connection, mockedClock ); Record record = new InternalRecord( asList( "ttl", "servers" ), new Value[]{ value( 100 ), value( asList( serverInfo( "READ", "one:1337", "two:1337" ), serverInfo( "WRITE", "one:1337" ), serverInfo( "ROUTE", "one:1337", "two:1337" ) ) ) } ); RoutingProcedureResponse routingResponse = newRoutingResponse( record ); when( mockedRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ) .thenReturn( completedFuture( routingResponse ) ); when( mockedClock.millis() ).thenReturn( 12345L ); ClusterComposition cluster = await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ); assertEquals( 12345 + 100_000, cluster.expirationTimestamp() ); assertEquals( serverSet( "one:1337", "two:1337" ), cluster.readers() ); assertEquals( serverSet( "one:1337" ), cluster.writers() ); assertEquals( serverSet( "one:1337", "two:1337" ), cluster.routers() ); }
@Test void shouldReturnFailureWhenProcedureRunnerFails() { RoutingProcedureRunner procedureRunner = newProcedureRunnerMock(); Connection connection = mock( Connection.class ); RuntimeException error = new RuntimeException( "hi" ); when( procedureRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ) .thenReturn( completedFuture( newRoutingResponse( error ) ) ); RoutingProcedureClusterCompositionProvider provider = newClusterCompositionProvider( procedureRunner, connection ); RuntimeException e = assertThrows( RuntimeException.class, () -> await( provider.getClusterComposition( connection, defaultDatabase(), empty() ) ) ); assertEquals( error, e ); }
@Test void shouldUseMultiDBProcedureRunnerWhenConnectingWith40Server() throws Throwable { MultiDatabasesRoutingProcedureRunner procedureRunner = newMultiDBProcedureRunnerMock(); Connection connection = mock( Connection.class ); RoutingProcedureClusterCompositionProvider provider = newClusterCompositionProvider( procedureRunner, connection ); when( procedureRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ).thenReturn( completedWithNull() ); provider.getClusterComposition( connection, defaultDatabase(), empty() ); verify( procedureRunner ).run( eq( connection ), any( DatabaseName.class ), any( InternalBookmark.class ) ); }
@Test void shouldUseProcedureRunnerWhenConnectingWith35AndPreviousServers() throws Throwable { RoutingProcedureRunner procedureRunner = newProcedureRunnerMock(); Connection connection = mock( Connection.class ); RoutingProcedureClusterCompositionProvider provider = newClusterCompositionProvider( procedureRunner, connection ); when( procedureRunner.run( eq( connection ), any( DatabaseName.class ), any(InternalBookmark.class ) ) ).thenReturn( completedWithNull() ); provider.getClusterComposition( connection, defaultDatabase(), empty() ); verify( procedureRunner ).run( eq( connection ), any( DatabaseName.class ), any( InternalBookmark.class ) ); }