Spring Stomp Over Websocket:流式传输大文件

Question

我的sockjs客户端在网页中，发送帧大小为16k的消息。消息大小限制是决定我可以传输文件的最大大小的最大大小。

以下是我在文档中发现的。

/**
 * Configure the maximum size for an incoming sub-protocol message.
 * For example a STOMP message may be received as multiple WebSocket messages
 * or multiple HTTP POST requests when SockJS fallback options are in use.
 *
 * <p>In theory a WebSocket message can be almost unlimited in size.
 * In practice WebSocket servers impose limits on incoming message size.
 * STOMP clients for example tend to split large messages around 16K
 * boundaries. Therefore a server must be able to buffer partial content
 * and decode when enough data is received. Use this property to configure
 * the max size of the buffer to use.
 *
 * <p>The default value is 64K (i.e. 64 * 1024).
 *
 * <p><strong>NOTE</strong> that the current version 1.2 of the STOMP spec
 * does not specifically discuss how to send STOMP messages over WebSocket.
 * Version 2 of the spec will but in the mean time existing client libraries
 * have already established a practice that servers must handle.
 */
public WebSocketTransportRegistration setMessageSizeLimit(int messageSizeLimit) {
    this.messageSizeLimit = messageSizeLimit;
    return this;
}

我的问题：我可以设置部分消息传递，以便将文件部分传输，并且在现在完成时不会被传输为单个消息？

更新：仍在寻找通过部分消息传递的解决方案同时，现在使用http进行大消息（我的应用程序中的文件上传/下载）。

Answer 1

我可以设置部分消息传递，以便将文件部分传输到部分，并且在现在完成时不会被传输为单个消息？

是。这是我的Spring Boot实验项目中的相关配置 - 基本上是UploadWSHandler已注册并设置了WebSocketTransportRegistration.setMessageSizeLimit。

@Configuration
@EnableWebSocket
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer implements WebSocketConfigurer {
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new UploadWSHandler(), "/binary");
    }
    
    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
        registration.setMessageSizeLimit(50 * 1024 * 1024);
    }
}

uploadwshandler如下。对不起，这里的代码太多 - 要点

• 返回true的 supportsPartialMessage
• handleBinaryMessage将被多次调用，并带有部分消息，因此我们需要组装字节。因此，afterConnectionEstablished使用WebSocket URL查询建立身份。但是您不必使用这种机制。之所以选择这种机制是要使客户端简单地使客户端简单，以便我只在javascript端将文件blob对象切成 webSocket.send(files[0])。（侧点：我想在客户端使用普通的Websocket -no Stomp/socks）
• Iternal客户端块机制提供message.isLast()最后消息
• 出于演示目的，我将其写入文件系统并用FileUploadInFlight在内存中累积这些字节，但您不必这样做，并且可以在其他地方进行流式传输。

public class UploadWSHandler extends BinaryWebSocketHandler {
    Map<WebSocketSession, FileUploadInFlight> sessionToFileMap = new WeakHashMap<>();
    @Override
    public boolean supportsPartialMessages() {
        return true;
    }
    @Override
    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
        ByteBuffer payload = message.getPayload();
        FileUploadInFlight inflightUpload = sessionToFileMap.get(session);
        if (inflightUpload == null) {
            throw new IllegalStateException("This is not expected");
        }
        inflightUpload.append(payload);
        if (message.isLast()) {
            Path basePath = Paths.get(".", "uploads", UUID.randomUUID().toString());
            Files.createDirectories(basePath);
            FileChannel channel = new FileOutputStream(
                    Paths.get(basePath.toString() ,inflightUpload.name).toFile(), false).getChannel();
            channel.write(ByteBuffer.wrap(inflightUpload.bos.toByteArray()));
            channel.close();
            session.sendMessage(new TextMessage("UPLOAD "+inflightUpload.name));
            session.close();
            sessionToFileMap.remove(session);
        }
        String response = "Upload Chunk: size "+ payload.array().length;
        System.out.println(response);
    }
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        sessionToFileMap.put(session, new FileUploadInFlight(session));
    }
    static class FileUploadInFlight {
        String name;
        String uniqueUploadId;
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        /**
         * Fragile constructor - beware not prod ready
         * @param session
         */
        FileUploadInFlight(WebSocketSession session) {
            String query = session.getUri().getQuery();
            String uploadSessionIdBase64 = query.split("=")[1];
            String uploadSessionId = new String(Base64Utils.decodeUrlSafe(uploadSessionIdBase64.getBytes()));
            System.out.println(uploadSessionId);
            List<String> sessionIdentifiers = Splitter.on("\").splitToList(uploadSessionId);
            String uniqueUploadId = session.getRemoteAddress().toString()+sessionIdentifiers.get(0);
            String fileName = sessionIdentifiers.get(1);
            this.name = fileName;
            this.uniqueUploadId = uniqueUploadId;
        }
        public void append(ByteBuffer byteBuffer) throws IOException{
            bos.write(byteBuffer.array());
        }
    }
}

顺便说一句，一个工作项目也是sprint-boot-with-websocked-chunking-sembly-emembly and fetch in with-websocked-chunking-assembly-and-fetch branch