FtpConnector


1   /*
2    * $Id: FtpConnector.java 3798 2006-11-04 04:07:14Z aperepel $
3    * --------------------------------------------------------------------------------------
4    * Copyright (c) MuleSource, Inc.  All rights reserved.  http://www.mulesource.com
5    *
6    * The software in this package is published under the terms of the MuleSource MPL
7    * license, a copy of which has been included with this distribution in the
8    * LICENSE.txt file.
9    */
10  
11  package org.mule.providers.ftp;
12  
13  import java.io.IOException  ;
14  import java.util.HashMap  ;
15  import java.util.Iterator  ;
16  import java.util.Map  ;
17  
18  import org.apache.commons.net.ftp.FTP;
19  import org.apache.commons.net.ftp.FTPClient;
20  import org.apache.commons.net.ftp.FTPReply;
21  import org.apache.commons.pool.ObjectPool;
22  import org.apache.commons.pool.PoolableObjectFactory;
23  import org.apache.commons.pool.impl.GenericObjectPool;
24  import org.mule.config.i18n.Message;
25  import org.mule.config.i18n.Messages;
26  import org.mule.providers.AbstractServiceEnabledConnector;
27  import org.mule.providers.file.FilenameParser;
28  import org.mule.providers.file.SimpleFilenameParser;
29  import org.mule.umo.UMOComponent;
30  import org.mule.umo.UMOException;
31  import org.mule.umo.endpoint.UMOEndpoint;
32  import org.mule.umo.endpoint.UMOEndpointURI;
33  import org.mule.umo.endpoint.UMOImmutableEndpoint;
34  import org.mule.umo.provider.ConnectorException;
35  import org.mule.umo.provider.UMOMessageReceiver;
36  
37  public class FtpConnector extends AbstractServiceEnabledConnector
38  {
39      public static final String   PROPERTY_POLLING_FREQUENCY = "pollingFrequency";
40      public static final String   PROPERTY_FILENAME = "filename";
41      public static final String   PROPERTY_OUTPUT_PATTERN = "outputPattern";
42      public static final String   PROPERTY_PASSIVE_MODE = "passive";
43      public static final String   PROPERTY_BINARY_TRANSFER = "binary";
44  
45      /**
46       * Time in milliseconds to poll. On each poll the poll() method is called
47       */
48      private long pollingFrequency = 0;
49  
50      private String   outputPattern = null;
51  
52      private FilenameParser filenameParser = new SimpleFilenameParser();
53  
54      private boolean passive = true;
55  
56      private boolean binary = true;
57  
58      /**
59       * Whether to test FTP connection on each take from pool.
60       */
61      private boolean validateConnections = true;
62  
63      private Map   pools = new HashMap  ();
64  
65      public String   getProtocol()
66      {
67          return "ftp";
68      }
69  
70      public UMOMessageReceiver createReceiver(UMOComponent component, UMOEndpoint endpoint) throws Exception  
71      {
72          long polling = pollingFrequency;
73          Map   props = endpoint.getProperties();
74          if (props != null)
75          {
76              // Override properties on the endpoint for the specific endpoint
77              String   tempPolling = (String  )props.get(PROPERTY_POLLING_FREQUENCY);
78              if (tempPolling != null)
79              {
80                  polling = Long.parseLong(tempPolling);
81              }
82          }
83          if (polling <= 0)
84          {
85              polling = 1000;
86          }
87          logger.debug("set polling frequency to: " + polling);
88          return serviceDescriptor.createMessageReceiver(this, component, endpoint, new Object  []{new Long  (
89              polling)});
90      }
91  
92      /**
93       * @return Returns the pollingFrequency.
94       */
95      public long getPollingFrequency()
96      {
97          return pollingFrequency;
98      }
99  
100     /**
101      * @param pollingFrequency The pollingFrequency to set.
102      */
103     public void setPollingFrequency(long pollingFrequency)
104     {
105         this.pollingFrequency = pollingFrequency;
106     }
107 
108     public FTPClient getFtp(UMOEndpointURI uri) throws Exception  
109     {
110         ObjectPool pool = getFtpPool(uri);
111         return (FTPClient)pool.borrowObject();
112     }
113 
114     public void releaseFtp(UMOEndpointURI uri, FTPClient client) throws Exception  
115     {
116         if (isCreateDispatcherPerRequest())
117         {
118             destroyFtp(uri, client);
119         }
120         else
121         {
122             if (client != null && client.isConnected())
123             {
124                 ObjectPool pool = getFtpPool(uri);
125                 pool.returnObject(client);
126             }
127         }
128     }
129 
130     public void destroyFtp(UMOEndpointURI uri, FTPClient client) throws Exception  
131     {
132         if (client != null && client.isConnected())
133         {
134             ObjectPool pool = getFtpPool(uri);
135             pool.invalidateObject(client);
136         }
137     }
138 
139     protected synchronized ObjectPool getFtpPool(UMOEndpointURI uri)
140     {
141         String   key = uri.getUsername() + ":" + uri.getPassword() + "@" + uri.getHost() + ":" + uri.getPort();
142         ObjectPool pool = (ObjectPool)pools.get(key);
143         if (pool == null)
144         {
145             pool = new GenericObjectPool(new FtpConnectionFactory(uri));
146             ((GenericObjectPool)pool).setTestOnBorrow(this.validateConnections);
147             pools.put(key, pool);
148         }
149         return pool;
150     }
151 
152     /**
153      * @author <a HREF="mailto:gnt@codehaus.org">Guillaume Nodet</a>
154      */
155     protected class FtpConnectionFactory implements PoolableObjectFactory
156     {
157         private UMOEndpointURI uri;
158 
159         public FtpConnectionFactory(UMOEndpointURI uri)
160         {
161             this.uri = uri;
162         }
163 
164         public Object   makeObject() throws Exception  
165         {
166             FTPClient client = new FTPClient();
167             try
168             {
169                 if (uri.getPort() > 0)
170                 {
171                     client.connect(uri.getHost(), uri.getPort());
172                 }
173                 else
174                 {
175                     client.connect(uri.getHost());
176                 }
177                 if (!FTPReply.isPositiveCompletion(client.getReplyCode()))
178                 {
179                     throw new IOException  ("Ftp error: " + client.getReplyCode());
180                 }
181                 if (!client.login(uri.getUsername(), uri.getPassword()))
182                 {
183                     throw new IOException  ("Ftp error: " + client.getReplyCode());
184                 }
185                 if (!client.setFileType(FTP.BINARY_FILE_TYPE))
186                 {
187                     throw new IOException  ("Ftp error. Couldn't set BINARY transfer type.");
188                 }
189             }
190             catch (Exception   e)
191             {
192                 if (client.isConnected())
193                 {
194                     client.disconnect();
195                 }
196                 throw e;
197             }
198             return client;
199         }
200 
201         public void destroyObject(Object   obj) throws Exception  
202         {
203             FTPClient client = (FTPClient)obj;
204             client.logout();
205             client.disconnect();
206         }
207 
208         public boolean validateObject(Object   obj)
209         {
210             FTPClient client = (FTPClient)obj;
211             try
212             {
213                 client.sendNoOp();
214                 return true;
215             }
216             catch (IOException   e)
217             {
218                 return false;
219             }
220         }
221 
222         public void activateObject(Object   obj) throws Exception  
223         {
224             FTPClient client = (FTPClient)obj;
225             client.setReaderThread(true);
226         }
227 
228         public void passivateObject(Object   obj) throws Exception  
229         {
230             FTPClient client = (FTPClient)obj;
231             client.setReaderThread(false);
232         }
233     }
234 
235     protected void doStop() throws UMOException
236     {
237         try
238         {
239             for (Iterator   it = pools.values().iterator(); it.hasNext();)
240             {
241                 ObjectPool pool = (ObjectPool)it.next();
242                 pool.close();
243             }
244         }
245         catch (Exception   e)
246         {
247             throw new ConnectorException(new Message(Messages.FAILED_TO_STOP_X, "FTP Connector"), this, e);
248         }
249     }
250 
251     /**
252      * @return Returns the outputPattern.
253      */
254     public String   getOutputPattern()
255     {
256         return outputPattern;
257     }
258 
259     /**
260      * @param outputPattern The outputPattern to set.
261      */
262     public void setOutputPattern(String   outputPattern)
263     {
264         this.outputPattern = outputPattern;
265     }
266 
267     /**
268      * @return Returns the filenameParser.
269      */
270     public FilenameParser getFilenameParser()
271     {
272         return filenameParser;
273     }
274 
275     /**
276      * @param filenameParser The filenameParser to set.
277      */
278     public void setFilenameParser(FilenameParser filenameParser)
279     {
280         this.filenameParser = filenameParser;
281     }
282 
283     /**
284      * Getter for FTP passive mode.
285      * 
286      * @return true if using FTP passive mode
287      */
288     public boolean isPassive()
289     {
290         return passive;
291     }
292 
293     /**
294      * Setter for FTP passive mode.
295      * 
296      * @param passive passive mode flag
297      */
298     public void setPassive(final boolean passive)
299     {
300         this.passive = passive;
301     }
302 
303     /**
304      * Passive mode is OFF by default. The value is taken from the connector
305      * settings. In case there are any overriding properties set on the endpoint,
306      * those will be used.
307      * 
308      * @see #setPassive(boolean)
309      */
310     public void enterActiveOrPassiveMode(FTPClient client, UMOImmutableEndpoint endpoint)
311     {
312         // well, no endpoint URI here, as we have to use the most common denominator
313         // in API :(
314         final String   passiveString = (String  )endpoint.getProperty(FtpConnector.PROPERTY_PASSIVE_MODE);
315         if (passiveString == null)
316         {
317             // try the connector properties then
318             if (isPassive())
319             {
320                 if (logger.isTraceEnabled())
321                 {
322                     logger.trace("Entering FTP passive mode");
323                 }
324                 client.enterLocalPassiveMode();
325             }
326             else
327             {
328                 if (logger.isTraceEnabled())
329                 {
330                     logger.trace("Entering FTP active mode");
331                 }
332                 client.enterLocalActiveMode();
333             }
334         }
335         else
336         {
337             // override with endpoint's definition
338             final boolean passiveMode = Boolean.valueOf(passiveString).booleanValue();
339             if (passiveMode)
340             {
341                 if (logger.isTraceEnabled())
342                 {
343                     logger.trace("Entering FTP passive mode (endpoint override)");
344                 }
345                 client.enterLocalPassiveMode();
346             }
347             else
348             {
349                 if (logger.isTraceEnabled())
350                 {
351                     logger.trace("Entering FTP active mode (endpoint override)");
352                 }
353                 client.enterLocalActiveMode();
354             }
355         }
356     }
357 
358     /**
359      * Whether to test FTP connection on each take from pool.
360      */
361     public boolean isValidateConnections()
362     {
363         return validateConnections;
364     }
365 
366     /**
367      * Whether to test FTP connection on each take from pool. This takes care of a
368      * failed (or restarted) FTP server at the expense of an additional NOOP command
369      * packet being sent, but increases overall availability. <p/> Disable to gain
370      * slight performance gain or if you are absolutely sure of the FTP server
371      * availability. <p/> The default value is <code>true</code>
372      */
373     public void setValidateConnections(final boolean validateConnections)
374     {
375         this.validateConnections = validateConnections;
376     }
377 
378     /**
379      * Getter for FTP transfer type.
380      * 
381      * @return true if using FTP binary type
382      */
383     public boolean isBinary()
384     {
385         return binary;
386     }
387 
388     /**
389      * Setter for FTP transfer type.
390      * 
391      * @param binary binary type flag
392      */
393     public void setBinary(final boolean binary)
394     {
395         this.binary = binary;
396     }
397 
398     /**
399      * Transfer type is BINARY by default. The value is taken from the connector
400      * settings. In case there are any overriding properties set on the endpoint,
401      * those will be used. <p/> The alternative type is ASCII. <p/>
402      * 
403      * @see #setBinary(boolean)
404      */
405     public void setupFileType(FTPClient client, UMOImmutableEndpoint endpoint) throws Exception  
406     {
407         int type;
408 
409         // well, no endpoint URI here, as we have to use the most common denominator
410         // in API :(
411         final String   binaryTransferString = (String  )endpoint.getProperty(FtpConnector.PROPERTY_BINARY_TRANSFER);
412         if (binaryTransferString == null)
413         {
414             // try the connector properties then
415             if (isBinary())
416             {
417                 if (logger.isTraceEnabled())
418                 {
419                     logger.trace("Using FTP BINARY type");
420                 }
421                 type = FTP.BINARY_FILE_TYPE;
422             }
423             else
424             {
425                 if (logger.isTraceEnabled())
426                 {
427                     logger.trace("Using FTP ASCII type");
428                 }
429                 type = FTP.ASCII_FILE_TYPE;
430             }
431         }
432         else
433         {
434             // override with endpoint's definition
435             final boolean binaryTransfer = Boolean.valueOf(binaryTransferString).booleanValue();
436             if (binaryTransfer)
437             {
438                 if (logger.isTraceEnabled())
439                 {
440                     logger.trace("Using FTP BINARY type (endpoint override)");
441                 }
442                 type = FTP.BINARY_FILE_TYPE;
443             }
444             else
445             {
446                 if (logger.isTraceEnabled())
447                 {
448                     logger.trace("Using FTP ASCII type (endpoint override)");
449                 }
450                 type = FTP.ASCII_FILE_TYPE;
451             }
452         }
453 
454         client.setFileType(type);
455     }
456 
457 }
458
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags