Skip to content

Commit

Permalink
Address code review comments.
Browse files Browse the repository at this point in the history
  • Loading branch information
mikkokar committed Nov 14, 2018
1 parent f7f53aa commit fad7bd1
Show file tree
Hide file tree
Showing 2 changed files with 7 additions and 9 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,8 @@ Publisher<Connection> borrowConnection2() {
}
}
}
}).timeout(Duration.ofMillis(poolSettings.pendingConnectionTimeoutMillis()),
}).timeout(
Duration.ofMillis(poolSettings.pendingConnectionTimeoutMillis()),
Mono.error(new MaxPendingConnectionTimeoutException(origin, connectionSettings.connectTimeoutMillis())));
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ public void borrowsReturnedConnection() {


@Test
public void emitsConnectionWhenOnceEstablished() throws ExecutionException, InterruptedException {
public void emitsConnectionWhenOnceEstablished() {
PublishSubject<Connection> subject = PublishSubject.create();
when(connectionFactory.createConnection(any(Origin.class), any(ConnectionSettings.class)))
.thenReturn(subject);
Expand Down Expand Up @@ -293,7 +293,6 @@ public void unsubscribingRemovesTheWaitingSubscriber() throws ExecutionException
}


// 2. The connection pool limits
@Test
public void limitsPendingConnectionsDueToConnectionEstablishment() {
when(connectionFactory.createConnection(any(Origin.class), any(ConnectionSettings.class)))
Expand All @@ -307,8 +306,8 @@ public void limitsPendingConnectionsDueToConnectionEstablishment() {

SimpleConnectionPool pool = new SimpleConnectionPool(origin, poolSettings, connectionFactory);

CompletableFuture<Connection> pending1 = Mono.from(pool.borrowConnection2()).toFuture();
CompletableFuture<Connection> pending2 = Mono.from(pool.borrowConnection2()).toFuture();
Mono.from(pool.borrowConnection2()).toFuture();
Mono.from(pool.borrowConnection2()).toFuture();

StepVerifier.create(pool.borrowConnection2())
.expectError(MaxPendingConnectionsExceededException.class)
Expand All @@ -319,7 +318,6 @@ public void limitsPendingConnectionsDueToConnectionEstablishment() {

}

// 2. The connection pool limits
@Test
public void limitsPendingConnectionsDueToPoolSaturation() {
when(connectionFactory.createConnection(any(Origin.class), any(ConnectionSettings.class)))
Expand Down Expand Up @@ -362,7 +360,6 @@ public void limitsPendingConnectionsDueToPoolSaturation() {
}


// 2. The connection pool limits
@Test
public void returnConnectionDecrementsConnectionCount() throws ExecutionException, InterruptedException, TimeoutException {
when(connectionFactory.createConnection(any(Origin.class), any(ConnectionSettings.class)))
Expand Down Expand Up @@ -627,7 +624,7 @@ public void borrowRetriesThreeTimesOnConnectionEstablishmentFailure() {
}

@Test
public void connectionEstablishmentFailureRetryThreeTimesAtConnectionClosure() {
public void borrowRetriesThreeTimesOnFailureDueToConnectionClosure() {
when(connectionFactory.createConnection(any(Origin.class), any(ConnectionSettings.class)))
.thenReturn(Observable.just(connection1))
.thenReturn(Observable.error(new OriginUnreachableException(origin, new RuntimeException())))
Expand Down Expand Up @@ -720,7 +717,7 @@ public void connectionEstablishmentFailureRetryThreeTimesOnlyAtConnectionClosure
}

@Test
public void pendingConnectionTimeout() {
public void emitsExceptionWhenPendingConnectionTimesOut() {
PublishSubject<Connection> subject = PublishSubject.create();
when(connectionFactory.createConnection(any(Origin.class), any(ConnectionSettings.class)))
.thenReturn(subject);
Expand Down

0 comments on commit fad7bd1

Please sign in to comment.