merge: 打通后台转码链路
This commit is contained in:
Zhou Weidong committed
commit
f66eae5baf
9 files changed
+604
No files matched your search
@@ -0,0 +1,40 @@
|
||||
# 后台转码链路打通
|
||||
|
||||
## 现状取证
|
||||
|
||||
- 上传完成时 `MediaService.complete` 已为视频调用 `TranscodeService.enqueue`。
|
||||
- 分集创建/替换视频地址时 `EpisodeService` 已调用
|
||||
`EpisodeTranscodeBindingService.bind`，会复用并绑定同一媒体的未完成任务。
|
||||
- 转码失败原因已写入 `transcode_task.error_msg`，最终状态写入 `status`。
|
||||
- 缺口是后台手动触发/重转接口，以及对应幂等和链路测试。
|
||||
|
||||
## 实现
|
||||
|
||||
1. 保留上传完成和改片自动入队逻辑，补测试锁定行为。
|
||||
2. `TranscodeService` 增加手动触发：
|
||||
- QUEUED/RUNNING 直接返回现有任务；
|
||||
- FAILED 重置原任务为 QUEUED；
|
||||
- SUCCESS 才创建新任务，紧接着重复触发会复用新 QUEUED 任务。
|
||||
3. 新增 `POST /admin/media/{id}/transcode`，校验媒资存在，复用 `episode:upload`。
|
||||
4. 返回任务 id 与状态，便于后台提示。
|
||||
|
||||
实现时对同一媒资行使用 `FOR UPDATE`，让重复请求在事务内串行判定；
|
||||
平台不存在的媒资保持 `DATA_NOT_FOUND`，不产生任务或 500。
|
||||
|
||||
## TDD 测试矩阵
|
||||
|
||||
| 场景 | 断言 |
|
||||
| --- | --- |
|
||||
| 视频上传完成 | `MediaService.complete` 调用一次 `enqueue(assetId, 0, 0, sourceUrl)` |
|
||||
| 图片上传完成 | 不创建转码任务 |
|
||||
| 手动触发成功媒资 | 新建一个 QUEUED 任务，重复调用复用同一任务 |
|
||||
| 手动触发进行中媒资 | 直接返回 QUEUED/RUNNING 任务，不新增记录 |
|
||||
| 手动触发失败媒资 | 原任务重置为 QUEUED，保留任务 id |
|
||||
| 媒资不存在 | 返回既有 `DATA_NOT_FOUND`，控制器不产生 500 |
|
||||
| worker 最终失败 | 任务状态 FAILED，`error_msg` 保留 ffmpeg 原因 |
|
||||
|
||||
## 验证
|
||||
|
||||
- 先提交失败测试，再提交实现。
|
||||
- `mvn -pl fruitshorts-module-media,fruitshorts-transcode-worker test`
|
||||
- 不启停任何服务，不执行数据库变更，不改 Web/Flutter/specs/API 文档。
|
||||
@@ -25205,3 +25205,112 @@ dramaTitle/episodeNumber/totalEpisodes 非空、越界页空 list 且 total 保
|
||||
`limit` 旧 List 兼容均通过；服务层断言 `selectPage`、稳定排序和两批 IN 查询。
|
||||
未启停服务、未操作真机、未修改客户端。
|
||||
|
||||
## 2026-10-22 后台转码链路打通 P1
|
||||
|
||||
- 分支：`feature/media-transcode-pipeline`
|
||||
- 基线：`develop` / `origin/develop` = `cd51aabe`
|
||||
- 隔离工作树：`/private/tmp/fruitserver-media-pipeline`
|
||||
- 计划：`docs/plans/2026-10-22-media-transcode-pipeline.md`
|
||||
|
||||
### 现状取证
|
||||
|
||||
- `/admin/media/upload/complete`：视频完成合并后会调用
|
||||
`transcodeService.enqueue(assetId, 0L, 0L, sourceUrl)`；图片直接置 ready，不入队。
|
||||
- 分集创建/更新并保存 `videoUrl`：`EpisodeService` 调用
|
||||
`EpisodeTranscodeBindingService.bind`，按 objectKey 找到 `media_asset`，
|
||||
再把 `dramaId/episodeId/sourceUrl` 绑定到任务；未完成任务复用，不重复入队。
|
||||
- worker：`claimNext` 只认领 `episode_id > 0`；成功后回填媒资 HLS、分集播放地址与时长；
|
||||
失败超过 3 次写 `FAILED + error_msg`。这些既有链路已由新增测试锁定。
|
||||
- 实际缺口：后台没有手动触发/重转接口，也没有其幂等测试。
|
||||
|
||||
### TDD 红测
|
||||
|
||||
先只提交测试，不写生产实现。定向命令：
|
||||
|
||||
```text
|
||||
$ mvn -pl fruitshorts-transcode-worker -am \
|
||||
-Dtest=MediaServiceTranscodePipelineTest,TranscodeServiceTriggerTest \
|
||||
-Dsurefire.failIfNoSpecifiedTests=false test
|
||||
[ERROR] ... MediaServiceTranscodePipelineTest.java:[13,32] 找不到符号
|
||||
符号: 类 TranscodeTriggerRespVO
|
||||
[ERROR] ... TranscodeServiceTriggerTest.java:[15,32] 找不到符号
|
||||
符号: 类 TranscodeTriggerRespVO
|
||||
[INFO] fruitshorts-transcode-worker ....................... FAILURE
|
||||
[INFO] BUILD FAILURE
|
||||
```
|
||||
|
||||
解读：失败原因为预期的生产接口尚不存在，不是测试环境或既有代码编译问题。
|
||||
下一步再实现最小生产代码令红测转绿。
|
||||
|
||||
控制器契约红测同样先失败在缺失的响应类型：
|
||||
|
||||
```text
|
||||
$ mvn -pl fruitshorts-admin-api -am \
|
||||
-Dtest=MediaTranscodeTriggerContractTest \
|
||||
-Dsurefire.failIfNoSpecifiedTests=false test
|
||||
[ERROR] MediaTranscodeTriggerContractTest.java:[11,32] 找不到符号
|
||||
符号: 类 TranscodeTriggerRespVO
|
||||
[INFO] fruitshorts-admin-api .............................. FAILURE
|
||||
[INFO] BUILD FAILURE
|
||||
```
|
||||
|
||||
解读：控制器测试已经先于实现落地，现有上传/分集链路未被破坏。
|
||||
|
||||
### 实现
|
||||
|
||||
- `TranscodeService.trigger(mediaAssetId, sourceUrl)`：
|
||||
- QUEUED / RUNNING：直接返回当前任务；
|
||||
- FAILED / CANCELED：原任务重置为 QUEUED，清空进度、重试次数与错误；
|
||||
- SUCCESS：复制原任务的 drama/episode 绑定，新建一轮 QUEUED；
|
||||
- 同一次手动重转产生的 QUEUED 会被下一次请求直接复用。
|
||||
- `MediaService.triggerTranscode` 在事务内以 `SELECT ... FOR UPDATE` 锁定媒体行，
|
||||
保证同一媒体的并发请求串行；媒体不存在返回 `DATA_NOT_FOUND`，图片拒绝转码。
|
||||
- 新增 `POST /admin/media/{id}/transcode`，权限沿用 `episode:upload`，
|
||||
写操作受 `ImpersonationGuard` 保护，并记录 `触发媒资转码` 操作日志。
|
||||
- worker 既有认领、ffmpeg、HLS/缩略图产物、媒资与分集回填逻辑未改；
|
||||
失败任务仍由 worker 写入 `FAILED + error_msg`。
|
||||
|
||||
### 定向测试
|
||||
|
||||
```text
|
||||
$ mvn -pl fruitshorts-transcode-worker -am \
|
||||
-Dtest=MediaServiceTranscodePipelineTest,TranscodeServiceTriggerTest \
|
||||
-Dsurefire.failIfNoSpecifiedTests=false test
|
||||
[INFO] Tests run: 4, Failures: 0, Errors: 0, Skipped: 0
|
||||
[INFO] Tests run: 4, Failures: 0, Errors: 0, Skipped: 0
|
||||
[INFO] BUILD SUCCESS
|
||||
|
||||
$ mvn -pl fruitshorts-admin-api -am \
|
||||
-Dtest=MediaTranscodeTriggerContractTest \
|
||||
-Dsurefire.failIfNoSpecifiedTests=false test
|
||||
[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0
|
||||
[INFO] BUILD SUCCESS
|
||||
```
|
||||
|
||||
解读：上传完成自动入队、图片不入队、SUCCESS 重转只新建一次、
|
||||
QUEUED 重复触发复用、FAILED 原地重置、最终失败原因保留、
|
||||
媒资不存在错误语义与手动接口返回结构均通过。
|
||||
|
||||
### 最终门禁
|
||||
|
||||
```text
|
||||
$ mvn -pl fruitshorts-module-media,fruitshorts-transcode-worker test
|
||||
[INFO] Tests run: 32, Failures: 0, Errors: 0, Skipped: 0
|
||||
[INFO] fruitshorts-module-media ........................... SUCCESS
|
||||
[INFO] fruitshorts-transcode-worker ....................... SUCCESS
|
||||
[INFO] BUILD SUCCESS
|
||||
```
|
||||
|
||||
补充跑 admin-api 与依赖模块全量测试：
|
||||
|
||||
```text
|
||||
$ mvn -pl fruitshorts-admin-api -am test
|
||||
[INFO] Tests run: 43, Failures: 0, Errors: 0, Skipped: 0
|
||||
[INFO] fruitshorts-admin-api .............................. SUCCESS
|
||||
[INFO] BUILD SUCCESS
|
||||
```
|
||||
|
||||
解读：指定门禁含真实 ffmpeg smoke，worker 在本机成功生成 HLS 与缩略图；
|
||||
后台控制器改动连同既有 admin 契约、权限、服务测试全部通过。
|
||||
未启停服务、未执行数据库迁移、未操作真机。
|
||||
|
||||
+10
@@ -10,6 +10,7 @@ import com.fruitshorts.media.config.MediaTranscodeProfile;
|
||||
import com.fruitshorts.media.service.MediaService;
|
||||
import com.fruitshorts.media.service.MediaTranscodeProfileService;
|
||||
import com.fruitshorts.media.vo.MediaAssetRespVO;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import com.fruitshorts.media.vo.UploadCompleteReqVO;
|
||||
import com.fruitshorts.media.vo.UploadInitReqVO;
|
||||
import com.fruitshorts.media.vo.UploadInitRespVO;
|
||||
@@ -84,6 +85,15 @@ public class MediaController {
|
||||
return Result.ok(mediaService.get(id));
|
||||
}
|
||||
|
||||
@Operation(summary = "手动触发或重转媒资")
|
||||
@PostMapping("/{id}/transcode")
|
||||
@PreAuthorize("@ss.hasPermission('episode:upload')")
|
||||
@OperationLog(module = "后台管理", action = "触发媒资转码")
|
||||
public Result<TranscodeTriggerRespVO> transcode(@PathVariable Long id) {
|
||||
ImpersonationGuard.rejectWrite();
|
||||
return Result.ok(mediaService.triggerTranscode(id));
|
||||
}
|
||||
|
||||
@Operation(summary = "读取转码档位配置")
|
||||
@GetMapping("/transcode-profile")
|
||||
@PreAuthorize("@ss.hasPermission('episode:upload')")
|
||||
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
// Author: fruit.com.cn
|
||||
|
||||
package com.fruitshorts.admin.controller;
|
||||
|
||||
import com.fruitshorts.admin.support.ControllerContractSupport;
|
||||
import com.fruitshorts.common.exception.ErrorCode;
|
||||
import com.fruitshorts.common.exception.ServiceException;
|
||||
import com.fruitshorts.media.service.MediaService;
|
||||
import com.fruitshorts.media.service.MediaTranscodeProfileService;
|
||||
import com.fruitshorts.media.service.TranscodeService;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.test.web.servlet.MockMvc;
|
||||
|
||||
import static org.mockito.BDDMockito.given;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
|
||||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
|
||||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
|
||||
|
||||
/**
|
||||
* 后台手动触发转码的最小契约。
|
||||
*/
|
||||
class MediaTranscodeTriggerContractTest {
|
||||
|
||||
private MediaService mediaService;
|
||||
private MockMvc mockMvc;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
mediaService = mock(MediaService.class);
|
||||
MediaController controller = new MediaController(
|
||||
mediaService, mock(MediaTranscodeProfileService.class));
|
||||
mockMvc = ControllerContractSupport.mockMvcOf(controller);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_return_task_id_and_status_when_triggering_transcode() throws Exception {
|
||||
given(mediaService.triggerTranscode(88L))
|
||||
.willReturn(new TranscodeTriggerRespVO(100L, TranscodeService.STATUS_QUEUED));
|
||||
|
||||
mockMvc.perform(post("/admin/media/88/transcode"))
|
||||
.andExpect(status().isOk())
|
||||
.andExpect(jsonPath("$.code").value(0))
|
||||
.andExpect(jsonPath("$.data.taskId").value("100"))
|
||||
.andExpect(jsonPath("$.data.status").value(TranscodeService.STATUS_QUEUED));
|
||||
|
||||
verify(mediaService).triggerTranscode(88L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_keep_not_found_semantics_when_media_does_not_exist() throws Exception {
|
||||
given(mediaService.triggerTranscode(404L))
|
||||
.willThrow(new ServiceException(ErrorCode.DATA_NOT_FOUND, "媒资不存在"));
|
||||
|
||||
mockMvc.perform(post("/admin/media/404/transcode"))
|
||||
.andExpect(status().isOk())
|
||||
.andExpect(jsonPath("$.code").value(ErrorCode.DATA_NOT_FOUND.getCode()))
|
||||
.andExpect(jsonPath("$.message").value("媒资不存在"));
|
||||
}
|
||||
}
|
||||
+25
@@ -11,6 +11,7 @@ import com.fruitshorts.media.dal.entity.MediaUploadTask;
|
||||
import com.fruitshorts.media.dal.mapper.MediaAssetMapper;
|
||||
import com.fruitshorts.media.dal.mapper.MediaUploadTaskMapper;
|
||||
import com.fruitshorts.media.vo.MediaAssetRespVO;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import com.fruitshorts.media.vo.UploadCompleteReqVO;
|
||||
import com.fruitshorts.media.vo.UploadInitReqVO;
|
||||
import com.fruitshorts.media.vo.UploadInitRespVO;
|
||||
@@ -199,6 +200,30 @@ public class MediaService {
|
||||
return toResp(asset);
|
||||
}
|
||||
|
||||
/** 后台手动触发/重转：锁住媒资行，保证同一媒体的重复请求串行判定任务状态。 */
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public TranscodeTriggerRespVO triggerTranscode(Long id) {
|
||||
MediaAsset asset = id == null ? null : mediaAssetMapper.selectOne(
|
||||
Wrappers.<MediaAsset>lambdaQuery()
|
||||
.eq(MediaAsset::getId, id)
|
||||
.last("for update"));
|
||||
if (asset == null) {
|
||||
throw new ServiceException(ErrorCode.DATA_NOT_FOUND, "媒资不存在");
|
||||
}
|
||||
if (isImage(asset.getFileName())) {
|
||||
throw new ServiceException(ErrorCode.BAD_REQUEST, "图片媒资无需转码");
|
||||
}
|
||||
String sourceUrl = asset.getSourceUrl();
|
||||
if (sourceUrl == null || sourceUrl.isBlank()) {
|
||||
sourceUrl = asset.getMp4Url();
|
||||
}
|
||||
TranscodeTriggerRespVO result = transcodeService.trigger(asset.getId(), sourceUrl);
|
||||
asset.setStatus(STATUS_TRANSCODING);
|
||||
asset.setTranscodeStatus(result.getStatus());
|
||||
mediaAssetMapper.updateById(asset);
|
||||
return result;
|
||||
}
|
||||
|
||||
/** 断点续传用：回显传到第几片、还剩多少；会话归属同样校验。 */
|
||||
public UploadProgressRespVO progress(String uploadId, Long userId) {
|
||||
MediaUploadTask task = getTask(uploadId);
|
||||
|
||||
+56
@@ -11,6 +11,7 @@ import com.fruitshorts.media.dal.entity.TranscodeTask;
|
||||
import com.fruitshorts.media.dal.mapper.MediaRenditionMapper;
|
||||
import com.fruitshorts.media.dal.mapper.MediaThumbnailMapper;
|
||||
import com.fruitshorts.media.dal.mapper.TranscodeTaskMapper;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
@@ -71,6 +72,28 @@ public class TranscodeService {
|
||||
return task.getId();
|
||||
}
|
||||
|
||||
/**
|
||||
* 后台手动触发/重转。进行中任务直接复用；失败任务原地重置；成功任务才新建一轮。
|
||||
*/
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public TranscodeTriggerRespVO trigger(Long mediaAssetId, String sourceUrl) {
|
||||
TranscodeTask latest = latestTask(mediaAssetId);
|
||||
if (latest != null && isActive(latest.getStatus())) {
|
||||
return response(latest.getId(), latest.getStatus());
|
||||
}
|
||||
if (latest != null && isRetryable(latest.getStatus())) {
|
||||
resetFailedTask(latest, sourceUrl);
|
||||
return response(latest.getId(), STATUS_QUEUED);
|
||||
}
|
||||
|
||||
Long dramaId = latest == null ? 0L : latest.getDramaId();
|
||||
Long episodeId = latest == null ? 0L : latest.getEpisodeId();
|
||||
String retrySource = latest == null || latest.getSourceUrl() == null
|
||||
|| latest.getSourceUrl().isBlank() ? sourceUrl : latest.getSourceUrl();
|
||||
Long taskId = enqueue(mediaAssetId, dramaId, episodeId, retrySource);
|
||||
return response(taskId, STATUS_QUEUED);
|
||||
}
|
||||
|
||||
/**
|
||||
* 分集保存视频地址后绑定任务；成功任务复用并同步回填已有的零绑定产物。
|
||||
*/
|
||||
@@ -246,6 +269,39 @@ public class TranscodeService {
|
||||
.eq(TranscodeTask::getStatus, status));
|
||||
}
|
||||
|
||||
private TranscodeTask latestTask(Long mediaAssetId) {
|
||||
return transcodeTaskMapper.selectOne(Wrappers.<TranscodeTask>lambdaQuery()
|
||||
.eq(TranscodeTask::getMediaAssetId, mediaAssetId)
|
||||
.orderByDesc(TranscodeTask::getId)
|
||||
.last("limit 1"));
|
||||
}
|
||||
|
||||
private boolean isActive(String status) {
|
||||
return STATUS_QUEUED.equals(status) || STATUS_RUNNING.equals(status);
|
||||
}
|
||||
|
||||
private boolean isRetryable(String status) {
|
||||
return STATUS_FAILED.equals(status) || "CANCELED".equals(status);
|
||||
}
|
||||
|
||||
private void resetFailedTask(TranscodeTask task, String sourceUrl) {
|
||||
boolean hasSource = sourceUrl != null && !sourceUrl.isBlank();
|
||||
transcodeTaskMapper.update(null, Wrappers.<TranscodeTask>lambdaUpdate()
|
||||
.eq(TranscodeTask::getId, task.getId())
|
||||
.in(TranscodeTask::getStatus, List.of(STATUS_FAILED, "CANCELED"))
|
||||
.set(TranscodeTask::getStatus, STATUS_QUEUED)
|
||||
.set(TranscodeTask::getProgress, 0)
|
||||
.set(TranscodeTask::getRetryCount, 0)
|
||||
.set(TranscodeTask::getErrorMsg, "")
|
||||
.set(TranscodeTask::getStartedAt, null)
|
||||
.set(TranscodeTask::getFinishedAt, null)
|
||||
.set(hasSource, TranscodeTask::getSourceUrl, sourceUrl));
|
||||
}
|
||||
|
||||
private TranscodeTriggerRespVO response(Long taskId, String status) {
|
||||
return new TranscodeTriggerRespVO(taskId, status);
|
||||
}
|
||||
|
||||
private void updateTaskBinding(TranscodeTask task, Long dramaId,
|
||||
Long episodeId, String sourceUrl) {
|
||||
if (task == null) {
|
||||
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
// Author: fruit.com.cn
|
||||
|
||||
package com.fruitshorts.media.vo;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/** 后台手动触发转码后的任务状态。 */
|
||||
@Data
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class TranscodeTriggerRespVO {
|
||||
|
||||
private Long taskId;
|
||||
private String status;
|
||||
}
|
||||
+146
@@ -0,0 +1,146 @@
|
||||
// Author: fruit.com.cn
|
||||
|
||||
package com.fruitshorts.worker;
|
||||
|
||||
import com.fruitshorts.media.config.MediaProperties;
|
||||
import com.fruitshorts.media.dal.entity.MediaAsset;
|
||||
import com.fruitshorts.media.dal.entity.MediaUploadTask;
|
||||
import com.fruitshorts.media.dal.mapper.MediaAssetMapper;
|
||||
import com.fruitshorts.media.dal.mapper.MediaUploadTaskMapper;
|
||||
import com.fruitshorts.media.service.MediaService;
|
||||
import com.fruitshorts.media.service.MediaStorageService;
|
||||
import com.fruitshorts.media.service.TranscodeService;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import com.fruitshorts.media.vo.UploadCompleteReqVO;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.nio.file.Path;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.BDDMockito.given;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
/**
|
||||
* 媒资完成上传与后台手动重转的入队语义。
|
||||
*/
|
||||
class MediaServiceTranscodePipelineTest {
|
||||
|
||||
private static final Long ASSET_ID = 88L;
|
||||
private static final String SOURCE_URL = "/files/episode/202610/video.mp4";
|
||||
|
||||
private MediaAssetMapper mediaAssetMapper;
|
||||
private MediaUploadTaskMapper uploadTaskMapper;
|
||||
private MediaStorageService storageService;
|
||||
private TranscodeService transcodeService;
|
||||
private MediaService mediaService;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
mediaAssetMapper = mock(MediaAssetMapper.class);
|
||||
uploadTaskMapper = mock(MediaUploadTaskMapper.class);
|
||||
storageService = mock(MediaStorageService.class);
|
||||
transcodeService = mock(TranscodeService.class);
|
||||
mediaService = new MediaService(
|
||||
new MediaProperties(), storageService, mediaAssetMapper,
|
||||
uploadTaskMapper, transcodeService);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_enqueue_video_transcode_when_upload_completes() {
|
||||
givenCompletedUpload("video.mp4");
|
||||
given(storageService.countChunks("upload-1")).willReturn(1);
|
||||
given(storageService.merge("upload-1", "episode/202610/video.mp4", 1))
|
||||
.willReturn(new MediaStorageService.MergeResult(
|
||||
Path.of("/tmp/video.mp4"), 1024L, "md5"));
|
||||
given(storageService.toPublicUrl("episode/202610/video.mp4"))
|
||||
.willReturn(SOURCE_URL);
|
||||
given(mediaAssetMapper.selectById(ASSET_ID)).willReturn(asset("video.mp4"));
|
||||
|
||||
mediaService.complete(completeReq(), 10L);
|
||||
|
||||
verify(transcodeService).enqueue(ASSET_ID, 0L, 0L, SOURCE_URL);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_not_enqueue_image_when_upload_completes() {
|
||||
givenCompletedUpload("cover.jpg");
|
||||
given(storageService.countChunks("upload-1")).willReturn(1);
|
||||
given(storageService.merge("upload-1", "episode/202610/cover.jpg", 1))
|
||||
.willReturn(new MediaStorageService.MergeResult(
|
||||
Path.of("/tmp/cover.jpg"), 512L, "md5"));
|
||||
given(storageService.toPublicUrl("episode/202610/cover.jpg"))
|
||||
.willReturn("/files/episode/202610/cover.jpg");
|
||||
given(mediaAssetMapper.selectById(ASSET_ID)).willReturn(asset("cover.jpg"));
|
||||
|
||||
mediaService.complete(completeReq(), 10L);
|
||||
|
||||
verify(transcodeService, never()).enqueue(any(), any(), any(), anyString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_return_not_found_when_triggering_transcode_for_missing_media() {
|
||||
given(mediaAssetMapper.selectOne(any())).willReturn(null);
|
||||
|
||||
assertThatThrownBy(() -> mediaService.triggerTranscode(404L))
|
||||
.isInstanceOf(com.fruitshorts.common.exception.ServiceException.class)
|
||||
.hasMessage("媒资不存在");
|
||||
verify(transcodeService, never()).trigger(any(), anyString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_trigger_transcode_and_mark_asset_queued() {
|
||||
MediaAsset asset = asset("video.mp4");
|
||||
given(mediaAssetMapper.selectOne(any())).willReturn(asset);
|
||||
given(transcodeService.trigger(ASSET_ID, SOURCE_URL))
|
||||
.willReturn(new TranscodeTriggerRespVO(100L, TranscodeService.STATUS_QUEUED));
|
||||
|
||||
TranscodeTriggerRespVO result = mediaService.triggerTranscode(ASSET_ID);
|
||||
|
||||
assertThat(result.getTaskId()).isEqualTo(100L);
|
||||
assertThat(result.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
|
||||
assertThat(asset.getStatus()).isEqualTo("transcoding");
|
||||
assertThat(asset.getTranscodeStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
|
||||
verify(mediaAssetMapper).updateById(asset);
|
||||
}
|
||||
|
||||
private void givenCompletedUpload(String fileName) {
|
||||
String objectKey = "episode/202610/" + fileName;
|
||||
MediaUploadTask task = new MediaUploadTask();
|
||||
task.setUploadId("upload-1");
|
||||
task.setFileName(fileName);
|
||||
task.setObjectKey(objectKey);
|
||||
task.setTotalChunks(1);
|
||||
task.setStatus("uploading");
|
||||
task.setCreateBy(10L);
|
||||
given(uploadTaskMapper.selectOne(any())).willReturn(task);
|
||||
doAnswer(invocation -> {
|
||||
MediaAsset inserted = invocation.getArgument(0);
|
||||
inserted.setId(ASSET_ID);
|
||||
return 1;
|
||||
}).when(mediaAssetMapper).insert(any(MediaAsset.class));
|
||||
}
|
||||
|
||||
private UploadCompleteReqVO completeReq() {
|
||||
UploadCompleteReqVO req = new UploadCompleteReqVO();
|
||||
req.setUploadId("upload-1");
|
||||
req.setDurationSec(120);
|
||||
return req;
|
||||
}
|
||||
|
||||
private MediaAsset asset(String fileName) {
|
||||
MediaAsset asset = new MediaAsset();
|
||||
asset.setId(ASSET_ID);
|
||||
asset.setFileName(fileName);
|
||||
asset.setObjectKey("episode/202610/" + fileName);
|
||||
asset.setSourceUrl(SOURCE_URL);
|
||||
asset.setStatus("uploaded");
|
||||
return asset;
|
||||
}
|
||||
}
|
||||
+138
@@ -0,0 +1,138 @@
|
||||
// Author: fruit.com.cn
|
||||
|
||||
package com.fruitshorts.worker;
|
||||
|
||||
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||
import com.fruitshorts.media.dal.entity.MediaRendition;
|
||||
import com.fruitshorts.media.dal.entity.MediaThumbnail;
|
||||
import com.fruitshorts.media.dal.entity.TranscodeTask;
|
||||
import com.fruitshorts.media.dal.mapper.MediaRenditionMapper;
|
||||
import com.fruitshorts.media.dal.mapper.MediaThumbnailMapper;
|
||||
import com.fruitshorts.media.dal.mapper.TranscodeTaskMapper;
|
||||
import com.fruitshorts.media.service.ThumbnailRecordService;
|
||||
import com.fruitshorts.media.service.TranscodeService;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.BDDMockito.given;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
/**
|
||||
* 后台重转的幂等规则：进行中复用、失败重置、成功才新建。
|
||||
*/
|
||||
class TranscodeServiceTriggerTest {
|
||||
|
||||
private static final Long ASSET_ID = 88L;
|
||||
|
||||
private TranscodeTaskMapper taskMapper;
|
||||
private TranscodeService transcodeService;
|
||||
|
||||
@BeforeAll
|
||||
static void initTableInfoCache() {
|
||||
MapperBuilderAssistant assistant =
|
||||
new MapperBuilderAssistant(new MybatisConfiguration(), "");
|
||||
TableInfoHelper.initTableInfo(assistant, TranscodeTask.class);
|
||||
TableInfoHelper.initTableInfo(assistant, MediaRendition.class);
|
||||
TableInfoHelper.initTableInfo(assistant, MediaThumbnail.class);
|
||||
}
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
taskMapper = mock(TranscodeTaskMapper.class);
|
||||
transcodeService = new TranscodeService(
|
||||
taskMapper, mock(MediaRenditionMapper.class),
|
||||
mock(MediaThumbnailMapper.class), mock(ThumbnailRecordService.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_create_once_and_reuse_queued_task_when_retrying_successful_asset() {
|
||||
TranscodeTask success = task(10L, TranscodeService.STATUS_SUCCESS);
|
||||
TranscodeTask queued = task(100L, TranscodeService.STATUS_QUEUED);
|
||||
given(taskMapper.selectOne(any())).willReturn(success, null, queued);
|
||||
doAnswer(invocation -> {
|
||||
TranscodeTask inserted = invocation.getArgument(0);
|
||||
inserted.setId(100L);
|
||||
return 1;
|
||||
}).when(taskMapper).insert(any(TranscodeTask.class));
|
||||
|
||||
TranscodeTriggerRespVO first = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
|
||||
TranscodeTriggerRespVO second = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
|
||||
|
||||
assertThat(first.getTaskId()).isEqualTo(100L);
|
||||
assertThat(first.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
|
||||
assertThat(second.getTaskId()).isEqualTo(100L);
|
||||
assertThat(second.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
|
||||
verify(taskMapper).insert(any(TranscodeTask.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_reuse_queued_task_for_repeated_trigger() {
|
||||
TranscodeTask queued = task(21L, TranscodeService.STATUS_QUEUED);
|
||||
given(taskMapper.selectOne(any())).willReturn(queued);
|
||||
|
||||
TranscodeTriggerRespVO first = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
|
||||
TranscodeTriggerRespVO second = transcodeService.trigger(ASSET_ID, "/files/video.mp4");
|
||||
|
||||
assertThat(first.getTaskId()).isEqualTo(21L);
|
||||
assertThat(second.getTaskId()).isEqualTo(21L);
|
||||
assertThat(second.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
|
||||
verify(taskMapper, never()).insert(any(TranscodeTask.class));
|
||||
verify(taskMapper, never()).updateById(any(TranscodeTask.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_reset_the_same_failed_task_for_manual_retry() {
|
||||
TranscodeTask failed = task(31L, TranscodeService.STATUS_FAILED);
|
||||
failed.setRetryCount(3);
|
||||
failed.setErrorMsg("ffmpeg failed");
|
||||
given(taskMapper.selectOne(any())).willReturn(failed);
|
||||
given(taskMapper.update(any(), any())).willReturn(1);
|
||||
|
||||
TranscodeTriggerRespVO result =
|
||||
transcodeService.trigger(ASSET_ID, "/files/video.mp4");
|
||||
|
||||
assertThat(result.getTaskId()).isEqualTo(31L);
|
||||
assertThat(result.getStatus()).isEqualTo(TranscodeService.STATUS_QUEUED);
|
||||
verify(taskMapper).update(any(), any());
|
||||
verify(taskMapper, never()).insert(any(TranscodeTask.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_persist_failure_reason_and_status_after_max_retry() {
|
||||
TranscodeTask running = task(41L, TranscodeService.STATUS_RUNNING);
|
||||
running.setRetryCount(2);
|
||||
given(taskMapper.selectById(41L)).willReturn(running);
|
||||
|
||||
String status = transcodeService.fail(
|
||||
41L, "ffmpeg 退出码 1：Invalid data found");
|
||||
|
||||
assertThat(status).isEqualTo(TranscodeService.STATUS_FAILED);
|
||||
assertThat(running.getStatus()).isEqualTo(TranscodeService.STATUS_FAILED);
|
||||
assertThat(running.getRetryCount()).isEqualTo(3);
|
||||
assertThat(running.getErrorMsg()).contains("Invalid data found");
|
||||
verify(taskMapper).updateById(running);
|
||||
}
|
||||
|
||||
private TranscodeTask task(Long id, String status) {
|
||||
TranscodeTask task = new TranscodeTask();
|
||||
task.setId(id);
|
||||
task.setMediaAssetId(ASSET_ID);
|
||||
task.setDramaId(7L);
|
||||
task.setEpisodeId(31L);
|
||||
task.setSourceUrl("/files/episode/202610/video.mp4");
|
||||
task.setStatus(status);
|
||||
task.setProgress(100);
|
||||
task.setRetryCount(0);
|
||||
task.setErrorMsg("");
|
||||
return task;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user