src/test/java/net/pterodactylus/fcp/quelaton/DefaultFcpClientTest.java

   1 package net.pterodactylus.fcp.quelaton;
   2
   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;
  13
  14 import java.io.ByteArrayInputStream;
  15 import java.io.File;
  16 import java.io.IOException;
  17 import java.net.URL;
  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;
  34
  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;
  48
  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;
  60
  61 /**
  62  * Unit test for {@link DefaultFcpClient}.
  63  *
  64  * @author <a href="bombe@freenetproject.org">David ‘Bombe’ Roden</a>
  65  */
  66 @RunWith(NestedRunner.class)
  67 public class DefaultFcpClientTest {
  68
  69         private static final String INSERT_URI =
  70                 "SSK@RVCHbJdkkyTCeNN9AYukEg76eyqmiosSaNKgE3U9zUw,7SHH53gletBVb9JD7nBsyClbLQsBubDPEIcwg908r7Y,AQECAAE/";
  71         private static final String REQUEST_URI =
  72                 "SSK@wtbgd2loNcJCXvtQVOftl2tuWBomDQHfqS6ytpPRhfw,7SHH53gletBVb9JD7nBsyClbLQsBubDPEIcwg908r7Y,AQACAAE/";
  73
  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;
  79
  80         public DefaultFcpClientTest() throws IOException {
  81                 fcpServer = new FakeTcpServer(threadPool);
  82                 fcpClient = new DefaultFcpClient(threadPool, "localhost", fcpServer.getPort(), () -> "Test");
  83         }
  84
  85         @After
  86         public void tearDown() throws IOException {
  87                 fcpServer.close();
  88                 threadPool.shutdown();
  89         }
  90
  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",
  97                         "Testnet=false",
  98                         "Version=Fred,0.7,1.0,1466",
  99                         "Build=1466",
 100                         "ConnectionIdentifier=14318898267048452a81b36e7f13a3f0",
 101                         "Node=Fred",
 102                         "ExtBuild=29",
 103                         "FCPVersion=2.0",
 104                         "NodeLanguage=ENGLISH",
 105                         "ExtRevision=v29",
 106                         "EndMessage"
 107                 );
 108         }
 109
 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))
 114                         .findFirst()
 115                         .orElse("");
 116         }
 117
 118         private Matcher<List<String>> matchesFcpMessage(String name, String... requiredLines) {
 119                 return matchesFcpMessageWithTerminator(name, "EndMessage", requiredLines);
 120         }
 121
 122         private Matcher<Iterable<String>> hasHead(String firstElement) {
 123                 return new TypeSafeDiagnosingMatcher<Iterable<String>>() {
 124                         @Override
 125                         protected boolean matchesSafely(Iterable<String> iterable, Description mismatchDescription) {
 126                                 if (!iterable.iterator().hasNext()) {
 127                                         mismatchDescription.appendText("is empty");
 128                                         return false;
 129                                 }
 130                                 String element = iterable.iterator().next();
 131                                 if (!element.equals(firstElement)) {
 132                                         mismatchDescription.appendText("starts with ").appendValue(element);
 133                                         return false;
 134                                 }
 135                                 return true;
 136                         }
 137
 138                         @Override
 139                         public void describeTo(Description description) {
 140                                 description.appendText("starts with ").appendValue(firstElement);
 141                         }
 142                 };
 143         }
 144
 145         private Matcher<List<String>> matchesFcpMessageWithTerminator(
 146                 String name, String terminator, String... requiredLines) {
 147                 return allOf(hasHead(name), hasParameters(1, 1, requiredLines), hasTail(terminator));
 148         }
 149
 150         private Matcher<List<String>> hasParameters(int ignoreStart, int ignoreEnd, String... lines) {
 151                 return new TypeSafeDiagnosingMatcher<List<String>>() {
 152                         @Override
 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");
 156                                         return false;
 157                                 }
 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);
 161                                                 return false;
 162                                         }
 163                                 }
 164                                 return true;
 165                         }
 166
 167                         @Override
 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);
 172                         }
 173                 };
 174         }
 175
 176         private Matcher<List<String>> hasTail(String... lastElements) {
 177                 return new TypeSafeDiagnosingMatcher<List<String>>() {
 178                         @Override
 179                         protected boolean matchesSafely(List<String> list, Description mismatchDescription) {
 180                                 if (list.size() < lastElements.length) {
 181                                         mismatchDescription.appendText("is too small");
 182                                         return false;
 183                                 }
 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);
 187                                         return false;
 188                                 }
 189                                 return true;
 190                         }
 191
 192                         @Override
 193                         public void describeTo(Description description) {
 194                                 description.appendText("ends with ").appendValueList("(", ", ", ")", lastElements);
 195                         }
 196                 };
 197         }
 198
 199         private List<String> lines;
 200         private String identifier;
 201
 202         private void connectAndAssert(Supplier<Matcher<List<String>>> requestMatcher)
 203         throws InterruptedException, ExecutionException, IOException {
 204                 connectNode();
 205                 readMessage(requestMatcher);
 206         }
 207
 208         private void readMessage(Supplier<Matcher<List<String>>> requestMatcher) throws IOException {
 209                 readMessage("EndMessage", requestMatcher);
 210         }
 211
 212         private void readMessage(String terminator, Supplier<Matcher<List<String>>> requestMatcher) throws IOException {
 213                 lines = fcpServer.collectUntil(is(terminator));
 214                 identifier = extractIdentifier(lines);
 215                 assertThat(lines, requestMatcher.get());
 216         }
 217
 218         public class ConnectionsAndKeyPairs {
 219
 220                 public class Connections {
 221
 222                         @Test(expected = ExecutionException.class)
 223                         public void throwsExceptionOnFailure() throws IOException, ExecutionException, InterruptedException {
 224                                 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
 225                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 226                                 fcpServer.writeLine(
 227                                         "CloseConnectionDuplicateClientName",
 228                                         "EndMessage"
 229                                 );
 230                                 keyPairFuture.get();
 231                         }
 232
 233                         @Test(expected = ExecutionException.class)
 234                         public void throwsExceptionIfConnectionIsClosed() throws IOException, ExecutionException, InterruptedException {
 235                                 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
 236                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 237                                 fcpServer.close();
 238                                 keyPairFuture.get();
 239                         }
 240
 241                         @Test
 242                         public void connectionIsReused() throws InterruptedException, ExecutionException, IOException {
 243                                 Future<FcpKeyPair> keyPair = fcpClient.generateKeypair().execute();
 244                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 245                                 replyWithKeyPair();
 246                                 keyPair.get();
 247                                 keyPair = fcpClient.generateKeypair().execute();
 248                                 readMessage(() -> matchesFcpMessage("GenerateSSK"));
 249                                 identifier = extractIdentifier(lines);
 250                                 replyWithKeyPair();
 251                                 keyPair.get();
 252                         }
 253
 254                         @Test
 255                         public void defaultFcpClientCanReconnectAfterConnectionHasBeenClosed()
 256                         throws InterruptedException, ExecutionException, IOException {
 257                                 Future<FcpKeyPair> keyPair = fcpClient.generateKeypair().execute();
 258                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 259                                 fcpServer.close();
 260                                 try {
 261                                         keyPair.get();
 262                                         Assert.fail();
 263                                 } catch (ExecutionException e) {
 264                                         /* ignore. */
 265                                 }
 266                                 keyPair = fcpClient.generateKeypair().execute();
 267                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 268                                 replyWithKeyPair();
 269                                 keyPair.get();
 270                         }
 271
 272                 }
 273
 274                 public class GenerateKeyPair {
 275
 276                         @Test
 277                         public void defaultFcpClientCanGenerateKeypair()
 278                         throws ExecutionException, InterruptedException, IOException {
 279                                 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
 280                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 281                                 replyWithKeyPair();
 282                                 FcpKeyPair keyPair = keyPairFuture.get();
 283                                 assertThat(keyPair.getPublicKey(), is(REQUEST_URI));
 284                                 assertThat(keyPair.getPrivateKey(), is(INSERT_URI));
 285                         }
 286
 287                 }
 288
 289                 private void replyWithKeyPair() throws IOException {
 290                         fcpServer.writeLine("SSKKeypair",
 291                                 "InsertURI=" + INSERT_URI + "",
 292                                 "RequestURI=" + REQUEST_URI + "",
 293                                 "Identifier=" + identifier,
 294                                 "EndMessage");
 295                 }
 296
 297         }
 298
 299         public class Peers {
 300
 301                 public class PeerCommands {
 302
 303                         public class ListPeer {
 304
 305                                 @Test
 306                                 public void byIdentity() throws InterruptedException, ExecutionException, IOException {
 307                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byIdentity("id1").execute();
 308                                         connectAndAssert(() -> matchesListPeer("id1"));
 309                                         replyWithPeer("id1");
 310                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 311                                 }
 312
 313                                 @Test
 314                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 315                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byHostAndPort("host.free.net", 12345).execute();
 316                                         connectAndAssert(() -> matchesListPeer("host.free.net:12345"));
 317                                         replyWithPeer("id1");
 318                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 319                                 }
 320
 321                                 @Test
 322                                 public void byName() throws InterruptedException, ExecutionException, IOException {
 323                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byName("FriendNode").execute();
 324                                         connectAndAssert(() -> matchesListPeer("FriendNode"));
 325                                         replyWithPeer("id1");
 326                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 327                                 }
 328
 329                                 @Test
 330                                 public void unknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 331                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byIdentity("id2").execute();
 332                                         connectAndAssert(() -> matchesListPeer("id2"));
 333                                         replyWithUnknownNodeIdentifier();
 334                                         assertThat(peer.get().isPresent(), is(false));
 335                                 }
 336
 337                                 private Matcher<List<String>> matchesListPeer(String nodeId) {
 338                                         return matchesFcpMessage(
 339                                                 "ListPeer",
 340                                                 "Identifier=" + identifier,
 341                                                 "NodeIdentifier=" + nodeId
 342                                         );
 343                                 }
 344
 345                         }
 346
 347                         public class ListPeers {
 348
 349                                 @Test
 350                                 public void withoutMetadataOrVolatile() throws IOException, ExecutionException, InterruptedException {
 351                                         Future<Collection<Peer>> peers = fcpClient.listPeers().execute();
 352                                         connectAndAssert(() -> matchesListPeers(false, false));
 353                                         replyWithPeer("id1");
 354                                         replyWithPeer("id2");
 355                                         sendEndOfPeerList();
 356                                         assertThat(peers.get(), hasSize(2));
 357                                         assertThat(peers.get().stream().map(Peer::getIdentity).collect(Collectors.toList()),
 358                                                 containsInAnyOrder("id1", "id2"));
 359                                 }
 360
 361                                 @Test
 362                                 public void withMetadata() throws IOException, ExecutionException, InterruptedException {
 363                                         Future<Collection<Peer>> peers = fcpClient.listPeers().includeMetadata().execute();
 364                                         connectAndAssert(() -> matchesListPeers(false, true));
 365                                         replyWithPeer("id1", "metadata.foo=bar1");
 366                                         replyWithPeer("id2", "metadata.foo=bar2");
 367                                         sendEndOfPeerList();
 368                                         assertThat(peers.get(), hasSize(2));
 369                                         assertThat(peers.get().stream().map(peer -> peer.getMetadata("foo")).collect(Collectors.toList()),
 370                                                 containsInAnyOrder("bar1", "bar2"));
 371                                 }
 372
 373                                 @Test
 374                                 public void withVolatile() throws IOException, ExecutionException, InterruptedException {
 375                                         Future<Collection<Peer>> peers = fcpClient.listPeers().includeVolatile().execute();
 376                                         connectAndAssert(() -> matchesListPeers(true, false));
 377                                         replyWithPeer("id1", "volatile.foo=bar1");
 378                                         replyWithPeer("id2", "volatile.foo=bar2");
 379                                         sendEndOfPeerList();
 380                                         assertThat(peers.get(), hasSize(2));
 381                                         assertThat(peers.get().stream().map(peer -> peer.getVolatile("foo")).collect(Collectors.toList()),
 382                                                 containsInAnyOrder("bar1", "bar2"));
 383                                 }
 384
 385                                 private Matcher<List<String>> matchesListPeers(boolean withVolatile, boolean withMetadata) {
 386                                         return matchesFcpMessage(
 387                                                 "ListPeers",
 388                                                 "WithVolatile=" + withVolatile,
 389                                                 "WithMetadata=" + withMetadata
 390                                         );
 391                                 }
 392
 393                                 private void sendEndOfPeerList() throws IOException {
 394                                         fcpServer.writeLine(
 395                                                 "EndListPeers",
 396                                                 "Identifier=" + identifier,
 397                                                 "EndMessage"
 398                                         );
 399                                 }
 400
 401                         }
 402
 403                         public class AddPeer {
 404
 405                                 @Test
 406                                 public void fromFile() throws InterruptedException, ExecutionException, IOException {
 407                                         Future<Optional<Peer>> peer = fcpClient.addPeer().fromFile(new File("/tmp/ref.txt")).execute();
 408                                         connectAndAssert(() -> allOf(matchesAddPeer(), hasItem("File=/tmp/ref.txt")));
 409                                         replyWithPeer("id1");
 410                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 411                                 }
 412
 413                                 @Test
 414                                 public void fromUrl() throws InterruptedException, ExecutionException, IOException {
 415                                         Future<Optional<Peer>> peer = fcpClient.addPeer().fromURL(new URL("http://node.ref/")).execute();
 416                                         connectAndAssert(() -> allOf(matchesAddPeer(), hasItem("URL=http://node.ref/")));
 417                                         replyWithPeer("id1");
 418                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 419                                 }
 420
 421                                 @Test
 422                                 public void fromNodeRef() throws InterruptedException, ExecutionException, IOException {
 423                                         NodeRef nodeRef = createNodeRef();
 424                                         Future<Optional<Peer>> peer = fcpClient.addPeer().fromNodeRef(nodeRef).execute();
 425                                         connectAndAssert(() -> allOf(matchesAddPeer(), Matchers.<String>hasItems(
 426                                                 "myName=name",
 427                                                 "ark.pubURI=public",
 428                                                 "ark.number=1",
 429                                                 "dsaGroup.g=base",
 430                                                 "dsaGroup.p=prime",
 431                                                 "dsaGroup.q=subprime",
 432                                                 "dsaPubKey.y=dsa-public",
 433                                                 "physical.udp=1.2.3.4:5678",
 434                                                 "auth.negTypes=3;5",
 435                                                 "sig=sig"
 436                                         )));
 437                                         replyWithPeer("id1");
 438                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 439                                 }
 440
 441                                 private NodeRef createNodeRef() {
 442                                         NodeRef nodeRef = new NodeRef();
 443                                         nodeRef.setIdentity("id1");
 444                                         nodeRef.setName("name");
 445                                         nodeRef.setARK(new ARK("public", "1"));
 446                                         nodeRef.setDSAGroup(new DSAGroup("base", "prime", "subprime"));
 447                                         nodeRef.setNegotiationTypes(new int[] { 3, 5 });
 448                                         nodeRef.setPhysicalUDP("1.2.3.4:5678");
 449                                         nodeRef.setDSAPublicKey("dsa-public");
 450                                         nodeRef.setSignature("sig");
 451                                         return nodeRef;
 452                                 }
 453
 454                                 private Matcher<List<String>> matchesAddPeer() {
 455                                         return matchesFcpMessage(
 456                                                 "AddPeer",
 457                                                 "Identifier=" + identifier
 458                                         );
 459                                 }
 460
 461                         }
 462
 463                         public class ModifyPeer {
 464
 465                                 @Test
 466                                 public void defaultFcpClientCanEnablePeerByName()
 467                                 throws InterruptedException, ExecutionException, IOException {
 468                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byName("id1").execute();
 469                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
 470                                         replyWithPeer("id1");
 471                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 472                                 }
 473
 474                                 @Test
 475                                 public void defaultFcpClientCanDisablePeerByName()
 476                                 throws InterruptedException, ExecutionException, IOException {
 477                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().disable().byName("id1").execute();
 478                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", true));
 479                                         replyWithPeer("id1");
 480                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 481                                 }
 482
 483                                 @Test
 484                                 public void defaultFcpClientCanEnablePeerByIdentity()
 485                                 throws InterruptedException, ExecutionException, IOException {
 486                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byIdentity("id1").execute();
 487                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
 488                                         replyWithPeer("id1");
 489                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 490                                 }
 491
 492                                 @Test
 493                                 public void defaultFcpClientCanEnablePeerByHostAndPort()
 494                                 throws InterruptedException, ExecutionException, IOException {
 495                                         Future<Optional<Peer>> peer =
 496                                                 fcpClient.modifyPeer().enable().byHostAndPort("1.2.3.4", 5678).execute();
 497                                         connectAndAssert(() -> matchesModifyPeer("1.2.3.4:5678", "IsDisabled", false));
 498                                         replyWithPeer("id1");
 499                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 500                                 }
 501
 502                                 @Test
 503                                 public void allowLocalAddressesOfPeer() throws InterruptedException, ExecutionException, IOException {
 504                                         Future<Optional<Peer>> peer =
 505                                                 fcpClient.modifyPeer().allowLocalAddresses().byIdentity("id1").execute();
 506                                         connectAndAssert(() -> allOf(
 507                                                 matchesModifyPeer("id1", "AllowLocalAddresses", true),
 508                                                 not(contains(startsWith("IsDisabled=")))
 509                                         ));
 510                                         replyWithPeer("id1");
 511                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 512                                 }
 513
 514                                 @Test
 515                                 public void disallowLocalAddressesOfPeer()
 516                                 throws InterruptedException, ExecutionException, IOException {
 517                                         Future<Optional<Peer>> peer =
 518                                                 fcpClient.modifyPeer().disallowLocalAddresses().byIdentity("id1").execute();
 519                                         connectAndAssert(() -> allOf(
 520                                                 matchesModifyPeer("id1", "AllowLocalAddresses", false),
 521                                                 not(contains(startsWith("IsDisabled=")))
 522                                         ));
 523                                         replyWithPeer("id1");
 524                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 525                                 }
 526
 527                                 @Test
 528                                 public void setBurstOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
 529                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().setBurstOnly().byIdentity("id1").execute();
 530                                         connectAndAssert(() -> allOf(
 531                                                 matchesModifyPeer("id1", "IsBurstOnly", true),
 532                                                 not(contains(startsWith("AllowLocalAddresses="))),
 533                                                 not(contains(startsWith("IsDisabled=")))
 534                                         ));
 535                                         replyWithPeer("id1");
 536                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 537                                 }
 538
 539                                 @Test
 540                                 public void clearBurstOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
 541                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().clearBurstOnly().byIdentity("id1").execute();
 542                                         connectAndAssert(() -> allOf(
 543                                                 matchesModifyPeer("id1", "IsBurstOnly", false),
 544                                                 not(contains(startsWith("AllowLocalAddresses="))),
 545                                                 not(contains(startsWith("IsDisabled=")))
 546                                         ));
 547                                         replyWithPeer("id1");
 548                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 549                                 }
 550
 551                                 @Test
 552                                 public void defaultFcpClientCanSetListenOnlyForPeer()
 553                                 throws InterruptedException, ExecutionException, IOException {
 554                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().setListenOnly().byIdentity("id1").execute();
 555                                         connectAndAssert(() -> allOf(
 556                                                 matchesModifyPeer("id1", "IsListenOnly", true),
 557                                                 not(contains(startsWith("AllowLocalAddresses="))),
 558                                                 not(contains(startsWith("IsDisabled="))),
 559                                                 not(contains(startsWith("IsBurstOnly=")))
 560                                         ));
 561                                         replyWithPeer("id1");
 562                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 563                                 }
 564
 565                                 @Test
 566                                 public void clearListenOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
 567                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().clearListenOnly().byIdentity("id1").execute();
 568                                         connectAndAssert(() -> allOf(
 569                                                 matchesModifyPeer("id1", "IsListenOnly", false),
 570                                                 not(contains(startsWith("AllowLocalAddresses="))),
 571                                                 not(contains(startsWith("IsDisabled="))),
 572                                                 not(contains(startsWith("IsBurstOnly=")))
 573                                         ));
 574                                         replyWithPeer("id1");
 575                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 576                                 }
 577
 578                                 @Test
 579                                 public void ignoreSourceForPeer() throws InterruptedException, ExecutionException, IOException {
 580                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().ignoreSource().byIdentity("id1").execute();
 581                                         connectAndAssert(() -> allOf(
 582                                                 matchesModifyPeer("id1", "IgnoreSourcePort", true),
 583                                                 not(contains(startsWith("AllowLocalAddresses="))),
 584                                                 not(contains(startsWith("IsDisabled="))),
 585                                                 not(contains(startsWith("IsBurstOnly="))),
 586                                                 not(contains(startsWith("IsListenOnly=")))
 587                                         ));
 588                                         replyWithPeer("id1");
 589                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 590                                 }
 591
 592                                 @Test
 593                                 public void useSourceForPeer() throws InterruptedException, ExecutionException, IOException {
 594                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().useSource().byIdentity("id1").execute();
 595                                         connectAndAssert(() -> allOf(
 596                                                 matchesModifyPeer("id1", "IgnoreSourcePort", false),
 597                                                 not(contains(startsWith("AllowLocalAddresses="))),
 598                                                 not(contains(startsWith("IsDisabled="))),
 599                                                 not(contains(startsWith("IsBurstOnly="))),
 600                                                 not(contains(startsWith("IsListenOnly=")))
 601                                         ));
 602                                         replyWithPeer("id1");
 603                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 604                                 }
 605
 606                                 @Test
 607                                 public void unknownNode() throws InterruptedException, ExecutionException, IOException {
 608                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byIdentity("id1").execute();
 609                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
 610                                         replyWithUnknownNodeIdentifier();
 611                                         assertThat(peer.get().isPresent(), is(false));
 612                                 }
 613
 614                                 private Matcher<List<String>> matchesModifyPeer(String nodeIdentifier, String setting, boolean value) {
 615                                         return matchesFcpMessage(
 616                                                 "ModifyPeer",
 617                                                 "Identifier=" + identifier,
 618                                                 "NodeIdentifier=" + nodeIdentifier,
 619                                                 setting + "=" + value
 620                                         );
 621                                 }
 622
 623                         }
 624
 625                         public class RemovePeer {
 626
 627                                 @Test
 628                                 public void byName() throws InterruptedException, ExecutionException, IOException {
 629                                         Future<Boolean> peer = fcpClient.removePeer().byName("Friend1").execute();
 630                                         connectAndAssert(() -> matchesRemovePeer("Friend1"));
 631                                         replyWithPeerRemoved("Friend1");
 632                                         assertThat(peer.get(), is(true));
 633                                 }
 634
 635                                 @Test
 636                                 public void invalidName() throws InterruptedException, ExecutionException, IOException {
 637                                         Future<Boolean> peer = fcpClient.removePeer().byName("NotFriend1").execute();
 638                                         connectAndAssert(() -> matchesRemovePeer("NotFriend1"));
 639                                         replyWithUnknownNodeIdentifier();
 640                                         assertThat(peer.get(), is(false));
 641                                 }
 642
 643                                 @Test
 644                                 public void byIdentity() throws InterruptedException, ExecutionException, IOException {
 645                                         Future<Boolean> peer = fcpClient.removePeer().byIdentity("id1").execute();
 646                                         connectAndAssert(() -> matchesRemovePeer("id1"));
 647                                         replyWithPeerRemoved("id1");
 648                                         assertThat(peer.get(), is(true));
 649                                 }
 650
 651                                 @Test
 652                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 653                                         Future<Boolean> peer = fcpClient.removePeer().byHostAndPort("1.2.3.4", 5678).execute();
 654                                         connectAndAssert(() -> matchesRemovePeer("1.2.3.4:5678"));
 655                                         replyWithPeerRemoved("Friend1");
 656                                         assertThat(peer.get(), is(true));
 657                                 }
 658
 659                                 private Matcher<List<String>> matchesRemovePeer(String nodeIdentifier) {
 660                                         return matchesFcpMessage(
 661                                                 "RemovePeer",
 662                                                 "Identifier=" + identifier,
 663                                                 "NodeIdentifier=" + nodeIdentifier
 664                                         );
 665                                 }
 666
 667                                 private void replyWithPeerRemoved(String nodeIdentifier) throws IOException {
 668                                         fcpServer.writeLine(
 669                                                 "PeerRemoved",
 670                                                 "Identifier=" + identifier,
 671                                                 "NodeIdentifier=" + nodeIdentifier,
 672                                                 "EndMessage"
 673                                         );
 674                                 }
 675
 676                         }
 677
 678                         private void replyWithPeer(String peerId, String... additionalLines) throws IOException {
 679                                 fcpServer.writeLine(
 680                                         "Peer",
 681                                         "Identifier=" + identifier,
 682                                         "identity=" + peerId,
 683                                         "opennet=false",
 684                                         "ark.pubURI=SSK@3YEf.../ark",
 685                                         "ark.number=78",
 686                                         "auth.negTypes=2",
 687                                         "version=Fred,0.7,1.0,1466",
 688                                         "lastGoodVersion=Fred,0.7,1.0,1466"
 689                                 );
 690                                 fcpServer.writeLine(additionalLines);
 691                                 fcpServer.writeLine("EndMessage");
 692                         }
 693
 694                 }
 695
 696                 public class PeerNoteCommands {
 697
 698                         public class ListPeerNotes {
 699
 700                                 @Test
 701                                 public void onUnknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 702                                         Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byName("Friend1").execute();
 703                                         connectAndAssert(() -> matchesListPeerNotes("Friend1"));
 704                                         replyWithUnknownNodeIdentifier();
 705                                         assertThat(peerNote.get().isPresent(), is(false));
 706                                 }
 707
 708                                 @Test
 709                                 public void byNodeName() throws InterruptedException, ExecutionException, IOException {
 710                                         Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byName("Friend1").execute();
 711                                         connectAndAssert(() -> matchesListPeerNotes("Friend1"));
 712                                         replyWithPeerNote();
 713                                         replyWithEndListPeerNotes();
 714                                         assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
 715                                         assertThat(peerNote.get().get().getPeerNoteType(), is(1));
 716                                 }
 717
 718                                 @Test
 719                                 public void byNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 720                                         Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byIdentity("id1").execute();
 721                                         connectAndAssert(() -> matchesListPeerNotes("id1"));
 722                                         replyWithPeerNote();
 723                                         replyWithEndListPeerNotes();
 724                                         assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
 725                                         assertThat(peerNote.get().get().getPeerNoteType(), is(1));
 726                                 }
 727
 728                                 @Test
 729                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 730                                         Future<Optional<PeerNote>> peerNote =
 731                                                 fcpClient.listPeerNotes().byHostAndPort("1.2.3.4", 5678).execute();
 732                                         connectAndAssert(() -> matchesListPeerNotes("1.2.3.4:5678"));
 733                                         replyWithPeerNote();
 734                                         replyWithEndListPeerNotes();
 735                                         assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
 736                                         assertThat(peerNote.get().get().getPeerNoteType(), is(1));
 737                                 }
 738
 739                                 private Matcher<List<String>> matchesListPeerNotes(String nodeIdentifier) {
 740                                         return matchesFcpMessage(
 741                                                 "ListPeerNotes",
 742                                                 "NodeIdentifier=" + nodeIdentifier
 743                                         );
 744                                 }
 745
 746                                 private void replyWithEndListPeerNotes() throws IOException {
 747                                         fcpServer.writeLine(
 748                                                 "EndListPeerNotes",
 749                                                 "Identifier=" + identifier,
 750                                                 "EndMessage"
 751                                         );
 752                                 }
 753
 754                                 private void replyWithPeerNote() throws IOException {
 755                                         fcpServer.writeLine(
 756                                                 "PeerNote",
 757                                                 "Identifier=" + identifier,
 758                                                 "NodeIdentifier=Friend1",
 759                                                 "NoteText=RXhhbXBsZSBUZXh0Lg==",
 760                                                 "PeerNoteType=1",
 761                                                 "EndMessage"
 762                                         );
 763                                 }
 764
 765                         }
 766
 767                         public class ModifyPeerNotes {
 768
 769                                 @Test
 770                                 public void byName() throws InterruptedException, ExecutionException, IOException {
 771                                         Future<Boolean> noteUpdated =
 772                                                 fcpClient.modifyPeerNote().darknetComment("foo").byName("Friend1").execute();
 773                                         connectAndAssert(() -> matchesModifyPeerNote("Friend1"));
 774                                         replyWithPeerNote();
 775                                         assertThat(noteUpdated.get(), is(true));
 776                                 }
 777
 778                                 @Test
 779                                 public void onUnknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 780                                         Future<Boolean> noteUpdated =
 781                                                 fcpClient.modifyPeerNote().darknetComment("foo").byName("Friend1").execute();
 782                                         connectAndAssert(() -> matchesModifyPeerNote("Friend1"));
 783                                         replyWithUnknownNodeIdentifier();
 784                                         assertThat(noteUpdated.get(), is(false));
 785                                 }
 786
 787                                 @Test
 788                                 public void defaultFcpClientFailsToModifyPeerNoteWithoutPeerNote()
 789                                 throws InterruptedException, ExecutionException, IOException {
 790                                         Future<Boolean> noteUpdated = fcpClient.modifyPeerNote().byName("Friend1").execute();
 791                                         assertThat(noteUpdated.get(), is(false));
 792                                 }
 793
 794                                 @Test
 795                                 public void byIdentifier() throws InterruptedException, ExecutionException, IOException {
 796                                         Future<Boolean> noteUpdated =
 797                                                 fcpClient.modifyPeerNote().darknetComment("foo").byIdentifier("id1").execute();
 798                                         connectAndAssert(() -> matchesModifyPeerNote("id1"));
 799                                         replyWithPeerNote();
 800                                         assertThat(noteUpdated.get(), is(true));
 801                                 }
 802
 803                                 @Test
 804                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 805                                         Future<Boolean> noteUpdated =
 806                                                 fcpClient.modifyPeerNote().darknetComment("foo").byHostAndPort("1.2.3.4", 5678).execute();
 807                                         connectAndAssert(() -> matchesModifyPeerNote("1.2.3.4:5678"));
 808                                         replyWithPeerNote();
 809                                         assertThat(noteUpdated.get(), is(true));
 810                                 }
 811
 812                                 private Matcher<List<String>> matchesModifyPeerNote(String nodeIdentifier) {
 813                                         return matchesFcpMessage(
 814                                                 "ModifyPeerNote",
 815                                                 "Identifier=" + identifier,
 816                                                 "NodeIdentifier=" + nodeIdentifier,
 817                                                 "PeerNoteType=1",
 818                                                 "NoteText=Zm9v"
 819                                         );
 820                                 }
 821
 822                                 private void replyWithPeerNote() throws IOException {
 823                                         fcpServer.writeLine(
 824                                                 "PeerNote",
 825                                                 "Identifier=" + identifier,
 826                                                 "NodeIdentifier=Friend1",
 827                                                 "NoteText=Zm9v",
 828                                                 "PeerNoteType=1",
 829                                                 "EndMessage"
 830                                         );
 831                                 }
 832
 833                         }
 834
 835                 }
 836
 837                 private void replyWithUnknownNodeIdentifier() throws IOException {
 838                         fcpServer.writeLine(
 839                                 "UnknownNodeIdentifier",
 840                                 "Identifier=" + identifier,
 841                                 "NodeIdentifier=id2",
 842                                 "EndMessage"
 843                         );
 844                 }
 845
 846         }
 847
 848         public class PluginCommands {
 849
 850                 private static final String CLASS_NAME = "foo.plugin.Plugin";
 851
 852                 private void replyWithPluginInfo() throws IOException {
 853                         fcpServer.writeLine(
 854                                 "PluginInfo",
 855                                 "Identifier=" + identifier,
 856                                 "PluginName=superPlugin",
 857                                 "IsTalkable=true",
 858                                 "LongVersion=1.2.3",
 859                                 "Version=42",
 860                                 "OriginUri=superPlugin",
 861                                 "Started=true",
 862                                 "EndMessage"
 863                         );
 864                 }
 865
 866                 private void verifyPluginInfo(Future<Optional<PluginInfo>> pluginInfo)
 867                 throws InterruptedException, ExecutionException {
 868                         assertThat(pluginInfo.get().get().getPluginName(), is("superPlugin"));
 869                         assertThat(pluginInfo.get().get().getOriginalURI(), is("superPlugin"));
 870                         assertThat(pluginInfo.get().get().isTalkable(), is(true));
 871                         assertThat(pluginInfo.get().get().getVersion(), is("42"));
 872                         assertThat(pluginInfo.get().get().getLongVersion(), is("1.2.3"));
 873                         assertThat(pluginInfo.get().get().isStarted(), is(true));
 874                 }
 875
 876                 public class LoadPlugin {
 877
 878                         public class OfficialPlugins {
 879
 880                                 @Test
 881                                 public void fromFreenet() throws ExecutionException, InterruptedException, IOException {
 882                                         Future<Optional<PluginInfo>> pluginInfo =
 883                                                 fcpClient.loadPlugin().officialFromFreenet("superPlugin").execute();
 884                                         connectAndAssert(() -> createMatcherForOfficialSource("freenet"));
 885                                         assertThat(lines, not(contains(startsWith("Store="))));
 886                                         replyWithPluginInfo();
 887                                         verifyPluginInfo(pluginInfo);
 888                                 }
 889
 890                                 @Test
 891                                 public void persistentFromFreenet() throws ExecutionException, InterruptedException, IOException {
 892                                         Future<Optional<PluginInfo>> pluginInfo =
 893                                                 fcpClient.loadPlugin().addToConfig().officialFromFreenet("superPlugin").execute();
 894                                         connectAndAssert(() -> createMatcherForOfficialSource("freenet"));
 895                                         assertThat(lines, hasItem("Store=true"));
 896                                         replyWithPluginInfo();
 897                                         verifyPluginInfo(pluginInfo);
 898                                 }
 899
 900                                 @Test
 901                                 public void fromHttps() throws ExecutionException, InterruptedException, IOException {
 902                                         Future<Optional<PluginInfo>> pluginInfo =
 903                                                 fcpClient.loadPlugin().officialFromHttps("superPlugin").execute();
 904                                         connectAndAssert(() -> createMatcherForOfficialSource("https"));
 905                                         replyWithPluginInfo();
 906                                         verifyPluginInfo(pluginInfo);
 907                                 }
 908
 909                                 private Matcher<List<String>> createMatcherForOfficialSource(String officialSource) {
 910                                         return matchesFcpMessage(
 911                                                 "LoadPlugin",
 912                                                 "Identifier=" + identifier,
 913                                                 "PluginURL=superPlugin",
 914                                                 "URLType=official",
 915                                                 "OfficialSource=" + officialSource
 916                                         );
 917                                 }
 918
 919                         }
 920
 921                         public class FromOtherSources {
 922
 923                                 private static final String FILE_PATH = "/path/to/plugin.jar";
 924                                 private static final String URL = "http://server.com/plugin.jar";
 925                                 private static final String KEY = "KSK@plugin.jar";
 926
 927                                 @Test
 928                                 public void fromFile() throws ExecutionException, InterruptedException, IOException {
 929                                         Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromFile(FILE_PATH).execute();
 930                                         connectAndAssert(() -> createMatcher("file", FILE_PATH));
 931                                         replyWithPluginInfo();
 932                                         verifyPluginInfo(pluginInfo);
 933                                 }
 934
 935                                 @Test
 936                                 public void fromUrl() throws ExecutionException, InterruptedException, IOException {
 937                                         Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromUrl(URL).execute();
 938                                         connectAndAssert(() -> createMatcher("url", URL));
 939                                         replyWithPluginInfo();
 940                                         verifyPluginInfo(pluginInfo);
 941                                 }
 942
 943                                 @Test
 944                                 public void fromFreenet() throws ExecutionException, InterruptedException, IOException {
 945                                         Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromFreenet(KEY).execute();
 946                                         connectAndAssert(() -> createMatcher("freenet", KEY));
 947                                         replyWithPluginInfo();
 948                                         verifyPluginInfo(pluginInfo);
 949                                 }
 950
 951                                 private Matcher<List<String>> createMatcher(String urlType, String url) {
 952                                         return matchesFcpMessage(
 953                                                 "LoadPlugin",
 954                                                 "Identifier=" + identifier,
 955                                                 "PluginURL=" + url,
 956                                                 "URLType=" + urlType
 957                                         );
 958                                 }
 959
 960                         }
 961
 962                         public class Failed {
 963
 964                                 @Test
 965                                 public void failedLoad() throws ExecutionException, InterruptedException, IOException {
 966                                         Future<Optional<PluginInfo>> pluginInfo =
 967                                                 fcpClient.loadPlugin().officialFromFreenet("superPlugin").execute();
 968                                         connectAndAssert(() -> matchesFcpMessage("LoadPlugin"));
 969                                         replyWithProtocolError();
 970                                         assertThat(pluginInfo.get().isPresent(), is(false));
 971                                 }
 972
 973                         }
 974
 975                 }
 976
 977                 private void replyWithProtocolError() throws IOException {
 978                         fcpServer.writeLine(
 979                                 "ProtocolError",
 980                                 "Identifier=" + identifier,
 981                                 "EndMessage"
 982                         );
 983                 }
 984
 985                 public class ReloadPlugin {
 986
 987                         @Test
 988                         public void reloadingPluginWorks() throws InterruptedException, ExecutionException, IOException {
 989                                 Future<Optional<PluginInfo>> pluginInfo = fcpClient.reloadPlugin().plugin(CLASS_NAME).execute();
 990                                 connectAndAssert(this::matchReloadPluginMessage);
 991                                 replyWithPluginInfo();
 992                                 verifyPluginInfo(pluginInfo);
 993                         }
 994
 995                         @Test
 996                         public void reloadingPluginWithMaxWaitTimeWorks()
 997                         throws InterruptedException, ExecutionException, IOException {
 998                                 Future<Optional<PluginInfo>> pluginInfo =
 999                                         fcpClient.reloadPlugin().waitFor(1234).plugin(CLASS_NAME).execute();
1000                                 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("MaxWaitTime=1234")));
1001                                 replyWithPluginInfo();
1002                                 verifyPluginInfo(pluginInfo);
1003                         }
1004
1005                         @Test
1006                         public void reloadingPluginWithPurgeWorks()
1007                         throws InterruptedException, ExecutionException, IOException {
1008                                 Future<Optional<PluginInfo>> pluginInfo =
1009                                         fcpClient.reloadPlugin().purge().plugin(CLASS_NAME).execute();
1010                                 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("Purge=true")));
1011                                 replyWithPluginInfo();
1012                                 verifyPluginInfo(pluginInfo);
1013                         }
1014
1015                         @Test
1016                         public void reloadingPluginWithStoreWorks()
1017                         throws InterruptedException, ExecutionException, IOException {
1018                                 Future<Optional<PluginInfo>> pluginInfo =
1019                                         fcpClient.reloadPlugin().addToConfig().plugin(CLASS_NAME).execute();
1020                                 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("Store=true")));
1021                                 replyWithPluginInfo();
1022                                 verifyPluginInfo(pluginInfo);
1023                         }
1024
1025                         private Matcher<List<String>> matchReloadPluginMessage() {
1026                                 return matchesFcpMessage(
1027                                         "ReloadPlugin",
1028                                         "Identifier=" + identifier,
1029                                         "PluginName=" + CLASS_NAME
1030                                 );
1031                         }
1032
1033                 }
1034
1035                 public class RemovePlugin {
1036
1037                         @Test
1038                         public void removingPluginWorks() throws InterruptedException, ExecutionException, IOException {
1039                                 Future<Boolean> pluginRemoved = fcpClient.removePlugin().plugin(CLASS_NAME).execute();
1040                                 connectAndAssert(this::matchPluginRemovedMessage);
1041                                 replyWithPluginRemoved();
1042                                 assertThat(pluginRemoved.get(), is(true));
1043                         }
1044
1045                         @Test
1046                         public void removingPluginWithMaxWaitTimeWorks()
1047                         throws InterruptedException, ExecutionException, IOException {
1048                                 Future<Boolean> pluginRemoved = fcpClient.removePlugin().waitFor(1234).plugin(CLASS_NAME).execute();
1049                                 connectAndAssert(() -> allOf(matchPluginRemovedMessage(), hasItem("MaxWaitTime=1234")));
1050                                 replyWithPluginRemoved();
1051                                 assertThat(pluginRemoved.get(), is(true));
1052                         }
1053
1054                         @Test
1055                         public void removingPluginWithPurgeWorks()
1056                         throws InterruptedException, ExecutionException, IOException {
1057                                 Future<Boolean> pluginRemoved = fcpClient.removePlugin().purge().plugin(CLASS_NAME).execute();
1058                                 connectAndAssert(() -> allOf(matchPluginRemovedMessage(), hasItem("Purge=true")));
1059                                 replyWithPluginRemoved();
1060                                 assertThat(pluginRemoved.get(), is(true));
1061                         }
1062
1063                         private void replyWithPluginRemoved() throws IOException {
1064                                 fcpServer.writeLine(
1065                                         "PluginRemoved",
1066                                         "Identifier=" + identifier,
1067                                         "PluginName=" + CLASS_NAME,
1068                                         "EndMessage"
1069                                 );
1070                         }
1071
1072                         private Matcher<List<String>> matchPluginRemovedMessage() {
1073                                 return matchesFcpMessage(
1074                                         "RemovePlugin",
1075                                         "Identifier=" + identifier,
1076                                         "PluginName=" + CLASS_NAME
1077                                 );
1078                         }
1079
1080                 }
1081
1082                 public class GetPluginInfo {
1083
1084                         @Test
1085                         public void gettingPluginInfoWorks() throws InterruptedException, ExecutionException, IOException {
1086                                 Future<Optional<PluginInfo>> pluginInfo = fcpClient.getPluginInfo().plugin(CLASS_NAME).execute();
1087                                 connectAndAssert(this::matchGetPluginInfoMessage);
1088                                 replyWithPluginInfo();
1089                                 verifyPluginInfo(pluginInfo);
1090                         }
1091
1092                         @Test
1093                         public void gettingPluginInfoWithDetailsWorks()
1094                         throws InterruptedException, ExecutionException, IOException {
1095                                 Future<Optional<PluginInfo>> pluginInfo =
1096                                         fcpClient.getPluginInfo().detailed().plugin(CLASS_NAME).execute();
1097                                 connectAndAssert(() -> allOf(matchGetPluginInfoMessage(), hasItem("Detailed=true")));
1098                                 replyWithPluginInfo();
1099                                 verifyPluginInfo(pluginInfo);
1100                         }
1101
1102                         @Test
1103                         public void protocolErrorIsRecognizedAsFailure()
1104                         throws InterruptedException, ExecutionException, IOException {
1105                                 Future<Optional<PluginInfo>> pluginInfo =
1106                                         fcpClient.getPluginInfo().detailed().plugin(CLASS_NAME).execute();
1107                                 connectAndAssert(() -> allOf(matchGetPluginInfoMessage(), hasItem("Detailed=true")));
1108                                 replyWithProtocolError();
1109                                 assertThat(pluginInfo.get(), is(Optional.empty()));
1110                         }
1111
1112                         private Matcher<List<String>> matchGetPluginInfoMessage() {
1113                                 return matchesFcpMessage(
1114                                         "GetPluginInfo",
1115                                         "Identifier=" + identifier,
1116                                         "PluginName=" + CLASS_NAME
1117                                 );
1118                         }
1119
1120                 }
1121
1122         }
1123
1124         public class UskSubscriptionCommands {
1125
1126                 private static final String URI = "USK@some,uri/file.txt";
1127
1128                 @Test
1129                 public void subscriptionWorks() throws InterruptedException, ExecutionException, IOException {
1130                         Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1131                         connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1132                         replyWithSubscribed();
1133                         assertThat(uskSubscription.get().get().getUri(), is(URI));
1134                         AtomicInteger edition = new AtomicInteger();
1135                         CountDownLatch updated = new CountDownLatch(2);
1136                         uskSubscription.get().get().onUpdate(e -> {
1137                                 edition.set(e);
1138                                 updated.countDown();
1139                         });
1140                         sendUpdateNotification(23);
1141                         sendUpdateNotification(24);
1142                         assertThat("updated in time", updated.await(5, TimeUnit.SECONDS), is(true));
1143                         assertThat(edition.get(), is(24));
1144                 }
1145
1146                 @Test
1147                 public void subscriptionUpdatesMultipleTimes() throws InterruptedException, ExecutionException, IOException {
1148                         Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1149                         connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1150                         replyWithSubscribed();
1151                         assertThat(uskSubscription.get().get().getUri(), is(URI));
1152                         AtomicInteger edition = new AtomicInteger();
1153                         CountDownLatch updated = new CountDownLatch(2);
1154                         uskSubscription.get().get().onUpdate(e -> {
1155                                 edition.set(e);
1156                                 updated.countDown();
1157                         });
1158                         uskSubscription.get().get().onUpdate(e -> updated.countDown());
1159                         sendUpdateNotification(23);
1160                         assertThat("updated in time", updated.await(5, TimeUnit.SECONDS), is(true));
1161                         assertThat(edition.get(), is(23));
1162                 }
1163
1164                 @Test
1165                 public void subscriptionCanBeCancelled() throws InterruptedException, ExecutionException, IOException {
1166                         Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1167                         connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1168                         replyWithSubscribed();
1169                         assertThat(uskSubscription.get().get().getUri(), is(URI));
1170                         AtomicBoolean updated = new AtomicBoolean();
1171                         uskSubscription.get().get().onUpdate(e -> updated.set(true));
1172                         uskSubscription.get().get().cancel();
1173                         readMessage(() -> matchesFcpMessage("UnsubscribeUSK", "Identifier=" + identifier));
1174                         sendUpdateNotification(23);
1175                         assertThat(updated.get(), is(false));
1176                 }
1177
1178                 private void replyWithSubscribed() throws IOException {
1179                         fcpServer.writeLine(
1180                                 "SubscribedUSK",
1181                                 "Identifier=" + identifier,
1182                                 "URI=" + URI,
1183                                 "DontPoll=false",
1184                                 "EndMessage"
1185                         );
1186                 }
1187
1188                 private void sendUpdateNotification(int edition, String... additionalLines) throws IOException {
1189                         fcpServer.writeLine(
1190                                 "SubscribedUSKUpdate",
1191                                 "Identifier=" + identifier,
1192                                 "URI=" + URI,
1193                                 "Edition=" + edition
1194                         );
1195                         fcpServer.writeLine(additionalLines);
1196                         fcpServer.writeLine("EndMessage");
1197                 }
1198
1199         }
1200
1201         public class ClientGet {
1202
1203                 @Test
1204                 public void works() throws InterruptedException, ExecutionException, IOException {
1205                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1206                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "ReturnType=direct"));
1207                         replyWithAllData("not-test", "Hello World", "text/plain;charset=latin-9");
1208                         replyWithAllData(identifier, "Hello", "text/plain;charset=utf-8");
1209                         Optional<Data> data = dataFuture.get();
1210                         verifyData(data);
1211                 }
1212
1213                 @Test
1214                 public void getFailedIsRecognized() throws InterruptedException, ExecutionException, IOException {
1215                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1216                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1217                         replyWithGetFailed("not-test");
1218                         replyWithGetFailed(identifier);
1219                         Optional<Data> data = dataFuture.get();
1220                         assertThat(data.isPresent(), is(false));
1221                 }
1222
1223                 @Test
1224                 public void getFailedForDifferentIdentifierIsIgnored()
1225                 throws InterruptedException, ExecutionException, IOException {
1226                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1227                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1228                         replyWithGetFailed("not-test");
1229                         replyWithAllData(identifier, "Hello", "text/plain;charset=utf-8");
1230                         Optional<Data> data = dataFuture.get();
1231                         verifyData(data);
1232                 }
1233
1234                 @Test(expected = ExecutionException.class)
1235                 public void connectionClosedIsRecognized() throws InterruptedException, ExecutionException, IOException {
1236                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1237                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1238                         fcpServer.close();
1239                         dataFuture.get();
1240                 }
1241
1242                 @Test
1243                 public void withIgnoreData() throws InterruptedException, ExecutionException, IOException {
1244                         fcpClient.clientGet().ignoreDataStore().uri("KSK@foo.txt").execute();
1245                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "IgnoreDS=true"));
1246                 }
1247
1248                 @Test
1249                 public void withDataStoreOnly() throws InterruptedException, ExecutionException, IOException {
1250                         fcpClient.clientGet().dataStoreOnly().uri("KSK@foo.txt").execute();
1251                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "DSonly=true"));
1252                 }
1253
1254                 @Test
1255                 public void clientGetWithMaxSizeSettingSendsCorrectCommands()
1256                 throws InterruptedException, ExecutionException, IOException {
1257                         fcpClient.clientGet().maxSize(1048576).uri("KSK@foo.txt").execute();
1258                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "MaxSize=1048576"));
1259                 }
1260
1261                 @Test
1262                 public void clientGetWithPrioritySettingSendsCorrectCommands()
1263                 throws InterruptedException, ExecutionException, IOException {
1264                         fcpClient.clientGet().priority(Priority.interactive).uri("KSK@foo.txt").execute();
1265                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "PriorityClass=1"));
1266                 }
1267
1268                 @Test
1269                 public void clientGetWithRealTimeSettingSendsCorrectCommands()
1270                 throws InterruptedException, ExecutionException, IOException {
1271                         fcpClient.clientGet().realTime().uri("KSK@foo.txt").execute();
1272                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "RealTimeFlag=true"));
1273                 }
1274
1275                 @Test
1276                 public void clientGetWithGlobalSettingSendsCorrectCommands()
1277                 throws InterruptedException, ExecutionException, IOException {
1278                         fcpClient.clientGet().global().uri("KSK@foo.txt").execute();
1279                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "Global=true"));
1280                 }
1281
1282                 private void replyWithGetFailed(String identifier) throws IOException {
1283                         fcpServer.writeLine(
1284                                 "GetFailed",
1285                                 "Identifier=" + identifier,
1286                                 "Code=3",
1287                                 "EndMessage"
1288                         );
1289                 }
1290
1291                 private void replyWithAllData(String identifier, String text, String contentType) throws IOException {
1292                         fcpServer.writeLine(
1293                                 "AllData",
1294                                 "Identifier=" + identifier,
1295                                 "DataLength=" + (text.length() + 1),
1296                                 "StartupTime=1435610539000",
1297                                 "CompletionTime=1435610540000",
1298                                 "Metadata.ContentType=" + contentType,
1299                                 "Data",
1300                                 text
1301                         );
1302                 }
1303
1304                 private void verifyData(Optional<Data> data) throws IOException {
1305                         assertThat(data.get().getMimeType(), is("text/plain;charset=utf-8"));
1306                         assertThat(data.get().size(), is(6L));
1307                         assertThat(ByteStreams.toByteArray(data.get().getInputStream()),
1308                                 is("Hello\n".getBytes(StandardCharsets.UTF_8)));
1309                 }
1310
1311         }
1312
1313         public class ClientPut {
1314
1315                 @Test
1316                 public void sendsCorrectCommand() throws IOException, ExecutionException, InterruptedException {
1317                         fcpClient.clientPut()
1318                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1319                                 .length(6)
1320                                 .uri("KSK@foo.txt")
1321                                 .execute();
1322                         connectNode();
1323                         readMessage("Hello", this::matchesDirectClientPut);
1324                 }
1325
1326                 @Test
1327                 public void succeedsOnCorrectIdentifier() throws InterruptedException, ExecutionException, IOException {
1328                         Future<Optional<Key>> key = fcpClient.clientPut()
1329                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1330                                 .length(6)
1331                                 .uri("KSK@foo.txt")
1332                                 .execute();
1333                         connectNode();
1334                         readMessage("Hello", this::matchesDirectClientPut);
1335                         replyWithPutFailed("not-the-right-one");
1336                         replyWithPutSuccessful(identifier);
1337                         assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1338                 }
1339
1340                 @Test
1341                 public void failsOnCorrectIdentifier() throws InterruptedException, ExecutionException, IOException {
1342                         Future<Optional<Key>> key = fcpClient.clientPut()
1343                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1344                                 .length(6)
1345                                 .uri("KSK@foo.txt")
1346                                 .execute();
1347                         connectNode();
1348                         readMessage("Hello", this::matchesDirectClientPut);
1349                         replyWithPutSuccessful("not-the-right-one");
1350                         replyWithPutFailed(identifier);
1351                         assertThat(key.get().isPresent(), is(false));
1352                 }
1353
1354                 @Test
1355                 public void renameIsSentCorrectly() throws InterruptedException, ExecutionException, IOException {
1356                         fcpClient.clientPut()
1357                                 .named("otherName.txt")
1358                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1359                                 .length(6)
1360                                 .uri("KSK@foo.txt")
1361                                 .execute();
1362                         connectNode();
1363                         readMessage("Hello", () -> allOf(
1364                                 hasHead("ClientPut"),
1365                                 hasParameters(1, 2, "TargetFilename=otherName.txt", "UploadFrom=direct", "DataLength=6",
1366                                         "URI=KSK@foo.txt"),
1367                                 hasTail("EndMessage", "Hello")
1368                         ));
1369                 }
1370
1371                 @Test
1372                 public void redirectIsSentCorrecly() throws IOException, ExecutionException, InterruptedException {
1373                         fcpClient.clientPut().redirectTo("KSK@bar.txt").uri("KSK@foo.txt").execute();
1374                         connectAndAssert(() ->
1375                                 matchesFcpMessage("ClientPut", "UploadFrom=redirect", "URI=KSK@foo.txt", "TargetURI=KSK@bar.txt"));
1376                 }
1377
1378                 @Test
1379                 public void withFileIsSentCorrectly() throws InterruptedException, ExecutionException, IOException {
1380                         fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1381                         connectAndAssert(() ->
1382                                 matchesFcpMessage("ClientPut", "UploadFrom=disk", "URI=KSK@foo.txt", "Filename=/tmp/data.txt"));
1383                 }
1384
1385                 public class DDA {
1386
1387                         private final File ddaFile;
1388                         private final File fileToUpload;
1389
1390                         public DDA() throws IOException {
1391                                 ddaFile = createDdaFile();
1392                                 fileToUpload = new File(ddaFile.getParent(), "test.dat");
1393                         }
1394
1395                         private Matcher<List<String>> matchesFileClientPut(File file) {
1396                                 return matchesFcpMessage("ClientPut", "UploadFrom=disk", "URI=KSK@foo.txt", "Filename=" + file);
1397                         }
1398
1399                         @Test
1400                         public void completeDda() throws IOException, ExecutionException, InterruptedException {
1401                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1402                                 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1403                                 sendDdaRequired(identifier);
1404                                 readMessage(() -> matchesTestDDARequest(ddaFile));
1405                                 sendTestDDAReply(ddaFile.getParent(), ddaFile);
1406                                 readMessage(() -> matchesTestDDAResponse(ddaFile));
1407                                 writeTestDDAComplete(ddaFile);
1408                                 readMessage(() -> matchesFileClientPut(fileToUpload));
1409                         }
1410
1411                         @Test
1412                         public void ignoreOtherDda() throws IOException, ExecutionException, InterruptedException {
1413                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1414                                 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1415                                 sendDdaRequired(identifier);
1416                                 readMessage(() -> matchesTestDDARequest(ddaFile));
1417                                 sendTestDDAReply("/some-other-directory", ddaFile);
1418                                 sendTestDDAReply(ddaFile.getParent(), ddaFile);
1419                                 readMessage(() -> matchesTestDDAResponse(ddaFile));
1420                         }
1421
1422                         @Test
1423                         public void sendResponseIfFileUnreadable() throws IOException, ExecutionException, InterruptedException {
1424                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1425                                 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1426                                 sendDdaRequired(identifier);
1427                                 readMessage(() -> matchesTestDDARequest(ddaFile));
1428                                 sendTestDDAReply(ddaFile.getParent(), new File(ddaFile + ".foo"));
1429                                 readMessage(this::matchesFailedToReadResponse);
1430                         }
1431
1432                         @Test
1433                         public void clientPutDoesNotResendOriginalClientPutOnTestDDACompleteWithWrongDirectory()
1434                         throws IOException, ExecutionException, InterruptedException {
1435                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1436                                 connectNode();
1437                                 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1438                                 String identifier = extractIdentifier(lines);
1439                                 fcpServer.writeLine(
1440                                         "TestDDAComplete",
1441                                         "Directory=/some-other-directory",
1442                                         "EndMessage"
1443                                 );
1444                                 sendDdaRequired(identifier);
1445                                 lines = fcpServer.collectUntil(is("EndMessage"));
1446                                 assertThat(lines, matchesFcpMessage(
1447                                         "TestDDARequest",
1448                                         "Directory=" + ddaFile.getParent(),
1449                                         "WantReadDirectory=true",
1450                                         "WantWriteDirectory=false"
1451                                 ));
1452                         }
1453
1454                         private Matcher<List<String>> matchesFailedToReadResponse() {
1455                                 return matchesFcpMessage(
1456                                         "TestDDAResponse",
1457                                         "Directory=" + ddaFile.getParent(),
1458                                         "ReadContent=failed-to-read"
1459                                 );
1460                         }
1461
1462                         private void writeTestDDAComplete(File tempFile) throws IOException {
1463                                 fcpServer.writeLine(
1464                                         "TestDDAComplete",
1465                                         "Directory=" + tempFile.getParent(),
1466                                         "ReadDirectoryAllowed=true",
1467                                         "EndMessage"
1468                                 );
1469                         }
1470
1471                         private Matcher<List<String>> matchesTestDDAResponse(File tempFile) {
1472                                 return matchesFcpMessage(
1473                                         "TestDDAResponse",
1474                                         "Directory=" + tempFile.getParent(),
1475                                         "ReadContent=test-content"
1476                                 );
1477                         }
1478
1479                         private void sendTestDDAReply(String directory, File tempFile) throws IOException {
1480                                 fcpServer.writeLine(
1481                                         "TestDDAReply",
1482                                         "Directory=" + directory,
1483                                         "ReadFilename=" + tempFile,
1484                                         "EndMessage"
1485                                 );
1486                         }
1487
1488                         private Matcher<List<String>> matchesTestDDARequest(File tempFile) {
1489                                 return matchesFcpMessage(
1490                                         "TestDDARequest",
1491                                         "Directory=" + tempFile.getParent(),
1492                                         "WantReadDirectory=true",
1493                                         "WantWriteDirectory=false"
1494                                 );
1495                         }
1496
1497                         private void sendDdaRequired(String identifier) throws IOException {
1498                                 fcpServer.writeLine(
1499                                         "ProtocolError",
1500                                         "Identifier=" + identifier,
1501                                         "Code=25",
1502                                         "EndMessage"
1503                                 );
1504                         }
1505
1506                 }
1507
1508                 private void replyWithPutSuccessful(String identifier) throws IOException {
1509                         fcpServer.writeLine(
1510                                 "PutSuccessful",
1511                                 "URI=KSK@foo.txt",
1512                                 "Identifier=" + identifier,
1513                                 "EndMessage"
1514                         );
1515                 }
1516
1517                 private void replyWithPutFailed(String identifier) throws IOException {
1518                         fcpServer.writeLine(
1519                                 "PutFailed",
1520                                 "Identifier=" + identifier,
1521                                 "EndMessage"
1522                         );
1523                 }
1524
1525                 private Matcher<List<String>> matchesDirectClientPut() {
1526                         return allOf(
1527                                 hasHead("ClientPut"),
1528                                 hasParameters(1, 2, "UploadFrom=direct", "DataLength=6", "URI=KSK@foo.txt"),
1529                                 hasTail("EndMessage", "Hello")
1530                         );
1531                 }
1532
1533                 private File createDdaFile() throws IOException {
1534                         File tempFile = File.createTempFile("test-dda-", ".dat");
1535                         tempFile.deleteOnExit();
1536                         Files.write("test-content", tempFile, StandardCharsets.UTF_8);
1537                         return tempFile;
1538                 }
1539
1540                 @Test
1541                 public void clientPutDoesNotReactToProtocolErrorForDifferentIdentifier()
1542                 throws InterruptedException, ExecutionException, IOException {
1543                         Future<Optional<Key>> key = fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1544                         connectNode();
1545                         List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1546                         String identifier = extractIdentifier(lines);
1547                         fcpServer.writeLine(
1548                                 "ProtocolError",
1549                                 "Identifier=not-the-right-one",
1550                                 "Code=25",
1551                                 "EndMessage"
1552                         );
1553                         fcpServer.writeLine(
1554                                 "PutSuccessful",
1555                                 "Identifier=" + identifier,
1556                                 "URI=KSK@foo.txt",
1557                                 "EndMessage"
1558                         );
1559                         assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1560                 }
1561
1562                 @Test
1563                 public void clientPutAbortsOnProtocolErrorOtherThan25()
1564                 throws InterruptedException, ExecutionException, IOException {
1565                         Future<Optional<Key>> key = fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1566                         connectNode();
1567                         List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1568                         String identifier = extractIdentifier(lines);
1569                         fcpServer.writeLine(
1570                                 "ProtocolError",
1571                                 "Identifier=" + identifier,
1572                                 "Code=1",
1573                                 "EndMessage"
1574                         );
1575                         assertThat(key.get().isPresent(), is(false));
1576                 }
1577
1578                 @Test
1579                 public void clientPutSendsNotificationsForGeneratedKeys()
1580                 throws InterruptedException, ExecutionException, IOException {
1581                         List<String> generatedKeys = new CopyOnWriteArrayList<>();
1582                         Future<Optional<Key>> key = fcpClient.clientPut()
1583                                 .onKeyGenerated(generatedKeys::add)
1584                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1585                                 .length(6)
1586                                 .uri("KSK@foo.txt")
1587                                 .execute();
1588                         connectNode();
1589                         List<String> lines = fcpServer.collectUntil(is("Hello"));
1590                         String identifier = extractIdentifier(lines);
1591                         fcpServer.writeLine(
1592                                 "URIGenerated",
1593                                 "Identifier=" + identifier,
1594                                 "URI=KSK@foo.txt",
1595                                 "EndMessage"
1596                         );
1597                         replyWithPutSuccessful(identifier);
1598                         assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1599                         assertThat(generatedKeys, contains("KSK@foo.txt"));
1600                 }
1601
1602         }
1603
1604         public class ConfigCommand {
1605
1606                 public class GetConfig {
1607
1608                         @Test
1609                         public void defaultFcpClientCanGetConfigWithoutDetails()
1610                         throws InterruptedException, ExecutionException, IOException {
1611                                 Future<ConfigData> configData = fcpClient.getConfig().execute();
1612                                 connectAndAssert(() -> matchesFcpMessage("GetConfig", "Identifier=" + identifier));
1613                                 replyWithConfigData();
1614                                 assertThat(configData.get(), notNullValue());
1615                         }
1616
1617                         @Test
1618                         public void defaultFcpClientCanGetConfigWithCurrent()
1619                         throws InterruptedException, ExecutionException, IOException {
1620                                 Future<ConfigData> configData = fcpClient.getConfig().withCurrent().execute();
1621                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithCurrent"));
1622                                 replyWithConfigData("current.foo=bar");
1623                                 assertThat(configData.get().getCurrent("foo"), is("bar"));
1624                         }
1625
1626                         @Test
1627                         public void defaultFcpClientCanGetConfigWithDefaults()
1628                         throws InterruptedException, ExecutionException, IOException {
1629                                 Future<ConfigData> configData = fcpClient.getConfig().withDefaults().execute();
1630                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithDefaults"));
1631                                 replyWithConfigData("default.foo=bar");
1632                                 assertThat(configData.get().getDefault("foo"), is("bar"));
1633                         }
1634
1635                         @Test
1636                         public void defaultFcpClientCanGetConfigWithSortOrder()
1637                         throws InterruptedException, ExecutionException, IOException {
1638                                 Future<ConfigData> configData = fcpClient.getConfig().withSortOrder().execute();
1639                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithSortOrder"));
1640                                 replyWithConfigData("sortOrder.foo=17");
1641                                 assertThat(configData.get().getSortOrder("foo"), is(17));
1642                         }
1643
1644                         @Test
1645                         public void defaultFcpClientCanGetConfigWithExpertFlag()
1646                         throws InterruptedException, ExecutionException, IOException {
1647                                 Future<ConfigData> configData = fcpClient.getConfig().withExpertFlag().execute();
1648                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithExpertFlag"));
1649                                 replyWithConfigData("expertFlag.foo=true");
1650                                 assertThat(configData.get().getExpertFlag("foo"), is(true));
1651                         }
1652
1653                         @Test
1654                         public void defaultFcpClientCanGetConfigWithForceWriteFlag()
1655                         throws InterruptedException, ExecutionException, IOException {
1656                                 Future<ConfigData> configData = fcpClient.getConfig().withForceWriteFlag().execute();
1657                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithForceWriteFlag"));
1658                                 replyWithConfigData("forceWriteFlag.foo=true");
1659                                 assertThat(configData.get().getForceWriteFlag("foo"), is(true));
1660                         }
1661
1662                         @Test
1663                         public void defaultFcpClientCanGetConfigWithShortDescription()
1664                         throws InterruptedException, ExecutionException, IOException {
1665                                 Future<ConfigData> configData = fcpClient.getConfig().withShortDescription().execute();
1666                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithShortDescription"));
1667                                 replyWithConfigData("shortDescription.foo=bar");
1668                                 assertThat(configData.get().getShortDescription("foo"), is("bar"));
1669                         }
1670
1671                         @Test
1672                         public void defaultFcpClientCanGetConfigWithLongDescription()
1673                         throws InterruptedException, ExecutionException, IOException {
1674                                 Future<ConfigData> configData = fcpClient.getConfig().withLongDescription().execute();
1675                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithLongDescription"));
1676                                 replyWithConfigData("longDescription.foo=bar");
1677                                 assertThat(configData.get().getLongDescription("foo"), is("bar"));
1678                         }
1679
1680                         @Test
1681                         public void defaultFcpClientCanGetConfigWithDataTypes()
1682                         throws InterruptedException, ExecutionException, IOException {
1683                                 Future<ConfigData> configData = fcpClient.getConfig().withDataTypes().execute();
1684                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithDataTypes"));
1685                                 replyWithConfigData("dataType.foo=number");
1686                                 assertThat(configData.get().getDataType("foo"), is("number"));
1687                         }
1688
1689                         private Matcher<List<String>> matchesGetConfigWithAdditionalParameter(String additionalParameter) {
1690                                 return matchesFcpMessage(
1691                                         "GetConfig",
1692                                         "Identifier=" + identifier,
1693                                         additionalParameter + "=true"
1694                                 );
1695                         }
1696
1697                 }
1698
1699                 public class ModifyConfig {
1700
1701                         @Test
1702                         public void defaultFcpClientCanModifyConfigData()
1703                         throws InterruptedException, ExecutionException, IOException {
1704                                 Future<ConfigData> newConfigData = fcpClient.modifyConfig().set("foo.bar").to("baz").execute();
1705                                 connectAndAssert(() -> matchesFcpMessage(
1706                                         "ModifyConfig",
1707                                         "Identifier=" + identifier,
1708                                         "foo.bar=baz"
1709                                 ));
1710                                 replyWithConfigData("current.foo.bar=baz");
1711                                 assertThat(newConfigData.get().getCurrent("foo.bar"), is("baz"));
1712                         }
1713
1714                 }
1715
1716                 private void replyWithConfigData(String... additionalLines) throws IOException {
1717                         fcpServer.writeLine("ConfigData", "Identifier=" + identifier);
1718                         fcpServer.writeLine(additionalLines);
1719                         fcpServer.writeLine("EndMessage");
1720                 }
1721
1722         }
1723
1724         public class NodeInformation {
1725
1726                 @Test
1727                 public void defaultFcpClientCanGetNodeInformation() throws InterruptedException, ExecutionException, IOException {
1728                         Future<NodeData> nodeData = fcpClient.getNode().execute();
1729                         connectAndAssert(() -> matchesGetNode(false, false, false));
1730                         replyWithNodeData();
1731                         assertThat(nodeData.get(), notNullValue());
1732                         assertThat(nodeData.get().getNodeRef().isOpennet(), is(false));
1733                 }
1734
1735                 @Test
1736                 public void defaultFcpClientCanGetNodeInformationWithOpennetRef()
1737                 throws InterruptedException, ExecutionException, IOException {
1738                         Future<NodeData> nodeData = fcpClient.getNode().opennetRef().execute();
1739                         connectAndAssert(() -> matchesGetNode(true, false, false));
1740                         replyWithNodeData("opennet=true");
1741                         assertThat(nodeData.get().getVersion().toString(), is("Fred,0.7,1.0,1466"));
1742                         assertThat(nodeData.get().getNodeRef().isOpennet(), is(true));
1743                 }
1744
1745                 @Test
1746                 public void defaultFcpClientCanGetNodeInformationWithPrivateData()
1747                 throws InterruptedException, ExecutionException, IOException {
1748                         Future<NodeData> nodeData = fcpClient.getNode().includePrivate().execute();
1749                         connectAndAssert(() -> matchesGetNode(false, true, false));
1750                         replyWithNodeData("ark.privURI=SSK@XdHMiRl");
1751                         assertThat(nodeData.get().getARK().getPrivateURI(), is("SSK@XdHMiRl"));
1752                 }
1753
1754                 @Test
1755                 public void defaultFcpClientCanGetNodeInformationWithVolatileData()
1756                 throws InterruptedException, ExecutionException, IOException {
1757                         Future<NodeData> nodeData = fcpClient.getNode().includeVolatile().execute();
1758                         connectAndAssert(() -> matchesGetNode(false, false, true));
1759                         replyWithNodeData("volatile.freeJavaMemory=205706528");
1760                         assertThat(nodeData.get().getVolatile("freeJavaMemory"), is("205706528"));
1761                 }
1762
1763                 private Matcher<List<String>> matchesGetNode(boolean withOpennetRef, boolean withPrivate, boolean withVolatile) {
1764                         return matchesFcpMessage(
1765                                 "GetNode",
1766                                 "Identifier=" + identifier,
1767                                 "GiveOpennetRef=" + withOpennetRef,
1768                                 "WithPrivate=" + withPrivate,
1769                                 "WithVolatile=" + withVolatile
1770                         );
1771                 }
1772
1773                 private void replyWithNodeData(String... additionalLines) throws IOException {
1774                         fcpServer.writeLine(
1775                                 "NodeData",
1776                                 "Identifier=" + identifier,
1777                                 "ark.pubURI=SSK@3YEf.../ark",
1778                                 "ark.number=78",
1779                                 "auth.negTypes=2",
1780                                 "version=Fred,0.7,1.0,1466",
1781                                 "lastGoodVersion=Fred,0.7,1.0,1466"
1782                         );
1783                         fcpServer.writeLine(additionalLines);
1784                         fcpServer.writeLine("EndMessage");
1785                 }
1786
1787         }
1788
1789 }