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

Fix ChannelSetTest async issues #979

Merged
merged 1 commit into from
Mar 23, 2020
Merged

Fix ChannelSetTest async issues #979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
97 changes: 97 additions & 0 deletions .../testFixtures/java/io/servicetalk/concurrent/api/TestCollectingCompletableSubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
/*
* Copyright © 2020 Apple Inc. and the ServiceTalk project 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.
*/
package io.servicetalk.concurrent.api;

import io.servicetalk.concurrent.Cancellable;
import io.servicetalk.concurrent.CompletableSource.Subscriber;
import io.servicetalk.concurrent.PublisherSource;

import java.util.concurrent.TimeUnit;

/**
* A {@link Subscriber} that enqueues signals and provides blocking methods to consume them.
*/
public final class TestCollectingCompletableSubscriber implements Subscriber {
private final TestCollectingPublisherSubscriber<Void> publisherSubscriber =
new TestCollectingPublisherSubscriber<>();
@Override
public void onSubscribe(final Cancellable cancellable) {
publisherSubscriber.onSubscribe(new PublisherSource.Subscription() {
@Override
public void request(final long n) {
}

@Override
public void cancel() {
cancellable.cancel();
}
});
}

@Override
public void onComplete() {
publisherSubscriber.onComplete();
}

@Override
public void onError(final Throwable t) {
publisherSubscriber.onError(t);
}

/**
* Block until {@link #onSubscribe(Cancellable)}.
*
* @return The {@link PublisherSource.Subscription} from {@link #onSubscribe(Cancellable)}.
* @throws InterruptedException if an interrupt occurs while blocking for waiting for
* {@link #onSubscribe(Cancellable)}.
*/
public Cancellable awaitSubscription() throws InterruptedException {
return publisherSubscriber.awaitSubscription();
}

/**
* Block until a terminal signal is received, throws if {@link #onComplete()} and returns normally if
* {@link #onError(Throwable)}.
*
* @return the exception received by {@link #onError(Throwable)}.
* @throws InterruptedException If an interrupt occurs while blocking for the terminal event.
*/
public Throwable awaitOnError() throws InterruptedException {
return publisherSubscriber.awaitOnError();
}

/**
* Block until a terminal signal is received, throws if {@link #onError(Throwable)} and returns normally if
* {@link #onComplete()}.
*
* @throws InterruptedException If an interrupt occurs while blocking for the terminal event.
*/
public void awaitOnComplete() throws InterruptedException {
publisherSubscriber.awaitOnComplete();
}

/**
* Block for a terminal event.
*
* @param timeout The duration of time to wait.
* @param unit The unit of time to apply to the duration.
* @return {@code true} if a terminal event has been received before the timeout duration.
* @throws InterruptedException If an interrupt occurs while blocking for the terminal event.
*/
public boolean pollTerminal(long timeout, TimeUnit unit) throws InterruptedException {
return publisherSubscriber.pollTerminal(timeout, unit);
}
}
96 changes: 51 additions & 45 deletions ...-netty-internal/src/test/java/io/servicetalk/transport/netty/internal/ChannelSetTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import io.servicetalk.concurrent.CompletableSource.Processor;
import io.servicetalk.concurrent.api.AsyncCloseable;
import io.servicetalk.concurrent.api.Completable;
import io.servicetalk.concurrent.api.LegacyMockedCompletableListenerRule;
import io.servicetalk.concurrent.api.TestCollectingCompletableSubscriber;
import io.servicetalk.concurrent.internal.ServiceTalkTestTimeout;

