Как исправить неполную загрузку с сервера Netty - PullRequest
Новая
       17

Как исправить неполную загрузку с сервера Netty

1 голос
/ 30 апреля 2019

Я реализую простой сервер и клиент netty для отправки и восстановления файлов. Нечто похожее на облачное хранилище.

У меня есть сервер, который обрабатывает входящие запросы и отправляет файлы обратно клиенту. Я также хочу, чтобы мои приложения могли обрабатывать большие файлы, поэтому я делю такие файлы на части и отправляю их по частям. Но есть проблема, которую я не могу решить.

Допустим,

  • У нас есть файл 4 gb на сервере.
  • Он разделен на 40 000 кусков.
  • Затем они отправляются клиентскому приложению, и я вижу, что все куски на сервере записываются в сокет, так как я использую поле int в качестве номера сообщения (номера чанка) и помещаю в журнал номер сообщения, который пишется.

Но тогда, когда клиент получает сообщения (чанки), в случае больших файлов процесс не завершается успешно, и клиент получает только некоторые (зависит от размера файла) чанки.

Клиент начинает получать последовательные сообщения - 1, 2, 3, 4 ... 27878, 27879, а затем останавливается без исключения , хотя последнее сообщение от сервера было, например, 40000.

Почти забыл сказать, что я использую JavaFX для клиентского приложения.

Итак, я попытался поиграть с опциями xms xmx java vm, но это не помогло.

Сервер 

public class Server {
    public void run() throws Exception {
        EventLoopGroup mainGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(mainGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(
                                    new ObjectDecoder(Constants.FRAME_SIZE, ClassResolvers.cacheDisabled(null)),
                                    new ObjectEncoder(),
                                    new MainHandler()
                            );
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture future = b.bind(8189).sync();
            future.channel().closeFuture().sync();
        } finally {
            mainGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new Server().run();

    }
}

Серверный обработчик 

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            if (msg == null) {
                return;
            }
            if (msg instanceof FileRequest) {
                FileRequest fr = (FileRequest) msg;
                switch (fr.getFileCommand()) {
                    case DOWNLOAD:
                        sendFileToClient(ctx, fr.getFilename());
                        break;
                    case LIST_FILES:
                        listFiles(ctx);
                        break;
                    case DELETE:
                        deleteFileOnServer(fr);
                        listFiles(ctx);
                        break;
                    case SEND:
                        saveFileOnServer(fr);
                        listFiles(ctx);
                        break;
                    case SEND_PARTIAL_DATA:
                        savePartialDataOnServer(fr);
                        break;
                }
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

Способы отправки файлов кусками 

private void sendFileToClient(ChannelHandlerContext ctx, String fileName) throws IOException {
        Path path = Paths.get("server_storage/" + fileName);
        if (Files.exists(path)) {
            if (Files.size(path) > Constants.FRAME_SIZE) {
                sendServerDataFrames(ctx, path);
                ctx.writeAndFlush(new FileRequest(FileCommand.LIST_FILES));
            } else {
                FileMessage fm = new FileMessage(path);
                ctx.writeAndFlush(fm);
            }
        }
    }

    private void sendServerDataFrames(ChannelHandlerContext ctx, Path path) throws IOException {
        byte[] byteBuf = new byte[Constants.FRAME_CHUNK_SIZE];

        FileMessage fileMessage = new FileMessage(path, byteBuf, 1);
        FileRequest fileRequest = new FileRequest(FileCommand.SEND_PARTIAL_DATA, fileMessage);

        FileInputStream fis = new FileInputStream(path.toFile());
        int read;
        while ((read = fis.read(byteBuf)) > 0) {
            if (read < Constants.FRAME_CHUNK_SIZE) {
                byteBuf = Arrays.copyOf(byteBuf, read);
                fileMessage.setData(byteBuf);
            }
            ctx.writeAndFlush(fileRequest);
            fileMessage.setMessageNumber(fileMessage.getMessageNumber() + 1);
        }

        System.out.println("server_storage/" + path.getFileName() +  ", server last frame number: " + fileMessage.getMessageNumber());
        System.out.println("server_storage/" + path.getFileName() +  ": closing file stream.");

        fis.close();
    }

клиентские обработчики 

@Override
    public void initialize(URL location, ResourceBundle resources) {
        Network.start();
        Thread t = new Thread(() -> {
            try {
                while (true) {
                    AbstractMessage am = Network.readObject();
                    if (am instanceof FileMessage) {
                        FileMessage fm = (FileMessage) am;
                        Files.write(Paths.get("client_storage/" + fm.getFilename()), fm.getData(), StandardOpenOption.CREATE);
                        refreshLocalFilesList();
                    }
                    if (am instanceof FilesListMessage) {
                        FilesListMessage flm = (FilesListMessage) am;
                        refreshServerFilesList(flm.getFilesList());
                    }
                    if (am instanceof FileRequest) {
                        FileRequest fr = (FileRequest) am;
                        switch (fr.getFileCommand()) {
                            case DELETE:
                                deleteFile(fr.getFilename());
                                break;
                            case SEND_PARTIAL_DATA:
                                receiveFrames(fr);
                                break;
                            case LIST_FILES:
                                refreshLocalFilesList();
                                break;
                        }
                    }
                }
            } catch (ClassNotFoundException | IOException e) {
                e.printStackTrace();
            } finally {
                Network.stop();
            }
        });
        t.setDaemon(true);
        t.start();
        refreshLocalFilesList();
        Network.sendMsg(new FileRequest(FileCommand.LIST_FILES));
    }

 private void receiveFrames(FileRequest fm) throws IOException {
        Utils.processBytes(fm.getFileMessage(), "client_storage/");
    }


public final class Utils {

    public static void processBytes(FileMessage fm, String pathPart) {
        Path path = Paths.get(pathPart + fm.getFilename());
        byte[] data = fm.getData();

        System.out.println(pathPart + path.getFileName() + ": " + fm.getMessageNumber());

        try {
            if (fm.getMessageNumber() == 1) {
                Files.write(path, data, StandardOpenOption.CREATE_NEW);
            } else {
                Files.write(path, data, StandardOpenOption.WRITE, StandardOpenOption.APPEND);
            }
        }
        catch (IOException e) {
            e.printStackTrace();
        }

    }

}

То, что я вижу на сервере. 

server_storage/DVD5_OFFICE_2010_SE_SP2_VOLUME_X86_RU-KROKOZ.iso: 42151
server_storage/DVD5_OFFICE_2010_SE_SP2_VOLUME_X86_RU-KROKOZ.iso: 42152
server_storage/DVD5_OFFICE_2010_SE_SP2_VOLUME_X86_RU-KROKOZ.iso, server last frame number: 42153
server_storage/DVD5_OFFICE_2010_SE_SP2_VOLUME_X86_RU-KROKOZ.iso: closing file stream.

А это на клиенте. 

client_storage/DVD5_OFFICE_2010_SE_SP2_VOLUME_X86_RU-KROKOZ.iso: 29055
client_storage/DVD5_OFFICE_2010_SE_SP2_VOLUME_X86_RU-KROKOZ.iso: 29056
client_storage/DVD5_OFFICE_2010_SE_SP2_VOLUME_X86_RU-KROKOZ.iso: 29057

И при отправке файлов с клиента на сервер проблем не возникает. В отладчике и в диспетчере задач Windows я вижу, что оба процесса работают одновременно, но это не так, когда файл отправляется с сервера на клиент. Сначала все фрагменты читаются, а затем отправляются клиенту, и он начинает получать их, но не может получить все из них.
Пожалуйста помоги. Я понятия не имею, что это может быть. Заранее спасибо.

...