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