import io.netty.channel.Channel;
Expand All @@ -41,13 +41,11 @@
import static io.servicetalk.concurrent.api.Executors.immediate;
import static io.servicetalk.concurrent.api.Processors.newCompletableProcessor;
import static io.servicetalk.concurrent.api.SourceAdapters.fromSource;
import static io.servicetalk.concurrent.api.SourceAdapters.toSource;
import static io.servicetalk.transport.netty.internal.ChannelSet.CHANNEL_CLOSEABLE_KEY;
import static java.lang.Boolean.FALSE;
import static java.lang.Boolean.parseBoolean;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.hamcrest.Matchers.is;
import static org.junit.Assume.assumeThat;
import static org.junit.Assert.assertFalse;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
Expand All @@ -63,12 +61,6 @@ public class ChannelSetTest {
public final ExpectedException thrown = ExpectedException.none();
@Rule
public final MockitoRule rule = MockitoJUnit.rule();
@Rule
public final LegacyMockedCompletableListenerRule subscriberRule1 = new LegacyMockedCompletableListenerRule();
@Rule
public final LegacyMockedCompletableListenerRule subscriberRule2 = new LegacyMockedCompletableListenerRule();
@Rule
public final LegacyMockedCompletableListenerRule subscriberRule3 = new LegacyMockedCompletableListenerRule();

@Mock
private Channel channel;
Expand Down Expand Up @@ -109,111 +101,123 @@ public void setupMocks() {
}

@Test
public void closeAsync() {
public void closeAsync() throws InterruptedException {
Completable completable = fixture.closeAsync();
verify(channel, never()).close();
subscriberRule1.listen(completable);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(completable).subscribe(subscriber);
verify(channel).close();
subscriberRule1.verifyCompletion();
subscriber.awaitOnComplete();
}

@Test
public void closeAsyncGracefullyWithNettyConnectionChannelHandler() throws Exception {
Completable completable = closeAsyncGracefully(fixture, 100, SECONDS);
verify(nettyConnection, never()).closeAsyncGracefully();
subscriberRule1.listen(completable);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(completable).subscribe(subscriber);
verify(nettyConnection).closeAsyncGracefully();
verify(channel, never()).close();
subscriberRule1.verifyNoEmissions();
assertFalse(subscriber.pollTerminal(10, MILLISECONDS));
closeAsyncGracefullyCompletable.onComplete();
subscriberRule1.verifyNoEmissions();
assertFalse(subscriber.pollTerminal(10, MILLISECONDS));
listener.operationComplete(channelCloseFuture);
subscriberRule1.verifyCompletion();
subscriber.awaitOnComplete();
}

@Test
public void closeAsyncGracefullyWithoutNettyConnectionChannelHandler() {
public void closeAsyncGracefullyWithoutNettyConnectionChannelHandler() throws InterruptedException {
when(mockClosableAttribute.getAndSet(any())).thenReturn(null);
Completable completable = closeAsyncGracefully(fixture, 100, SECONDS);
verify(channel, never()).close();
subscriberRule1.listen(completable);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(completable).subscribe(subscriber);
verify(channel).close();
subscriberRule1.verifyCompletion();
subscriber.awaitOnComplete();
}

@Test
public void testCloseAsyncGracefullyThenCloseAsync() throws Exception {
Completable gracefulCompletable = closeAsyncGracefully(fixture, 100, SECONDS);
Completable closeCompletable = fixture.closeAsync();

subscriberRule1.listen(gracefulCompletable);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(gracefulCompletable).subscribe(subscriber);
verify(nettyConnection).closeAsyncGracefully();

subscriberRule2.listen(closeCompletable);
TestCollectingCompletableSubscriber subscriber2 = new TestCollectingCompletableSubscriber();
toSource(closeCompletable).subscribe(subscriber2);
verify(channel).close();
// once closeCompletable being subscribed to closes the channel, the Completable returned from
// closeAsyncGracefully must complete.
closeAsyncGracefullyCompletable.onComplete();

fixture.onClose().toFuture().get();

subscriberRule1.verifyCompletion();
subscriberRule2.verifyCompletion();
subscriber.awaitOnComplete();
subscriber2.awaitOnComplete();
}

@Test
public void testCloseAsyncThenCloseAsyncGracefully() throws Exception {
Completable closeCompletable = fixture.closeAsync();
Completable gracefulCompletable = closeAsyncGracefully(fixture, 100, SECONDS);

subscriberRule2.listen(closeCompletable);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(closeCompletable).subscribe(subscriber);
verify(channel).close();

subscriberRule1.listen(gracefulCompletable);
TestCollectingCompletableSubscriber subscriber2 = new TestCollectingCompletableSubscriber();
toSource(gracefulCompletable).subscribe(subscriber2);
verify(nettyConnection, never()).closeAsyncGracefully();

fixture.onClose().toFuture().get();

subscriberRule1.verifyCompletion();
subscriberRule2.verifyCompletion();
subscriber.awaitOnComplete();
subscriber2.awaitOnComplete();
}

@Test
public void testCloseAsyncGracefullyTwice() throws Exception {
Completable gracefulCompletable1 = closeAsyncGracefully(fixture, 60, SECONDS);
Completable gracefulCompletable2 = closeAsyncGracefully(fixture, 60, SECONDS);

subscriberRule1.listen(gracefulCompletable1);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(gracefulCompletable1).subscribe(subscriber);
verify(nettyConnection).closeAsyncGracefully();

subscriberRule2.listen(gracefulCompletable2);
TestCollectingCompletableSubscriber subscriber2 = new TestCollectingCompletableSubscriber();
toSource(gracefulCompletable2).subscribe(subscriber2);
verify(nettyConnection, times(1)).closeAsyncGracefully();

subscriberRule1.verifyNoEmissions();
assertFalse(subscriber.pollTerminal(10, MILLISECONDS));
closeAsyncGracefullyCompletable.onComplete();
subscriberRule1.verifyNoEmissions();
assertFalse(subscriber.pollTerminal(10, MILLISECONDS));

listener.operationComplete(channelCloseFuture);

fixture.onClose().toFuture().get();

subscriberRule1.verifyCompletion();
subscriberRule2.verifyCompletion();
subscriber.awaitOnComplete();
subscriber2.awaitOnComplete();
}

@Test
public void testCloseAsyncGracefullyTwiceTimesOut() throws Exception {
Completable gracefulCompletable1 = closeAsyncGracefully(fixture, 100, MILLISECONDS);
Completable gracefulCompletable2 = closeAsyncGracefully(fixture, 1000, MILLISECONDS);

subscriberRule1.listen(gracefulCompletable1);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(gracefulCompletable1).subscribe(subscriber);
verify(nettyConnection).closeAsyncGracefully();

subscriberRule2.listen(gracefulCompletable2);
TestCollectingCompletableSubscriber subscriber2 = new TestCollectingCompletableSubscriber();
toSource(gracefulCompletable2).subscribe(subscriber2);
verify(nettyConnection, times(1)).closeAsyncGracefully();

gracefulCompletable1.toFuture().get();
subscriberRule2.verifyCompletion();
subscriber.awaitOnComplete();
subscriber2.awaitOnComplete();
verify(channel).close();
}

Expand All @@ -222,26 +226,28 @@ public void testCloseAsyncTwice() throws Exception {
Completable closeCompletable1 = fixture.closeAsync();
Completable closeCompletable2 = fixture.closeAsync();

subscriberRule1.listen(closeCompletable1);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(closeCompletable1).subscribe(subscriber);
verify(channel).close();

subscriberRule2.listen(closeCompletable2);
TestCollectingCompletableSubscriber subscriber2 = new TestCollectingCompletableSubscriber();
toSource(closeCompletable2).subscribe(subscriber2);
verify(channel, times(1)).close();

fixture.onClose().toFuture().get();

subscriberRule1.verifyCompletion();
subscriberRule2.verifyCompletion();
subscriber.awaitOnComplete();
subscriber2.awaitOnComplete();
}

@Test
public void closeAsyncGracefullyClosesAfterTimeout() throws Exception {
assumeThat("Ignored flaky test", parseBoolean(System.getenv("CI")), is(FALSE));
Completable completable = closeAsyncGracefully(fixture, 100, MILLISECONDS);
subscriberRule1.listen(completable);
TestCollectingCompletableSubscriber subscriber = new TestCollectingCompletableSubscriber();
toSource(completable).subscribe(subscriber);
verify(nettyConnection).closeAsyncGracefully();
fixture.onClose().toFuture().get();
verify(channel).close();
subscriberRule1.verifyCompletion();
subscriber.awaitOnComplete();
}
}