Skip to content

adding callback on shutdown #82

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

Merged
merged 1 commit into from
May 5, 2016
Merged

adding callback on shutdown #82

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
30 changes: 23 additions & 7 deletions src/main/java/io/reactivesocket/DefaultReactiveSocket.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.reactivestreams.Subscription;

import java.io.IOException;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

Expand All @@ -55,6 +56,7 @@ public class DefaultReactiveSocket implements ReactiveSocket {
private final RequestHandler clientRequestHandler;
private final ConnectionSetupHandler responderConnectionHandler;
private final LeaseGovernor leaseGovernor;
private final CopyOnWriteArrayList<Completable> shutdownListeners;

private DefaultReactiveSocket(
DuplexConnection connection,
Expand All @@ -72,6 +74,7 @@ private DefaultReactiveSocket(
this.responderConnectionHandler = responderConnectionHandler;
this.leaseGovernor = leaseGovernor;
this.errorStream = errorStream;
this.shutdownListeners = new CopyOnWriteArrayList<>();
}

/**
Expand Down Expand Up @@ -439,15 +442,28 @@ public void addOutput(Frame f, Completable callback) {

};

@Override
public void onShutdown(Completable c) {
shutdownListeners.add(c);
}

@Override
public void close() throws Exception {
connection.close();
leaseGovernor.unregister(responder);
if (requester != null) {
requester.shutdown();
}
if (responder != null) {
responder.shutdown();
try {
connection.close();
leaseGovernor.unregister(responder);
if (requester != null) {
requester.shutdown();
}
if (responder != null) {
responder.shutdown();
}

shutdownListeners.forEach(Completable::success);

} catch (Throwable t) {
shutdownListeners.forEach(c -> c.error(t));
throw t;
}
}

Expand Down
2 changes: 2 additions & 0 deletions src/main/java/io/reactivesocket/DuplexConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
*/
package io.reactivesocket;

import io.reactivesocket.internal.rx.EmptySubscription;
import io.reactivesocket.rx.Completable;
import io.reactivesocket.rx.Observable;
import org.reactivestreams.Publisher;
Expand All @@ -32,6 +33,7 @@ public interface DuplexConnection extends Closeable {

default void addOutput(Frame frame, Completable callback) {
addOutput(s -> {
s.onSubscribe(EmptySubscription.INSTANCE);
s.onNext(frame);
s.onComplete();
}, callback);
Expand Down
5 changes: 5 additions & 0 deletions src/main/java/io/reactivesocket/ReactiveSocket.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,11 @@ public void error(Throwable e) {
*/
void onRequestReady(Completable c);

/**
* Registers a completable to be run when an ReactiveSocket is closed
*/
void onShutdown(Completable c);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of returning void, I would return the previous Completable (and allow changing the callback).

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, why to ask for a Completable? This is like a close future for the socket, which is a single instance that anyone can subscriber/listen to. So, why not have something like:

Completable shutdownListener();

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it's a good idea.


/**
* Server granting new lease information to client
*
Expand Down
71 changes: 70 additions & 1 deletion src/test/java/io/reactivesocket/ReactiveSocketTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
package io.reactivesocket;

import io.reactivesocket.lease.FairLeaseGovernor;
import io.reactivesocket.rx.Completable;
import io.reactivex.disposables.Disposable;
import io.reactivex.observables.ConnectableObservable;
import io.reactivex.subscribers.TestSubscriber;
Expand Down Expand Up @@ -256,6 +257,74 @@ private void awaitSocketAvailability(ReactiveSocket socket, long timeout, TimeUn
assertTrue("client socket has positive avaibility", socket.availability() > 0.0);
}

@Test(timeout = 2000)
public void testShutdownListener() throws Exception {
socketClient = DefaultReactiveSocket.fromClientConnection(
clientConnection,
ConnectionSetupPayload.create("UTF-8", "UTF-8", NO_FLAGS),
err -> err.printStackTrace()
);

CountDownLatch latch = new CountDownLatch(1);

socketClient.onShutdown(new Completable() {
@Override
public void success() {
latch.countDown();
}

@Override
public void error(Throwable e) {

}
});

socketClient.close();

latch.await();
}

@Test(timeout = 2000)
public void testMultipleShutdownListeners() throws Exception {
socketClient = DefaultReactiveSocket.fromClientConnection(
clientConnection,
ConnectionSetupPayload.create("UTF-8", "UTF-8", NO_FLAGS),
err -> err.printStackTrace()
);

CountDownLatch latch = new CountDownLatch(2);

socketClient
.onShutdown(new Completable() {
@Override
public void success() {
latch.countDown();
}

@Override
public void error(Throwable e) {

}
});

socketClient
.onShutdown(new Completable() {
@Override
public void success() {
latch.countDown();
}

@Override
public void error(Throwable e) {

}
});

socketClient.close();

latch.await();
}

@Test(timeout=2000)
@Theory
public void testRequestResponse(int setupFlag) throws InterruptedException {
Expand All @@ -269,7 +338,7 @@ public void testRequestResponse(int setupFlag) throws InterruptedException {
ts.assertNoErrors();
ts.assertValue(TestUtil.utf8EncodedPayload("hello world", null));
}

@Test(timeout=2000, expected=IllegalStateException.class)
public void testRequestResponsePremature() throws InterruptedException {
socketClient = DefaultReactiveSocket.fromClientConnection(
Expand Down