ftpconnector.java

提供ESB 应用mule源代码 提供ESB 应用mule源代码

    public FilenameParser getFilenameParser()
    {
        return filenameParser;
    }

    /**
     * @param filenameParser The filenameParser to set.
     */
    public void setFilenameParser(FilenameParser filenameParser)
    {
        this.filenameParser = filenameParser;
    }

    /**
     * Getter for FTP passive mode.
     * 
     * @return true if using FTP passive mode
     */
    public boolean isPassive()
    {
        return passive;
    }

    /**
     * Setter for FTP passive mode.
     * 
     * @param passive passive mode flag
     */
    public void setPassive(final boolean passive)
    {
        this.passive = passive;
    }

    /**
     * Passive mode is OFF by default. The value is taken from the connector
     * settings. In case there are any overriding properties set on the endpoint,
     * those will be used.
     * 
     * @see #setPassive(boolean)
     */
    public void enterActiveOrPassiveMode(FTPClient client, ImmutableEndpoint endpoint)
    {
        // well, no endpoint URI here, as we have to use the most common denominator
        // in API :(
        final String passiveString = (String)endpoint.getProperty(FtpConnector.PROPERTY_PASSIVE_MODE);
        if (passiveString == null)
        {
            // try the connector properties then
            if (isPassive())
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Entering FTP passive mode");
                }
                client.enterLocalPassiveMode();
            }
            else
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Entering FTP active mode");
                }
                client.enterLocalActiveMode();
            }
        }
        else
        {
            // override with endpoint's definition
            final boolean passiveMode = Boolean.valueOf(passiveString).booleanValue();
            if (passiveMode)
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Entering FTP passive mode (endpoint override)");
                }
                client.enterLocalPassiveMode();
            }
            else
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Entering FTP active mode (endpoint override)");
                }
                client.enterLocalActiveMode();
            }
        }
    }

    /**
     * Whether to test FTP connection on each take from pool.
     */
    public boolean isValidateConnections()
    {
        return validateConnections;
    }

    /**
     * Whether to test FTP connection on each take from pool. This takes care of a
     * failed (or restarted) FTP server at the expense of an additional NOOP command
     * packet being sent, but increases overall availability. <p/> Disable to gain
     * slight performance gain or if you are absolutely sure of the FTP server
     * availability. <p/> The default value is <code>true</code>
     */
    public void setValidateConnections(final boolean validateConnections)
    {
        this.validateConnections = validateConnections;
    }

    /**
     * Getter for FTP transfer type.
     * 
     * @return true if using FTP binary type
     */
    public boolean isBinary()
    {
        return binary;
    }

    /**
     * Setter for FTP transfer type.
     * 
     * @param binary binary type flag
     */
    public void setBinary(final boolean binary)
    {
        this.binary = binary;
    }

    /**
     * Transfer type is BINARY by default. The value is taken from the connector
     * settings. In case there are any overriding properties set on the endpoint,
     * those will be used. <p/> The alternative type is ASCII. <p/>
     * 
     * @see #setBinary(boolean)
     */
    public void setupFileType(FTPClient client, ImmutableEndpoint endpoint) throws Exception
    {
        int type;

        // well, no endpoint URI here, as we have to use the most common denominator
        // in API :(
        final String binaryTransferString = (String)endpoint.getProperty(FtpConnector.PROPERTY_BINARY_TRANSFER);
        if (binaryTransferString == null)
        {
            // try the connector properties then
            if (isBinary())
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Using FTP BINARY type");
                }
                type = org.apache.commons.net.ftp.FTP.BINARY_FILE_TYPE;
            }
            else
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Using FTP ASCII type");
                }
                type = org.apache.commons.net.ftp.FTP.ASCII_FILE_TYPE;
            }
        }
        else
        {
            // override with endpoint's definition
            final boolean binaryTransfer = Boolean.valueOf(binaryTransferString).booleanValue();
            if (binaryTransfer)
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Using FTP BINARY type (endpoint override)");
                }
                type = org.apache.commons.net.ftp.FTP.BINARY_FILE_TYPE;
            }
            else
            {
                if (logger.isTraceEnabled())
                {
                    logger.trace("Using FTP ASCII type (endpoint override)");
                }
                type = org.apache.commons.net.ftp.FTP.ASCII_FILE_TYPE;
            }
        }

        client.setFileType(type);
    }

    /**
     * Well get the output stream (if any) for this type of transport. Typically this
     * will be called only when Streaming is being used on an outbound endpoint
     *
     * @param endpoint the endpoint that releates to this Dispatcher
     * @param message the current message being processed
     * @return the output stream to use for this request or null if the transport
     *         does not support streaming
     * @throws org.mule.api.MuleException
     */
    public OutputStream getOutputStream(ImmutableEndpoint endpoint, MuleMessage message)
        throws MuleException
    {
        try
        {
            final EndpointURI uri = endpoint.getEndpointURI();
            String filename = getFilename(endpoint, message);

            final FTPClient client = this.createFtpClient(endpoint);
            try
            {
                OutputStream out = client.storeFileStream(filename);
                return new CallbackOutputStream(out,
                        new CallbackOutputStream.Callback()
                        {
                            public void onClose() throws Exception
                            {
                                try
                                {
                                    if (!client.completePendingCommand())
                                    {
                                        client.logout();
                                        client.disconnect();
                                        throw new IOException("FTP Stream failed to complete pending request");
                                    }
                                }
                                finally
                                {
                                    releaseFtp(uri, client);
                                }
                            }
                        });
            }
            catch (Exception e)
            {
                logger.debug("Error getting output stream: ", e);
                releaseFtp(uri, client);
                throw e;
            }
        }
        catch (Exception e)
        {
            throw new DispatchException(CoreMessages.streamingFailedNoStream(), message, endpoint, e);
        }
    }

    private String getFilename(ImmutableEndpoint endpoint, MuleMessage message) throws IOException
    {
        String filename = (String) message.getProperty(FtpConnector.PROPERTY_FILENAME);
        String outPattern = (String) endpoint.getProperty(FtpConnector.PROPERTY_OUTPUT_PATTERN);
        if (outPattern == null)
        {
            outPattern = message.getStringProperty(FtpConnector.PROPERTY_OUTPUT_PATTERN, getOutputPattern());
        }
        if (outPattern != null || filename == null)
        {
            filename = generateFilename(message, outPattern);
        }
        if (filename == null)
        {
            throw new IOException("Filename is null");
        }
        return filename;
    }
    
    private String generateFilename(MuleMessage message, String pattern)
    {
        if (pattern == null)
        {
            pattern = getOutputPattern();
        }
        return getFilenameParser().getFilename(message, pattern);
    }

    /**
     * Creates a new FTPClient that logs in and changes the working directory using the data
     * provided in <code>endpoint</code>.
     */
    protected FTPClient createFtpClient(ImmutableEndpoint endpoint) throws Exception
    {
        EndpointURI uri = endpoint.getEndpointURI();
        FTPClient client = this.getFtp(uri);

        this.enterActiveOrPassiveMode(client, endpoint);
        this.setupFileType(client, endpoint);

        String path = uri.getPath();
        // MULE-2400: if the path begins with '~' we must strip the first '/' to make things
        // work with FTPClient
        if ((path.length() >= 2) && (path.charAt(1) == '~'))
        {
            path = path.substring(1);
        }
        
        if (!client.changeWorkingDirectory(path))
        {
            throw new IOException(MessageFormat.format("Failed to change working directory to {0}. Ftp error: {1}",
                                                       new Object[] {path, new Integer(client.getReplyCode())}));
        }
        return client;
    }

    /**
     * Override this method to do extra checking on the file.
     */
    protected boolean validateFile(FTPFile file)
    {
        return true;
    }
}