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

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