src/net/pterodactylus/util/fcp/FcpConnection.java

   1 /*
   2  * jSite2 - FpcConnection.java -
   3  * Copyright © 2008 David Roden
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program; if not, write to the Free Software
  17  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  18  */
  19
  20 package net.pterodactylus.util.fcp;
  21
  22 import java.io.IOException;
  23 import java.io.InputStream;
  24 import java.io.OutputStream;
  25 import java.net.InetAddress;
  26 import java.net.Socket;
  27 import java.net.UnknownHostException;
  28 import java.util.ArrayList;
  29 import java.util.Collections;
  30 import java.util.HashMap;
  31 import java.util.List;
  32 import java.util.Map;
  33
  34 import net.pterodactylus.util.io.Closer;
  35 import net.pterodactylus.util.io.LimitedInputStream;
  36
  37 /**
  38  * An FCP connection to a Freenet node.
  39  *
  40  * @author David ‘Bombe’ Roden &lt;bombe@freenetproject.org&gt;
  41  * @version $Id$
  42  */
  43 public class FcpConnection {
  44
  45         /** The default port for FCP v2. */
  46         public static final int DEFAULT_PORT = 9481;
  47
  48         /** The list of FCP listeners. */
  49         private final List<FcpListener> fcpListeners = new ArrayList<FcpListener>();
  50
  51         /** The address of the node. */
  52         private final InetAddress address;
  53
  54         /** The port number of the node’s FCP port. */
  55         private final int port;
  56
  57         /** The remote socket. */
  58         private Socket remoteSocket;
  59
  60         /** The input stream from the node. */
  61         private InputStream remoteInputStream;
  62
  63         /** The output stream to the node. */
  64         private OutputStream remoteOutputStream;
  65
  66         /** The connection handler. */
  67         private FcpConnectionHandler connectionHandler;
  68
  69         /** Incoming message statistics. */
  70         private Map<String, Integer> incomingMessageStatistics = Collections.synchronizedMap(new HashMap<String, Integer>());
  71
  72         /**
  73          * Creates a new FCP connection to the freenet node running on localhost,
  74          * using the default port.
  75          *
  76          * @throws UnknownHostException
  77          *             if the hostname can not be resolved
  78          */
  79         public FcpConnection() throws UnknownHostException {
  80                 this(InetAddress.getLocalHost());
  81         }
  82
  83         /**
  84          * Creates a new FCP connection to the Freenet node running on the given
  85          * host, listening on the default port.
  86          *
  87          * @param host
  88          *            The hostname of the Freenet node
  89          * @throws UnknownHostException
  90          *             if <code>host</code> can not be resolved
  91          */
  92         public FcpConnection(String host) throws UnknownHostException {
  93                 this(host, DEFAULT_PORT);
  94         }
  95
  96         /**
  97          * Creates a new FCP connection to the Freenet node running on the given
  98          * host, listening on the given port.
  99          *
 100          * @param host
 101          *            The hostname of the Freenet node
 102          * @param port
 103          *            The port number of the node’s FCP port
 104          * @throws UnknownHostException
 105          *             if <code>host</code> can not be resolved
 106          */
 107         public FcpConnection(String host, int port) throws UnknownHostException {
 108                 this(InetAddress.getByName(host), port);
 109         }
 110
 111         /**
 112          * Creates a new FCP connection to the Freenet node running at the given
 113          * address, listening on the default port.
 114          *
 115          * @param address
 116          *            The address of the Freenet node
 117          */
 118         public FcpConnection(InetAddress address) {
 119                 this(address, DEFAULT_PORT);
 120         }
 121
 122         /**
 123          * Creates a new FCP connection to the Freenet node running at the given
 124          * address, listening on the given port.
 125          *
 126          * @param address
 127          *            The address of the Freenet node
 128          * @param port
 129          *            The port number of the node’s FCP port
 130          */
 131         public FcpConnection(InetAddress address, int port) {
 132                 this.address = address;
 133                 this.port = port;
 134         }
 135
 136         //
 137         // LISTENER MANAGEMENT
 138         //
 139
 140         /**
 141          * Adds the given listener to the list of listeners.
 142          *
 143          * @param fcpListener
 144          *            The listener to add
 145          */
 146         public void addFcpListener(FcpListener fcpListener) {
 147                 fcpListeners.add(fcpListener);
 148         }
 149
 150         /**
 151          * Removes the given listener from the list of listeners.
 152          *
 153          * @param fcpListener
 154          *            The listener to remove
 155          */
 156         public void removeFcpListener(FcpListener fcpListener) {
 157                 fcpListeners.remove(fcpListener);
 158         }
 159
 160         /**
 161          * Notifies listeners that a “NodeHello” message was received.
 162          *
 163          * @see FcpListener#receivedNodeHello(FcpConnection, NodeHello)
 164          * @param nodeHello
 165          *            The “NodeHello” message
 166          */
 167         private void fireReceivedNodeHello(NodeHello nodeHello) {
 168                 for (FcpListener fcpListener: fcpListeners) {
 169                         fcpListener.receivedNodeHello(this, nodeHello);
 170                 }
 171         }
 172
 173         /**
 174          * Notifies listeners that a “CloseConnectionDuplicateClientName” message
 175          * was received.
 176          *
 177          * @see FcpListener#receivedCloseConnectionDuplicateClientName(FcpConnection,
 178          *      CloseConnectionDuplicateClientName)
 179          * @param closeConnectionDuplicateClientName
 180          *            The “CloseConnectionDuplicateClientName” message
 181          */
 182         private void fireReceivedCloseConnectionDuplicateClientName(CloseConnectionDuplicateClientName closeConnectionDuplicateClientName) {
 183                 for (FcpListener fcpListener: fcpListeners) {
 184                         fcpListener.receivedCloseConnectionDuplicateClientName(this, closeConnectionDuplicateClientName);
 185                 }
 186         }
 187
 188         /**
 189          * Notifies listeners that a “SSKKeypair” message was received.
 190          *
 191          * @see FcpListener#receivedSSKKeypair(FcpConnection, SSKKeypair)
 192          * @param sskKeypair
 193          *            The “SSKKeypair” message
 194          */
 195         private void fireReceivedSSKKeypair(SSKKeypair sskKeypair) {
 196                 for (FcpListener fcpListener: fcpListeners) {
 197                         fcpListener.receivedSSKKeypair(this, sskKeypair);
 198                 }
 199         }
 200
 201         /**
 202          * Notifies listeners that a “Peer” message was received.
 203          *
 204          * @see FcpListener#receivedPeer(FcpConnection, Peer)
 205          * @param peer
 206          *            The “Peer” message
 207          */
 208         private void fireReceivedPeer(Peer peer) {
 209                 for (FcpListener fcpListener: fcpListeners) {
 210                         fcpListener.receivedPeer(this, peer);
 211                 }
 212         }
 213
 214         /**
 215          * Notifies all listeners that an “EndListPeers” message was received.
 216          *
 217          * @see FcpListener#receivedEndListPeers(FcpConnection, EndListPeers)
 218          * @param endListPeers
 219          *            The “EndListPeers” message
 220          */
 221         private void fireReceivedEndListPeers(EndListPeers endListPeers) {
 222                 for (FcpListener fcpListener: fcpListeners) {
 223                         fcpListener.receivedEndListPeers(this, endListPeers);
 224                 }
 225         }
 226
 227         /**
 228          * Notifies all listeners that a “PeerNote” message was received.
 229          *
 230          * @see FcpListener#receivedPeerNote(FcpConnection, PeerNote)
 231          * @param peerNote
 232          */
 233         private void fireReceivedPeerNote(PeerNote peerNote) {
 234                 for (FcpListener fcpListener: fcpListeners) {
 235                         fcpListener.receivedPeerNote(this, peerNote);
 236                 }
 237         }
 238
 239         /**
 240          * Notifies all listeners that an “EndListPeerNotes” message was received.
 241          *
 242          * @see FcpListener#receivedEndListPeerNotes(FcpConnection,
 243          *      EndListPeerNotes)
 244          * @param endListPeerNotes
 245          *            The “EndListPeerNotes” message
 246          */
 247         private void fireReceivedEndListPeerNotes(EndListPeerNotes endListPeerNotes) {
 248                 for (FcpListener fcpListener: fcpListeners) {
 249                         fcpListener.receivedEndListPeerNotes(this, endListPeerNotes);
 250                 }
 251         }
 252
 253         /**
 254          * Notifies all listeners that a “PeerRemoved” message was received.
 255          *
 256          * @see FcpListener#receivedPeerRemoved(FcpConnection, PeerRemoved)
 257          * @param peerRemoved
 258          *            The “PeerRemoved” message
 259          */
 260         private void fireReceivedPeerRemoved(PeerRemoved peerRemoved) {
 261                 for (FcpListener fcpListener: fcpListeners) {
 262                         fcpListener.receivedPeerRemoved(this, peerRemoved);
 263                 }
 264         }
 265
 266         /**
 267          * Notifies all listeners that a “NodeData” message was received.
 268          *
 269          * @see FcpListener#receivedNodeData(FcpConnection, NodeData)
 270          * @param nodeData
 271          *            The “NodeData” message
 272          */
 273         private void fireReceivedNodeData(NodeData nodeData) {
 274                 for (FcpListener fcpListener: fcpListeners) {
 275                         fcpListener.receivedNodeData(this, nodeData);
 276                 }
 277         }
 278
 279         /**
 280          * Notifies all listeners that a “TestDDAReply” message was received.
 281          *
 282          * @see FcpListener#receivedTestDDAReply(FcpConnection, TestDDAReply)
 283          * @param testDDAReply
 284          *            The “TestDDAReply” message
 285          */
 286         private void fireReceivedTestDDAReply(TestDDAReply testDDAReply) {
 287                 for (FcpListener fcpListener: fcpListeners) {
 288                         fcpListener.receivedTestDDAReply(this, testDDAReply);
 289                 }
 290         }
 291
 292         /**
 293          * Notifies all listeners that a “TestDDAComplete” message was received.
 294          *
 295          * @see FcpListener#receivedTestDDAComplete(FcpConnection, TestDDAComplete)
 296          * @param testDDAComplete
 297          *            The “TestDDAComplete” message
 298          */
 299         private void fireReceivedTestDDAComplete(TestDDAComplete testDDAComplete) {
 300                 for (FcpListener fcpListener: fcpListeners) {
 301                         fcpListener.receivedTestDDAComplete(this, testDDAComplete);
 302                 }
 303         }
 304
 305         /**
 306          * Notifies all listeners that a “PersistentGet” message was received.
 307          *
 308          * @param persistentGet
 309          *            The “PersistentGet” message
 310          */
 311         private void fireReceivedPersistentGet(PersistentGet persistentGet) {
 312                 for (FcpListener fcpListener: fcpListeners) {
 313                         fcpListener.receivedPersistentGet(this, persistentGet);
 314                 }
 315         }
 316
 317         /**
 318          * Notifies all listeners that a “PersistentPut” message was received.
 319          *
 320          * @see FcpListener#receivedPersistentPut(FcpConnection, PersistentPut)
 321          * @param persistentPut
 322          *            The “PersistentPut” message
 323          */
 324         private void fireReceivedPersistentPut(PersistentPut persistentPut) {
 325                 for (FcpListener fcpListener: fcpListeners) {
 326                         fcpListener.receivedPersistentPut(this, persistentPut);
 327                 }
 328         }
 329
 330         /**
 331          * Notifies all listeners that a “EndListPersistentRequests” message was
 332          * received.
 333          *
 334          * @param endListPersistentRequests
 335          *            The “EndListPersistentRequests” message
 336          */
 337         private void fireReceivedEndListPersistentRequests(EndListPersistentRequests endListPersistentRequests) {
 338                 for (FcpListener fcpListener: fcpListeners) {
 339                         fcpListener.receivedEndListPersistentRequests(this, endListPersistentRequests);
 340                 }
 341         }
 342
 343         /**
 344          * Notifies all listeners that a “URIGenerated” message was received.
 345          *
 346          * @param uriGenerated
 347          *            The “URIGenerated” message
 348          */
 349         private void fireReceivedURIGenerated(URIGenerated uriGenerated) {
 350                 for (FcpListener fcpListener: fcpListeners) {
 351                         fcpListener.receivedURIGenerated(this, uriGenerated);
 352                 }
 353         }
 354
 355         /**
 356          * Notifies all listeners that a “DataFound” message was received.
 357          *
 358          * @param dataFound
 359          *            The “DataFound” message
 360          */
 361         private void fireReceivedDataFound(DataFound dataFound) {
 362                 for (FcpListener fcpListener: fcpListeners) {
 363                         fcpListener.receivedDataFound(this, dataFound);
 364                 }
 365         }
 366
 367         /**
 368          * Notifies all listeners that an “AllData” message was received.
 369          *
 370          * @param allData
 371          *            The “AllData” message
 372          */
 373         private void fireReceivedAllData(AllData allData) {
 374                 for (FcpListener fcpListener: fcpListeners) {
 375                         fcpListener.receivedAllData(this, allData);
 376                 }
 377         }
 378
 379         /**
 380          * Notifies all listeners that a “SimpleProgress” message was received.
 381          *
 382          * @param simpleProgress
 383          *            The “SimpleProgress” message
 384          */
 385         private void fireReceivedSimpleProgress(SimpleProgress simpleProgress) {
 386                 for (FcpListener fcpListener: fcpListeners) {
 387                         fcpListener.receivedSimpleProgress(this, simpleProgress);
 388                 }
 389         }
 390
 391         /**
 392          * Notifies all listeners that a “StartedCompression” message was received.
 393          *
 394          * @param startedCompression
 395          *            The “StartedCompression” message
 396          */
 397         private void fireReceivedStartedCompression(StartedCompression startedCompression) {
 398                 for (FcpListener fcpListener: fcpListeners) {
 399                         fcpListener.receivedStartedCompression(this, startedCompression);
 400                 }
 401         }
 402
 403         /**
 404          * Notifies all listeners that a “FinishedCompression” message was received.
 405          *
 406          * @param finishedCompression
 407          *            The “FinishedCompression” message
 408          */
 409         private void fireReceivedFinishedCompression(FinishedCompression finishedCompression) {
 410                 for (FcpListener fcpListener: fcpListeners) {
 411                         fcpListener.receviedFinishedCompression(this, finishedCompression);
 412                 }
 413         }
 414
 415         /**
 416          * Notifies all listeners that an “UnknownPeerNoteType” message was
 417          * received.
 418          *
 419          * @param unknownPeerNoteType
 420          *            The “UnknownPeerNoteType” message
 421          */
 422         private void fireReceivedUnknownPeerNoteType(UnknownPeerNoteType unknownPeerNoteType) {
 423                 for (FcpListener fcpListener: fcpListeners) {
 424                         fcpListener.receivedUnknownPeerNoteType(this, unknownPeerNoteType);
 425                 }
 426         }
 427
 428         /**
 429          * Notifies all listeners that an “UnknownNodeIdentifier” message was
 430          * received.
 431          *
 432          * @param unknownNodeIdentifier
 433          *            The “UnknownNodeIdentifier” message
 434          */
 435         private void fireReceivedUnknownNodeIdentifier(UnknownNodeIdentifier unknownNodeIdentifier) {
 436                 for (FcpListener fcpListener: fcpListeners) {
 437                         fcpListener.receivedUnknownNodeIdentifier(this, unknownNodeIdentifier);
 438                 }
 439         }
 440
 441         /**
 442          * Notifies all listeners that a “ConfigData” message was received.
 443          *
 444          * @param configData
 445          *            The “ConfigData” message
 446          */
 447         private void fireReceivedConfigData(ConfigData configData) {
 448                 for (FcpListener fcpListener: fcpListeners) {
 449                         fcpListener.receivedConfigData(this, configData);
 450                 }
 451         }
 452
 453         /**
 454          * Notifies all listeners that a “GetFailed” message was received.
 455          *
 456          * @param getFailed
 457          *            The “GetFailed” message
 458          */
 459         private void fireReceivedGetFailed(GetFailed getFailed) {
 460                 for (FcpListener fcpListener: fcpListeners) {
 461                         fcpListener.receivedGetFailed(this, getFailed);
 462                 }
 463         }
 464
 465         /**
 466          * Notifies all listeners that a “PutFailed” message was received.
 467          *
 468          * @param putFailed
 469          *            The “PutFailed” message
 470          */
 471         private void fireReceivedPutFailed(PutFailed putFailed) {
 472                 for (FcpListener fcpListener: fcpListeners) {
 473                         fcpListener.receivedPutFailed(this, putFailed);
 474                 }
 475         }
 476
 477         /**
 478          * Notifies all listeners that an “IdentifierCollision” message was
 479          * received.
 480          *
 481          * @param identifierCollision
 482          *            The “IdentifierCollision” message
 483          */
 484         private void fireReceivedIdentifierCollision(IdentifierCollision identifierCollision) {
 485                 for (FcpListener fcpListener: fcpListeners) {
 486                         fcpListener.receivedIdentifierCollision(this, identifierCollision);
 487                 }
 488         }
 489
 490         /**
 491          * Notifies all listeners that an “PersistentPutDir” message was received.
 492          *
 493          * @param persistentPutDir
 494          *            The “PersistentPutDir” message
 495          */
 496         private void fireReceivedPersistentPutDir(PersistentPutDir persistentPutDir) {
 497                 for (FcpListener fcpListener: fcpListeners) {
 498                         fcpListener.receivedPersistentPutDir(this, persistentPutDir);
 499                 }
 500         }
 501
 502         /**
 503          * Notifies all listeners that a “PersistentRequestRemoved” message was
 504          * received.
 505          *
 506          * @param persistentRequestRemoved
 507          *            The “PersistentRequestRemoved” message
 508          */
 509         private void fireReceivedPersistentRequestRemoved(PersistentRequestRemoved persistentRequestRemoved) {
 510                 for (FcpListener fcpListener: fcpListeners) {
 511                         fcpListener.receivedPersistentRequestRemoved(this, persistentRequestRemoved);
 512                 }
 513         }
 514
 515         /**
 516          * Notifies all listeners that a “SubscribedUSKUpdate” message was received.
 517          *
 518          * @param subscribedUSKUpdate
 519          *            The “SubscribedUSKUpdate” message
 520          */
 521         private void fireReceivedSubscribedUSKUpdate(SubscribedUSKUpdate subscribedUSKUpdate) {
 522                 for (FcpListener fcpListener: fcpListeners) {
 523                         fcpListener.receivedSubscribedUSKUpdate(this, subscribedUSKUpdate);
 524                 }
 525         }
 526
 527         /**
 528          * Notifies all listeners that a “PluginInfo” message was received.
 529          *
 530          * @param pluginInfo
 531          *            The “PluginInfo” message
 532          */
 533         private void fireReceivedPluginInfo(PluginInfo pluginInfo) {
 534                 for (FcpListener fcpListener: fcpListeners) {
 535                         fcpListener.receivedPluginInfo(this, pluginInfo);
 536                 }
 537         }
 538
 539         /**
 540          * Notifies all listeners that an “FCPPluginReply” message was received.
 541          *
 542          * @param fcpPluginReply
 543          *            The “FCPPluginReply” message
 544          */
 545         private void fireReceivedFCPPluginReply(FCPPluginReply fcpPluginReply) {
 546                 for (FcpListener fcpListener: fcpListeners) {
 547                         fcpListener.receivedFCPPluginReply(this, fcpPluginReply);
 548                 }
 549         }
 550
 551         /**
 552          * Notifies all listeners that a “PersistentRequestModified” message was
 553          * received.
 554          *
 555          * @param persistentRequestModified
 556          *            The “PersistentRequestModified” message
 557          */
 558         private void fireReceivedPersistentRequestModified(PersistentRequestModified persistentRequestModified) {
 559                 for (FcpListener fcpListener: fcpListeners) {
 560                         fcpListener.receivedPersistentRequestModified(this, persistentRequestModified);
 561                 }
 562         }
 563
 564         /**
 565          * Notifies all listeners that a “PutSuccessful” message was received.
 566          *
 567          * @param putSuccessful
 568          *            The “PutSuccessful” message
 569          */
 570         private void fireReceivedPutSuccessful(PutSuccessful putSuccessful) {
 571                 for (FcpListener fcpListener: fcpListeners) {
 572                         fcpListener.receivedPutSuccessful(this, putSuccessful);
 573                 }
 574         }
 575
 576         /**
 577          * Notifies all listeners that a “PutFetchable” message was received.
 578          *
 579          * @param putFetchable
 580          *            The “PutFetchable” message
 581          */
 582         private void fireReceivedPutFetchable(PutFetchable putFetchable) {
 583                 for (FcpListener fcpListener: fcpListeners) {
 584                         fcpListener.receivedPutFetchable(this, putFetchable);
 585                 }
 586         }
 587
 588         /**
 589          * Notifies all listeners that a “ProtocolError” message was received.
 590          *
 591          * @param protocolError
 592          *            The “ProtocolError” message
 593          */
 594         private void fireReceivedProtocolError(ProtocolError protocolError) {
 595                 for (FcpListener fcpListener: fcpListeners) {
 596                         fcpListener.receivedProtocolError(this, protocolError);
 597                 }
 598         }
 599
 600         /**
 601          * Notifies all registered listeners that a message has been received.
 602          *
 603          * @see FcpListener#receivedMessage(FcpConnection, FcpMessage)
 604          * @param fcpMessage
 605          *            The message that was received
 606          */
 607         private void fireMessageReceived(FcpMessage fcpMessage) {
 608                 for (FcpListener fcpListener: fcpListeners) {
 609                         fcpListener.receivedMessage(this, fcpMessage);
 610                 }
 611         }
 612
 613         //
 614         // ACTIONS
 615         //
 616
 617         /**
 618          * Connects to the node.
 619          *
 620          * @throws IOException
 621          *             if an I/O error occurs
 622          * @throws IllegalStateException
 623          *             if there is already a connection to the node
 624          */
 625         public synchronized void connect() throws IOException, IllegalStateException {
 626                 if (connectionHandler != null) {
 627                         throw new IllegalStateException("already connected, disconnect first");
 628                 }
 629                 remoteSocket = new Socket(address, port);
 630                 remoteInputStream = remoteSocket.getInputStream();
 631                 remoteOutputStream = remoteSocket.getOutputStream();
 632                 new Thread(connectionHandler = new FcpConnectionHandler(this, remoteInputStream)).start();
 633         }
 634
 635         /**
 636          * Disconnects from the node. If there is no connection to the node, this
 637          * method does nothing.
 638          */
 639         public synchronized void disconnect() {
 640                 if (connectionHandler == null) {
 641                         return;
 642                 }
 643                 Closer.close(remoteSocket);
 644                 connectionHandler.stop();
 645                 connectionHandler = null;
 646         }
 647
 648         /**
 649          * Sends the given FCP message.
 650          *
 651          * @param fcpMessage
 652          *            The FCP message to send
 653          * @throws IOException
 654          *             if an I/O error occurs
 655          */
 656         public synchronized void sendMessage(FcpMessage fcpMessage) throws IOException {
 657                 System.out.println("sending message: " + fcpMessage.getName());
 658                 fcpMessage.write(remoteOutputStream);
 659         }
 660
 661         //
 662         // PACKAGE-PRIVATE METHODS
 663         //
 664
 665         /**
 666          * Handles the given message, notifying listeners. This message should only
 667          * be called by {@link FcpConnectionHandler}.
 668          *
 669          * @param fcpMessage
 670          *            The received message
 671          */
 672         void handleMessage(FcpMessage fcpMessage) {
 673                 String messageName = fcpMessage.getName();
 674                 countMessage(messageName);
 675                 if ("SimpleProgress".equals(messageName)) {
 676                         fireReceivedSimpleProgress(new SimpleProgress(fcpMessage));
 677                 } else if ("ProtocolError".equals(messageName)) {
 678                         fireReceivedProtocolError(new ProtocolError(fcpMessage));
 679                 } else if ("PersistentGet".equals(messageName)) {
 680                         fireReceivedPersistentGet(new PersistentGet(fcpMessage));
 681                 } else if ("PersistentPut".equals(messageName)) {
 682                         fireReceivedPersistentPut(new PersistentPut(fcpMessage));
 683                 } else if ("PersistentPutDir".equals(messageName)) {
 684                         fireReceivedPersistentPutDir(new PersistentPutDir(fcpMessage));
 685                 } else if ("URIGenerated".equals(messageName)) {
 686                         fireReceivedURIGenerated(new URIGenerated(fcpMessage));
 687                 } else if ("EndListPersistentRequests".equals(messageName)) {
 688                         fireReceivedEndListPersistentRequests(new EndListPersistentRequests(fcpMessage));
 689                 } else if ("Peer".equals(messageName)) {
 690                         fireReceivedPeer(new Peer(fcpMessage));
 691                 } else if ("PeerNote".equals(messageName)) {
 692                         fireReceivedPeerNote(new PeerNote(fcpMessage));
 693                 } else if ("StartedCompression".equals(messageName)) {
 694                         fireReceivedStartedCompression(new StartedCompression(fcpMessage));
 695                 } else if ("FinishedCompression".equals(messageName)) {
 696                         fireReceivedFinishedCompression(new FinishedCompression(fcpMessage));
 697                 } else if ("GetFailed".equals(messageName)) {
 698                         fireReceivedGetFailed(new GetFailed(fcpMessage));
 699                 } else if ("PutFetchable".equals(messageName)) {
 700                         fireReceivedPutFetchable(new PutFetchable(fcpMessage));
 701                 } else if ("PutSuccessful".equals(messageName)) {
 702                         fireReceivedPutSuccessful(new PutSuccessful(fcpMessage));
 703                 } else if ("PutFailed".equals(messageName)) {
 704                         fireReceivedPutFailed(new PutFailed(fcpMessage));
 705                 } else if ("DataFound".equals(messageName)) {
 706                         fireReceivedDataFound(new DataFound(fcpMessage));
 707                 } else if ("SubscribedUSKUpdate".equals(messageName)) {
 708                         fireReceivedSubscribedUSKUpdate(new SubscribedUSKUpdate(fcpMessage));
 709                 } else if ("IdentifierCollision".equals(messageName)) {
 710                         fireReceivedIdentifierCollision(new IdentifierCollision(fcpMessage));
 711                 } else if ("AllData".equals(messageName)) {
 712                         LimitedInputStream payloadInputStream = getInputStream(FcpUtils.safeParseLong(fcpMessage.getField("DataLength")));
 713                         fireReceivedAllData(new AllData(fcpMessage, payloadInputStream));
 714                         try {
 715                                 payloadInputStream.consume();
 716                         } catch (IOException ioe1) {
 717                                 /* well, ignore. when the connection handler fails, all fails. */
 718                         }
 719                 } else if ("EndListPeerNotes".equals(messageName)) {
 720                         fireReceivedEndListPeerNotes(new EndListPeerNotes(fcpMessage));
 721                 } else if ("EndListPeers".equals(messageName)) {
 722                         fireReceivedEndListPeers(new EndListPeers(fcpMessage));
 723                 } else if ("SSKKeypair".equals(messageName)) {
 724                         fireReceivedSSKKeypair(new SSKKeypair(fcpMessage));
 725                 } else if ("PeerRemoved".equals(messageName)) {
 726                         fireReceivedPeerRemoved(new PeerRemoved(fcpMessage));
 727                 } else if ("PersistentRequestModified".equals(messageName)) {
 728                         fireReceivedPersistentRequestModified(new PersistentRequestModified(fcpMessage));
 729                 } else if ("PersistentRequestRemoved".equals(messageName)) {
 730                         fireReceivedPersistentRequestRemoved(new PersistentRequestRemoved(fcpMessage));
 731                 } else if ("UnknownPeerNoteType".equals(messageName)) {
 732                         fireReceivedUnknownPeerNoteType(new UnknownPeerNoteType(fcpMessage));
 733                 } else if ("UnknownNodeIdentifier".equals(messageName)) {
 734                         fireReceivedUnknownNodeIdentifier(new UnknownNodeIdentifier(fcpMessage));
 735                 } else if ("FCPPluginReply".equals(messageName)) {
 736                         LimitedInputStream payloadInputStream = getInputStream(FcpUtils.safeParseLong(fcpMessage.getField("DataLength")));
 737                         fireReceivedFCPPluginReply(new FCPPluginReply(fcpMessage, payloadInputStream));
 738                         try {
 739                                 payloadInputStream.consume();
 740                         } catch (IOException ioe1) {
 741                                 /* ignore. */
 742                         }
 743                 } else if ("PluginInfo".equals(messageName)) {
 744                         fireReceivedPluginInfo(new PluginInfo(fcpMessage));
 745                 } else if ("NodeData".equals(messageName)) {
 746                         fireReceivedNodeData(new NodeData(fcpMessage));
 747                 } else if ("TestDDAReply".equals(messageName)) {
 748                         fireReceivedTestDDAReply(new TestDDAReply(fcpMessage));
 749                 } else if ("TestDDAComplete".equals(messageName)) {
 750                         fireReceivedTestDDAComplete(new TestDDAComplete(fcpMessage));
 751                 } else if ("ConfigData".equals(messageName)) {
 752                         fireReceivedConfigData(new ConfigData(fcpMessage));
 753                 } else if ("NodeHello".equals(messageName)) {
 754                         fireReceivedNodeHello(new NodeHello(fcpMessage));
 755                 } else if ("CloseConnectionDuplicateClientName".equals(messageName)) {
 756                         fireReceivedCloseConnectionDuplicateClientName(new CloseConnectionDuplicateClientName(fcpMessage));
 757                 } else {
 758                         fireMessageReceived(fcpMessage);
 759                 }
 760         }
 761
 762         /**
 763          * Handles a disconnect from the node.
 764          */
 765         synchronized void handleDisconnect() {
 766                 Closer.close(remoteInputStream);
 767                 Closer.close(remoteOutputStream);
 768                 Closer.close(remoteSocket);
 769                 connectionHandler = null;
 770         }
 771
 772         //
 773         // PRIVATE METHODS
 774         //
 775
 776         /**
 777          * Incremets the counter in {@link #incomingMessageStatistics} by <cod>1</code>
 778          * for the given message name.
 779          *
 780          * @param name
 781          *            The name of the message to count
 782          */
 783         private void countMessage(String name) {
 784                 int oldValue = 0;
 785                 if (incomingMessageStatistics.containsKey(name)) {
 786                         oldValue = incomingMessageStatistics.get(name);
 787                 }
 788                 incomingMessageStatistics.put(name, oldValue + 1);
 789         }
 790
 791         /**
 792          * Returns a limited input stream from the node’s input stream.
 793          *
 794          * @param dataLength
 795          *            The length of the stream
 796          * @return The limited input stream
 797          */
 798         private LimitedInputStream getInputStream(long dataLength) {
 799                 if (dataLength <= 0) {
 800                         return new LimitedInputStream(null, 0);
 801                 }
 802                 return new LimitedInputStream(remoteInputStream, dataLength);
 803         }
 804
 805 }