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.File;
   20   import java.io.FileInputStream;
   21   import java.io.IOException;
   22   import java.io.InputStream;
   23   import java.net.ConnectException;
   24   import java.net.MalformedURLException;
   25   import java.net.URL;
   26   
   27   import javax.jms.JMSException;
   28   
   29   import org.apache.activemq.command.ActiveMQBlobMessage;
   30   import org.apache.commons.net.ftp.FTPClient;
   31   
   32   /**
   33    * A FTP implementation of {@link BlobUploadStrategy}.
   34    */
   35   public class FTPBlobUploadStrategy implements BlobUploadStrategy {
   36   	
   37   	private URL url;
   38   	private String ftpUser = "";
   39   	private String ftpPass = "";
   40   	private BlobTransferPolicy transferPolicy;
   41   	
   42   	public FTPBlobUploadStrategy(BlobTransferPolicy transferPolicy) throws MalformedURLException {
   43   		this.transferPolicy = transferPolicy;
   44   		this.url = new URL(this.transferPolicy.getUploadUrl());
   45   		
   46   		setUserInformation(url.getUserInfo());
   47   	}
   48   
   49   	public URL uploadFile(ActiveMQBlobMessage message, File file)
   50   			throws JMSException, IOException {
   51   		return uploadStream(message, new FileInputStream(file));
   52   	}
   53   
   54   	public URL uploadStream(ActiveMQBlobMessage message, InputStream in)
   55   			throws JMSException, IOException {
   56   		String connectUrl = url.getHost();
   57   		int port = url.getPort() < 1 ? 21 : url.getPort();
   58   		
   59   		FTPClient ftp = new FTPClient();
   60   		try {
   61           	ftp.connect(connectUrl, port);
   62           } catch(ConnectException e) {
   63           	throw new JMSException("Problem connecting the FTP-server");
   64           }
   65   		if(!ftp.login(ftpUser, ftpPass)) {
   66   			ftp.quit();
   67   			ftp.disconnect();
   68   			throw new JMSException("Cant Authentificate to FTP-Server");
   69   		}
   70   		String path = url.getPath();
   71           String workingDir = path.substring(0, path.lastIndexOf("/"));
   72   		String filename = message.getMessageId().toString().replaceAll(":", "_");
   73           ftp.setFileType(FTPClient.BINARY_FILE_TYPE);
   74           
   75           String url;
   76           if(!ftp.changeWorkingDirectory(workingDir)) {
   77           	url = this.url.toString().replaceFirst(this.url.getPath(), "")+"/";
   78           } else {
   79           	url = this.url.toString();
   80           }
   81           
   82   		ftp.storeFile(filename, in);
   83   		ftp.quit();
   84   		ftp.disconnect();
   85   		
   86   		return new URL(url + filename);
   87   	}
   88   	
   89   	private void setUserInformation(String userInfo) {
   90   		if(userInfo != null) {
   91   			String[] userPass = userInfo.split(":");
   92   			if(userPass.length > 0) this.ftpUser = userPass[0];
   93   			if(userPass.length > 1) this.ftpPass = userPass[1];
   94   		} else {
   95   			this.ftpUser = "anonymous";
   96   			this.ftpPass = "anonymous";
   97   		}
   98   	}
   99   
  100   }