如何使用Java中的Apache Mina SSHD SFTP服务器处理传入文件

Question

目前我正在研究SFTP协议.我使用Apache Mina Sshd库创建了SFTP客户端使用Jsch库和SFTP服务器.我已经建立了它们之间的连接,并且可以成功地将文件发送到SFTP服务器.现在我正在创建一个SFTP服务器处理传入文件的辅助文件处理程序.举个例子,假设SFTP服务器可以从SFTP客户端接收文件,但是当前在我的实现中,当文件到达服务器时没有办法通知.我只是去服务器根文件夹,看看是否有是一个可用的文件.这就是我知道文件是否到达的方式.

我想实现当文件到达服务器时它将通知用户文件到达和文件内容.(文件名和其他详细信息).但问题是我是Apache Mina的新手sshd API.i已经通过文档但我无法弄明白.

请问我是否知道如果在Apache Mina Sshd服务器中有任何已经实现的侦听器来处理传入文件,或者如果没有,我如何为传入文件实现我自己的侦听器.

SFTP服务器代码

public class SftpServerStarter {

    private SshServer sshd;
    private final static Logger logger = LoggerFactory.getLogger(SftpServerStarter.class);

    public void start(){


        sshd = SshServer.setUpDefaultServer();
        sshd.setPort(22);
        sshd.setHost("localhost");

        sshd.setPasswordAuthenticator(new MyPasswordAuthenticator());
        sshd.setPublickeyAuthenticator(new MyPublickeyAuthenticator());
        sshd.setKeyPairProvider(new SimpleGeneratorHostKeyProvider());
        sshd.setSubsystemFactories(Arrays.<NamedFactory<Command>>asList(new SftpSubsystem.Factory()));
        sshd.setCommandFactory(new ScpCommandFactory());
        sshd.setFileSystemFactory(new VirtualFileSystemFactory("C:/root"));


        try {
            logger.info("Starting ...");
            sshd.start();
            logger.info("Started");
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            logger.info("Can not Start Server");
        }
    }

}

Answer 1

我开始使用@gihan的建议,但是在用户完成在某些客户端上传之前,我遇到了文件监视器处理文件的一些问题.

这是我在Mina的源代码中找到的解决方案.尽管在Apache米娜的站点稀疏,无用的文件,我觉得这是他们旨在让开发者使用他们的图书馆该怎么走.

注意:由于您的需求可能与我的不同,请记住,这可能不是复制粘贴解决方案.您可能需要调整此代码以满足您的需求,但我相信这段代码确实为您正在寻找的解决方案提供了关键.

第1步:实现SftpEventListener

创建自己的实现类org.apache.sshd.server.subsystem.sftp.SftpEventListener.这是我的例子.我的实现是建立运行一系列的注册FileUploadCompleteListener方法,每当一个文件被新上载或覆盖,并阻止用户尝试导航或创建目录.

public class SFTPServiceSFTPEventListener implements SftpEventListener {

    Logger logger = Logger.getLogger(SFTPServiceSFTPEventListener.class);

    SFTPService service;

    public SFTPServiceSFTPEventListener(SFTPService service) {
        this.service = service;
    }

    public interface FileUploadCompleteListener {
        void onFileReady(File file);
    }

    private List<FileUploadCompleteListener> fileReadyListeners = new ArrayList<FileUploadCompleteListener>();

    public void addFileUploadCompleteListener(FileUploadCompleteListener listener) {
        fileReadyListeners.add(listener);
    }

    public void removeFileUploadCompleteListener(FileUploadCompleteListener listener) {
        fileReadyListeners.remove(listener);
    }

    @Override
    public void initialized(ServerSession serverSession, int version) {

    }

    @Override
    public void destroying(ServerSession serverSession) {

    }

    @Override
    public void open(ServerSession serverSession, String remoteHandle, Handle localHandle) {
        File openedFile = localHandle.getFile().toFile();
        if (openedFile.exists() && openedFile.isFile()) {
        }
    }

    @Override
    public void read(ServerSession serverSession, String remoteHandle, DirectoryHandle localHandle, Map<String,Path> entries) {

    }

    @Override
    public void read(ServerSession serverSession, String remoteHandle, FileHandle localHandle, long offset, byte[] data, int dataOffset, int dataLen, int readLen) {

    }

    @Override
    public void write(ServerSession serverSession, String remoteHandle, FileHandle localHandle, long offset, byte[] data, int dataOffset, int dataLen) {
    }

    @Override
    public void blocking(ServerSession serverSession,  String remoteHandle, FileHandle localHandle, long offset, long length, int mask) {
    }

    @Override
    public void blocked(ServerSession serverSession, String remoteHandle, FileHandle localHandle, long offset, long length, int mask, Throwable thrown) {
    }

    @Override
    public void unblocking(ServerSession serverSession, String remoteHandle, FileHandle localHandle, long offset, long length) {
    }

    @Override
    public void unblocked(ServerSession serverSession, String remoteHandle, FileHandle localHandle, long offset, long length, Boolean result, Throwable thrown) {
    }

    @Override
    public void close(ServerSession serverSession, String remoteHandle, Handle localHandle) {
        File closedFile = localHandle.getFile().toFile();
        if (closedFile.exists() && closedFile.isFile()) {
            logger.info(String.format("User %s closed file: \"%s\"", serverSession.getUsername(), localHandle.getFile().toAbsolutePath()));
            this.service.UserWroteFile(serverSession.getUsername(), localHandle.getFile());

            for (FileUploadCompleteListener fileReadyListener : fileReadyListeners) {
                fileReadyListener.onFileReady(closedFile);
            }
        }
    }

