1 package net.pterodactylus.fcp.quelaton;
3 import static org.hamcrest.MatcherAssert.assertThat;
4 import static org.hamcrest.Matchers.allOf;
5 import static org.hamcrest.Matchers.contains;
6 import static org.hamcrest.Matchers.containsInAnyOrder;
7 import static org.hamcrest.Matchers.hasItem;
8 import static org.hamcrest.Matchers.hasSize;
9 import static org.hamcrest.Matchers.is;
10 import static org.hamcrest.Matchers.not;
11 import static org.hamcrest.Matchers.notNullValue;
12 import static org.hamcrest.Matchers.startsWith;
14 import java.io.ByteArrayInputStream;
16 import java.io.IOException;
18 import java.nio.charset.StandardCharsets;
19 import java.util.Arrays;
20 import java.util.Collection;
21 import java.util.List;
22 import java.util.Optional;
23 import java.util.concurrent.CopyOnWriteArrayList;
24 import java.util.concurrent.CountDownLatch;
25 import java.util.concurrent.ExecutionException;
26 import java.util.concurrent.ExecutorService;
27 import java.util.concurrent.Executors;
28 import java.util.concurrent.Future;
29 import java.util.concurrent.TimeUnit;
30 import java.util.concurrent.atomic.AtomicBoolean;
31 import java.util.concurrent.atomic.AtomicInteger;
32 import java.util.function.Supplier;
33 import java.util.stream.Collectors;
35 import net.pterodactylus.fcp.ARK;
36 import net.pterodactylus.fcp.ConfigData;
37 import net.pterodactylus.fcp.DSAGroup;
38 import net.pterodactylus.fcp.FcpKeyPair;
39 import net.pterodactylus.fcp.Key;
40 import net.pterodactylus.fcp.NodeData;
41 import net.pterodactylus.fcp.NodeRef;
42 import net.pterodactylus.fcp.Peer;
43 import net.pterodactylus.fcp.PeerNote;
44 import net.pterodactylus.fcp.PluginInfo;
45 import net.pterodactylus.fcp.Priority;
46 import net.pterodactylus.fcp.fake.FakeTcpServer;
47 import net.pterodactylus.fcp.quelaton.ClientGetCommand.Data;
49 import com.google.common.io.ByteStreams;
50 import com.google.common.io.Files;
51 import com.nitorcreations.junit.runners.NestedRunner;
52 import org.hamcrest.Description;
53 import org.hamcrest.Matcher;
54 import org.hamcrest.Matchers;
55 import org.hamcrest.TypeSafeDiagnosingMatcher;
56 import org.junit.After;
57 import org.junit.Assert;
58 import org.junit.Test;
59 import org.junit.runner.RunWith;
62 * Unit test for {@link DefaultFcpClient}.
64 * @author <a href="bombe@freenetproject.org">David ‘Bombe’ Roden</a>
66 @RunWith(NestedRunner.class)
67 public class DefaultFcpClientTest {
69 private static final String INSERT_URI =
70 "SSK@RVCHbJdkkyTCeNN9AYukEg76eyqmiosSaNKgE3U9zUw,7SHH53gletBVb9JD7nBsyClbLQsBubDPEIcwg908r7Y,AQECAAE/";
71 private static final String REQUEST_URI =
72 "SSK@wtbgd2loNcJCXvtQVOftl2tuWBomDQHfqS6ytpPRhfw,7SHH53gletBVb9JD7nBsyClbLQsBubDPEIcwg908r7Y,AQACAAE/";
74 private int threadCounter = 0;
75 private final ExecutorService threadPool =
76 Executors.newCachedThreadPool(r -> new Thread(r, "Test-Thread-" + threadCounter++));
77 private final FakeTcpServer fcpServer;
78 private final DefaultFcpClient fcpClient;
80 public DefaultFcpClientTest() throws IOException {
81 fcpServer = new FakeTcpServer(threadPool);
82 fcpClient = new DefaultFcpClient(threadPool, "localhost", fcpServer.getPort(), () -> "Test");
86 public void tearDown() throws IOException {
88 threadPool.shutdown();
91 private void connectNode() throws InterruptedException, ExecutionException, IOException {
92 fcpServer.connect().get();
93 fcpServer.collectUntil(is("EndMessage"));
94 fcpServer.writeLine("NodeHello",
95 "CompressionCodecs=4 - GZIP(0), BZIP2(1), LZMA(2), LZMA_NEW(3)",
96 "Revision=build01466",
98 "Version=Fred,0.7,1.0,1466",
100 "ConnectionIdentifier=14318898267048452a81b36e7f13a3f0",
104 "NodeLanguage=ENGLISH",
110 private String extractIdentifier(List<String> lines) {
111 return lines.stream()
112 .filter(s -> s.startsWith("Identifier="))
113 .map(s -> s.substring(s.indexOf('=') + 1))
118 private Matcher<List<String>> matchesFcpMessage(String name, String... requiredLines) {
119 return matchesFcpMessageWithTerminator(name, "EndMessage", requiredLines);
122 private Matcher<Iterable<String>> hasHead(String firstElement) {
123 return new TypeSafeDiagnosingMatcher<Iterable<String>>() {
125 protected boolean matchesSafely(Iterable<String> iterable, Description mismatchDescription) {
126 if (!iterable.iterator().hasNext()) {
127 mismatchDescription.appendText("is empty");
130 String element = iterable.iterator().next();
131 if (!element.equals(firstElement)) {
132 mismatchDescription.appendText("starts with ").appendValue(element);
139 public void describeTo(Description description) {
140 description.appendText("starts with ").appendValue(firstElement);
145 private Matcher<List<String>> matchesFcpMessageWithTerminator(
146 String name, String terminator, String... requiredLines) {
147 return allOf(hasHead(name), hasParameters(1, 1, requiredLines), hasTail(terminator));
150 private Matcher<List<String>> hasParameters(int ignoreStart, int ignoreEnd, String... lines) {
151 return new TypeSafeDiagnosingMatcher<List<String>>() {
153 protected boolean matchesSafely(List<String> item, Description mismatchDescription) {
154 if (item.size() < (ignoreStart + ignoreEnd)) {
155 mismatchDescription.appendText("has only ").appendValue(item.size()).appendText(" elements");
158 for (String line : lines) {
159 if ((item.indexOf(line) < ignoreStart) || (item.indexOf(line) >= (item.size() - ignoreEnd))) {
160 mismatchDescription.appendText("does not contains ").appendValue(line);
168 public void describeTo(Description description) {
169 description.appendText("contains ").appendValueList("(", ", ", ")", lines);
170 description.appendText(", ignoring the first ").appendValue(ignoreStart);
171 description.appendText(" and the last ").appendValue(ignoreEnd);
176 private Matcher<List<String>> hasTail(String... lastElements) {
177 return new TypeSafeDiagnosingMatcher<List<String>>() {
179 protected boolean matchesSafely(List<String> list, Description mismatchDescription) {
180 if (list.size() < lastElements.length) {
181 mismatchDescription.appendText("is too small");
184 List<String> tail = list.subList(list.size() - lastElements.length, list.size());
185 if (!tail.equals(Arrays.asList(lastElements))) {
186 mismatchDescription.appendText("ends with ").appendValueList("(", ", ", ")", tail);
193 public void describeTo(Description description) {
194 description.appendText("ends with ").appendValueList("(", ", ", ")", lastElements);
200 public void defaultFcpClientCanGetNodeInformation() throws InterruptedException, ExecutionException, IOException {
201 Future<NodeData> nodeData = fcpClient.getNode().execute();
203 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
204 String identifier = extractIdentifier(lines);
205 assertThat(lines, matchesFcpMessage(
207 "Identifier=" + identifier,
208 "GiveOpennetRef=false",
214 "Identifier=" + identifier,
215 "ark.pubURI=SSK@3YEf.../ark",
218 "version=Fred,0.7,1.0,1466",
219 "lastGoodVersion=Fred,0.7,1.0,1466",
222 assertThat(nodeData.get(), notNullValue());
226 public void defaultFcpClientCanGetNodeInformationWithOpennetRef()
227 throws InterruptedException, ExecutionException, IOException {
228 Future<NodeData> nodeData = fcpClient.getNode().opennetRef().execute();
230 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
231 String identifier = extractIdentifier(lines);
232 assertThat(lines, matchesFcpMessage(
234 "Identifier=" + identifier,
235 "GiveOpennetRef=true",
241 "Identifier=" + identifier,
243 "ark.pubURI=SSK@3YEf.../ark",
246 "version=Fred,0.7,1.0,1466",
247 "lastGoodVersion=Fred,0.7,1.0,1466",
250 assertThat(nodeData.get().getVersion().toString(), is("Fred,0.7,1.0,1466"));
254 public void defaultFcpClientCanGetNodeInformationWithPrivateData()
255 throws InterruptedException, ExecutionException, IOException {
256 Future<NodeData> nodeData = fcpClient.getNode().includePrivate().execute();
258 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
259 String identifier = extractIdentifier(lines);
260 assertThat(lines, matchesFcpMessage(
262 "Identifier=" + identifier,
263 "GiveOpennetRef=false",
269 "Identifier=" + identifier,
271 "ark.pubURI=SSK@3YEf.../ark",
274 "version=Fred,0.7,1.0,1466",
275 "lastGoodVersion=Fred,0.7,1.0,1466",
276 "ark.privURI=SSK@XdHMiRl",
279 assertThat(nodeData.get().getARK().getPrivateURI(), is("SSK@XdHMiRl"));
283 public void defaultFcpClientCanGetNodeInformationWithVolatileData()
284 throws InterruptedException, ExecutionException, IOException {
285 Future<NodeData> nodeData = fcpClient.getNode().includeVolatile().execute();
287 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
288 String identifier = extractIdentifier(lines);
289 assertThat(lines, matchesFcpMessage(
291 "Identifier=" + identifier,
292 "GiveOpennetRef=false",
298 "Identifier=" + identifier,
300 "ark.pubURI=SSK@3YEf.../ark",
303 "version=Fred,0.7,1.0,1466",
304 "lastGoodVersion=Fred,0.7,1.0,1466",
305 "volatile.freeJavaMemory=205706528",
308 assertThat(nodeData.get().getVolatile("freeJavaMemory"), is("205706528"));
312 public void defaultFcpClientCanGetConfigWithoutDetails()
313 throws InterruptedException, ExecutionException, IOException {
314 Future<ConfigData> configData = fcpClient.getConfig().execute();
316 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
317 String identifier = extractIdentifier(lines);
318 assertThat(lines, matchesFcpMessage(
320 "Identifier=" + identifier
324 "Identifier=" + identifier,
327 assertThat(configData.get(), notNullValue());
331 public void defaultFcpClientCanGetConfigWithCurrent()
332 throws InterruptedException, ExecutionException, IOException {
333 Future<ConfigData> configData = fcpClient.getConfig().withCurrent().execute();
335 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
336 String identifier = extractIdentifier(lines);
337 assertThat(lines, matchesFcpMessage(
339 "Identifier=" + identifier,
344 "Identifier=" + identifier,
348 assertThat(configData.get().getCurrent("foo"), is("bar"));
352 public void defaultFcpClientCanGetConfigWithDefaults()
353 throws InterruptedException, ExecutionException, IOException {
354 Future<ConfigData> configData = fcpClient.getConfig().withDefaults().execute();
356 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
357 String identifier = extractIdentifier(lines);
358 assertThat(lines, matchesFcpMessage(
360 "Identifier=" + identifier,
365 "Identifier=" + identifier,
369 assertThat(configData.get().getDefault("foo"), is("bar"));
373 public void defaultFcpClientCanGetConfigWithSortOrder()
374 throws InterruptedException, ExecutionException, IOException {
375 Future<ConfigData> configData = fcpClient.getConfig().withSortOrder().execute();
377 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
378 String identifier = extractIdentifier(lines);
379 assertThat(lines, matchesFcpMessage(
381 "Identifier=" + identifier,
386 "Identifier=" + identifier,
390 assertThat(configData.get().getSortOrder("foo"), is(17));
394 public void defaultFcpClientCanGetConfigWithExpertFlag()
395 throws InterruptedException, ExecutionException, IOException {
396 Future<ConfigData> configData = fcpClient.getConfig().withExpertFlag().execute();
398 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
399 String identifier = extractIdentifier(lines);
400 assertThat(lines, matchesFcpMessage(
402 "Identifier=" + identifier,
403 "WithExpertFlag=true"
407 "Identifier=" + identifier,
408 "expertFlag.foo=true",
411 assertThat(configData.get().getExpertFlag("foo"), is(true));
415 public void defaultFcpClientCanGetConfigWithForceWriteFlag()
416 throws InterruptedException, ExecutionException, IOException {
417 Future<ConfigData> configData = fcpClient.getConfig().withForceWriteFlag().execute();
419 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
420 String identifier = extractIdentifier(lines);
421 assertThat(lines, matchesFcpMessage(
423 "Identifier=" + identifier,
424 "WithForceWriteFlag=true"
428 "Identifier=" + identifier,
429 "forceWriteFlag.foo=true",
432 assertThat(configData.get().getForceWriteFlag("foo"), is(true));
436 public void defaultFcpClientCanGetConfigWithShortDescription()
437 throws InterruptedException, ExecutionException, IOException {
438 Future<ConfigData> configData = fcpClient.getConfig().withShortDescription().execute();
440 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
441 String identifier = extractIdentifier(lines);
442 assertThat(lines, matchesFcpMessage(
444 "Identifier=" + identifier,
445 "WithShortDescription=true"
449 "Identifier=" + identifier,
450 "shortDescription.foo=bar",
453 assertThat(configData.get().getShortDescription("foo"), is("bar"));
457 public void defaultFcpClientCanGetConfigWithLongDescription()
458 throws InterruptedException, ExecutionException, IOException {
459 Future<ConfigData> configData = fcpClient.getConfig().withLongDescription().execute();
461 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
462 String identifier = extractIdentifier(lines);
463 assertThat(lines, matchesFcpMessage(
465 "Identifier=" + identifier,
466 "WithLongDescription=true"
470 "Identifier=" + identifier,
471 "longDescription.foo=bar",
474 assertThat(configData.get().getLongDescription("foo"), is("bar"));
478 public void defaultFcpClientCanGetConfigWithDataTypes()
479 throws InterruptedException, ExecutionException, IOException {
480 Future<ConfigData> configData = fcpClient.getConfig().withDataTypes().execute();
482 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
483 String identifier = extractIdentifier(lines);
484 assertThat(lines, matchesFcpMessage(
486 "Identifier=" + identifier,
491 "Identifier=" + identifier,
492 "dataType.foo=number",
495 assertThat(configData.get().getDataType("foo"), is("number"));
499 public void defaultFcpClientCanModifyConfigData() throws InterruptedException, ExecutionException, IOException {
500 Future<ConfigData> newConfigData = fcpClient.modifyConfig().set("foo.bar").to("baz").execute();
502 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
503 String identifier = extractIdentifier(lines);
504 assertThat(lines, matchesFcpMessage(
506 "Identifier=" + identifier,
511 "Identifier=" + identifier,
512 "current.foo.bar=baz",
515 assertThat(newConfigData.get().getCurrent("foo.bar"), is("baz"));
518 private List<String> lines;
519 private String identifier;
521 private void connectAndAssert(Supplier<Matcher<List<String>>> requestMatcher)
522 throws InterruptedException, ExecutionException, IOException {
524 readMessage(requestMatcher);
527 private void readMessage(Supplier<Matcher<List<String>>> requestMatcher) throws IOException {
528 readMessage("EndMessage", requestMatcher);
531 private void readMessage(String terminator, Supplier<Matcher<List<String>>> requestMatcher) throws IOException {
532 lines = fcpServer.collectUntil(is(terminator));
533 identifier = extractIdentifier(lines);
534 assertThat(lines, requestMatcher.get());
537 public class ConnectionsAndKeyPairs {
539 public class Connections {
541 @Test(expected = ExecutionException.class)
542 public void throwsExceptionOnFailure() throws IOException, ExecutionException, InterruptedException {
543 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
544 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
546 "CloseConnectionDuplicateClientName",
552 @Test(expected = ExecutionException.class)
553 public void throwsExceptionIfConnectionIsClosed() throws IOException, ExecutionException, InterruptedException {
554 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
555 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
561 public void connectionIsReused() throws InterruptedException, ExecutionException, IOException {
562 Future<FcpKeyPair> keyPair = fcpClient.generateKeypair().execute();
563 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
566 keyPair = fcpClient.generateKeypair().execute();
567 readMessage(() -> matchesFcpMessage("GenerateSSK"));
568 identifier = extractIdentifier(lines);
574 public void defaultFcpClientCanReconnectAfterConnectionHasBeenClosed()
575 throws InterruptedException, ExecutionException, IOException {
576 Future<FcpKeyPair> keyPair = fcpClient.generateKeypair().execute();
577 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
582 } catch (ExecutionException e) {
584 keyPair = fcpClient.generateKeypair().execute();
585 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
592 public class GenerateKeyPair {
595 public void defaultFcpClientCanGenerateKeypair()
596 throws ExecutionException, InterruptedException, IOException {
597 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
598 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
600 FcpKeyPair keyPair = keyPairFuture.get();
601 assertThat(keyPair.getPublicKey(), is(REQUEST_URI));
602 assertThat(keyPair.getPrivateKey(), is(INSERT_URI));
607 private void replyWithKeyPair() throws IOException {
608 fcpServer.writeLine("SSKKeypair",
609 "InsertURI=" + INSERT_URI + "",
610 "RequestURI=" + REQUEST_URI + "",
611 "Identifier=" + identifier,
619 public class PeerCommands {
621 public class ListPeer {
624 public void byIdentity() throws InterruptedException, ExecutionException, IOException {
625 Future<Optional<Peer>> peer = fcpClient.listPeer().byIdentity("id1").execute();
626 connectAndAssert(() -> matchesListPeer("id1"));
627 replyWithPeer("id1");
628 assertThat(peer.get().get().getIdentity(), is("id1"));
632 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
633 Future<Optional<Peer>> peer = fcpClient.listPeer().byHostAndPort("host.free.net", 12345).execute();
634 connectAndAssert(() -> matchesListPeer("host.free.net:12345"));
635 replyWithPeer("id1");
636 assertThat(peer.get().get().getIdentity(), is("id1"));
640 public void byName() throws InterruptedException, ExecutionException, IOException {
641 Future<Optional<Peer>> peer = fcpClient.listPeer().byName("FriendNode").execute();
642 connectAndAssert(() -> matchesListPeer("FriendNode"));
643 replyWithPeer("id1");
644 assertThat(peer.get().get().getIdentity(), is("id1"));
648 public void unknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
649 Future<Optional<Peer>> peer = fcpClient.listPeer().byIdentity("id2").execute();
650 connectAndAssert(() -> matchesListPeer("id2"));
651 replyWithUnknownNodeIdentifier();
652 assertThat(peer.get().isPresent(), is(false));
655 private Matcher<List<String>> matchesListPeer(String nodeId) {
656 return matchesFcpMessage(
658 "Identifier=" + identifier,
659 "NodeIdentifier=" + nodeId
665 public class ListPeers {
668 public void withoutMetadataOrVolatile() throws IOException, ExecutionException, InterruptedException {
669 Future<Collection<Peer>> peers = fcpClient.listPeers().execute();
670 connectAndAssert(() -> matchesListPeers(false, false));
671 replyWithPeer("id1");
672 replyWithPeer("id2");
674 assertThat(peers.get(), hasSize(2));
675 assertThat(peers.get().stream().map(Peer::getIdentity).collect(Collectors.toList()),
676 containsInAnyOrder("id1", "id2"));
680 public void withMetadata() throws IOException, ExecutionException, InterruptedException {
681 Future<Collection<Peer>> peers = fcpClient.listPeers().includeMetadata().execute();
682 connectAndAssert(() -> matchesListPeers(false, true));
683 replyWithPeer("id1", "metadata.foo=bar1");
684 replyWithPeer("id2", "metadata.foo=bar2");
686 assertThat(peers.get(), hasSize(2));
687 assertThat(peers.get().stream().map(peer -> peer.getMetadata("foo")).collect(Collectors.toList()),
688 containsInAnyOrder("bar1", "bar2"));
692 public void withVolatile() throws IOException, ExecutionException, InterruptedException {
693 Future<Collection<Peer>> peers = fcpClient.listPeers().includeVolatile().execute();
694 connectAndAssert(() -> matchesListPeers(true, false));
695 replyWithPeer("id1", "volatile.foo=bar1");
696 replyWithPeer("id2", "volatile.foo=bar2");
698 assertThat(peers.get(), hasSize(2));
699 assertThat(peers.get().stream().map(peer -> peer.getVolatile("foo")).collect(Collectors.toList()),
700 containsInAnyOrder("bar1", "bar2"));
703 private Matcher<List<String>> matchesListPeers(boolean withVolatile, boolean withMetadata) {
704 return matchesFcpMessage(
706 "WithVolatile=" + withVolatile,
707 "WithMetadata=" + withMetadata
711 private void sendEndOfPeerList() throws IOException {
714 "Identifier=" + identifier,
721 public class AddPeer {
724 public void fromFile() throws InterruptedException, ExecutionException, IOException {
725 Future<Optional<Peer>> peer = fcpClient.addPeer().fromFile(new File("/tmp/ref.txt")).execute();
726 connectAndAssert(() -> allOf(matchesAddPeer(), hasItem("File=/tmp/ref.txt")));
727 replyWithPeer("id1");
728 assertThat(peer.get().get().getIdentity(), is("id1"));
732 public void fromUrl() throws InterruptedException, ExecutionException, IOException {
733 Future<Optional<Peer>> peer = fcpClient.addPeer().fromURL(new URL("http://node.ref/")).execute();
734 connectAndAssert(() -> allOf(matchesAddPeer(), hasItem("URL=http://node.ref/")));
735 replyWithPeer("id1");
736 assertThat(peer.get().get().getIdentity(), is("id1"));
740 public void fromNodeRef() throws InterruptedException, ExecutionException, IOException {
741 NodeRef nodeRef = createNodeRef();
742 Future<Optional<Peer>> peer = fcpClient.addPeer().fromNodeRef(nodeRef).execute();
743 connectAndAssert(() -> allOf(matchesAddPeer(), Matchers.<String>hasItems(
749 "dsaGroup.q=subprime",
750 "dsaPubKey.y=dsa-public",
751 "physical.udp=1.2.3.4:5678",
755 replyWithPeer("id1");
756 assertThat(peer.get().get().getIdentity(), is("id1"));
759 private NodeRef createNodeRef() {
760 NodeRef nodeRef = new NodeRef();
761 nodeRef.setIdentity("id1");
762 nodeRef.setName("name");
763 nodeRef.setARK(new ARK("public", "1"));
764 nodeRef.setDSAGroup(new DSAGroup("base", "prime", "subprime"));
765 nodeRef.setNegotiationTypes(new int[] { 3, 5 });
766 nodeRef.setPhysicalUDP("1.2.3.4:5678");
767 nodeRef.setDSAPublicKey("dsa-public");
768 nodeRef.setSignature("sig");
772 private Matcher<List<String>> matchesAddPeer() {
773 return matchesFcpMessage(
775 "Identifier=" + identifier
781 public class ModifyPeer {
784 public void defaultFcpClientCanEnablePeerByName()
785 throws InterruptedException, ExecutionException, IOException {
786 Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byName("id1").execute();
787 connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
788 replyWithPeer("id1");
789 assertThat(peer.get().get().getIdentity(), is("id1"));
793 public void defaultFcpClientCanDisablePeerByName()
794 throws InterruptedException, ExecutionException, IOException {
795 Future<Optional<Peer>> peer = fcpClient.modifyPeer().disable().byName("id1").execute();
796 connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", true));
797 replyWithPeer("id1");
798 assertThat(peer.get().get().getIdentity(), is("id1"));
802 public void defaultFcpClientCanEnablePeerByIdentity()
803 throws InterruptedException, ExecutionException, IOException {
804 Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byIdentity("id1").execute();
805 connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
806 replyWithPeer("id1");
807 assertThat(peer.get().get().getIdentity(), is("id1"));
811 public void defaultFcpClientCanEnablePeerByHostAndPort()
812 throws InterruptedException, ExecutionException, IOException {
813 Future<Optional<Peer>> peer =
814 fcpClient.modifyPeer().enable().byHostAndPort("1.2.3.4", 5678).execute();
815 connectAndAssert(() -> matchesModifyPeer("1.2.3.4:5678", "IsDisabled", false));
816 replyWithPeer("id1");
817 assertThat(peer.get().get().getIdentity(), is("id1"));
821 public void allowLocalAddressesOfPeer() throws InterruptedException, ExecutionException, IOException {
822 Future<Optional<Peer>> peer =
823 fcpClient.modifyPeer().allowLocalAddresses().byIdentity("id1").execute();
824 connectAndAssert(() -> allOf(
825 matchesModifyPeer("id1", "AllowLocalAddresses", true),
826 not(contains(startsWith("IsDisabled=")))
828 replyWithPeer("id1");
829 assertThat(peer.get().get().getIdentity(), is("id1"));
833 public void disallowLocalAddressesOfPeer()
834 throws InterruptedException, ExecutionException, IOException {
835 Future<Optional<Peer>> peer =
836 fcpClient.modifyPeer().disallowLocalAddresses().byIdentity("id1").execute();
837 connectAndAssert(() -> allOf(
838 matchesModifyPeer("id1", "AllowLocalAddresses", false),
839 not(contains(startsWith("IsDisabled=")))
841 replyWithPeer("id1");
842 assertThat(peer.get().get().getIdentity(), is("id1"));
846 public void setBurstOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
847 Future<Optional<Peer>> peer = fcpClient.modifyPeer().setBurstOnly().byIdentity("id1").execute();
848 connectAndAssert(() -> allOf(
849 matchesModifyPeer("id1", "IsBurstOnly", true),
850 not(contains(startsWith("AllowLocalAddresses="))),
851 not(contains(startsWith("IsDisabled=")))
853 replyWithPeer("id1");
854 assertThat(peer.get().get().getIdentity(), is("id1"));
858 public void clearBurstOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
859 Future<Optional<Peer>> peer = fcpClient.modifyPeer().clearBurstOnly().byIdentity("id1").execute();
860 connectAndAssert(() -> allOf(
861 matchesModifyPeer("id1", "IsBurstOnly", false),
862 not(contains(startsWith("AllowLocalAddresses="))),
863 not(contains(startsWith("IsDisabled=")))
865 replyWithPeer("id1");
866 assertThat(peer.get().get().getIdentity(), is("id1"));
870 public void defaultFcpClientCanSetListenOnlyForPeer()
871 throws InterruptedException, ExecutionException, IOException {
872 Future<Optional<Peer>> peer = fcpClient.modifyPeer().setListenOnly().byIdentity("id1").execute();
873 connectAndAssert(() -> allOf(
874 matchesModifyPeer("id1", "IsListenOnly", true),
875 not(contains(startsWith("AllowLocalAddresses="))),
876 not(contains(startsWith("IsDisabled="))),
877 not(contains(startsWith("IsBurstOnly=")))
879 replyWithPeer("id1");
880 assertThat(peer.get().get().getIdentity(), is("id1"));
884 public void clearListenOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
885 Future<Optional<Peer>> peer = fcpClient.modifyPeer().clearListenOnly().byIdentity("id1").execute();
886 connectAndAssert(() -> allOf(
887 matchesModifyPeer("id1", "IsListenOnly", false),
888 not(contains(startsWith("AllowLocalAddresses="))),
889 not(contains(startsWith("IsDisabled="))),
890 not(contains(startsWith("IsBurstOnly=")))
892 replyWithPeer("id1");
893 assertThat(peer.get().get().getIdentity(), is("id1"));
897 public void ignoreSourceForPeer() throws InterruptedException, ExecutionException, IOException {
898 Future<Optional<Peer>> peer = fcpClient.modifyPeer().ignoreSource().byIdentity("id1").execute();
899 connectAndAssert(() -> allOf(
900 matchesModifyPeer("id1", "IgnoreSourcePort", true),
901 not(contains(startsWith("AllowLocalAddresses="))),
902 not(contains(startsWith("IsDisabled="))),
903 not(contains(startsWith("IsBurstOnly="))),
904 not(contains(startsWith("IsListenOnly=")))
906 replyWithPeer("id1");
907 assertThat(peer.get().get().getIdentity(), is("id1"));
911 public void useSourceForPeer() throws InterruptedException, ExecutionException, IOException {
912 Future<Optional<Peer>> peer = fcpClient.modifyPeer().useSource().byIdentity("id1").execute();
913 connectAndAssert(() -> allOf(
914 matchesModifyPeer("id1", "IgnoreSourcePort", false),
915 not(contains(startsWith("AllowLocalAddresses="))),
916 not(contains(startsWith("IsDisabled="))),
917 not(contains(startsWith("IsBurstOnly="))),
918 not(contains(startsWith("IsListenOnly=")))
920 replyWithPeer("id1");
921 assertThat(peer.get().get().getIdentity(), is("id1"));
925 public void unknownNode() throws InterruptedException, ExecutionException, IOException {
926 Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byIdentity("id1").execute();
927 connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
928 replyWithUnknownNodeIdentifier();
929 assertThat(peer.get().isPresent(), is(false));
932 private Matcher<List<String>> matchesModifyPeer(String nodeIdentifier, String setting, boolean value) {
933 return matchesFcpMessage(
935 "Identifier=" + identifier,
936 "NodeIdentifier=" + nodeIdentifier,
937 setting + "=" + value
943 public class RemovePeer {
946 public void byName() throws InterruptedException, ExecutionException, IOException {
947 Future<Boolean> peer = fcpClient.removePeer().byName("Friend1").execute();
948 connectAndAssert(() -> matchesRemovePeer("Friend1"));
949 replyWithPeerRemoved("Friend1");
950 assertThat(peer.get(), is(true));
954 public void invalidName() throws InterruptedException, ExecutionException, IOException {
955 Future<Boolean> peer = fcpClient.removePeer().byName("NotFriend1").execute();
956 connectAndAssert(() -> matchesRemovePeer("NotFriend1"));
957 replyWithUnknownNodeIdentifier();
958 assertThat(peer.get(), is(false));
962 public void byIdentity() throws InterruptedException, ExecutionException, IOException {
963 Future<Boolean> peer = fcpClient.removePeer().byIdentity("id1").execute();
964 connectAndAssert(() -> matchesRemovePeer("id1"));
965 replyWithPeerRemoved("id1");
966 assertThat(peer.get(), is(true));
970 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
971 Future<Boolean> peer = fcpClient.removePeer().byHostAndPort("1.2.3.4", 5678).execute();
972 connectAndAssert(() -> matchesRemovePeer("1.2.3.4:5678"));
973 replyWithPeerRemoved("Friend1");
974 assertThat(peer.get(), is(true));
977 private Matcher<List<String>> matchesRemovePeer(String nodeIdentifier) {
978 return matchesFcpMessage(
980 "Identifier=" + identifier,
981 "NodeIdentifier=" + nodeIdentifier
985 private void replyWithPeerRemoved(String nodeIdentifier) throws IOException {
988 "Identifier=" + identifier,
989 "NodeIdentifier=" + nodeIdentifier,
996 private void replyWithPeer(String peerId, String... additionalLines) throws IOException {
999 "Identifier=" + identifier,
1000 "identity=" + peerId,
1002 "ark.pubURI=SSK@3YEf.../ark",
1005 "version=Fred,0.7,1.0,1466",
1006 "lastGoodVersion=Fred,0.7,1.0,1466"
1008 fcpServer.writeLine(additionalLines);
1009 fcpServer.writeLine("EndMessage");
1014 public class PeerNoteCommands {
1016 public class ListPeerNotes {
1019 public void onUnknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
1020 Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byName("Friend1").execute();
1021 connectAndAssert(() -> matchesListPeerNotes("Friend1"));
1022 replyWithUnknownNodeIdentifier();
1023 assertThat(peerNote.get().isPresent(), is(false));
1027 public void byNodeName() throws InterruptedException, ExecutionException, IOException {
1028 Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byName("Friend1").execute();
1029 connectAndAssert(() -> matchesListPeerNotes("Friend1"));
1030 replyWithPeerNote();
1031 replyWithEndListPeerNotes();
1032 assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
1033 assertThat(peerNote.get().get().getPeerNoteType(), is(1));
1037 public void byNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
1038 Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byIdentity("id1").execute();
1039 connectAndAssert(() -> matchesListPeerNotes("id1"));
1040 replyWithPeerNote();
1041 replyWithEndListPeerNotes();
1042 assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
1043 assertThat(peerNote.get().get().getPeerNoteType(), is(1));
1047 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
1048 Future<Optional<PeerNote>> peerNote =
1049 fcpClient.listPeerNotes().byHostAndPort("1.2.3.4", 5678).execute();
1050 connectAndAssert(() -> matchesListPeerNotes("1.2.3.4:5678"));
1051 replyWithPeerNote();
1052 replyWithEndListPeerNotes();
1053 assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
1054 assertThat(peerNote.get().get().getPeerNoteType(), is(1));
1057 private Matcher<List<String>> matchesListPeerNotes(String nodeIdentifier) {
1058 return matchesFcpMessage(
1060 "NodeIdentifier=" + nodeIdentifier
1064 private void replyWithEndListPeerNotes() throws IOException {
1065 fcpServer.writeLine(
1067 "Identifier=" + identifier,
1072 private void replyWithPeerNote() throws IOException {
1073 fcpServer.writeLine(
1075 "Identifier=" + identifier,
1076 "NodeIdentifier=Friend1",
1077 "NoteText=RXhhbXBsZSBUZXh0Lg==",
1085 public class ModifyPeerNotes {
1088 public void byName() throws InterruptedException, ExecutionException, IOException {
1089 Future<Boolean> noteUpdated =
1090 fcpClient.modifyPeerNote().darknetComment("foo").byName("Friend1").execute();
1091 connectAndAssert(() -> matchesModifyPeerNote("Friend1"));
1092 replyWithPeerNote();
1093 assertThat(noteUpdated.get(), is(true));
1097 public void onUnknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
1098 Future<Boolean> noteUpdated =
1099 fcpClient.modifyPeerNote().darknetComment("foo").byName("Friend1").execute();
1100 connectAndAssert(() -> matchesModifyPeerNote("Friend1"));
1101 replyWithUnknownNodeIdentifier();
1102 assertThat(noteUpdated.get(), is(false));
1106 public void defaultFcpClientFailsToModifyPeerNoteWithoutPeerNote()
1107 throws InterruptedException, ExecutionException, IOException {
1108 Future<Boolean> noteUpdated = fcpClient.modifyPeerNote().byName("Friend1").execute();
1109 assertThat(noteUpdated.get(), is(false));
1113 public void byIdentifier() throws InterruptedException, ExecutionException, IOException {
1114 Future<Boolean> noteUpdated =
1115 fcpClient.modifyPeerNote().darknetComment("foo").byIdentifier("id1").execute();
1116 connectAndAssert(() -> matchesModifyPeerNote("id1"));
1117 replyWithPeerNote();
1118 assertThat(noteUpdated.get(), is(true));
1122 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
1123 Future<Boolean> noteUpdated =
1124 fcpClient.modifyPeerNote().darknetComment("foo").byHostAndPort("1.2.3.4", 5678).execute();
1125 connectAndAssert(() -> matchesModifyPeerNote("1.2.3.4:5678"));
1126 replyWithPeerNote();
1127 assertThat(noteUpdated.get(), is(true));
1130 private Matcher<List<String>> matchesModifyPeerNote(String nodeIdentifier) {
1131 return matchesFcpMessage(
1133 "Identifier=" + identifier,
1134 "NodeIdentifier=" + nodeIdentifier,
1140 private void replyWithPeerNote() throws IOException {
1141 fcpServer.writeLine(
1143 "Identifier=" + identifier,
1144 "NodeIdentifier=Friend1",
1155 private void replyWithUnknownNodeIdentifier() throws IOException {
1156 fcpServer.writeLine(
1157 "UnknownNodeIdentifier",
1158 "Identifier=" + identifier,
1159 "NodeIdentifier=id2",
1166 public class PluginCommands {
1168 private static final String CLASS_NAME = "foo.plugin.Plugin";
1170 private void replyWithPluginInfo() throws IOException {
1171 fcpServer.writeLine(
1173 "Identifier=" + identifier,
1174 "PluginName=superPlugin",
1176 "LongVersion=1.2.3",
1178 "OriginUri=superPlugin",
1184 private void verifyPluginInfo(Future<Optional<PluginInfo>> pluginInfo)
1185 throws InterruptedException, ExecutionException {
1186 assertThat(pluginInfo.get().get().getPluginName(), is("superPlugin"));
1187 assertThat(pluginInfo.get().get().getOriginalURI(), is("superPlugin"));
1188 assertThat(pluginInfo.get().get().isTalkable(), is(true));
1189 assertThat(pluginInfo.get().get().getVersion(), is("42"));
1190 assertThat(pluginInfo.get().get().getLongVersion(), is("1.2.3"));
1191 assertThat(pluginInfo.get().get().isStarted(), is(true));
1194 public class LoadPlugin {
1196 public class OfficialPlugins {
1199 public void fromFreenet() throws ExecutionException, InterruptedException, IOException {
1200 Future<Optional<PluginInfo>> pluginInfo =
1201 fcpClient.loadPlugin().officialFromFreenet("superPlugin").execute();
1202 connectAndAssert(() -> createMatcherForOfficialSource("freenet"));
1203 assertThat(lines, not(contains(startsWith("Store="))));
1204 replyWithPluginInfo();
1205 verifyPluginInfo(pluginInfo);
1209 public void persistentFromFreenet() throws ExecutionException, InterruptedException, IOException {
1210 Future<Optional<PluginInfo>> pluginInfo =
1211 fcpClient.loadPlugin().addToConfig().officialFromFreenet("superPlugin").execute();
1212 connectAndAssert(() -> createMatcherForOfficialSource("freenet"));
1213 assertThat(lines, hasItem("Store=true"));
1214 replyWithPluginInfo();
1215 verifyPluginInfo(pluginInfo);
1219 public void fromHttps() throws ExecutionException, InterruptedException, IOException {
1220 Future<Optional<PluginInfo>> pluginInfo =
1221 fcpClient.loadPlugin().officialFromHttps("superPlugin").execute();
1222 connectAndAssert(() -> createMatcherForOfficialSource("https"));
1223 replyWithPluginInfo();
1224 verifyPluginInfo(pluginInfo);
1227 private Matcher<List<String>> createMatcherForOfficialSource(String officialSource) {
1228 return matchesFcpMessage(
1230 "Identifier=" + identifier,
1231 "PluginURL=superPlugin",
1233 "OfficialSource=" + officialSource
1239 public class FromOtherSources {
1241 private static final String FILE_PATH = "/path/to/plugin.jar";
1242 private static final String URL = "http://server.com/plugin.jar";
1243 private static final String KEY = "KSK@plugin.jar";
1246 public void fromFile() throws ExecutionException, InterruptedException, IOException {
1247 Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromFile(FILE_PATH).execute();
1248 connectAndAssert(() -> createMatcher("file", FILE_PATH));
1249 replyWithPluginInfo();
1250 verifyPluginInfo(pluginInfo);
1254 public void fromUrl() throws ExecutionException, InterruptedException, IOException {
1255 Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromUrl(URL).execute();
1256 connectAndAssert(() -> createMatcher("url", URL));
1257 replyWithPluginInfo();
1258 verifyPluginInfo(pluginInfo);
1262 public void fromFreenet() throws ExecutionException, InterruptedException, IOException {
1263 Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromFreenet(KEY).execute();
1264 connectAndAssert(() -> createMatcher("freenet", KEY));
1265 replyWithPluginInfo();
1266 verifyPluginInfo(pluginInfo);
1269 private Matcher<List<String>> createMatcher(String urlType, String url) {
1270 return matchesFcpMessage(
1272 "Identifier=" + identifier,
1274 "URLType=" + urlType
1280 public class Failed {
1283 public void failedLoad() throws ExecutionException, InterruptedException, IOException {
1284 Future<Optional<PluginInfo>> pluginInfo =
1285 fcpClient.loadPlugin().officialFromFreenet("superPlugin").execute();
1286 connectAndAssert(() -> matchesFcpMessage("LoadPlugin"));
1287 replyWithProtocolError();
1288 assertThat(pluginInfo.get().isPresent(), is(false));
1295 private void replyWithProtocolError() throws IOException {
1296 fcpServer.writeLine(
1298 "Identifier=" + identifier,
1303 public class ReloadPlugin {
1306 public void reloadingPluginWorks() throws InterruptedException, ExecutionException, IOException {
1307 Future<Optional<PluginInfo>> pluginInfo = fcpClient.reloadPlugin().plugin(CLASS_NAME).execute();
1308 connectAndAssert(() -> matchReloadPluginMessage());
1309 replyWithPluginInfo();
1310 verifyPluginInfo(pluginInfo);
1314 public void reloadingPluginWithMaxWaitTimeWorks()
1315 throws InterruptedException, ExecutionException, IOException {
1316 Future<Optional<PluginInfo>> pluginInfo =
1317 fcpClient.reloadPlugin().waitFor(1234).plugin(CLASS_NAME).execute();
1318 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("MaxWaitTime=1234")));
1319 replyWithPluginInfo();
1320 verifyPluginInfo(pluginInfo);
1324 public void reloadingPluginWithPurgeWorks()
1325 throws InterruptedException, ExecutionException, IOException {
1326 Future<Optional<PluginInfo>> pluginInfo =
1327 fcpClient.reloadPlugin().purge().plugin(CLASS_NAME).execute();
1328 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("Purge=true")));
1329 replyWithPluginInfo();
1330 verifyPluginInfo(pluginInfo);
1334 public void reloadingPluginWithStoreWorks()
1335 throws InterruptedException, ExecutionException, IOException {
1336 Future<Optional<PluginInfo>> pluginInfo =
1337 fcpClient.reloadPlugin().addToConfig().plugin(CLASS_NAME).execute();
1338 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("Store=true")));
1339 replyWithPluginInfo();
1340 verifyPluginInfo(pluginInfo);
1343 private Matcher<List<String>> matchReloadPluginMessage() {
1344 return matchesFcpMessage(
1346 "Identifier=" + identifier,
1347 "PluginName=" + CLASS_NAME
1353 public class RemovePlugin {
1356 public void removingPluginWorks() throws InterruptedException, ExecutionException, IOException {
1357 Future<Boolean> pluginRemoved = fcpClient.removePlugin().plugin(CLASS_NAME).execute();
1358 connectAndAssert(() -> matchPluginRemovedMessage());
1359 replyWithPluginRemoved();
1360 assertThat(pluginRemoved.get(), is(true));
1364 public void removingPluginWithMaxWaitTimeWorks()
1365 throws InterruptedException, ExecutionException, IOException {
1366 Future<Boolean> pluginRemoved = fcpClient.removePlugin().waitFor(1234).plugin(CLASS_NAME).execute();
1367 connectAndAssert(() -> allOf(matchPluginRemovedMessage(), hasItem("MaxWaitTime=1234")));
1368 replyWithPluginRemoved();
1369 assertThat(pluginRemoved.get(), is(true));
1373 public void removingPluginWithPurgeWorks()
1374 throws InterruptedException, ExecutionException, IOException {
1375 Future<Boolean> pluginRemoved = fcpClient.removePlugin().purge().plugin(CLASS_NAME).execute();
1376 connectAndAssert(() -> allOf(matchPluginRemovedMessage(), hasItem("Purge=true")));
1377 replyWithPluginRemoved();
1378 assertThat(pluginRemoved.get(), is(true));
1381 private void replyWithPluginRemoved() throws IOException {
1382 fcpServer.writeLine(
1384 "Identifier=" + identifier,
1385 "PluginName=" + CLASS_NAME,
1390 private Matcher<List<String>> matchPluginRemovedMessage() {
1391 return matchesFcpMessage(
1393 "Identifier=" + identifier,
1394 "PluginName=" + CLASS_NAME
1400 public class GetPluginInfo {
1403 public void gettingPluginInfoWorks() throws InterruptedException, ExecutionException, IOException {
1404 Future<Optional<PluginInfo>> pluginInfo = fcpClient.getPluginInfo().plugin(CLASS_NAME).execute();
1405 connectAndAssert(() -> matchGetPluginInfoMessage());
1406 replyWithPluginInfo();
1407 verifyPluginInfo(pluginInfo);
1411 public void gettingPluginInfoWithDetailsWorks()
1412 throws InterruptedException, ExecutionException, IOException {
1413 Future<Optional<PluginInfo>> pluginInfo =
1414 fcpClient.getPluginInfo().detailed().plugin(CLASS_NAME).execute();
1415 connectAndAssert(() -> allOf(matchGetPluginInfoMessage(), hasItem("Detailed=true")));
1416 replyWithPluginInfo();
1417 verifyPluginInfo(pluginInfo);
1421 public void protocolErrorIsRecognizedAsFailure()
1422 throws InterruptedException, ExecutionException, IOException {
1423 Future<Optional<PluginInfo>> pluginInfo =
1424 fcpClient.getPluginInfo().detailed().plugin(CLASS_NAME).execute();
1425 connectAndAssert(() -> allOf(matchGetPluginInfoMessage(), hasItem("Detailed=true")));
1426 replyWithProtocolError();
1427 assertThat(pluginInfo.get(), is(Optional.empty()));
1430 private Matcher<List<String>> matchGetPluginInfoMessage() {
1431 return matchesFcpMessage(
1433 "Identifier=" + identifier,
1434 "PluginName=" + CLASS_NAME
1442 public class UskSubscriptionCommands {
1444 private static final String URI = "USK@some,uri/file.txt";
1447 public void subscriptionWorks() throws InterruptedException, ExecutionException, IOException {
1448 Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1449 connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1450 replyWithSubscribed();
1451 assertThat(uskSubscription.get().get().getUri(), is(URI));
1452 AtomicInteger edition = new AtomicInteger();
1453 CountDownLatch updated = new CountDownLatch(2);
1454 uskSubscription.get().get().onUpdate(e -> {
1456 updated.countDown();
1458 sendUpdateNotification(23);
1459 sendUpdateNotification(24);
1460 assertThat("updated in time", updated.await(5, TimeUnit.SECONDS), is(true));
1461 assertThat(edition.get(), is(24));
1465 public void subscriptionUpdatesMultipleTimes() throws InterruptedException, ExecutionException, IOException {
1466 Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1467 connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1468 replyWithSubscribed();
1469 assertThat(uskSubscription.get().get().getUri(), is(URI));
1470 AtomicInteger edition = new AtomicInteger();
1471 CountDownLatch updated = new CountDownLatch(2);
1472 uskSubscription.get().get().onUpdate(e -> {
1474 updated.countDown();
1476 uskSubscription.get().get().onUpdate(e -> updated.countDown());
1477 sendUpdateNotification(23);
1478 assertThat("updated in time", updated.await(5, TimeUnit.SECONDS), is(true));
1479 assertThat(edition.get(), is(23));
1483 public void subscriptionCanBeCancelled() throws InterruptedException, ExecutionException, IOException {
1484 Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1485 connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1486 replyWithSubscribed();
1487 assertThat(uskSubscription.get().get().getUri(), is(URI));
1488 AtomicBoolean updated = new AtomicBoolean();
1489 uskSubscription.get().get().onUpdate(e -> updated.set(true));
1490 uskSubscription.get().get().cancel();
1491 readMessage(() -> matchesFcpMessage("UnsubscribeUSK", "Identifier=" + identifier));
1492 sendUpdateNotification(23);
1493 assertThat(updated.get(), is(false));
1496 private void replyWithSubscribed() throws IOException {
1497 fcpServer.writeLine(
1499 "Identifier=" + identifier,
1506 private void sendUpdateNotification(int edition, String... additionalLines) throws IOException {
1507 fcpServer.writeLine(
1508 "SubscribedUSKUpdate",
1509 "Identifier=" + identifier,
1511 "Edition=" + edition
1513 fcpServer.writeLine(additionalLines);
1514 fcpServer.writeLine("EndMessage");
1519 public class ClientGet {
1522 public void works() throws InterruptedException, ExecutionException, IOException {
1523 Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1524 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "ReturnType=direct"));
1525 replyWithAllData("not-test", "Hello World", "text/plain;charset=latin-9");
1526 replyWithAllData(identifier, "Hello", "text/plain;charset=utf-8");
1527 Optional<Data> data = dataFuture.get();
1532 public void getFailedIsRecognized() throws InterruptedException, ExecutionException, IOException {
1533 Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1534 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1535 replyWithGetFailed("not-test");
1536 replyWithGetFailed(identifier);
1537 Optional<Data> data = dataFuture.get();
1538 assertThat(data.isPresent(), is(false));
1542 public void getFailedForDifferentIdentifierIsIgnored()
1543 throws InterruptedException, ExecutionException, IOException {
1544 Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1545 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1546 replyWithGetFailed("not-test");
1547 replyWithAllData(identifier, "Hello", "text/plain;charset=utf-8");
1548 Optional<Data> data = dataFuture.get();
1552 @Test(expected = ExecutionException.class)
1553 public void connectionClosedIsRecognized() throws InterruptedException, ExecutionException, IOException {
1554 Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1555 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1561 public void withIgnoreData() throws InterruptedException, ExecutionException, IOException {
1562 fcpClient.clientGet().ignoreDataStore().uri("KSK@foo.txt").execute();
1563 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "IgnoreDS=true"));
1567 public void withDataStoreOnly() throws InterruptedException, ExecutionException, IOException {
1568 fcpClient.clientGet().dataStoreOnly().uri("KSK@foo.txt").execute();
1569 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "DSonly=true"));
1573 public void clientGetWithMaxSizeSettingSendsCorrectCommands()
1574 throws InterruptedException, ExecutionException, IOException {
1575 fcpClient.clientGet().maxSize(1048576).uri("KSK@foo.txt").execute();
1576 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "MaxSize=1048576"));
1580 public void clientGetWithPrioritySettingSendsCorrectCommands()
1581 throws InterruptedException, ExecutionException, IOException {
1582 fcpClient.clientGet().priority(Priority.interactive).uri("KSK@foo.txt").execute();
1583 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "PriorityClass=1"));
1587 public void clientGetWithRealTimeSettingSendsCorrectCommands()
1588 throws InterruptedException, ExecutionException, IOException {
1589 fcpClient.clientGet().realTime().uri("KSK@foo.txt").execute();
1590 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "RealTimeFlag=true"));
1594 public void clientGetWithGlobalSettingSendsCorrectCommands()
1595 throws InterruptedException, ExecutionException, IOException {
1596 fcpClient.clientGet().global().uri("KSK@foo.txt").execute();
1597 connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "Global=true"));
1600 private void replyWithGetFailed(String identifier) throws IOException {
1601 fcpServer.writeLine(
1603 "Identifier=" + identifier,
1609 private void replyWithAllData(String identifier, String text, String contentType) throws IOException {
1610 fcpServer.writeLine(
1612 "Identifier=" + identifier,
1613 "DataLength=" + (text.length() + 1),
1614 "StartupTime=1435610539000",
1615 "CompletionTime=1435610540000",
1616 "Metadata.ContentType=" + contentType,
1622 private void verifyData(Optional<Data> data) throws IOException {
1623 assertThat(data.get().getMimeType(), is("text/plain;charset=utf-8"));
1624 assertThat(data.get().size(), is(6L));
1625 assertThat(ByteStreams.toByteArray(data.get().getInputStream()),
1626 is("Hello\n".getBytes(StandardCharsets.UTF_8)));
1631 public class ClientPut {
1634 public void sendsCorrectCommand() throws IOException, ExecutionException, InterruptedException {
1635 fcpClient.clientPut()
1636 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1641 readMessage("Hello", () -> matchesDirectClientPut());
1645 public void succeedsOnCorrectIdentifier() throws InterruptedException, ExecutionException, IOException {
1646 Future<Optional<Key>> key = fcpClient.clientPut()
1647 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1652 readMessage("Hello", () -> matchesDirectClientPut());
1653 replyWithPutFailed("not-the-right-one");
1654 replyWithPutSuccessful(identifier);
1655 assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1659 public void failsOnCorrectIdentifier() throws InterruptedException, ExecutionException, IOException {
1660 Future<Optional<Key>> key = fcpClient.clientPut()
1661 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1666 readMessage("Hello", () -> matchesDirectClientPut());
1667 replyWithPutSuccessful("not-the-right-one");
1668 replyWithPutFailed(identifier);
1669 assertThat(key.get().isPresent(), is(false));
1673 public void renameIsSentCorrectly() throws InterruptedException, ExecutionException, IOException {
1674 fcpClient.clientPut()
1675 .named("otherName.txt")
1676 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1681 readMessage("Hello", () -> allOf(
1682 hasHead("ClientPut"),
1683 hasParameters(1, 2, "TargetFilename=otherName.txt", "UploadFrom=direct", "DataLength=6",
1685 hasTail("EndMessage", "Hello")
1690 public void redirectIsSentCorrecly() throws IOException, ExecutionException, InterruptedException {
1691 fcpClient.clientPut().redirectTo("KSK@bar.txt").uri("KSK@foo.txt").execute();
1692 connectAndAssert(() ->
1693 matchesFcpMessage("ClientPut", "UploadFrom=redirect", "URI=KSK@foo.txt", "TargetURI=KSK@bar.txt"));
1697 public void withFileIsSentCorrectly() throws InterruptedException, ExecutionException, IOException {
1698 fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1699 connectAndAssert(() ->
1700 matchesFcpMessage("ClientPut", "UploadFrom=disk", "URI=KSK@foo.txt", "Filename=/tmp/data.txt"));
1705 private final File ddaFile;
1706 private final File fileToUpload;
1708 public DDA() throws IOException {
1709 ddaFile = createDdaFile();
1710 fileToUpload = new File(ddaFile.getParent(), "test.dat");
1713 private Matcher<List<String>> matchesFileClientPut(File file) {
1714 return matchesFcpMessage("ClientPut", "UploadFrom=disk", "URI=KSK@foo.txt", "Filename=" + file);
1718 public void completeDda() throws IOException, ExecutionException, InterruptedException {
1719 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1720 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1721 sendDdaRequired(identifier);
1722 readMessage(() -> matchesTestDDARequest(ddaFile));
1723 sendTestDDAReply(ddaFile.getParent(), ddaFile);
1724 readMessage(() -> matchesTestDDAResponse(ddaFile));
1725 writeTestDDAComplete(ddaFile);
1726 readMessage(() -> matchesFileClientPut(fileToUpload));
1730 public void ignoreOtherDda() throws IOException, ExecutionException, InterruptedException {
1731 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1732 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1733 sendDdaRequired(identifier);
1734 readMessage(() -> matchesTestDDARequest(ddaFile));
1735 sendTestDDAReply("/some-other-directory", ddaFile);
1736 sendTestDDAReply(ddaFile.getParent(), ddaFile);
1737 readMessage(() -> matchesTestDDAResponse(ddaFile));
1741 public void sendResponseIfFileUnreadable() throws IOException, ExecutionException, InterruptedException {
1742 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1743 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1744 sendDdaRequired(identifier);
1745 readMessage(() -> matchesTestDDARequest(ddaFile));
1746 sendTestDDAReply(ddaFile.getParent(), new File(ddaFile + ".foo"));
1747 readMessage(() -> matchesFailedToReadResponse());
1751 public void clientPutDoesNotResendOriginalClientPutOnTestDDACompleteWithWrongDirectory()
1752 throws IOException, ExecutionException, InterruptedException {
1753 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1755 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1756 String identifier = extractIdentifier(lines);
1757 fcpServer.writeLine(
1759 "Directory=/some-other-directory",
1762 sendDdaRequired(identifier);
1763 lines = fcpServer.collectUntil(is("EndMessage"));
1764 assertThat(lines, matchesFcpMessage(
1766 "Directory=" + ddaFile.getParent(),
1767 "WantReadDirectory=true",
1768 "WantWriteDirectory=false"
1772 private Matcher<List<String>> matchesFailedToReadResponse() {
1773 return matchesFcpMessage(
1775 "Directory=" + ddaFile.getParent(),
1776 "ReadContent=failed-to-read"
1780 private void writeTestDDAComplete(File tempFile) throws IOException {
1781 fcpServer.writeLine(
1783 "Directory=" + tempFile.getParent(),
1784 "ReadDirectoryAllowed=true",
1789 private Matcher<List<String>> matchesTestDDAResponse(File tempFile) {
1790 return matchesFcpMessage(
1792 "Directory=" + tempFile.getParent(),
1793 "ReadContent=test-content"
1797 private void sendTestDDAReply(String directory, File tempFile) throws IOException {
1798 fcpServer.writeLine(
1800 "Directory=" + directory,
1801 "ReadFilename=" + tempFile,
1806 private Matcher<List<String>> matchesTestDDARequest(File tempFile) {
1807 return matchesFcpMessage(
1809 "Directory=" + tempFile.getParent(),
1810 "WantReadDirectory=true",
1811 "WantWriteDirectory=false"
1815 private void sendDdaRequired(String identifier) throws IOException {
1816 fcpServer.writeLine(
1818 "Identifier=" + identifier,
1826 private void replyWithPutSuccessful(String identifier) throws IOException {
1827 fcpServer.writeLine(
1830 "Identifier=" + identifier,
1835 private void replyWithPutFailed(String identifier) throws IOException {
1836 fcpServer.writeLine(
1838 "Identifier=" + identifier,
1843 private Matcher<List<String>> matchesDirectClientPut() {
1845 hasHead("ClientPut"),
1846 hasParameters(1, 2, "UploadFrom=direct", "DataLength=6", "URI=KSK@foo.txt"),
1847 hasTail("EndMessage", "Hello")
1851 private File createDdaFile() throws IOException {
1852 File tempFile = File.createTempFile("test-dda-", ".dat");
1853 tempFile.deleteOnExit();
1854 Files.write("test-content", tempFile, StandardCharsets.UTF_8);
1859 public void clientPutDoesNotReactToProtocolErrorForDifferentIdentifier()
1860 throws InterruptedException, ExecutionException, IOException {
1861 Future<Optional<Key>> key = fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1863 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1864 String identifier = extractIdentifier(lines);
1865 fcpServer.writeLine(
1867 "Identifier=not-the-right-one",
1871 fcpServer.writeLine(
1873 "Identifier=" + identifier,
1877 assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1881 public void clientPutAbortsOnProtocolErrorOtherThan25()
1882 throws InterruptedException, ExecutionException, IOException {
1883 Future<Optional<Key>> key = fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1885 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1886 String identifier = extractIdentifier(lines);
1887 fcpServer.writeLine(
1889 "Identifier=" + identifier,
1893 assertThat(key.get().isPresent(), is(false));
1897 public void clientPutSendsNotificationsForGeneratedKeys()
1898 throws InterruptedException, ExecutionException, IOException {
1899 List<String> generatedKeys = new CopyOnWriteArrayList<>();
1900 Future<Optional<Key>> key = fcpClient.clientPut()
1901 .onKeyGenerated(generatedKeys::add)
1902 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1907 List<String> lines = fcpServer.collectUntil(is("Hello"));
1908 String identifier = extractIdentifier(lines);
1909 fcpServer.writeLine(
1911 "Identifier=" + identifier,
1915 replyWithPutSuccessful(identifier);
1916 assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1917 assertThat(generatedKeys, contains("KSK@foo.txt"));