src/main/java/de/todesbaum/util/freenet/fcp2/Connection.java

   1 /*
   2  * jSite - Connection.java - Copyright © 2006–2012 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 de.todesbaum.util.freenet.fcp2;
  20
  21 import static java.util.logging.Level.FINE;
  22 import static java.util.logging.Logger.getLogger;
  23
  24 import java.io.File;
  25 import java.io.FileOutputStream;
  26 import java.io.IOException;
  27 import java.io.InputStream;
  28 import java.io.OutputStream;
  29 import java.io.OutputStreamWriter;
  30 import java.io.Writer;
  31 import java.net.Socket;
  32 import java.nio.charset.Charset;
  33 import java.util.ArrayList;
  34 import java.util.List;
  35 import java.util.logging.Logger;
  36
  37 import net.pterodactylus.util.io.Closer;
  38 import net.pterodactylus.util.io.StreamCopier;
  39 import net.pterodactylus.util.io.StreamCopier.ProgressListener;
  40 import de.todesbaum.util.io.LineInputStream;
  41 import de.todesbaum.util.io.TempFileInputStream;
  42
  43 /**
  44  * A physical connection to a Freenet node.
  45  *
  46  * @author David Roden &lt;droden@gmail.com&gt;
  47  * @version $Id$
  48  */
  49 public class Connection {
  50
  51         private static final Logger logger = getLogger(Connection.class.getName());
  52
  53         /** The listeners that receive events from this connection. */
  54         private List<ConnectionListener> connectionListeners = new ArrayList<ConnectionListener>();
  55
  56         /** The node this connection is connected to. */
  57         private final Node node;
  58
  59         /** The name of this connection. */
  60         private final String name;
  61
  62         /** The network socket of this connection. */
  63         private Socket nodeSocket;
  64
  65         /** The input stream that reads from the socket. */
  66         private InputStream nodeInputStream;
  67
  68         /** The output stream that writes to the socket. */
  69         private OutputStream nodeOutputStream;
  70
  71         /** The thread that reads from the socket. */
  72         private NodeReader nodeReader;
  73
  74         /** A writer for the output stream. */
  75         private Writer nodeWriter;
  76
  77         /** The NodeHello message sent by the node on connect. */
  78         protected Message nodeHello;
  79
  80         /** The temp directory to use. */
  81         private String tempDirectory;
  82
  83         /**
  84          * Creates a new connection to the specified node with the specified name.
  85          *
  86          * @param node
  87          *            The node to connect to
  88          * @param name
  89          *            The name of this connection
  90          */
  91         public Connection(Node node, String name) {
  92                 this.node = node;
  93                 this.name = name;
  94         }
  95
  96         /**
  97          * Adds a listener that gets notified on connection events.
  98          *
  99          * @param connectionListener
 100          *            The listener to add
 101          */
 102         public void addConnectionListener(ConnectionListener connectionListener) {
 103                 connectionListeners.add(connectionListener);
 104         }
 105
 106         /**
 107          * Removes a listener from the list of registered listeners. Only the first
 108          * matching listener is removed.
 109          *
 110          * @param connectionListener
 111          *            The listener to remove
 112          * @see List#remove(java.lang.Object)
 113          */
 114         public void removeConnectionListener(ConnectionListener connectionListener) {
 115                 connectionListeners.remove(connectionListener);
 116         }
 117
 118         /**
 119          * Notifies listeners about a received message.
 120          *
 121          * @param message
 122          *            The received message
 123          */
 124         protected void fireMessageReceived(Message message) {
 125                 for (ConnectionListener connectionListener : connectionListeners) {
 126                         connectionListener.messageReceived(this, message);
 127                 }
 128         }
 129
 130         /**
 131          * Notifies listeners about the loss of the connection.
 132          */
 133         protected void fireConnectionTerminated() {
 134                 for (ConnectionListener connectionListener : connectionListeners) {
 135                         connectionListener.connectionTerminated(this);
 136                 }
 137         }
 138
 139         /**
 140          * Returns the name of the connection.
 141          *
 142          * @return The name of the connection
 143          */
 144         public String getName() {
 145                 return name;
 146         }
 147
 148         /**
 149          * Sets the temp directory to use for creation of temporary files.
 150          *
 151          * @param tempDirectory
 152          *            The temp directory to use, or {@code null} to use the default
 153          *            temp directory
 154          */
 155         public void setTempDirectory(String tempDirectory) {
 156                 this.tempDirectory = tempDirectory;
 157         }
 158
 159         /**
 160          * Connects to the node.
 161          *
 162          * @return <code>true</code> if the connection succeeded and the node
 163          *         returned a NodeHello message
 164          * @throws IOException
 165          *             if an I/O error occurs
 166          * @see #getNodeHello()
 167          */
 168         public synchronized boolean connect() throws IOException {
 169                 nodeSocket = null;
 170                 nodeInputStream = null;
 171                 nodeOutputStream = null;
 172                 nodeWriter = null;
 173                 nodeReader = null;
 174                 try {
 175                         nodeSocket = new Socket(node.getHostname(), node.getPort());
 176                         nodeSocket.setReceiveBufferSize(65535);
 177                         nodeInputStream = nodeSocket.getInputStream();
 178                         nodeOutputStream = nodeSocket.getOutputStream();
 179                         nodeWriter = new OutputStreamWriter(nodeOutputStream, Charset.forName("UTF-8"));
 180                         nodeReader = new NodeReader(nodeInputStream);
 181                         Thread nodeReaderThread = new Thread(nodeReader);
 182                         nodeReaderThread.setDaemon(true);
 183                         nodeReaderThread.start();
 184                         ClientHello clientHello = new ClientHello();
 185                         clientHello.setName(name);
 186                         clientHello.setExpectedVersion("2.0");
 187                         execute(clientHello);
 188                         synchronized (this) {
 189                                 try {
 190                                         wait();
 191                                 } catch (InterruptedException e) {
 192                                 }
 193                         }
 194                         return nodeHello != null;
 195                 } catch (IOException ioe1) {
 196                         disconnect();
 197                         throw ioe1;
 198                 }
 199         }
 200
 201         /**
 202          * Returns whether this connection is still connected to the node.
 203          *
 204          * @return <code>true</code> if this connection is still valid,
 205          *         <code>false</code> otherwise
 206          */
 207         public boolean isConnected() {
 208                 return (nodeHello != null) && (nodeSocket != null) && (nodeSocket.isConnected());
 209         }
 210
 211         /**
 212          * Returns the NodeHello message the node sent on connection.
 213          *
 214          * @return The NodeHello message of the node
 215          */
 216         public Message getNodeHello() {
 217                 return nodeHello;
 218         }
 219
 220         /**
 221          * Disconnects from the node.
 222          */
 223         public void disconnect() {
 224                 Closer.close(nodeWriter);
 225                 nodeWriter = null;
 226                 Closer.close(nodeOutputStream);
 227                 nodeOutputStream = null;
 228                 Closer.close(nodeInputStream);
 229                 nodeInputStream = null;
 230                 nodeInputStream = null;
 231                 Closer.close(nodeSocket);
 232                 nodeSocket = null;
 233                 synchronized (this) {
 234                         notify();
 235                 }
 236                 logger.log(FINE, "Connection terminated.", new Exception());
 237                 fireConnectionTerminated();
 238         }
 239
 240         /**
 241          * Executes the specified command.
 242          *
 243          * @param command
 244          *            The command to execute
 245          * @throws IllegalStateException
 246          *             if the connection is not connected
 247          * @throws IOException
 248          *             if an I/O error occurs
 249          */
 250         public synchronized void execute(Command command) throws IllegalStateException, IOException {
 251                 execute(command, null);
 252         }
 253
 254         /**
 255          * Executes the specified command.
 256          *
 257          * @param command
 258          *            The command to execute
 259          * @param progressListener
 260          *            A progress listener for a payload transfer
 261          * @throws IllegalStateException
 262          *             if the connection is not connected
 263          * @throws IOException
 264          *             if an I/O error occurs
 265          */
 266         public synchronized void execute(Command command, ProgressListener progressListener) throws IllegalStateException, IOException {
 267                 if (nodeSocket == null) {
 268                         throw new IllegalStateException("connection is not connected");
 269                 }
 270                 nodeWriter.write(command.getCommandName() + Command.LINEFEED);
 271                 command.write(nodeWriter);
 272                 nodeWriter.write("EndMessage" + Command.LINEFEED);
 273                 nodeWriter.flush();
 274                 if (command.hasPayload()) {
 275                         InputStream payloadInputStream = null;
 276                         try {
 277                                 payloadInputStream = command.getPayload();
 278                                 StreamCopier.copy(payloadInputStream, nodeOutputStream, progressListener, command.getPayloadLength());
 279                         } finally {
 280                                 Closer.close(payloadInputStream);
 281                         }
 282                         nodeOutputStream.flush();
 283                 }
 284         }
 285
 286         /**
 287          * The reader thread for this connection. This is essentially a thread that
 288          * reads lines from the node, creates messages from them and notifies
 289          * listeners about the messages.
 290          *
 291          * @author David Roden &lt;droden@gmail.com&gt;
 292          * @version $Id$
 293          */
 294         private class NodeReader implements Runnable {
 295
 296                 /** The input stream to read from. */
 297                 @SuppressWarnings("hiding")
 298                 private InputStream nodeInputStream;
 299
 300                 /**
 301                  * Creates a new reader that reads from the specified input stream.
 302                  *
 303                  * @param nodeInputStream
 304                  *            The input stream to read from
 305                  */
 306                 public NodeReader(InputStream nodeInputStream) {
 307                         this.nodeInputStream = nodeInputStream;
 308                 }
 309
 310                 /**
 311                  * Main loop of the reader. Lines are read and converted into
 312                  * {@link Message} objects.
 313                  */
 314                 @SuppressWarnings("synthetic-access")
 315                 public void run() {
 316                         LineInputStream nodeReader = null;
 317                         try {
 318                                 nodeReader = new LineInputStream(nodeInputStream);
 319                                 String line = "";
 320                                 Message message = null;
 321                                 while (line != null) {
 322                                         line = nodeReader.readLine();
 323                                         // System.err.println("> " + line);
 324                                         if (line == null) {
 325                                                 break;
 326                                         }
 327                                         if (message == null) {
 328                                                 message = new Message(line);
 329                                                 continue;
 330                                         }
 331                                         if ("Data".equals(line)) {
 332                                                 /* need to read message from stream now */
 333                                                 File tempFile = null;
 334                                                 try {
 335                                                         tempFile = File.createTempFile("fcpv2", "data", (tempDirectory != null) ? new File(tempDirectory) : null);
 336                                                         tempFile.deleteOnExit();
 337                                                         FileOutputStream tempFileOutputStream = new FileOutputStream(tempFile);
 338                                                         long dataLength = Long.parseLong(message.get("DataLength"));
 339                                                         StreamCopier.copy(nodeInputStream, tempFileOutputStream, dataLength);
 340                                                         tempFileOutputStream.close();
 341                                                         message.setPayloadInputStream(new TempFileInputStream(tempFile));
 342                                                 } catch (IOException ioe1) {
 343                                                         ioe1.printStackTrace();
 344                                                 }
 345                                         }
 346                                         if ("Data".equals(line) || "EndMessage".equals(line)) {
 347                                                 if (message.getName().equals("NodeHello")) {
 348                                                         nodeHello = message;
 349                                                         synchronized (Connection.this) {
 350                                                                 Connection.this.notify();
 351                                                         }
 352                                                 } else {
 353                                                         fireMessageReceived(message);
 354                                                 }
 355                                                 message = null;
 356                                                 continue;
 357                                         }
 358                                         int equalsPosition = line.indexOf('=');
 359                                         if (equalsPosition > -1) {
 360                                                 String key = line.substring(0, equalsPosition).trim();
 361                                                 String value = line.substring(equalsPosition + 1).trim();
 362                                                 if (key.equals("Identifier")) {
 363                                                         message.setIdentifier(value);
 364                                                 } else {
 365                                                         message.put(key, value);
 366                                                 }
 367                                                 continue;
 368                                         }
 369                                         /* skip lines consisting of whitespace only */
 370                                         if (line.trim().length() == 0) {
 371                                                 continue;
 372                                         }
 373                                         /* if we got here, some error occured! */
 374                                         throw new IOException("Unexpected line: " + line);
 375                                 }
 376                         } catch (IOException ioe1) {
 377                                 // ioe1.printStackTrace();
 378                         } finally {
 379                                 if (nodeReader != null) {
 380                                         try {
 381                                                 nodeReader.close();
 382                                         } catch (IOException ioe1) {
 383                                         }
 384                                 }
 385                                 if (nodeInputStream != null) {
 386                                         try {
 387                                                 nodeInputStream.close();
 388                                         } catch (IOException ioe1) {
 389                                         }
 390                                 }
 391                         }
 392                         Connection.this.disconnect();
 393                 }
 394
 395         }
 396
 397 }