1   /**
    2    * Licensed to the Apache Software Foundation (ASF) under one or more
    3    * contributor license agreements.  See the NOTICE file distributed with
    4    * this work for additional information regarding copyright ownership.
    5    * The ASF licenses this file to You under the Apache License, Version 2.0
    6    * (the "License"); you may not use this file except in compliance with
    7    * the License.  You may obtain a copy of the License at
    8    *
    9    *      http://www.apache.org/licenses/LICENSE-2.0
   10    *
   11    * Unless required by applicable law or agreed to in writing, software
   12    * distributed under the License is distributed on an "AS IS" BASIS,
   13    * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   14    * See the License for the specific language governing permissions and
   15    * limitations under the License.
   16    */
   17   package org.apache.activemq.blob;
   18   
   19   import java.io.IOException;
   20   import java.io.InputStream;
   21   import java.io.FilterInputStream;
   22   import java.net.ConnectException;
   23   import java.net.URL;
   24   
   25   import javax.jms.JMSException;
   26   
   27   import org.apache.activemq.command.ActiveMQBlobMessage;
   28   import org.apache.commons.net.ftp.FTPClient;
   29   
   30   /**
   31    * A FTP implementation for {@link BlobDownloadStrategy}.
   32    */
   33   public class FTPBlobDownloadStrategy implements BlobDownloadStrategy {
   34       private String ftpUser;
   35       private String ftpPass;
   36   
   37       public InputStream getInputStream(ActiveMQBlobMessage message) throws IOException, JMSException {
   38           URL url = message.getURL();
   39   
   40           setUserInformation(url.getUserInfo());
   41           String connectUrl = url.getHost();
   42           int port = url.getPort() < 1 ? 21 : url.getPort();
   43   
   44           final FTPClient ftp = new FTPClient();
   45           try {
   46               ftp.connect(connectUrl, port);
   47           } catch(ConnectException e) {
   48               throw new JMSException("Problem connecting the FTP-server");
   49           }
   50   
   51           if(!ftp.login(ftpUser, ftpPass)) {
   52               ftp.quit();
   53               ftp.disconnect();
   54               throw new JMSException("Cant Authentificate to FTP-Server");
   55           }
   56           String path = url.getPath();
   57           String workingDir = path.substring(0, path.lastIndexOf("/"));
   58           String file = path.substring(path.lastIndexOf("/")+1);
   59   
   60           ftp.changeWorkingDirectory(workingDir);
   61           ftp.setFileType(FTPClient.BINARY_FILE_TYPE);
   62   
   63           InputStream input = new FilterInputStream(ftp.retrieveFileStream(file)) {
   64   
   65               public void close() throws IOException {
   66                   in.close();
   67                   ftp.quit();
   68                   ftp.disconnect();
   69               }
   70           };
   71   
   72           return input;
   73       }
   74   
   75       private void setUserInformation(String userInfo) {
   76           if(userInfo != null) {
   77               String[] userPass = userInfo.split(":");
   78               if(userPass.length > 0) this.ftpUser = userPass[0];
   79               if(userPass.length > 1) this.ftpPass = userPass[1];
   80           } else {
   81               this.ftpUser = "anonymous";
   82               this.ftpPass = "anonymous";
   83           }
   84       }
   85   
   86   }