Throw an exception when the node is not connected.
[jSite.git] / 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 <droden@gmail.com>
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 }