import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.stream.Collectors;
private void connectAndAssert(Supplier<Matcher<List<String>>> requestMatcher)
throws InterruptedException, ExecutionException, IOException {
connectNode();
+ readMessage(requestMatcher);
+ }
+
+ private void readMessage(Supplier<Matcher<List<String>>> requestMatcher) throws IOException {
lines = fcpServer.collectUntil(is("EndMessage"));
identifier = extractIdentifier(lines);
assertThat(lines, requestMatcher.get());
assertThat(edition.get(), is(23));
}
+ @Test
+ public void subscriptionCanBeCancelled() throws InterruptedException, ExecutionException, IOException {
+ Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
+ connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI, "EndMessage"));
+ replyWithSubscribed();
+ assertThat(uskSubscription.get().get().getUri(), is(URI));
+ AtomicBoolean updated = new AtomicBoolean();
+ uskSubscription.get().get().onUpdate(e -> updated.set(true));
+ uskSubscription.get().get().cancel();
+ readMessage(() -> matchesFcpMessage("UnsubscribeUSK", "Identifier=" + identifier, "EndMessage"));
+ sendUpdateNotification(23);
+ assertThat(updated.get(), is(false));
+ }
+
private void replyWithSubscribed() throws IOException {
fcpServer.writeLine(
"SubscribedUSK",