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

   1 /*
   2  * jFCPlib - FpcConnection.java - Copyright © 2008 David Roden
   3  *
   4  * This program is free software; you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License as published by
   6  * the Free Software Foundation; either version 2 of the License, or
   7  * (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  * GNU General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License
  15  * along with this program; if not, write to the Free Software
  16  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  17  */
  18
  19 package net.pterodactylus.fcp;
  20
  21 import java.io.Closeable;
  22 import java.io.FilterInputStream;
  23 import java.io.IOException;
  24 import java.io.InputStream;
  25 import java.io.OutputStream;
  26 import java.net.InetAddress;
  27 import java.net.Socket;
  28 import java.net.UnknownHostException;
  29 import java.util.Collections;
  30 import java.util.HashMap;
  31 import java.util.Map;
  32 import java.util.logging.Logger;
  33
  34 import net.pterodactylus.util.logging.Logging;
  35
  36 /**
  37  * An FCP connection to a Freenet node.
  38  *
  39  * @author David ‘Bombe’ Roden &lt;bombe@freenetproject.org&gt;
  40  */
  41 public class FcpConnection implements Closeable {
  42
  43         /** Logger. */
  44         private static final Logger logger = Logging.getLogger(FcpConnection.class.getName());
  45
  46         /** The default port for FCP v2. */
  47         public static final int DEFAULT_PORT = 9481;
  48
  49         /** Listener management. */
  50         private final FcpListenerManager fcpListenerManager = new FcpListenerManager(this);
  51
  52         /** The address of the node. */
  53         private final InetAddress address;
  54
  55         /** The port number of the node’s FCP port. */
  56         private final int port;
  57
  58         /** The remote socket. */
  59         private Socket remoteSocket;
  60
  61         /** The input stream from the node. */
  62         private InputStream remoteInputStream;
  63
  64         /** The output stream to the node. */
  65         private OutputStream remoteOutputStream;
  66
  67         /** The connection handler. */
  68         private FcpConnectionHandler connectionHandler;
  69
  70         /** Incoming message statistics. */
  71         private Map<String, Integer> incomingMessageStatistics = Collections.synchronizedMap(new HashMap<String, Integer>());
  72
  73         /**
  74          * Creates a new FCP connection to the freenet node running on localhost,
  75          * using the default port.
  76          *
  77          * @throws UnknownHostException
  78          *             if the hostname can not be resolved
  79          */
  80         public FcpConnection() throws UnknownHostException {
  81                 this(InetAddress.getLocalHost());
  82         }
  83
  84         /**
  85          * Creates a new FCP connection to the Freenet node running on the given
  86          * host, listening on the default port.
  87          *
  88          * @param host
  89          *            The hostname of the Freenet node
  90          * @throws UnknownHostException
  91          *             if <code>host</code> can not be resolved
  92          */
  93         public FcpConnection(String host) throws UnknownHostException {
  94                 this(host, DEFAULT_PORT);
  95         }
  96
  97         /**
  98          * Creates a new FCP connection to the Freenet node running on the given
  99          * host, listening on the given port.
 100          *
 101          * @param host
 102          *            The hostname of the Freenet node
 103          * @param port
 104          *            The port number of the node’s FCP port
 105          * @throws UnknownHostException
 106          *             if <code>host</code> can not be resolved
 107          */
 108         public FcpConnection(String host, int port) throws UnknownHostException {
 109                 this(InetAddress.getByName(host), port);
 110         }
 111
 112         /**
 113          * Creates a new FCP connection to the Freenet node running at the given
 114          * address, listening on the default port.
 115          *
 116          * @param address
 117          *            The address of the Freenet node
 118          */
 119         public FcpConnection(InetAddress address) {
 120                 this(address, DEFAULT_PORT);
 121         }
 122
 123         /**
 124          * Creates a new FCP connection to the Freenet node running at the given
 125          * address, listening on the given port.
 126          *
 127          * @param address
 128          *            The address of the Freenet node
 129          * @param port
 130          *            The port number of the node’s FCP port
 131          */
 132         public FcpConnection(InetAddress address, int port) {
 133                 this.address = address;
 134                 this.port = port;
 135         }
 136
 137         //
 138         // LISTENER MANAGEMENT
 139         //
 140
 141         /**
 142          * Adds the given listener to the list of listeners.
 143          *
 144          * @param fcpListener
 145          *            The listener to add
 146          */
 147         public void addFcpListener(FcpListener fcpListener) {
 148                 fcpListenerManager.addListener(fcpListener);
 149         }
 150
 151         /**
 152          * Removes the given listener from the list of listeners.
 153          *
 154          * @param fcpListener
 155          *            The listener to remove
 156          */
 157         public void removeFcpListener(FcpListener fcpListener) {
 158                 fcpListenerManager.removeListener(fcpListener);
 159         }
 160
 161         //
 162         // ACTIONS
 163         //
 164
 165         /**
 166          * Connects to the node.
 167          *
 168          * @throws IOException
 169          *             if an I/O error occurs
 170          * @throws IllegalStateException
 171          *             if there is already a connection to the node
 172          */
 173         public synchronized void connect() throws IOException, IllegalStateException {
 174                 if (connectionHandler != null) {
 175                         throw new IllegalStateException("already connected, disconnect first");
 176                 }
 177                 logger.info("connecting to " + address + ":" + port + "…");
 178                 remoteSocket = new Socket(address, port);
 179                 remoteInputStream = remoteSocket.getInputStream();
 180                 remoteOutputStream = remoteSocket.getOutputStream();
 181                 new Thread(connectionHandler = new FcpConnectionHandler(this, remoteInputStream)).start();
 182         }
 183
 184         /**
 185          * Disconnects from the node. If there is no connection to the node, this
 186          * method does nothing.
 187          *
 188          * @deprecated Use {@link #close()} instead
 189          */
 190         @Deprecated
 191         public synchronized void disconnect() {
 192                 close();
 193         }
 194
 195         /**
 196          * Closes the connection. If there is no connection to the node, this method
 197          * does nothing.
 198          */
 199         public void close() {
 200                 handleDisconnect(null);
 201         }
 202
 203         /**
 204          * Sends the given FCP message.
 205          *
 206          * @param fcpMessage
 207          *            The FCP message to send
 208          * @throws IOException
 209          *             if an I/O error occurs
 210          */
 211         public synchronized void sendMessage(FcpMessage fcpMessage) throws IOException {
 212                 logger.fine("sending message: " + fcpMessage.getName());
 213                 fcpMessage.write(remoteOutputStream);
 214         }
 215
 216         //
 217         // PACKAGE-PRIVATE METHODS
 218         //
 219
 220         /**
 221          * Handles the given message, notifying listeners. This message should only
 222          * be called by {@link FcpConnectionHandler}.
 223          *
 224          * @param fcpMessage
 225          *            The received message
 226          */
 227         void handleMessage(FcpMessage fcpMessage) {
 228                 logger.fine("received message: " + fcpMessage.getName());
 229                 String messageName = fcpMessage.getName();
 230                 countMessage(messageName);
 231                 if ("SimpleProgress".equals(messageName)) {
 232                         fcpListenerManager.fireReceivedSimpleProgress(new SimpleProgress(fcpMessage));
 233                 } else if ("ProtocolError".equals(messageName)) {
 234                         fcpListenerManager.fireReceivedProtocolError(new ProtocolError(fcpMessage));
 235                 } else if ("PersistentGet".equals(messageName)) {
 236                         fcpListenerManager.fireReceivedPersistentGet(new PersistentGet(fcpMessage));
 237                 } else if ("PersistentPut".equals(messageName)) {
 238                         fcpListenerManager.fireReceivedPersistentPut(new PersistentPut(fcpMessage));
 239                 } else if ("PersistentPutDir".equals(messageName)) {
 240                         fcpListenerManager.fireReceivedPersistentPutDir(new PersistentPutDir(fcpMessage));
 241                 } else if ("URIGenerated".equals(messageName)) {
 242                         fcpListenerManager.fireReceivedURIGenerated(new URIGenerated(fcpMessage));
 243                 } else if ("EndListPersistentRequests".equals(messageName)) {
 244                         fcpListenerManager.fireReceivedEndListPersistentRequests(new EndListPersistentRequests(fcpMessage));
 245                 } else if ("Peer".equals(messageName)) {
 246                         fcpListenerManager.fireReceivedPeer(new Peer(fcpMessage));
 247                 } else if ("PeerNote".equals(messageName)) {
 248                         fcpListenerManager.fireReceivedPeerNote(new PeerNote(fcpMessage));
 249                 } else if ("StartedCompression".equals(messageName)) {
 250                         fcpListenerManager.fireReceivedStartedCompression(new StartedCompression(fcpMessage));
 251                 } else if ("FinishedCompression".equals(messageName)) {
 252                         fcpListenerManager.fireReceivedFinishedCompression(new FinishedCompression(fcpMessage));
 253                 } else if ("GetFailed".equals(messageName)) {
 254                         fcpListenerManager.fireReceivedGetFailed(new GetFailed(fcpMessage));
 255                 } else if ("PutFetchable".equals(messageName)) {
 256                         fcpListenerManager.fireReceivedPutFetchable(new PutFetchable(fcpMessage));
 257                 } else if ("PutSuccessful".equals(messageName)) {
 258                         fcpListenerManager.fireReceivedPutSuccessful(new PutSuccessful(fcpMessage));
 259                 } else if ("PutFailed".equals(messageName)) {
 260                         fcpListenerManager.fireReceivedPutFailed(new PutFailed(fcpMessage));
 261                 } else if ("DataFound".equals(messageName)) {
 262                         fcpListenerManager.fireReceivedDataFound(new DataFound(fcpMessage));
 263                 } else if ("SubscribedUSKUpdate".equals(messageName)) {
 264                         fcpListenerManager.fireReceivedSubscribedUSKUpdate(new SubscribedUSKUpdate(fcpMessage));
 265                 } else if ("IdentifierCollision".equals(messageName)) {
 266                         fcpListenerManager.fireReceivedIdentifierCollision(new IdentifierCollision(fcpMessage));
 267                 } else if ("AllData".equals(messageName)) {
 268                         LimitedInputStream payloadInputStream = getInputStream(FcpUtils.safeParseLong(fcpMessage.getField("DataLength")));
 269                         fcpListenerManager.fireReceivedAllData(new AllData(fcpMessage, payloadInputStream));
 270                         try {
 271                                 payloadInputStream.consume();
 272                         } catch (IOException ioe1) {
 273                                 /* well, ignore. when the connection handler fails, all fails. */
 274                         }
 275                 } else if ("EndListPeerNotes".equals(messageName)) {
 276                         fcpListenerManager.fireReceivedEndListPeerNotes(new EndListPeerNotes(fcpMessage));
 277                 } else if ("EndListPeers".equals(messageName)) {
 278                         fcpListenerManager.fireReceivedEndListPeers(new EndListPeers(fcpMessage));
 279                 } else if ("SSKKeypair".equals(messageName)) {
 280                         fcpListenerManager.fireReceivedSSKKeypair(new SSKKeypair(fcpMessage));
 281                 } else if ("PeerRemoved".equals(messageName)) {
 282                         fcpListenerManager.fireReceivedPeerRemoved(new PeerRemoved(fcpMessage));
 283                 } else if ("PersistentRequestModified".equals(messageName)) {
 284                         fcpListenerManager.fireReceivedPersistentRequestModified(new PersistentRequestModified(fcpMessage));
 285                 } else if ("PersistentRequestRemoved".equals(messageName)) {
 286                         fcpListenerManager.fireReceivedPersistentRequestRemoved(new PersistentRequestRemoved(fcpMessage));
 287                 } else if ("UnknownPeerNoteType".equals(messageName)) {
 288                         fcpListenerManager.fireReceivedUnknownPeerNoteType(new UnknownPeerNoteType(fcpMessage));
 289                 } else if ("UnknownNodeIdentifier".equals(messageName)) {
 290                         fcpListenerManager.fireReceivedUnknownNodeIdentifier(new UnknownNodeIdentifier(fcpMessage));
 291                 } else if ("FCPPluginReply".equals(messageName)) {
 292                         LimitedInputStream payloadInputStream = getInputStream(FcpUtils.safeParseLong(fcpMessage.getField("DataLength")));
 293                         fcpListenerManager.fireReceivedFCPPluginReply(new FCPPluginReply(fcpMessage, payloadInputStream));
 294                         try {
 295                                 payloadInputStream.consume();
 296                         } catch (IOException ioe1) {
 297                                 /* ignore. */
 298                         }
 299                 } else if ("PluginInfo".equals(messageName)) {
 300                         fcpListenerManager.fireReceivedPluginInfo(new PluginInfo(fcpMessage));
 301                 } else if ("NodeData".equals(messageName)) {
 302                         fcpListenerManager.fireReceivedNodeData(new NodeData(fcpMessage));
 303                 } else if ("TestDDAReply".equals(messageName)) {
 304                         fcpListenerManager.fireReceivedTestDDAReply(new TestDDAReply(fcpMessage));
 305                 } else if ("TestDDAComplete".equals(messageName)) {
 306                         fcpListenerManager.fireReceivedTestDDAComplete(new TestDDAComplete(fcpMessage));
 307                 } else if ("ConfigData".equals(messageName)) {
 308                         fcpListenerManager.fireReceivedConfigData(new ConfigData(fcpMessage));
 309                 } else if ("NodeHello".equals(messageName)) {
 310                         fcpListenerManager.fireReceivedNodeHello(new NodeHello(fcpMessage));
 311                 } else if ("CloseConnectionDuplicateClientName".equals(messageName)) {
 312                         fcpListenerManager.fireReceivedCloseConnectionDuplicateClientName(new CloseConnectionDuplicateClientName(fcpMessage));
 313                 } else if ("SentFeed".equals(messageName)) {
 314                         fcpListenerManager.fireSentFeed(new SentFeed(fcpMessage));
 315                 } else if ("ReceivedBookmarkFeed".equals(messageName)) {
 316                         fcpListenerManager.fireReceivedBookmarkFeed(new ReceivedBookmarkFeed(fcpMessage));
 317                 } else {
 318                         fcpListenerManager.fireMessageReceived(fcpMessage);
 319                 }
 320         }
 321
 322         /**
 323          * Handles a disconnect from the node.
 324          *
 325          * @param throwable
 326          *            The exception that caused the disconnect, or <code>null</code>
 327          *            if there was no exception
 328          */
 329         synchronized void handleDisconnect(Throwable throwable) {
 330                 FcpUtils.close(remoteInputStream);
 331                 FcpUtils.close(remoteOutputStream);
 332                 FcpUtils.close(remoteSocket);
 333                 if (connectionHandler != null) {
 334                         connectionHandler.stop();
 335                         connectionHandler = null;
 336                         fcpListenerManager.fireConnectionClosed(throwable);
 337                 }
 338         }
 339
 340         //
 341         // PRIVATE METHODS
 342         //
 343
 344         /**
 345          * Incremets the counter in {@link #incomingMessageStatistics} by
 346          * <cod>1</code> for the given message name.
 347          *
 348          * @param name
 349          *            The name of the message to count
 350          */
 351         private void countMessage(String name) {
 352                 int oldValue = 0;
 353                 if (incomingMessageStatistics.containsKey(name)) {
 354                         oldValue = incomingMessageStatistics.get(name);
 355                 }
 356                 incomingMessageStatistics.put(name, oldValue + 1);
 357                 logger.finest("count for " + name + ": " + (oldValue + 1));
 358         }
 359
 360         /**
 361          * Returns a limited input stream from the node’s input stream.
 362          *
 363          * @param dataLength
 364          *            The length of the stream
 365          * @return The limited input stream
 366          */
 367         private synchronized LimitedInputStream getInputStream(long dataLength) {
 368                 if (dataLength <= 0) {
 369                         return new LimitedInputStream(null, 0);
 370                 }
 371                 return new LimitedInputStream(remoteInputStream, dataLength);
 372         }
 373
 374         /**
 375          * A wrapper around an {@link InputStream} that only supplies a limit number
 376          * of bytes from the underlying input stream.
 377          *
 378          * @author David ‘Bombe’ Roden &lt;bombe@freenetproject.org&gt;
 379          */
 380         private static class LimitedInputStream extends FilterInputStream {
 381
 382                 /** The remaining number of bytes that can be read. */
 383                 private long remaining;
 384
 385                 /**
 386                  * Creates a new LimitedInputStream that supplies at most
 387                  * <code>length</code> bytes from the given input stream.
 388                  *
 389                  * @param inputStream
 390                  *            The input stream
 391                  * @param length
 392                  *            The number of bytes to read
 393                  */
 394                 public LimitedInputStream(InputStream inputStream, long length) {
 395                         super(inputStream);
 396                         remaining = length;
 397                 }
 398
 399                 /**
 400                  * @see java.io.FilterInputStream#available()
 401                  */
 402                 @Override
 403                 public synchronized int available() throws IOException {
 404                         if (remaining == 0) {
 405                                 return 0;
 406                         }
 407                         return (int) Math.min(super.available(), Math.min(Integer.MAX_VALUE, remaining));
 408                 }
 409
 410                 /**
 411                  * @see java.io.FilterInputStream#read()
 412                  */
 413                 @Override
 414                 public synchronized int read() throws IOException {
 415                         int read = -1;
 416                         if (remaining > 0) {
 417                                 read = super.read();
 418                                 remaining--;
 419                         }
 420                         return read;
 421                 }
 422
 423                 /**
 424                  * @see java.io.FilterInputStream#read(byte[], int, int)
 425                  */
 426                 @Override
 427                 public synchronized int read(byte[] b, int off, int len) throws IOException {
 428                         if (remaining == 0) {
 429                                 return -1;
 430                         }
 431                         int toCopy = (int) Math.min(len, Math.min(remaining, Integer.MAX_VALUE));
 432                         int read = super.read(b, off, toCopy);
 433                         remaining -= read;
 434                         return read;
 435                 }
 436
 437                 /**
 438                  * @see java.io.FilterInputStream#skip(long)
 439                  */
 440                 @Override
 441                 public synchronized long skip(long n) throws IOException {
 442                         if ((n < 0) || (remaining == 0)) {
 443                                 return 0;
 444                         }
 445                         long skipped = super.skip(Math.min(n, remaining));
 446                         remaining -= skipped;
 447                         return skipped;
 448                 }
 449
 450                 /**
 451                  * {@inheritDoc} This method does nothing, as {@link #mark(int)} and
 452                  * {@link #reset()} are not supported.
 453                  *
 454                  * @see java.io.FilterInputStream#mark(int)
 455                  */
 456                 @Override
 457                 public void mark(int readlimit) {
 458                         /* do nothing. */
 459                 }
 460
 461                 /**
 462                  * {@inheritDoc}
 463                  *
 464                  * @see java.io.FilterInputStream#markSupported()
 465                  * @return <code>false</code>
 466                  */
 467                 @Override
 468                 public boolean markSupported() {
 469                         return false;
 470                 }
 471
 472                 /**
 473                  * {@inheritDoc} This method does nothing, as {@link #mark(int)} and
 474                  * {@link #reset()} are not supported.
 475                  *
 476                  * @see java.io.FilterInputStream#reset()
 477                  */
 478                 @Override
 479                 public void reset() throws IOException {
 480                         /* do nothing. */
 481                 }
 482
 483                 /**
 484                  * Consumes the input stream, i.e. read all bytes until the limit is
 485                  * reached.
 486                  *
 487                  * @throws IOException
 488                  *             if an I/O error occurs
 489                  */
 490                 public synchronized void consume() throws IOException {
 491                         while (remaining > 0) {
 492                                 skip(remaining);
 493                         }
 494                 }
 495
 496         }
 497
 498 }