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