    @Override
    public void creating(ServerSession serverSession, Path path, Map<String,?> attrs) throws UnsupportedOperationException {
        logger.warn(String.format("Blocked user %s attempt to create a directory \"%s\"", serverSession.getUsername(), path.toString()));
        throw new UnsupportedOperationException("Creating sub-directories is not permitted.");
    }

    @Override
    public void created(ServerSession serverSession, Path path, Map<String,?> attrs, Throwable thrown) {
        String username = serverSession.getUsername();
        logger.info(String.format("User %s created: \"%s\"", username, path.toString()));
        service.UserWroteFile(username, path);
    }

    @Override
    public void moving(ServerSession serverSession, Path path, Path path1, Collection<CopyOption> collection) {

    }

    @Override
    public void moved(ServerSession serverSession, Path source, Path destination, Collection<CopyOption> collection, Throwable throwable) {
        String username = serverSession.getUsername();
        logger.info(String.format("User %s moved: \"%s\" to \"%s\"", username, source.toString(), destination.toString()));
        service.UserWroteFile(username, destination);
    }

    @Override
    public void removing(ServerSession serverSession, Path path) {

    }

    @Override
    public void removed(ServerSession serverSession, Path path, Throwable thrown) {

    }

    @Override
    public void linking(ServerSession serverSession, Path source, Path target, boolean symLink) throws UnsupportedOperationException {
        logger.warn(String.format("Blocked user %s attempt to create a link to \"%s\" at \"%s\"", serverSession.getUsername(), target.toString(), source.toString()));
        throw new UnsupportedOperationException("Creating links is not permitted");
    }

    @Override
    public void linked(ServerSession serverSession, Path source, Path target, boolean symLink, Throwable thrown) {

    }

    @Override
    public void modifyingAttributes(ServerSession serverSession, Path path, Map<String,?> attrs) {

    }

    @Override
    public void modifiedAttributes(ServerSession serverSession, Path path, Map<String,?> attrs, Throwable thrown) {
        String username = serverSession.getUsername();
        service.UserWroteFile(username, path);
    }
}

第2步:将侦听器的实例添加到服务器

一旦实现了类,您需要做的就是实例化它并使用SftpSubsystemFactory之前调用服务器将其添加到start()服务器:

// Your SSHD Server
SshServer sshd = SshServer.setUpDefaultServer();

SftpSubsystemFactory sftpSubsystemFactory= new SftpSubsystemFactory();

// This is where to put your implementation of SftpEventListener
SFTPServiceSFTPEventListener sftpEventListener = new SFTPServiceSFTPEventListener(this);
sftpEventListener.addFileUploadCompleteListener(new SFTPServiceSFTPEventListener.FileUploadCompleteListener() {
    @Override
    public void onFileReady(File file) {
        try {
            doThingsWithFile(file);
        } catch (Exception e) {
            logger.warn(String.format("An error occurred while attempting to do things with the file: \"%s\"", file.getName()), e);
        }
    }
});
sftpSubsystemFactory.addSftpEventListener(sftpEventListener);

List<NamedFactory<Command>> namedFactoryList = new ArrayList<NamedFactory<Command>>();
namedFactoryList.add(sftpSubsystemFactory);
sshd.setSubsystemFactories(namedFactoryList);

// Do your other init stuff...

sshd.start();

完成后,您的实施SftpEventListener将自动响应您已实施的事件.煤矿基本上只响应当用户关闭文件(文件上传完成后发生),但正如我所说,你可以随时来实现其他方法对其他事件作出响应.

Answer 2

我终于找到了一个解决方案，但它不是来自 Apache Mina SSHD API。概念是这样的：我们可以监控服务器根目录的文件变化。如果服务器文件夹中的文件发生更改，则会触发事件。有大量 API 可用于执行此操作。在我的代码片段中，我使用org.apache.commons.io.monitor.

SFTPFileListner 类

public static void startMonitor(String rootFolder) throws Exception {

        //every 5 seconds it will start monitoring
        final long pollingInterval = 5 * 1000;

        File folder = new File(rootFolder);

        if (!folder.exists()) {

            throw new RuntimeException("ERROR : Server root directory not found: " + rootFolder);
        }

        FileAlterationObserver observer = new FileAlterationObserver(folder);
        FileAlterationMonitor monitor = new FileAlterationMonitor(pollingInterval);
        FileAlterationListener listener = new FileAlterationListenerAdaptor() {

            @Override
            public void onFileCreate(File file) {
                try {

                    System.out.println("[SFTPFileListner] Received :"+ file.getName());
                    System.out.println("[SFTPFileListner] Received File Path :"+ file.getCanonicalPath());



                } catch (IOException e) {
                    throw new RuntimeException("ERROR: Unrecoverable error when creating files " + e.getMessage(),e);
                }
            }

        };

        observer.addListener(listener);
        monitor.addObserver(observer);
        monitor.start();
    }

创建监视器类后，您可以调用 SFTP 服务器类中实现的方法。

SFTP服务器类

//pass server root directory 
SFTPFileListner.startMonitor("C:/root");