完成对范围流读取的支持

Author: chivehao

CHANGELOG.MD

@@ -2,6 +2,8 @@
 
 更新日志文档，版本顺序从新到旧，最新版本在最前(上)面。
 
+# 1.0.8
+
 # 1.0.7
 
 - 添加对应115文件流读取的支持

gradle.properties

@@ -1,3 +1,3 @@
 group=run.ikaros.plugin
 description=A pan115 plugin for ikaros.
-version=1.0.7
+version=1.0.8

lib/api-1.0.4-sources.jar

@@ -172,6 +172,17 @@ public Flux<DataBuffer> getSteam(Attachment attachment) {
         return checkoutMono.flatMapMany(driver -> pan115Repository.openUFileSteam(attachment.getUrl()));
     }
 
+    @Override
+    public Flux<DataBuffer> getSteam(Attachment attachment, long start, long end) {
+        Assert.notNull(attachment, "'attachment' must not null.");
+        AttachmentType type = attachment.getType();
+        if (AttachmentType.Driver_Directory.equals(type)) return Flux.empty();
+        Long driverId = attachment.getDriverId();
+        Mono<AttachmentDriver> checkoutMono = checkoutToken(driverId);
+        return checkoutMono.flatMapMany(driver ->
+                pan115Repository.openUFileSteamWithRange(attachment.getUrl(), start, end));
+    }
+
     private void applyPan115Token(AttachmentDriver driver) {
         Assert.notNull(driver, "'driver' must not null.");
         pan115Repository.refreshToken(driver);

lib/api-1.0.4.jar

@@ -258,28 +258,105 @@ public Flux<DataBuffer> openUFileSteam(String pickCode) {
         Assert.hasText(pickCode, "'pickCode' must has text.");
         final String uFileDownUrl = openUFileDownUrl(pickCode);
         return streamFileWithRestTemplate(uFileDownUrl);
-        // return Flux.create(sink -> {
-        //     Schedulers.boundedElastic().schedule(() -> {
-        //         try {
-        //             ResponseEntity<byte[]> response = restTemplate.getForEntity(
-        //                     uFileDownUrl, byte[].class);
-        //             if (response.getStatusCode() == HttpStatus.OK &&
-        //                     response.getBody() != null) {
-        //                 // 一次性发送所有数据（适合小文件）
-        //                 DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(
-        //                         response.getBody().length);
-        //                 dataBuffer.write(response.getBody());
-        //                 sink.next(dataBuffer);
-        //                 sink.complete();
-        //             } else {
-        //                 sink.error(new RuntimeException(
-        //                         "请求失败，状态码: " + response.getStatusCode()));
-        //             }
-        //         } catch (Exception e) {
-        //             sink.error(e);
-        //         }
-        //     });
-        // });
+    }
+
+    @Override
+    public Flux<DataBuffer> openUFileSteamWithRange(String pickCode, long start, long end) {
+        Assert.hasText(pickCode, "'pickCode' must has text.");
+        final String uFileDownUrl = openUFileDownUrl(pickCode);
+        return streamFileWithRestTemplateWithRange(uFileDownUrl, start, end, 8192);
+    }
+
+    private Flux<DataBuffer> streamFileWithRestTemplateWithRange(String uFileDownUrl, Long start, Long end, int chunkSize) {
+        if (chunkSize <= 0) {
+            chunkSize = 8192; // 默认8KB
+        }
+
+        final int finalChunkSize = chunkSize;
+
+        // 设置 Range 请求头
+        String rangeHeader;
+        if (start != null && end != null) {
+            rangeHeader = String.format("bytes=%d-%d", start, end);
+        } else if (start != null) {
+            rangeHeader = String.format("bytes=%d-", start);
+        } else {
+            rangeHeader = "bytes=0-"; // 默认从头开始
+        }
+
+        return Flux.create(sink -> {
+            // 使用 boundedElastic 调度器执行阻塞操作
+            Schedulers.boundedElastic().schedule(() -> {
+                try {
+                    String newUrl = uFileDownUrl;
+                    if (StringUtils.isNoneBlank(newUrl)) {
+                        newUrl = URLDecoder.decode(newUrl, StandardCharsets.UTF_8);
+                    }
+                    log.debug("Open stream range from {} to {} for url: {}", start, end, newUrl);
+                    restTemplate.execute(newUrl, HttpMethod.GET, new RequestCallback() {
+                                @Override
+                                public void doWithRequest(ClientHttpRequest request) throws IOException {
+                                    request.getHeaders().set(HttpHeaders.RANGE, rangeHeader);
+                                    request.getHeaders().addAll(headers);
+                                }
+                            },
+                            (ResponseExtractor<Void>) response -> {
+                                if (response.getStatusCode() != HttpStatus.PARTIAL_CONTENT) {
+                                    sink.error(new RuntimeException(
+                                            "请求失败，状态码: " + response.getStatusCode()));
+                                    return null;
+                                }
+
+                                try (InputStream inputStream = response.getBody()) {
+                                    byte[] buffer = new byte[finalChunkSize];
+                                    int bytesRead;
+                                    AtomicBoolean isCanceled = new AtomicBoolean(false);
+
+                                    sink.onCancel(() -> {
+                                        isCanceled.set(true);
+                                        try {
+                                            inputStream.close();
+                                        } catch (IOException e) {
+                                            // 忽略关闭异常
+                                        }
+                                    });
+
+                                    while (!isCanceled.get() &&
+                                            (bytesRead = inputStream.read(buffer)) != -1) {
+
+                                        // 创建 DataBuffer
+                                        DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bytesRead);
+                                        dataBuffer.write(buffer, 0, bytesRead);
+
+                                        // 发布数据块
+                                        sink.next(dataBuffer);
+
+                                        // 检查是否取消
+                                        if (sink.isCancelled()) {
+                                            log.debug("Close stream range from {} to {} for url: {}", start, end, uFileDownUrl);
+                                            break;
+                                        }
+                                    }
+
+                                    if (!isCanceled.get()) {
+                                        sink.complete();
+                                    }
+                                } catch (Exception e) {
+                                    if (!sink.isCancelled()) {
+                                        sink.error(e);
+                                    }
+                                }
+                                return null;
+                            });
+                } catch (Exception e) {
+                    if (!sink.isCancelled()) {
+                        sink.error(e);
+                    }
+                } finally {
+                    headers.remove(HttpHeaders.RANGE);
+                }
+            });
+        }, FluxSink.OverflowStrategy.BUFFER);
     }
 
 

lib/api-1.0.5-sources.jar

@@ -1,5 +1,6 @@
 package run.ikaros.plugin.pan115.repository;
 
+import org.reactivestreams.Publisher;
 import org.springframework.core.io.buffer.DataBuffer;
 import reactor.core.publisher.Flux;
 import run.ikaros.api.core.attachment.AttachmentDriver;
@@ -25,4 +26,6 @@ List<Pan115Attachment> openUFileFiles(String cid, Integer limit, Integer cur,
     String openVideoPlay(String pickCode);
 
     Flux<DataBuffer> openUFileSteam(String pickCode);
+
+    Flux<DataBuffer> openUFileSteamWithRange(String pickCode, long start, long end);
 }