core/src/main/java/org/apache/ftpserver/impl/IODataConnection.java - mina-ftpserver - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *  http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 package org.apache.ftpserver.impl;

 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
 import java.io.Writer;
 import java.net.Socket;
 import java.nio.charset.StandardCharsets;
 import java.util.zip.DeflaterOutputStream;
 import java.util.zip.InflaterInputStream;

 import org.apache.ftpserver.ftplet.DataConnection;
 import org.apache.ftpserver.ftplet.DataType;
 import org.apache.ftpserver.ftplet.FtpSession;
 import org.apache.ftpserver.usermanager.impl.TransferRateRequest;
 import org.apache.ftpserver.util.IoUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * <strong>Internal class, do not use directly.</strong>
  *
  * An active open data connection, used for transfering data over the data
  * connection.
  *
  * @author <a href="http://mina.apache.org">Apache MINA Project</a>
  */
 public class IODataConnection implements DataConnection {

     private final Logger LOG = LoggerFactory
     .getLogger(IODataConnection.class);


     private static final byte[] EOL = System.getProperty("line.separator").getBytes();

     private final FtpIoSession session;

     private final Socket socket;

     private final ServerDataConnectionFactory factory;

     public IODataConnection(final Socket socket, final FtpIoSession session,
             final ServerDataConnectionFactory factory) {
         this.session = session;
         this.socket = socket;
         this.factory = factory;
     }

     /**
      * Get data input stream. The return value will never be null.
      */
     private InputStream getDataInputStream() throws IOException {
         try {

             // get data socket
             Socket dataSoc = socket;
             if (dataSoc == null) {
                 throw new IOException("Cannot open data connection.");
             }

             // create input stream
             InputStream is = dataSoc.getInputStream();
             if (factory.isZipMode()) {
                 is = new InflaterInputStream(is);
             }
             return is;
         } catch (IOException ex) {
             factory.closeDataConnection();
             throw ex;
         }
     }

     /**
      * Get data output stream. The return value will never be null.
      */
     private OutputStream getDataOutputStream() throws IOException {
         try {

             // get data socket
             Socket dataSoc = socket;
             if (dataSoc == null) {
                 throw new IOException("Cannot open data connection.");
             }

             // create output stream
             OutputStream os = dataSoc.getOutputStream();
             if (factory.isZipMode()) {
                 os = new DeflaterOutputStream(os);
             }
             return os;
         } catch (IOException ex) {
             factory.closeDataConnection();
             throw ex;
         }
     }

     /*
      * (non-Javadoc)
      *
      * @seeorg.apache.ftpserver.FtpDataConnection2#transferFromClient(java.io.
      * OutputStream)
      */
     public final long transferFromClient(FtpSession session,
             final OutputStream out) throws IOException {
         TransferRateRequest transferRateRequest = new TransferRateRequest();
         transferRateRequest = (TransferRateRequest) session.getUser()
                 .authorize(transferRateRequest);
         int maxRate = 0;
         if (transferRateRequest != null) {
             maxRate = transferRateRequest.getMaxUploadRate();
         }

         InputStream is = getDataInputStream();
         try {
             return transfer(session, false, is, out, maxRate);
         } finally {
             IoUtils.close(is);
         }
     }

     /*
      * (non-Javadoc)
      *
      * @see
      * org.apache.ftpserver.FtpDataConnection2#transferToClient(java.io.InputStream
      * )
      */
     public final long transferToClient(FtpSession session, final InputStream in)
             throws IOException {
         TransferRateRequest transferRateRequest = new TransferRateRequest();
         transferRateRequest = (TransferRateRequest) session.getUser()
                 .authorize(transferRateRequest);
         int maxRate = 0;
         if (transferRateRequest != null) {
             maxRate = transferRateRequest.getMaxDownloadRate();
         }

         OutputStream out = getDataOutputStream();
         try {
             return transfer(session, true, in, out, maxRate);
         } finally {
             IoUtils.close(out);
         }
     }

     /*
      * (non-Javadoc)
      *
      * @see
      * org.apache.ftpserver.FtpDataConnection2#transferToClient(java.lang.String
      * )
      */
     public final void transferToClient(FtpSession session, final String str)
             throws IOException {
         OutputStream out = getDataOutputStream();
         Writer writer = null;
         try {
             writer = new OutputStreamWriter(out, StandardCharsets.UTF_8);
             writer.write(str);

             // update session
             if (session instanceof DefaultFtpSession) {
                 ((DefaultFtpSession) session).increaseWrittenDataBytes(str
                         .getBytes(StandardCharsets.UTF_8).length);
             }
         } finally {
             if (writer != null) {
                 writer.flush();
             }
             IoUtils.close(writer);
         }

     }

     private final long transfer(FtpSession session, boolean isWrite,
             final InputStream in, final OutputStream out, final int maxRate)
             throws IOException {
         long transferredSize = 0L;

         boolean isAscii = session.getDataType() == DataType.ASCII;
         long startTime = System.currentTimeMillis();
         byte[] buff = new byte[4096];

         BufferedInputStream bis = null;
         BufferedOutputStream bos = null;
         try {
             bis = IoUtils.getBufferedInputStream(in);

             bos = IoUtils.getBufferedOutputStream(out);

             DefaultFtpSession defaultFtpSession = null;
             if (session instanceof DefaultFtpSession) {
                 defaultFtpSession = (DefaultFtpSession) session;
             }

             byte lastByte = 0;
             while (true) {

                 // if current rate exceeds the max rate, sleep for 50ms
                 // and again check the current transfer rate
                 if (maxRate > 0) {

                     // prevent "divide by zero" exception
                     long interval = System.currentTimeMillis() - startTime;
                     if (interval == 0) {
                         interval = 1;
                     }

                     // check current rate
                     long currRate = (transferredSize * 1000L) / interval;
                     if (currRate > maxRate) {
                         try {
                             Thread.sleep(50);
                         } catch (InterruptedException ex) {
                             break;
                         }
                         continue;
                     }
                 }

                 // read data
                 int count = bis.read(buff);

                 if (count == -1) {
                     break;
                 }

                 // update MINA session
                 if (defaultFtpSession != null) {
                     if (isWrite) {
                         defaultFtpSession.increaseWrittenDataBytes(count);
                     } else {
                         defaultFtpSession.increaseReadDataBytes(count);
                     }
                 }

                 // write data
                 // if ascii, replace \n by \r\n
                 if (isAscii) {
                     for (int i = 0; i < count; ++i) {
                         byte b = buff[i];
                         if(isWrite) {
                             if (b == '\n' && lastByte != '\r') {
                                 bos.write('\r');
                             }

                             bos.write(b);
                         } else {
                             if(b == '\n') {
                                 // for reads, we should always get \r\n
                                 // so what we do here is to ignore \n bytes
                                 // and on \r dump the system local line ending.
                                 // Some clients won't transform new lines into \r\n so we make sure we don't delete new lines
                                 if (lastByte != '\r'){
                                     bos.write(EOL);
                                 }
                             } else if(b == '\r') {
                                 bos.write(EOL);
                             } else {
                                 // not a line ending, just output
                                 bos.write(b);
                             }
                         }
                         // store this byte so that we can compare it for line endings
                         lastByte = b;
                     }
                 } else {
                     bos.write(buff, 0, count);
                 }

                 transferredSize += count;

                 notifyObserver();
             }
         } catch(IOException e) {
             LOG.warn("Exception during data transfer, closing data connection socket", e);
             factory.closeDataConnection();
             throw e;
         } catch(RuntimeException e) {
             LOG.warn("Exception during data transfer, closing data connection socket", e);
             factory.closeDataConnection();
             throw e;
         } finally {
             if (bos != null) {
                 bos.flush();
             }
         }

         return transferredSize;
     }

     /**
      * Notify connection manager observer.
      */
     protected void notifyObserver() {
         session.updateLastAccessTime();

         // TODO this has been moved from AbstractConnection, do we need to keep
         // it?
         // serverContext.getConnectionManager().updateConnection(this);
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	package org.apache.ftpserver.impl;

	import java.io.BufferedInputStream;
	import java.io.BufferedOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.io.OutputStreamWriter;
	import java.io.Writer;
	import java.net.Socket;
	import java.nio.charset.StandardCharsets;
	import java.util.zip.DeflaterOutputStream;
	import java.util.zip.InflaterInputStream;

	import org.apache.ftpserver.ftplet.DataConnection;
	import org.apache.ftpserver.ftplet.DataType;
	import org.apache.ftpserver.ftplet.FtpSession;
	import org.apache.ftpserver.usermanager.impl.TransferRateRequest;
	import org.apache.ftpserver.util.IoUtils;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* <strong>Internal class, do not use directly.</strong>
	*
	* An active open data connection, used for transfering data over the data
	* connection.
	*
	* @author <a href="http://mina.apache.org">Apache MINA Project</a>
	*/
	public class IODataConnection implements DataConnection {

	private final Logger LOG = LoggerFactory
	.getLogger(IODataConnection.class);


	private static final byte[] EOL = System.getProperty("line.separator").getBytes();

	private final FtpIoSession session;

	private final Socket socket;

	private final ServerDataConnectionFactory factory;

	public IODataConnection(final Socket socket, final FtpIoSession session,
	final ServerDataConnectionFactory factory) {
	this.session = session;
	this.socket = socket;
	this.factory = factory;
	}

	/**
	* Get data input stream. The return value will never be null.
	*/
	private InputStream getDataInputStream() throws IOException {
	try {

	// get data socket
	Socket dataSoc = socket;
	if (dataSoc == null) {
	throw new IOException("Cannot open data connection.");
	}

	// create input stream
	InputStream is = dataSoc.getInputStream();
	if (factory.isZipMode()) {
	is = new InflaterInputStream(is);
	}
	return is;
	} catch (IOException ex) {
	factory.closeDataConnection();
	throw ex;
	}
	}

	/**
	* Get data output stream. The return value will never be null.
	*/
	private OutputStream getDataOutputStream() throws IOException {
	try {

	// get data socket
	Socket dataSoc = socket;
	if (dataSoc == null) {
	throw new IOException("Cannot open data connection.");
	}

	// create output stream
	OutputStream os = dataSoc.getOutputStream();
	if (factory.isZipMode()) {
	os = new DeflaterOutputStream(os);
	}
	return os;
	} catch (IOException ex) {
	factory.closeDataConnection();
	throw ex;
	}
	}

	/*
	* (non-Javadoc)
	*
	* @seeorg.apache.ftpserver.FtpDataConnection2#transferFromClient(java.io.
	* OutputStream)
	*/
	public final long transferFromClient(FtpSession session,
	final OutputStream out) throws IOException {
	TransferRateRequest transferRateRequest = new TransferRateRequest();
	transferRateRequest = (TransferRateRequest) session.getUser()
	.authorize(transferRateRequest);
	int maxRate = 0;
	if (transferRateRequest != null) {
	maxRate = transferRateRequest.getMaxUploadRate();
	}

	InputStream is = getDataInputStream();
	try {
	return transfer(session, false, is, out, maxRate);
	} finally {
	IoUtils.close(is);
	}
	}

	/*
	* (non-Javadoc)
	*
	* @see
	* org.apache.ftpserver.FtpDataConnection2#transferToClient(java.io.InputStream
	* )
	*/
	public final long transferToClient(FtpSession session, final InputStream in)
	throws IOException {
	TransferRateRequest transferRateRequest = new TransferRateRequest();
	transferRateRequest = (TransferRateRequest) session.getUser()
	.authorize(transferRateRequest);
	int maxRate = 0;
	if (transferRateRequest != null) {
	maxRate = transferRateRequest.getMaxDownloadRate();
	}

	OutputStream out = getDataOutputStream();
	try {
	return transfer(session, true, in, out, maxRate);
	} finally {
	IoUtils.close(out);
	}
	}

	/*
	* (non-Javadoc)
	*
	* @see
	* org.apache.ftpserver.FtpDataConnection2#transferToClient(java.lang.String
	* )
	*/
	public final void transferToClient(FtpSession session, final String str)
	throws IOException {
	OutputStream out = getDataOutputStream();
	Writer writer = null;
	try {
	writer = new OutputStreamWriter(out, StandardCharsets.UTF_8);
	writer.write(str);

	// update session
	if (session instanceof DefaultFtpSession) {
	((DefaultFtpSession) session).increaseWrittenDataBytes(str
	.getBytes(StandardCharsets.UTF_8).length);
	}
	} finally {
	if (writer != null) {
	writer.flush();
	}
	IoUtils.close(writer);
	}

	}

	private final long transfer(FtpSession session, boolean isWrite,
	final InputStream in, final OutputStream out, final int maxRate)
	throws IOException {
	long transferredSize = 0L;

	boolean isAscii = session.getDataType() == DataType.ASCII;
	long startTime = System.currentTimeMillis();
	byte[] buff = new byte[4096];

	BufferedInputStream bis = null;
	BufferedOutputStream bos = null;
	try {
	bis = IoUtils.getBufferedInputStream(in);

	bos = IoUtils.getBufferedOutputStream(out);

	DefaultFtpSession defaultFtpSession = null;
	if (session instanceof DefaultFtpSession) {
	defaultFtpSession = (DefaultFtpSession) session;
	}

	byte lastByte = 0;
	while (true) {

	// if current rate exceeds the max rate, sleep for 50ms
	// and again check the current transfer rate
	if (maxRate > 0) {

	// prevent "divide by zero" exception
	long interval = System.currentTimeMillis() - startTime;
	if (interval == 0) {
	interval = 1;
	}

	// check current rate
	long currRate = (transferredSize * 1000L) / interval;
	if (currRate > maxRate) {
	try {
	Thread.sleep(50);
	} catch (InterruptedException ex) {
	break;
	}
	continue;
	}
	}

	// read data
	int count = bis.read(buff);

	if (count == -1) {
	break;
	}

	// update MINA session
	if (defaultFtpSession != null) {
	if (isWrite) {
	defaultFtpSession.increaseWrittenDataBytes(count);
	} else {
	defaultFtpSession.increaseReadDataBytes(count);
	}
	}

	// write data
	// if ascii, replace \n by \r\n
	if (isAscii) {
	for (int i = 0; i < count; ++i) {
	byte b = buff[i];
	if(isWrite) {
	if (b == '\n' && lastByte != '\r') {
	bos.write('\r');
	}

	bos.write(b);
	} else {
	if(b == '\n') {
	// for reads, we should always get \r\n
	// so what we do here is to ignore \n bytes
	// and on \r dump the system local line ending.
	// Some clients won't transform new lines into \r\n so we make sure we don't delete new lines
	if (lastByte != '\r'){
	bos.write(EOL);
	}
	} else if(b == '\r') {
	bos.write(EOL);
	} else {
	// not a line ending, just output
	bos.write(b);
	}
	}
	// store this byte so that we can compare it for line endings
	lastByte = b;
	}
	} else {
	bos.write(buff, 0, count);
	}

	transferredSize += count;

	notifyObserver();
	}
	} catch(IOException e) {
	LOG.warn("Exception during data transfer, closing data connection socket", e);
	factory.closeDataConnection();
	throw e;
	} catch(RuntimeException e) {
	LOG.warn("Exception during data transfer, closing data connection socket", e);
	factory.closeDataConnection();
	throw e;
	} finally {
	if (bos != null) {
	bos.flush();
	}
	}

	return transferredSize;
	}

	/**
	* Notify connection manager observer.
	*/
	protected void notifyObserver() {
	session.updateLastAccessTime();

	// TODO this has been moved from AbstractConnection, do we need to keep
	// it?
	// serverContext.getConnectionManager().updateConnection(this);
	}
	}