feat(webflux): 支持ResponseMessage流式响应 (#361)

* feat(webflux)!: 支持ResponseMessage流式响应

移除普通 JSON Flux 的无界收集,委托 Spring Jackson 编码器增量输出。

通过 Micrometer ContextSnapshot 恢复所有已注册的响应式上下文。

BREAKING CHANGE: 普通 JSON Flux 改为增量传输,不再保证 Content-Length;首元素写出后的异常将终止响应,无法改写为完整错误 ResponseMessage。

* feat(crud): 增加分页查询聚合保护策略

* build(starter): 补齐EasyORM测试驱动依赖
This commit is contained in:
老周
2026-08-07 14:18:45 +08:00
committed by GitHub
parent c87b0a6f39
commit 5c82100671
24 changed files with 3348 additions and 519 deletions

View File

@@ -0,0 +1,521 @@
# WebFlux ResponseMessage 流式响应优化计划
## 状态
- 阶段:已实现(定向测试通过;全量回归存在与本变更无关的既有阻断)
- owning modules`hsweb-commons/hsweb-commons-crud``hsweb-starter`
- 目标版本:`5.0.2-SNAPSHOT`
## 背景与当前事实
`ResponseMessageWrapper` 当前会把非 SSE、非 NDJSON 的 `Flux` 执行
`collectList()`,再包装为 `ResponseMessage<List<T>>`。这保持了统一响应结构,但会让
内存占用随结果总量线性增长,潜在无界流还可能永远不产生响应。
当前链路存在两个收集点:
1. `hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/web/ResponseMessageWrapper.java`
对待包装的 `Flux` 执行 `collectList()`
2. `hsweb-starter/src/main/java/org/hswebframework/web/starter/jackson/CustomJackson2jsonEncoder.java`
对普通 `application/json` 的多值 Publisher 再次执行 `collectList()`。因此即使通过
`X-Response-Wrapper: Ignore` 绕过响应包装,原始 Flux 仍可能被完整收集。
Spring Framework 6.2 自带的 `Jackson2JsonEncoder` 已经支持将多值 Publisher 增量编码成
标准 JSON 数组:首元素输出 `[`,后续元素输出分隔符,完成时输出 `]`,不需要
`collectList()`。现有 `CustomJackson2jsonEncoder` 保留的是较早版本的收集逻辑,应改为
委托 Spring 官方实现,而不是继续维护一份平行的 Jackson 数组编码算法。
hsweb 的自定义编码器同时承担同步序列化所需的 ThreadLocal 上下文恢复。仓库已经提供
`AuthenticationThreadLocalAccessor``LocaleThreadLocalAccessor` 和 Micrometer Context
Propagation 依赖。实现应基于当前完整 Reactor Context 与所有已注册的 `ThreadLocalAccessor`
统一传播,而不是让编码器识别 Authentication、Locale 或其他具体上下文类型。
## 目标
1. `Flux<T>``application/json` 下继续输出兼容的单个 `ResponseMessage`
```json
{
"message": "success",
"result": [
{"id": "device-001"},
{"id": "device-002"}
],
"status": 200,
"timestamp": 0
}
```
2. 不再按完整结果集执行 `collectList()`,内存边界收敛为单个元素序列化缓冲、少量
JSON 框架缓冲和网络写出缓冲。
3. 保持 Reactive Streams 的 demand、cancel、onError、onComplete 语义,不调用
`block()`、不嵌套 `subscribe()`、不引入额外线程切换。
4. 优先委托 Spring 6.2 官方 `Jackson2JsonEncoder`、`HttpMessageWriter`、
`ReactiveAdapterRegistry` 和内容协商机制。
5. 保持 hsweb 的 ObjectMapper 配置、EntityFactory 扩展 ResponseMessage以及所有通过
Micrometer `ThreadLocalAccessor` 注册的上下文;认证和 Locale/i18n 继续通过相同机制兼容。
6. SSE、NDJSON 和显式忽略包装的响应继续按原协议输出,不额外套 ResponseMessage。
## 非目标
- 不修改 MVC `ResponseMessageWrapperAdvice`Servlet 流式响应另行设计。
- 不把 SSE、NDJSON 改造成单一 JSON 外壳。
- 不为流中每个元素创建 `ResponseMessage<T>`,避免改变现有响应结构。
- 不新增数据库、事件、权限或业务接口行为。
- 不承诺客户端调用 `response.json()` 时也能逐元素消费;本次首先解决服务端无界收集和
HTTP 分块写出。客户端增量消费仍应优先使用 NDJSON/SSE 或流式 JSON 解析器。
## 推荐方案
### 1. 自定义 Jackson 编码器回归 Spring 官方实现
将 `CustomJackson2jsonEncoder` 调整为继承 Spring 官方 `Jackson2JsonEncoder`,移除本地维护的:
- 非流式 `collectList()`
- `SequenceWriter`、数组分隔符和 DataBuffer 拼装;
- 与 Spring 官方实现重复的媒体类型、ObjectWriter 和资源清理代码。
自定义类只保留一个职责:在编码信号进入 Spring 官方编码器前恢复已注册的同步上下文。
上下文恢复使用已经存在的 Micrometer `ContextSnapshotFactory` 与
`ThreadLocalAccessor`,通过 Reactor 官方 `Operators.liftPublisher` 装饰订阅者。在调用
下游 `onNext` 的同步作用域内,从当前完整 Reactor Context 恢复所有已注册 accessor 对应的
ThreadLocalSpring `Jackson2JsonEncoder` 的同步元素序列化完成后立即恢复旧值。每次订阅使用
Micrometer `ContextSnapshotFactory.captureAll(contextView)` 创建一次通用快照:先捕获订阅线程上
所有已注册 accessor 的兼容 ThreadLocal再合并 Reactor Context按官方覆盖顺序Reactor
Context 中的同 key 值优先。该装饰器必须完整透传 Subscription、demand、cancel 和终止信号,
不创建第二次订阅。
不能只在链路上追加 Reactor `contextCapture()`。该操作符解决的是相反方向:在订阅阶段把当前
线程中已注册 accessor 对应的 ThreadLocal 捕获到 Reactor Context。当前请求上下文已经以 Reactor
Context 为事实来源,实际缺口是在 `publishOn` 等异步边界之后,把 Reactor Context 恢复到执行
Jackson getter/serializer 的线程。默认传播模式下,`handle`、`tap` 等操作符提供的有限恢复只覆盖
它们自己的回调,也不能保证作用域覆盖后续 Spring encoder 的同步 `onNext`。只有由应用显式全局
启用 `Hooks.enableAutomaticContextPropagation()` 后,`contextCapture()` 才能与自动恢复机制组合
简化这部分桥接;框架组件不应为了一个 writer 改变整个应用的传播语义。因此这里保留局部
`ContextSnapshot` 作用域,并让它精确包围官方 Jackson encoder 的同步编码调用。
不通过以下方式实现:
- 不复制 Spring `AbstractJackson2Encoder` 源码;
- 不通过反射访问 Spring 私有方法;
- 不强制全局调用 `Hooks.enableAutomaticContextPropagation()`
- 不自己创建一套脱离 Spring 配置的 ObjectMapper。
### 2. 引入内部流式响应标记
在 `hsweb-commons-crud` 增加仅供响应处理链使用的内部类型,例如:
```java
final class StreamingResponseMessage<T> {
private final ResponseMessage<?> metadata;
private final Publisher<T> result;
private final ResolvableType elementType;
}
```
该类型不是新的 Controller 公共返回契约,只用于在 `ResponseMessageWrapper` 与专用
`HttpMessageWriter` 之间传递外壳元数据、元素类型和原始 Publisher。禁止让普通 Jackson
直接把嵌套 Publisher 当 JavaBean 属性序列化。
### 3. 使用专用 HttpMessageWriter 输出外层 ResponseMessage
新增 `ResponseMessageJacksonHttpMessageWriter`,实现 Spring 官方
`HttpMessageWriter<StreamingResponseMessage<?>>` 扩展点:
1. 只声明支持 `application/json` 和 `application/*+json`。
2. 从 `ResponseMessageWrapper` 已配置的 writers 中复用现有
`EncoderHttpMessageWriter` 所持有的 `Jackson2JsonEncoder`,不另建 ObjectMapper。
3. 将 `result` Publisher 委托给 Spring 官方编码器,得到增量 JSON 数组
`Flux<DataBuffer>`。
4. 在第一个数组 DataBuffer 前拼接 ResponseMessage JSON 前缀,在数组完成后拼接元数据
后缀,并通过 `ServerHttpResponse.writeWith(...)` 写出。
5. 使用官方 encoder 的 `getEncodeHints(...)` 保留 `@JsonView`、日志前缀、具体元素类型
和 Reactor Context hints。
专用 writer 只加入当前 `ResponseMessageWrapper` 的 writer 列表,并排在通用 Jackson
writer 前面。其余 Spring WebFlux handler 不需要认识内部标记类型,避免修改全局 Codec 顺序。
数据流如下:
```text
Controller Flux<T>
-> ResponseMessageWrapper
-> StreamingResponseMessage<T>
-> ResponseMessageJacksonHttpMessageWriter
-> Spring Jackson2JsonEncoder(result Flux<T>)
-> prefix + JSON array buffers + suffix
-> ServerHttpResponse.writeWith
```
### 4. 首个响应提交与 JSON 框架
不能先单独写出 `{"message":"success","result":`,否则源 Publisher 在首元素前失败时
响应已经提交。
writer 应等待官方数组编码器的第一个信号:
- 首元素成功编码:把外层前缀与第一个数组 DataBuffer 合并为首个输出 buffer
- 空 Flux官方编码器生成 `[]`,输出完整成功外壳;
- 首元素前 onError 或首元素序列化失败:不输出任何 buffer错误继续交给 WebFlux
异常处理链。
后续元素以官方编码器产生的逗号和元素 buffer 继续写出,完成后追加外层后缀。
DataBuffer 合并、丢弃和取消路径按 Spring `DataBufferUtils.release(...)` 规则处理。本地 writer
通过 discard hook 释放未写出的 DataBufferJSON generator 与 ByteArrayBuilder 的生命周期由
官方 encoder 管理。当前 Spring Framework 6.2.10 的官方实现内部使用 `doAfterTerminate`,且不
公开私有 generator 的关闭钩子,因此本地实现不通过复制源码或反射声称修复该内部 cancel 边界。
### 5. ResponseMessage 元数据兼容
- JSON 字段语义保持 `message/result/status/code/timestamp` 不变;`code == null` 继续遵循
`@JsonInclude` 省略。
- 外层元数据使用同一个 ObjectMapper 和 EntityFactory 创建的 `ResponseMessage` 实例,
兼容自定义 ResponseMessage 子类、Jackson Module、Mixin 和命名策略。
- writer 将 `result` 作为流式数组插入,其他元数据按 ObjectMapper 可见属性输出;扩展字段
不能被静默丢弃。
- JSON 对象字段顺序在语义上不构成协议,但实现和回归测试优先维持当前基础字段顺序,
降低快照测试和非规范客户端的兼容风险。
- `timestamp` 在 `ResponseMessageWrapper` 创建流式外壳元数据时生成。它不再等待完整结果收集,
因而语义从“收集完成时间”调整为“流式响应开始时间”;失败或取消时不会追加完整成功后缀。
### 6. Wrapper 使用 ReactiveAdapterRegistry 和官方内容协商
`ResponseMessageWrapper` 不再只通过 `instanceof Mono/Flux` 判断:
1. 使用 `ReactiveAdapterRegistry` 获取 adapter
2. 单值 Publisher 继续映射成单个 `ResponseMessage<T>`
3. 多值 Publisher 在协商结果为普通 JSON 且专用 writer 可用时转换为
`StreamingResponseMessage<T>`
4. no-value Publisher 保持空成功响应语义;
5. 已返回 `ResponseMessage`、`ResponseEntity` 或命中 excludes 时继续跳过。
媒体类型判断改为使用 Spring `selectMediaType(...)` 及 writer 声明的 streaming media types
不再依赖 `accept.contains(...)` 精确相等。以下类型继续直接交给原始 body
- `text/event-stream`
- `application/x-ndjson`
- Spring encoder 声明的其他 streaming media type包括兼容的 vendor `+x-ndjson`
- `X-Response-Wrapper: Ignore`。
### 7. 错误、取消和背压契约
#### 首个 buffer 前错误
响应未提交,错误原样传播,由现有 `CommonErrorControllerAdvice` 生成 HTTP 状态和错误
`ResponseMessage`。
#### 首个 buffer 后错误
HTTP 响应已经提交,无法再可靠改写状态码或完整 ResponseMessage。推荐行为是
- 不吞异常;
- 不把部分结果伪装成成功;
- 终止写出并让连接以不完整 JSON/传输错误结束;
- 日志保留原始异常,但不记录完整 payload。
如果调用方要求流内错误事件,应使用 NDJSON/SSE而不是普通 JSON 数组外壳。
#### 取消与背压
- 客户端断开或取消必须沿同一 Subscription 取消原始 result Publisher
- 不使用 `cache`、`replay`、`collectList`、无界 `buffer` 或额外 `subscribe`
- 元素序列化是同步一对一转换,使用 `map`/官方 encoder 即可,不引入并发 `flatMap`
- 默认使用 `writeWith`,不为每个元素强制 flush。若后续需要低延迟刷新应基于媒体类型或
有界批次单独设计,不能默认逐元素 `writeAndFlushWith`。
### 8. queryPager 有界聚合保护
`PagerResult.data` 是 `List<T>`,因此 `QueryHelper.queryPager(...)` 仍需要对当前页执行
`collectList()`。页大小必须保持跨请求、跨节点稳定,不能根据 JVM 实时剩余内存动态变化,否则
offset 分页可能重复或漏数。采用不可变 `PagerQueryPolicy` 统一规范化:
- 默认阈值为 1000保留 JVM 系统属性 `hsweb.max-pager-page-size`,并支持 Spring 配置
`hsweb.web.pageable.max-page-size`;配置值必须大于 0
- 默认溢出策略为兼容模式 `WARN`:显式 `pageSize > maxPageSize` 时记录不含查询条件的告警并保留
原值,避免 5.0.x 已有大页调用被静默截断;
- `CLAMP` 将超大页截断到最大值,`PagerResult.pageSize` 返回规范化后的实际值;`REJECT` 返回
`pageSize` 参数校验错误;配置为 `hsweb.web.pageable.overflow-policy`
- `pageSize < 1` 时回退到 easy-orm 默认页大小,再受最大页大小约束;
- `paging=false` 传入分页结果接口时,不受 `WARN` 兼容豁免,始终转换为最大受限页;真正的无分页
结果继续走返回 `Flux` 的 `/_query/no-paging`,大结果优先使用 NDJSON
- Spring WebFlux 将不可变策略 bean 写入 Reactor Context`QueryHelper` 在订阅时读取;非 HTTP
调用使用稳定默认策略,也可通过显式策略/最大值重载或 `contextWrite` 指定业务级规则;
- `ReactiveCrudService` 提供服务级 `resolvePagerQueryPolicy(ContextView)` 默认扩展点以及显式
`PagerQueryPolicy` 查询重载。策略优先级固定为“调用时显式策略 > 服务扩展点 > Reactor Context
> 框架默认策略”;默认扩展点只读取当前订阅上下文,不阻塞、不修改上下文,也不引入可变状态;
- 查询参数先 `clone()`,不修改调用方对象;查询侧分页限制之外,`collectList()` 前再使用
`take(effectivePageSize)`,即使自定义 `ReactiveQuery` 未正确应用分页也不会超过当前策略实际允许
的数量;不使用可变全局字段模拟请求级配置。
此保护只约束返回 `PagerResult` 的有界聚合链路,不限制显式流式查询,不修改数据库 SQL 方言、
排序、总数复用或重新分页语义。
## 兼容与发布策略
兼容对象来自已发布的 hsweb 响应协议和现有前端/外部调用方:
- 保持 `application/json` 的单个 ResponseMessage 外壳和 `result` 数组结构;
- 保持 Mono、显式 ResponseMessage、ResponseEntity、SSE、NDJSON、excludes 和
`X-Response-Wrapper: Ignore` 行为;
- 保持所有已注册 ThreadLocalAccessor 参与同步序列化;编码器不识别具体上下文类型;
- 传输方式从完成后一次性写出变为 chunked/增量写出;
- 首元素后的异常由“可返回完整错误 ResponseMessage”变为“连接终止”这是流式输出的固有
语义变化,必须在发布说明中明确。
推荐直接使用单一 canonical 流式实现,不长期维护 `collect` 与 `stream` 两套编码链。
如集成验证发现已发布调用方强依赖晚期错误仍返回完整 JSON再补充临时回滚配置该配置必须
有明确移除条件和最大收集条数,不能恢复无界 `collectList()`。
## 任务拆分
1. 新增测试,固定当前成功响应 JSON、Mono、空 Flux、显式 ResponseMessage、SSE/NDJSON、
excludes 和 Ignore 行为。
2. 重构 `CustomJackson2jsonEncoder`:继承官方 `Jackson2JsonEncoder`,增加基于
ContextSnapshot 的信号上下文装饰器,删除本地数组编码和 `collectList()`。
3. 为编码器补统一上下文、Reactor Context 优先级、异步线程切换、普通 JSON Flux 增量数组、
取消和首元素错误测试。
4. 新增内部 `StreamingResponseMessage` 与 `ResponseMessageJacksonHttpMessageWriter`。
5. 调整 `ResponseMessageWrapper`:使用 ReactiveAdapterRegistry、官方内容协商和专用 writer
删除 Flux `collectList()` 及无效 `switchIfEmpty()`。
6. 补 WebFlux 集成测试,验证首个 DataBuffer 在源完成前到达、JSON 外壳兼容、背压和取消。
7. 为 `QueryHelper.queryPager(...)` 增加不可变 `PagerQueryPolicy`、策略/最大页大小重载、订阅期
Reactor Context 解析,以及 `collectList()` 前的 `take(effectivePageSize)` 最终保护;为
`ReactiveCrudService` 增加服务级策略解析扩展点和显式策略重载。
8. 在 `CommonWebFluxConfiguration` 增加 `hsweb.web.pageable` 配置绑定和策略 WebFilter默认
`WARN` 保持显式大页兼容,`CLAMP/REJECT` 由部署按容量开启。
9. 补分页边界测试默认页、0/负数、最大值、超大值、`paging=false`、三种溢出策略、并行/串行/
复用 total、调用方参数不变、Reactor Context 覆盖、Spring 属性绑定、服务策略覆盖、显式策略
优先级,以及查询源忽略分页时的最终限制与取消。
10. 运行目标模块测试与上游聚合测试;若实现假设变化,先更新本设计并重新确认。
## 测试目标与验收标准
### 编码器单元测试
文件:
`hsweb-starter/src/test/java/org/hswebframework/web/starter/jackson/CustomJackson2jsonEncoderTest.java`
- 普通 `Flux<TestEntity>` + `application/json` 输出合法 JSON 数组,且首个 DataBuffer 在源
complete 前可被请求到。
- 空 Flux 输出 `[]`。
- `application/x-ndjson` 保持逐行编码和流式媒体类型声明。
- Locale 为 `zh-CN`、`en-US` 时,现有 EnumDict 文案序列化保持一致。
- Reactor Context 中的 Authentication 在序列化 getter/扩展字段中可见,完成后 ThreadLocal
恢复原值。
- Reactor Context 与订阅线程 ThreadLocal 存在相同 accessor key 时,以 Reactor Context 为准,
序列化完成后恢复原 ThreadLocal。
- 任意新增的 Micrometer `ThreadLocalAccessor` 无需修改 encoder经过 `publishOn` 切换线程后,
对应值在同步 Jackson getter 中可见,序列化后工作线程恢复原值且不污染线程池。
- 下游取消后,上游收到 cancel编码器资源完成清理。
- 首元素序列化失败时不产生任何 DataBuffer错误类型和原因不被吞掉。
### Writer 单元测试
建议新增:
`hsweb-commons/hsweb-commons-crud/src/test/java/org/hswebframework/web/crud/web/ResponseMessageJacksonHttpMessageWriterTest.java`
- 三个真实形态实体输出一个 ResponseMessage`result` 是按原顺序排列的 JSON 数组。
- 空 Flux 输出 `result: []`。
- `ResponseMessage` 的 status、message、timestamp、可选 code 及 EntityFactory 扩展字段均保留。
- 首元素前源错误、首元素序列化错误时响应未提交。
- 首元素后源错误时错误向下游传播writer 不追加成功后缀。
- StepVerifier 以逐次 request 验证 writer 不会主动收集完整 Publisher。
- cancel 传播到源 Publisher已经分配但未写出的 DataBuffer 被释放。
### Wrapper/WebFlux 集成测试
建议新增:
`hsweb-commons/hsweb-commons-crud/src/test/java/org/hswebframework/web/crud/web/ResponseMessageWrapperTest.java`
- `Mono<TestEntity>` 保持单对象 ResponseMessage。
- `Flux<TestEntity>` + `application/json` 在源完成前收到首块数据,完整后 JSON 与现有协议兼容。
- 空 Flux 返回成功且 `result` 为 `[]`。
- `Flux.never()` 不产生完整响应,但不会在服务端累计元素;客户端取消后源被取消。
- SSE、NDJSON、vendor `+x-ndjson`、Ignore header 和 excludes 均绕过包装。
- 显式 `Mono<ResponseMessage<T>>` 不被重复包装。
- 首元素前业务异常由 `CommonErrorControllerAdvice` 返回对应 HTTP 状态和错误响应。
### 回归与规模验证
- 使用可解释的 `TestEntity` 数据生成 100,000 个元素,验证链路能完成且不存在
`collectList()`;不以脆弱的瞬时堆内存数字作为唯一断言。
- 通过 TestPublisher/自定义 Publisher 记录 request 与 cancel验证没有
`Long.MAX_VALUE` 驱动的完整收集语义;允许网络 writer 使用有限预取。
- 不使用 `Thread.sleep`;使用 StepVerifier、TestPublisher 和 WebTestClient 的响应体订阅控制。
### 验证命令
```bash
mvn -pl hsweb-starter,hsweb-commons/hsweb-commons-crud -am \
-Dtest=CustomJackson2jsonEncoderTest,ResponseMessageJacksonHttpMessageWriterTest,ResponseMessageWrapperTest,ResponseMessageStreamingIntegrationTest \
-Dsurefire.failIfNoSpecifiedTests=false test
mvn -pl hsweb-starter,hsweb-commons/hsweb-commons-crud -am test
```
通过标准:相关测试全部通过;普通 JSON、NDJSON、SSE 和 i18n 回归均符合上述契约;目标生产
代码不再对可能无界的响应 Flux 使用 `collectList()`。
## 可观测性与运维判断
- 不新增 TraceHolder/MonoTracer/FluxTracer这是 HTTP 序列化基础设施,不是业务阶段,现有
WebFlux HTTP tracing 已覆盖请求生命周期;逐元素 span 会造成高频噪声。
- 不新增 MBean实现不维护常驻队列、缓存或后台线程背压和缓冲由 Reactor/Netty 管理。
- 不涉及 SQL、数据权限、事务、事件或跨模块远程调用。
## 代码注释目标
实现时仅在以下非显然边界增加注释:
- 首个数组 buffer 与外层前缀合并,用于保留提交前异常处理能力;
- 首元素后错误只能终止已提交响应;
- ContextSnapshot 作用域必须包围官方 Jackson encoder 的同步 onNext 编码;
- discard 路径负责本地 DataBuffer 的 cancel 释放;官方 encoder 的私有资源生命周期不重复实现。
普通 `map`、`switchIfEmpty`、writer 选择等自解释胶水不增加冗余注释。
## 风险与待确认点
1. **晚期错误语义**:推荐在首个 buffer 已写出后中断响应,不尝试输出“部分结果 + 错误
ResponseMessage”。需要用户确认接受该流式固有语义。
2. **传输变化**:普通 JSON Flux 将使用 chunked/增量写出,不再等完成后计算 Content-Length。
3. **自定义 ResponseMessage**:必须通过 ObjectMapper/EntityFactory 回归测试确认扩展字段与命名
策略不丢失;若某个扩展重定义了 `result` 的序列化形态,需要把它建模为 writer SPI而不是
硬编码兼容分支。
4. **Spring 升级边界**:只使用公开的 `Jackson2JsonEncoder`、`HttpMessageWriter`、
`ReactiveAdapterRegistry`、`Operators.liftPublisher` 和 Context Propagation API不依赖私有
方法或复制源码,以降低版本升级风险。
## 实施结果2026-08-07
### 实际代码落点
- `hsweb-starter/src/main/java/org/hswebframework/web/starter/jackson/CustomJackson2jsonEncoder.java`
已改为继承 Spring `Jackson2JsonEncoder`。普通 JSON 多值 Publisher 的数组 framing、逐元素
编码和错误语义全部委托官方实现;本地只用 `ContextSnapshotFactory` 与
`Operators.liftPublisher` 在同步 `onNext` 序列化范围恢复当前 Reactor Context 中所有已注册的
Micrometer `ThreadLocalAccessor`。每次订阅通过 `captureAll(contextView)` 通用合并兼容
ThreadLocal 与 Reactor Context同 key 由 Reactor Context 覆盖;编码器不依赖任何具体上下文
类型。
- `hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/web/StreamingResponseMessage.java`
作为 package-private 内部交接模型Publisher 不作为普通 JavaBean 属性交给 Jackson。
- `hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/web/ResponseMessageJacksonHttpMessageWriter.java`
使用现有 Jackson encoder 编码外层元数据,并把 `result` 委托同一 encoder 增量编码为 JSON
数组;首个数组 buffer 与外层前缀合并,首元素前错误不会提交响应,晚期错误不会追加成功后缀。
- `hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/web/ResponseMessageWrapper.java`
已删除目标 WebFlux 响应链路的 `collectList()`,改用 `ReactiveAdapterRegistry`、官方内容协商
和 encoder 声明的 streaming media types。专用 writer 仅加入当前 wrapper 的 writer 列表,
不改变全局 codec 顺序。
- `hsweb-starter/src/test/java/org/hswebframework/web/starter/jackson/ResponseMessageStreamingIntegrationTest.java`
启动最小 `@EnableWebFlux` 上下文与真实 Reactor Netty 随机端口服务,同时装配实际
`ResponseMessageWrapper` 和 `CustomJackson2jsonEncoder`,验证 HTTP 传输层而非直接调用 writer。
- `hsweb-starter/pom.xml` 显式声明项目已有的 `io.micrometer:context-propagation` 依赖,并增加
test-scope `reactor-netty-http` 用于真实 HTTP 集成测试,不改变发布依赖。
- `hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/query/QueryHelper.java`
的分页重载统一委托 `PagerQueryPolicy`,默认重载在订阅期从 Reactor Context 取策略;已知 total、
并行分页和串行分页三条 `collectList()` 链路前均增加 `take(effectivePageSize)`。显式
`maxPageSize` 重载固定采用 `CLAMP`,显式策略重载用于非 HTTP 或更严格的业务场景。
- `hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/query/PagerQueryPolicy.java`
是线程安全的不可变策略对象,统一处理默认页、非法小值、显式超大页和 `paging=false`;默认
`WARN` 保留已发布大页调用,`CLAMP` 截断,`REJECT` 返回 i18n 参数校验错误,并继续兼容
JVM 属性 `hsweb.max-pager-page-size`。
- `PagerQueryProperties` 与 `CommonWebFluxConfiguration` 绑定
`hsweb.web.pageable.max-page-size` / `overflow-policy`,创建可覆盖的策略 bean并由
WebFilter 写入每次请求的 Reactor Context不依赖可变静态 Holder 或 ThreadLocal。
- `ReactiveCrudService.queryPager(...)` 默认重载在每次订阅时调用
`resolvePagerQueryPolicy(ContextView)`;默认实现读取 Reactor Context服务实现可覆盖为稳定的
业务级策略。调用时显式传入 `PagerQueryPolicy` 的重载优先级最高,并直接委托 `QueryHelper`
不触发服务解析扩展点。
- `hsweb-commons/hsweb-commons-crud/src/test/java/org/hswebframework/web/crud/query/QueryHelperPagerTest.java`
覆盖正常页、0/负数、显式大页兼容、`paging=false`、`WARN/CLAMP/REJECT`、已知/零 total、
并行/串行分页、映射、参数不变、Reactor Context 覆盖和上游取消。
- `PagerQueryConfigurationTest` 使用 `ReactiveWebApplicationContextRunner` 验证 Spring 属性绑定、
自定义策略 bean back-off以及经过 `publishOn` 异步边界后的 Reactor Context 传播。
- `ReactiveCrudServicePagerPolicyTest` 验证默认服务经过 `publishOn` 后读取 Reactor Context、服务级
策略覆盖上下文、显式策略覆盖服务策略和上下文,以及显式策略便利重载。
### 已验证契约
- 普通 `application/json` Flux 在源完成前产生首个 DataBuffer最终仍是单个
`ResponseMessage`,且 `result` 为有序 JSON 数组。
- 空 Flux 输出 `result: []`Mono、显式 ResponseMessage、NDJSON、SSE、Ignore header 和
excludes 保持原有边界。
- Authentication、中文/英文 Locale 在同步 Jackson 序列化期间可见,完成后 ThreadLocal 恢复。
- 订阅线程 ThreadLocal 与 Reactor Context 存在相同 accessor key 时,序列化使用 Reactor Context
中的值,结束后恢复订阅线程原值。
- 真实 HTTP 请求在 `publishOn` 切换到专用 Scheduler 后Authentication、Locale 和测试动态注册
的 correlation-id accessor 都能在 Jackson getter 中读取;响应完成后再次检查同一工作线程,
correlation-id 与 Authentication 均已恢复,不存在请求上下文泄漏。
- 首元素前错误不输出外壳;首元素后错误原样传播且不追加成功后缀;取消传播到原 Publisher。
- EntityFactory 创建的 ResponseMessage 子类及其 Jackson 扩展字段通过同一个 ObjectMapper
保留。
- 真实 Reactor Netty 连接中,首个元素对应的 JSON 已在源 Publisher 完成前到达客户端;客户端
收到首元素后停止读取cancel 能继续传播到服务端原 Publisher。
- 100,000 个元素通过网络完整输出,客户端使用 Jackson non-blocking parser 按 ByteBuf 分片解析,
不把响应重新聚合为字符串或对象列表;最终元素数、`status`、`result` 数组和根对象闭合均正确。
- `application/vnd.hsweb+json` 保持 ResponseMessage 外壳Ignore header 返回原始数组NDJSON 和
SSE 保持不包装。
- 目标生产链路 `ResponseMessageWrapper` 和 `CustomJackson2jsonEncoder` 中已无
`collectList()`MVC `ResponseMessageWrapperAdvice` 不在本次范围内,仍保持原实现。
- `QueryHelper.queryPager(...)` 仍按 `PagerResult<List<T>>` 契约聚合当前页,但页大小已同时受查询
参数和 Reactor `take` 约束;自定义查询忽略分页参数时,测试确认在实际上限处取消上游。
- `pageSize < 1` 回退到受限默认值;显式超大页默认告警并保留旧值,可配置为截断或拒绝;
`paging=false` 无条件转换为最大受限页。规范化后的 `pageSize` 写入结果元数据,原始
`QueryParamEntity` 保持不变。
- Spring WebFlux 请求和默认 `ReactiveCrudService` 在异步调度后仍从同一 Reactor Context 读取
不可变策略;自定义 `PagerQueryPolicy` bean 会替代自动配置。服务可覆盖
`resolvePagerQueryPolicy(ContextView)`,非 HTTP 链路也可通过显式重载或 `contextWrite` 使用
相同契约;显式重载不会调用服务解析器。
定向验证命令通过:
```bash
mvn -pl hsweb-starter,hsweb-commons/hsweb-commons-crud -am \
-Dtest=CustomJackson2jsonEncoderTest,ResponseMessageJacksonHttpMessageWriterTest,ResponseMessageWrapperTest,ResponseMessageStreamingIntegrationTest \
-Dsurefire.failIfNoSpecifiedTests=false test
```
结果为 12/12 reactor modules success相关测试 29 个全部通过wrapper 10、writer 7、encoder 7、
真实 HTTP 集成测试 5。10 万元素测试属于可重复的规模集成验证,用于证明传输层确实增量工作;它
不等同于多并发、长时间运行并采集 JVM/Direct Memory 指标的正式容量压测。
分页聚合保护执行:
```bash
mvn -pl hsweb-commons/hsweb-commons-crud -am \
-Dtest=QueryHelperPagerTest,PagerQueryConfigurationTest,ReactiveCrudServicePagerPolicyTest \
-Dsurefire.failIfNoSpecifiedTests=false test
mvn -pl hsweb-commons/hsweb-commons-crud test
```
定向测试 17 个全部通过:`QueryHelperPagerTest` 11 个,
`PagerQueryConfigurationTest` 2 个,`ReactiveCrudServicePagerPolicyTest` 4 个。目标模块全量结果为
124 tests、0 failures、0 errors、1 skipped。JaCoCo 方法级结果:
`QueryHelper.doQueryPager` 25/25 lines、4/4 branches`PagerQueryPolicy.normalize` 15/15 lines、
6/6 branches`handleOverflow` 9/9 lines、3/3 branches`ReactiveCrudService` 默认策略入口、
两项显式策略重载和 `resolvePagerQueryPolicy` 均已覆盖。
全量 `-am test` 在进入目标模块前被既有的
`hsweb-datasource-api/DefaultSwitcherTest` 阻断(初始状态期望为空,实际为 `test`)。直接运行目标
模块时,`hsweb-commons-crud` 全量测试通过。`hsweb-starter/SystemInitializeTest` 最初因 EasyORM
初始化 H2 方言时同步初始化全部内置方言,而其 optional PostgreSQL 驱动不在测试类路径中,缺少
`io.r2dbc.postgresql.codec.PostgresqlObjectId`。`hsweb-starter` 已显式增加 test-scope
`r2dbc-postgresql`Java 17 下定向运行 `SystemInitializeTest` 为 1 test、0 failures、0 errors。
`hsweb-starter` 全量测试在本机 macOS 上继续运行到流式 HTTP 集成测试,其中
`testFirstHttpChunkArrivesBeforePublisherCompletes` 出现 30 秒读取超时;同一测试在增加上述测试依赖前
已通过 Linux CI且依赖树中的 Reactor Netty 版本仍由项目现有 `1.2.9` 管理,因此不对生产实现做
平台特调,以 PR 的 Linux CI 复跑结果作为最终判定。
### JavaBean 属性中嵌套 Flux 的结论
Spring WebFlux 通过 `ReactiveAdapterRegistry` 展开的是 Controller 顶层返回值。Jackson
serializer 是同步、单值的序列化扩展点,不能异步订阅 JavaBean 属性中的 Publisher因此
`PagerResult{total: 100, data: Flux<T>}` 不能依赖普通 Jackson 或一个内部 `subscribe()` 的
自定义 serializer 实现可靠的背压式输出。
需要该协议时应复用本次模式:定义显式的内部流式响应模型,由专用 `HttpMessageWriter` 先编码
`total` 等有界元数据,再把 `data` Publisher 委托官方 encoder 增量写为数组。该能力应作为后续
独立、通用的“嵌套 Publisher 响应 writer”设计不在本次修改 `PagerResult` 或 Jackson 全局行为。

View File

@@ -45,6 +45,7 @@ import java.util.List;
@Getter
@Setter
public class PagerResult<E> implements Serializable {
@Serial
private static final long serialVersionUID = -6171751136953308027L;
/**

View File

@@ -0,0 +1,176 @@
package org.hswebframework.web.crud.query;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.exception.ValidationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.util.context.Context;
import reactor.util.context.ContextView;
import java.util.Locale;
import java.util.Objects;
/**
* 响应式分页结果的页大小策略。
*
* <p>策略在订阅期从 Reactor Context 获取,保证异步链路使用稳定配置;它只约束必须聚合为
* {@code PagerResult<List<T>>} 的分页查询,不限制显式返回 {@code Flux<T>} 的流式查询。</p>
*
* @since 5.0.2
*/
public final class PagerQueryPolicy {
public static final String MAX_PAGE_SIZE_PROPERTY = "hsweb.web.pageable.max-page-size";
public static final String LEGACY_MAX_PAGE_SIZE_PROPERTY = "hsweb.max-pager-page-size";
public static final String OVERFLOW_POLICY_PROPERTY = "hsweb.web.pageable.overflow-policy";
public static final int DEFAULT_MAX_PAGE_SIZE = 1000;
private static final Logger log = LoggerFactory.getLogger(PagerQueryPolicy.class);
private static final PagerQueryPolicy DEFAULT = new PagerQueryPolicy(
resolveDefaultMaxPageSize(),
resolveDefaultOverflowPolicy());
private final int maxPageSize;
private final OverflowPolicy overflowPolicy;
public PagerQueryPolicy(int maxPageSize, OverflowPolicy overflowPolicy) {
if (maxPageSize < 1) {
throw new IllegalArgumentException("maxPageSize must be greater than 0");
}
this.maxPageSize = maxPageSize;
this.overflowPolicy = Objects.requireNonNull(overflowPolicy, "overflowPolicy");
}
/**
* 获取非 Spring 场景使用的稳定默认策略。
*
* @return 默认策略
*/
public static PagerQueryPolicy defaults() {
return DEFAULT;
}
/**
* 创建一个始终截断超大页的策略。
*
* @param maxPageSize 最大页大小
* @return CLAMP策略
*/
public static PagerQueryPolicy clamp(int maxPageSize) {
return new PagerQueryPolicy(maxPageSize, OverflowPolicy.CLAMP);
}
/**
* 从 Reactor Context 获取当前订阅使用的策略。
*
* @param contextView 当前订阅上下文
* @return 上下文策略;未设置时返回稳定默认策略
*/
public static PagerQueryPolicy from(ContextView contextView) {
return contextView.getOrDefault(PagerQueryPolicy.class, DEFAULT);
}
/**
* 将策略写入 Reactor Context。
*
* @param context 上下文
* @param policy 分页策略
* @return 包含策略的新上下文
*/
public static Context writeTo(Context context, PagerQueryPolicy policy) {
return context.put(PagerQueryPolicy.class, Objects.requireNonNull(policy, "policy"));
}
/**
* 复制并规范化分页参数。显式大页按溢出策略处理,{@code paging=false} 始终转换为最大受限页。
*
* @param source 原始查询参数,不会被修改
* @return 已开启分页并应用当前策略的查询参数副本
*/
public QueryParamEntity normalize(QueryParamEntity source) {
Objects.requireNonNull(source, "source");
QueryParamEntity normalized = source.clone();
int requestedPageSize = normalized.getPageSize();
int fallbackPageSize = Math.min(
Math.max(QueryParam.DEFAULT_PAGE_SIZE, 1),
maxPageSize);
int effectivePageSize;
if (!normalized.isPaging()) {
effectivePageSize = maxPageSize;
} else if (requestedPageSize < 1) {
effectivePageSize = fallbackPageSize;
} else if (requestedPageSize <= maxPageSize) {
effectivePageSize = requestedPageSize;
} else {
effectivePageSize = handleOverflow(requestedPageSize);
}
normalized.setPaging(true);
normalized.setPageSize(effectivePageSize);
return normalized;
}
public int getMaxPageSize() {
return maxPageSize;
}
public OverflowPolicy getOverflowPolicy() {
return overflowPolicy;
}
private int handleOverflow(int requestedPageSize) {
return switch (overflowPolicy) {
case WARN -> {
// 兼容已发布的大页调用;告警只包含数量,不记录查询条件或业务数据。
log.warn(
"Requested pageSize [{}] exceeds configured maxPageSize [{}], preserving it because overflow policy is WARN",
requestedPageSize,
maxPageSize);
yield requestedPageSize;
}
case CLAMP -> maxPageSize;
case REJECT -> throw new ValidationException.NoStackTrace(
"pageSize",
"error.page_size_exceeded",
requestedPageSize,
maxPageSize);
};
}
private static int resolveDefaultMaxPageSize() {
return Integer.getInteger(
MAX_PAGE_SIZE_PROPERTY,
Integer.getInteger(LEGACY_MAX_PAGE_SIZE_PROPERTY, DEFAULT_MAX_PAGE_SIZE));
}
private static OverflowPolicy resolveDefaultOverflowPolicy() {
String value = System.getProperty(OVERFLOW_POLICY_PROPERTY, OverflowPolicy.WARN.name());
try {
return OverflowPolicy.valueOf(value.trim().toUpperCase(Locale.ROOT));
} catch (IllegalArgumentException error) {
throw new IllegalArgumentException(
"Unsupported pager overflow policy: " + value,
error);
}
}
/**
* 显式分页大小超过阈值时的处理方式。
*/
public enum OverflowPolicy {
/** 记录告警并保留调用方页大小,用于兼容迁移。 */
WARN,
/** 截断为最大页大小。 */
CLAMP,
/** 返回参数校验错误。 */
REJECT
}
}

View File

@@ -13,6 +13,7 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import jakarta.validation.constraints.NotNull;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
@@ -442,7 +443,7 @@ public interface QueryHelper {
*
* @return 数据流
*/
Flux<R> fetch(int pageIndex,int pageSize);
Flux<R> fetch(int pageIndex, int pageSize);
/**
* 执行分页查询,默认返回第一页的25条数据.
@@ -746,17 +747,99 @@ public interface QueryHelper {
}
/**
* 指定ReactiveQuery和QueryParamEntity,执行查询并封装为分页查询结果.
* 使用指定的最大页大小执行分页查询
*
* @param param QueryParamEntity
* @param query ReactiveQuery
* @param maxPageSize 当前场景允许聚合的最大页大小
* @param <T> T
* @return PagerResult
*/
static <T> Mono<PagerResult<T>> queryPager(QueryParamEntity param,
Supplier<ReactiveQuery<T>> query,
int maxPageSize) {
return queryPager(param, query, Function.identity(), PagerQueryPolicy.clamp(maxPageSize));
}
/**
* 使用指定策略执行分页查询。
*
* @param param QueryParamEntity
* @param query ReactiveQuery
* @param mapper 转换结果类型
* @param policy 分页策略
* @param <T> T
* @return PagerResult
*/
static <T, R> Mono<PagerResult<R>> queryPager(QueryParamEntity param,
static <T> Mono<PagerResult<T>> queryPager(QueryParamEntity param,
Supplier<ReactiveQuery<T>> query,
PagerQueryPolicy policy) {
return queryPager(param, query, Function.identity(), policy);
}
/**
* 指定ReactiveQuery和QueryParamEntity,执行查询并封装为分页查询结果.
*
* @param queryParam QueryParamEntity
* @param query ReactiveQuery
* @param mapper 转换结果类型
* @param <T> T
* @param <R> R
* @return PagerResult
*/
static <T, R> Mono<PagerResult<R>> queryPager(QueryParamEntity queryParam,
Supplier<ReactiveQuery<T>> query,
Function<T, R> mapper) {
return Mono.deferContextual(contextView -> doQueryPager(
queryParam,
query,
mapper,
PagerQueryPolicy.from(contextView)));
}
/**
* 使用指定的最大页大小执行分页查询并转换结果类型。
*
* @param queryParam 原始查询参数,不会被修改
* @param query ReactiveQuery
* @param mapper 转换结果类型
* @param maxPageSize 当前场景允许聚合的最大页大小
* @param <T> 查询结果类型
* @param <R> 转换结果类型
* @return PagerResult
*/
static <T, R> Mono<PagerResult<R>> queryPager(QueryParamEntity queryParam,
Supplier<ReactiveQuery<T>> query,
Function<T, R> mapper,
int maxPageSize) {
return queryPager(queryParam, query, mapper, PagerQueryPolicy.clamp(maxPageSize));
}
/**
* 使用指定策略执行分页查询并转换结果类型。
*
* @param queryParam 原始查询参数,不会被修改
* @param query ReactiveQuery
* @param mapper 转换结果类型
* @param policy 分页策略
* @param <T> 查询结果类型
* @param <R> 转换结果类型
* @return PagerResult
*/
static <T, R> Mono<PagerResult<R>> queryPager(QueryParamEntity queryParam,
Supplier<ReactiveQuery<T>> query,
Function<T, R> mapper,
PagerQueryPolicy policy) {
return Mono.defer(() -> doQueryPager(queryParam, query, mapper, policy));
}
private static <T, R> Mono<PagerResult<R>> doQueryPager(QueryParamEntity queryParam,
Supplier<ReactiveQuery<T>> query,
Function<T, R> mapper,
PagerQueryPolicy policy) {
// PagerResult的数据必须聚合为List统一规范化后再查询真正的无分页结果由Flux接口承载。
QueryParamEntity param = policy.normalize(queryParam);
//如果查询参数指定了总数,表示不需要再进行count操作.
//建议前端在使用分页查询时,切换下一页时,将第一次查询到total结果传入查询参数,可以提升查询性能.
if (param.getTotal() != null) {
@@ -764,6 +847,7 @@ public interface QueryHelper {
.get()
.setParam(param.rePaging(param.getTotal()))
.fetch()
.take(param.getPageSize())
.map(mapper)
.collectList()
.map(list -> PagerResult.of(param.getTotal(), list, param));
@@ -773,7 +857,13 @@ public interface QueryHelper {
return Mono
.zip(
query.get().setParam(param.clone()).count(),
query.get().setParam(param.clone()).fetch().map(mapper).collectList(),
query
.get()
.setParam(param.clone())
.fetch()
.take(param.getPageSize())
.map(mapper)
.collectList(),
(total, data) -> PagerResult.of(total, data, param)
);
}
@@ -791,6 +881,7 @@ public interface QueryHelper {
.get()
.setParam(rePagingQuery)
.fetch()
.take(rePagingQuery.getPageSize())
.map(mapper)
.collectList()
.map(list -> PagerResult.of(total, list, rePagingQuery));

View File

@@ -8,12 +8,14 @@ import org.hswebframework.ezorm.rdb.mapping.defaults.SaveResult;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.api.crud.entity.TransactionManagers;
import org.hswebframework.web.crud.query.QueryHelper;
import org.hswebframework.web.crud.query.PagerQueryPolicy;
import org.reactivestreams.Publisher;
import org.springframework.transaction.annotation.Transactional;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.context.ContextView;
import java.util.ArrayList;
import java.util.Collection;
import java.util.function.Function;
@@ -90,103 +92,103 @@ public interface ReactiveCrudService<E, K> {
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<E> findById(K id) {
return getRepository()
.findById(id);
.findById(id);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Flux<E> findById(Collection<K> publisher) {
return getRepository()
.findById(publisher);
.findById(publisher);
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<E> findById(Mono<K> publisher) {
return getRepository()
.findById(publisher);
.findById(publisher);
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default Flux<E> findById(Flux<K> publisher) {
return getRepository()
.findById(publisher);
.findById(publisher);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<SaveResult> save(Publisher<E> entityPublisher) {
return getRepository()
.save(entityPublisher);
.save(entityPublisher);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<SaveResult> save(E data) {
return getRepository()
.save(data);
.save(data);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<SaveResult> save(Collection<E> collection) {
return getRepository()
.save(collection);
.save(collection);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> updateById(K id, Mono<E> entityPublisher) {
return getRepository()
.updateById(id, entityPublisher);
.updateById(id, entityPublisher);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> updateById(K id, E data) {
return getRepository()
.updateById(id, Mono.just(data));
.updateById(id, Mono.just(data));
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> insertBatch(Publisher<? extends Collection<E>> entityPublisher) {
return getRepository()
.insertBatch(entityPublisher);
.insertBatch(entityPublisher);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> insert(Publisher<E> entityPublisher) {
return getRepository()
.insert(entityPublisher);
.insert(entityPublisher);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> insert(E data) {
return getRepository()
.insert(Mono.just(data));
.insert(Mono.just(data));
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> deleteById(Publisher<K> idPublisher) {
return getRepository()
.deleteById(idPublisher);
.deleteById(idPublisher);
}
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
@Transactional(rollbackFor = Throwable.class, transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> deleteById(K id) {
return getRepository()
.deleteById(Mono.just(id));
.deleteById(Mono.just(id));
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default Flux<E> query(Mono<? extends QueryParamEntity> queryParamMono) {
return queryParamMono
.flatMapMany(this::query);
.flatMapMany(this::query);
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default Flux<E> query(QueryParamEntity param) {
return getRepository()
.createQuery()
.setParam(param)
.fetch();
.createQuery()
.setParam(param)
.fetch();
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
@@ -194,50 +196,66 @@ public interface ReactiveCrudService<E, K> {
return queryPager(queryParamMono, Function.identity());
}
/**
* 使用当前订阅解析出的分页策略执行分页查询。
*
* <p>策略在订阅期通过 {@link #resolvePagerQueryPolicy(ContextView)} 获取,因此服务实现可以
* 在保留 Reactor Context 默认行为的同时提供稳定的业务级限制。</p>
*
* @param query 查询参数,执行时不会修改原对象
* @param mapper 结果转换函数
* @param <T> 结果类型
* @return 分页查询结果
* @since 5.0.2
*/
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default <T> Mono<PagerResult<T>> queryPager(QueryParamEntity query, Function<E, T> mapper) {
//如果查询参数指定了总数,表示不需要再进行count操作.
//建议前端在使用分页查询时,切换下一页时,将第一次查询到total结果传入查询参数,可以提升查询性能.
if (query.getTotal() != null) {
return getRepository()
.createQuery()
.setParam(query.rePaging(query.getTotal()))
.fetch()
.map(mapper)
.collectList()
.map(list -> PagerResult.of(query.getTotal(), list, query));
}
//并行分页,更快,所在页码无数据时,会返回空list.
if (query.isParallelPager()) {
return Mono
.zip(
createQuery().setParam(query.clone()).count(),
createQuery().setParam(query.clone()).fetch().map(mapper).collectList(),
(total, data) -> PagerResult.of(total, data, query)
);
}
return getRepository()
.createQuery()
.setParam(query.clone())
.count()
.flatMap(total -> {
if (total == 0) {
return Mono.just(PagerResult.of(0, new ArrayList<>(), query));
}
//查询前根据数据总数进行重新分页:要跳转的页码没有数据则跳转到最后一页
QueryParamEntity rePagingQuery = query.clone().rePaging(total);
return query(rePagingQuery)
.map(mapper)
.collectList()
.map(list -> PagerResult.of(total, list, rePagingQuery));
});
return Mono.deferContextual(contextView -> queryPager(
query,
mapper,
resolvePagerQueryPolicy(contextView)));
}
/**
* 使用显式策略执行分页查询。显式策略优先于服务扩展点和 Reactor Context。
*
* @param query 查询参数,执行时不会修改原对象
* @param policy 本次查询使用的非空不可变策略
* @return 分页查询结果
* @since 5.0.2
*/
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<PagerResult<E>> queryPager(QueryParamEntity query,
PagerQueryPolicy policy) {
return queryPager(query, Function.identity(), policy);
}
/**
* 使用显式策略执行分页查询并转换结果。
*
* <p>该重载不调用 {@link #resolvePagerQueryPolicy(ContextView)};策略校验、错误传播、
* 有界收集和取消语义由 {@link QueryHelper#queryPager(QueryParamEntity, java.util.function.Supplier, Function, PagerQueryPolicy)}
* 统一处理。</p>
*
* @param query 查询参数,执行时不会修改原对象
* @param mapper 结果转换函数
* @param policy 本次查询使用的非空不可变策略
* @param <T> 结果类型
* @return 分页查询结果
* @since 5.0.2
*/
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default <T> Mono<PagerResult<T>> queryPager(QueryParamEntity query,
Function<E, T> mapper,
PagerQueryPolicy policy) {
return QueryHelper.queryPager(query, this::createQuery, mapper, policy);
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default <T> Mono<PagerResult<T>> queryPager(Mono<? extends QueryParamEntity> queryParamMono, Function<E, T> mapper) {
return queryParamMono
.cast(QueryParamEntity.class)
.flatMap(param -> queryPager(param, mapper));
.cast(QueryParamEntity.class)
.flatMap(param -> queryPager(param, mapper));
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
@@ -245,12 +263,28 @@ public interface ReactiveCrudService<E, K> {
return queryPager(queryParamMono, Function.identity());
}
/**
* 解析当前服务订阅使用的分页策略。
*
* <p>框架在每次 {@code queryPager} 订阅时调用本方法。默认实现读取只读 Reactor Context
* 未设置时回退到框架稳定默认策略。实现类可以返回服务级不可变策略,但必须保持同步、非阻塞且
* 返回非空值;抛出的异常会作为当前查询的响应式错误传播,不应在此执行查询或其他副作用。</p>
*
* @param contextView 当前订阅的只读 Reactor Context
* @return 本次订阅使用的非空不可变分页策略
* @see PagerQueryPolicy#from(ContextView)
* @since 5.0.2
*/
default PagerQueryPolicy resolvePagerQueryPolicy(ContextView contextView) {
return PagerQueryPolicy.from(contextView);
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
default Mono<Integer> count(QueryParamEntity queryParam) {
return getRepository()
.createQuery()
.setParam(queryParam)
.count();
.createQuery()
.setParam(queryParam)
.count();
}
@Transactional(transactionManager = TransactionManagers.reactiveTransactionManager)
@@ -258,5 +292,4 @@ public interface ReactiveCrudService<E, K> {
return queryParamMono.flatMap(this::count);
}
}

View File

@@ -1,5 +1,6 @@
package org.hswebframework.web.crud.web;
import org.hswebframework.web.crud.query.PagerQueryPolicy;
import org.hswebframework.web.i18n.WebFluxLocaleFilter;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -7,6 +8,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.ReactiveAdapterRegistry;
@@ -16,8 +18,25 @@ import org.springframework.web.server.WebFilter;
@AutoConfiguration
@ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE)
@EnableConfigurationProperties(PagerQueryProperties.class)
public class CommonWebFluxConfiguration {
@Bean
@ConditionalOnMissingBean
public PagerQueryPolicy pagerQueryPolicy(PagerQueryProperties properties) {
return properties.createPolicy();
}
/**
* 将应用级不可变策略写入每次订阅的 Reactor Context异步边界后仍由当前链路读取。
*/
@Bean
public WebFilter pagerQueryPolicyWebFilter(PagerQueryPolicy policy) {
return (exchange, chain) -> chain
.filter(exchange)
.contextWrite(context -> PagerQueryPolicy.writeTo(context, policy));
}
@Bean
@ConditionalOnMissingBean
public CommonErrorControllerAdvice commonErrorControllerAdvice() {

View File

@@ -0,0 +1,42 @@
package org.hswebframework.web.crud.web;
import org.hswebframework.web.crud.query.PagerQueryPolicy;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* 响应式分页结果的聚合保护配置。
*
* <p>配置只负责创建不可变的 {@link PagerQueryPolicy},请求级传播由 WebFlux 过滤器负责;
* 显式返回 Flux 的流式查询不受此配置限制。</p>
*
* @since 5.0.2
*/
@ConfigurationProperties(prefix = "hsweb.web.pageable")
public class PagerQueryProperties {
private int maxPageSize = PagerQueryPolicy.defaults().getMaxPageSize();
private PagerQueryPolicy.OverflowPolicy overflowPolicy = PagerQueryPolicy
.defaults()
.getOverflowPolicy();
public PagerQueryPolicy createPolicy() {
return new PagerQueryPolicy(maxPageSize, overflowPolicy);
}
public int getMaxPageSize() {
return maxPageSize;
}
public void setMaxPageSize(int maxPageSize) {
this.maxPageSize = maxPageSize;
}
public PagerQueryPolicy.OverflowPolicy getOverflowPolicy() {
return overflowPolicy;
}
public void setOverflowPolicy(PagerQueryPolicy.OverflowPolicy overflowPolicy) {
this.overflowPolicy = overflowPolicy;
}
}

View File

@@ -6,6 +6,7 @@ import lombok.Getter;
import lombok.Setter;
import org.hswebframework.web.api.crud.entity.EntityFactoryHolder;
import java.io.Serial;
import java.io.Serializable;
@Getter
@@ -13,6 +14,7 @@ import java.io.Serializable;
@JsonInclude(JsonInclude.Include.NON_NULL)
public class ResponseMessage<T> implements Serializable {
@Serial
private static final long serialVersionUID = 8992436576262574064L;
@Schema(description = "消息提示")

View File

@@ -0,0 +1,318 @@
package org.hswebframework.web.crud.web;
import com.fasterxml.jackson.core.JsonEncoding;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.BeanDescription;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.introspect.BeanPropertyDefinition;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.reactivestreams.Publisher;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.EncodingException;
import org.springframework.core.codec.Hints;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpOutputMessage;
import org.springframework.http.codec.HttpMessageWriter;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.lang.Nullable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
/**
* Writes one {@link ResponseMessage} whose {@code result} is an incremental JSON array.
*
* <p>Outer metadata is encoded with the configured Spring Jackson encoder. Array framing
* and element encoding are delegated to the same encoder, while this writer only joins
* the two JSON layers. It neither buffers the result Publisher nor subscribes separately.</p>
*/
final class ResponseMessageJacksonHttpMessageWriter
implements HttpMessageWriter<StreamingResponseMessage> {
private static final MediaType APPLICATION_PLUS_JSON =
MediaType.parseMediaType("application/*+json");
private static final List<MediaType> MEDIA_TYPES =
List.of(MediaType.APPLICATION_JSON, APPLICATION_PLUS_JSON);
private final Jackson2JsonEncoder encoder;
private final ObjectMapper mapper;
ResponseMessageJacksonHttpMessageWriter(Jackson2JsonEncoder encoder) {
this.encoder = encoder;
this.mapper = encoder.getObjectMapper();
}
@Override
public List<MediaType> getWritableMediaTypes() {
return MEDIA_TYPES;
}
@Override
public boolean canWrite(ResolvableType elementType, @Nullable MediaType mediaType) {
Class<?> resolved = elementType.resolve();
if (resolved == null || !StreamingResponseMessage.class.isAssignableFrom(resolved)) {
return false;
}
return mediaType == null || MEDIA_TYPES
.stream()
.anyMatch(candidate -> candidate.isCompatibleWith(mediaType));
}
@Override
public Mono<Void> write(Publisher<? extends StreamingResponseMessage> inputStream,
ResolvableType elementType,
@Nullable MediaType mediaType,
ReactiveHttpOutputMessage outputMessage,
Map<String, Object> hints) {
MediaType contentType = selectContentType(mediaType, outputMessage);
return Mono
.from(inputStream)
.flatMap(message -> outputMessage.writeWith(
encode(message, outputMessage.bufferFactory(), contentType, hints))
.thenReturn(true))
.defaultIfEmpty(false)
.flatMap(written -> written ? Mono.empty() : outputMessage.setComplete());
}
@Override
public Mono<Void> write(Publisher<? extends StreamingResponseMessage> inputStream,
ResolvableType actualType,
ResolvableType elementType,
@Nullable MediaType mediaType,
ServerHttpRequest request,
ServerHttpResponse response,
Map<String, Object> hints) {
MediaType contentType = selectContentType(mediaType, response);
return Mono
.from(inputStream)
.flatMap(message -> {
Map<String, Object> allHints = Hints.merge(
hints,
encoder.getEncodeHints(
message.getActualType(),
message.getElementType(),
contentType,
request,
response));
return response.writeWith(
encode(message, response.bufferFactory(), contentType, allHints))
.thenReturn(true);
})
.defaultIfEmpty(false)
.flatMap(written -> written ? Mono.empty() : response.setComplete());
}
Flux<DataBuffer> encode(StreamingResponseMessage message,
DataBufferFactory bufferFactory,
MediaType mediaType,
Map<String, Object> hints) {
Mono<EnvelopeFragments> fragments = encodeMetadata(
message.getMetadata(),
bufferFactory,
mediaType,
hints);
return fragments
.flatMapMany(parts -> joinEnvelope(
encoder.encode(
message.getResult(),
bufferFactory,
message.getElementType(),
mediaType,
hints),
parts,
bufferFactory))
.doOnDiscard(DataBuffer.class, DataBufferUtils::release);
}
private Mono<EnvelopeFragments> encodeMetadata(ResponseMessage<?> metadata,
DataBufferFactory bufferFactory,
MediaType mediaType,
Map<String, Object> hints) {
ResolvableType metadataType = ResolvableType.forInstance(metadata);
return encoder
.encode(Mono.just(metadata), bufferFactory, metadataType, mediaType, hints)
.single()
.map(buffer -> {
byte[] bytes = new byte[buffer.readableByteCount()];
try {
buffer.read(bytes);
return createEnvelopeFragments(
bytes,
metadata.getClass(),
jsonEncoding(mediaType));
} finally {
DataBufferUtils.release(buffer);
}
});
}
private Flux<DataBuffer> joinEnvelope(Flux<DataBuffer> encodedResult,
EnvelopeFragments parts,
DataBufferFactory bufferFactory) {
return encodedResult
.switchOnFirst((signal, result) -> {
if (signal.hasError()) {
return result;
}
if (!signal.hasValue()) {
return Flux.just(bufferFactory.wrap(parts.completeEmpty()));
}
return result
.index()
.map(indexed -> {
DataBuffer dataBuffer = indexed.getT2();
if (indexed.getT1() != 0) {
return dataBuffer;
}
// Delay the outer prefix until the first array buffer so an
// early source/serialization error remains uncommitted.
return bufferFactory.join(List.of(
bufferFactory.wrap(parts.prefix()),
dataBuffer));
})
.concatWith(Mono.fromSupplier(
() -> bufferFactory.wrap(parts.suffix())));
})
.doOnDiscard(DataBuffer.class, DataBufferUtils::release);
}
private EnvelopeFragments createEnvelopeFragments(byte[] metadata,
Class<?> metadataType,
JsonEncoding encoding) {
try {
JsonNode json = mapper.readTree(metadata);
if (!(json instanceof ObjectNode objectNode)) {
throw new EncodingException("ResponseMessage metadata must encode as a JSON object");
}
PropertyPosition property = resolveResultProperty(metadataType, objectNode);
ByteArrayOutputStream output = new ByteArrayOutputStream(metadata.length + 32);
int prefixEnd = -1;
int suffixStart = -1;
try (JsonGenerator generator = mapper.getFactory().createGenerator(output, encoding)) {
generator.setCodec(mapper);
generator.writeStartObject();
Iterator<Map.Entry<String, JsonNode>> fields = objectNode.fields();
while (fields.hasNext()) {
Map.Entry<String, JsonNode> field = fields.next();
String name = field.getKey();
if (prefixEnd < 0 &&
(name.equals(property.resultName()) ||
name.equals(property.nextPropertyName()))) {
generator.writeFieldName(property.resultName());
generator.writeStartArray();
generator.flush();
prefixEnd = output.size() - encoding.bits() / Byte.SIZE;
generator.writeEndArray();
generator.flush();
suffixStart = output.size();
}
if (name.equals(property.resultName())) {
continue;
}
generator.writeFieldName(name);
generator.writeTree(field.getValue());
}
if (prefixEnd < 0) {
generator.writeFieldName(property.resultName());
generator.writeStartArray();
generator.flush();
prefixEnd = output.size() - encoding.bits() / Byte.SIZE;
generator.writeEndArray();
generator.flush();
suffixStart = output.size();
}
generator.writeEndObject();
}
byte[] complete = output.toByteArray();
return new EnvelopeFragments(
Arrays.copyOfRange(complete, 0, prefixEnd),
Arrays.copyOfRange(complete, prefixEnd, suffixStart),
Arrays.copyOfRange(complete, suffixStart, complete.length));
} catch (IOException error) {
throw new EncodingException("Failed to create streaming ResponseMessage JSON", error);
}
}
private PropertyPosition resolveResultProperty(Class<?> metadataType, ObjectNode metadata) {
BeanDescription description = mapper
.getSerializationConfig()
.introspect(mapper.constructType(metadataType));
List<BeanPropertyDefinition> properties = description.findProperties();
for (int index = 0; index < properties.size(); index++) {
BeanPropertyDefinition property = properties.get(index);
if (!"result".equals(property.getInternalName())) {
continue;
}
for (int next = index + 1; next < properties.size(); next++) {
String nextName = properties.get(next).getName();
if (metadata.has(nextName)) {
return new PropertyPosition(property.getName(), nextName);
}
}
return new PropertyPosition(property.getName(), "");
}
return new PropertyPosition("result", metadata.has("status") ? "status" : "");
}
private JsonEncoding jsonEncoding(MediaType mediaType) {
Charset charset = mediaType.getCharset();
if (charset == null) {
return JsonEncoding.UTF8;
}
if (StandardCharsets.US_ASCII.equals(charset)) {
return JsonEncoding.UTF8;
}
for (JsonEncoding encoding : JsonEncoding.values()) {
if (encoding.getJavaName().equalsIgnoreCase(charset.name())) {
return encoding;
}
}
throw new EncodingException("Unsupported JSON charset: " + charset);
}
private MediaType selectContentType(@Nullable MediaType mediaType,
ReactiveHttpOutputMessage outputMessage) {
MediaType contentType = outputMessage.getHeaders().getContentType();
if (contentType == null) {
contentType = mediaType != null && mediaType.isConcrete()
? mediaType
: MediaType.APPLICATION_JSON;
outputMessage.getHeaders().setContentType(contentType);
}
return contentType;
}
private record PropertyPosition(String resultName, String nextPropertyName) {
}
private record EnvelopeFragments(byte[] prefix, byte[] emptyArray, byte[] suffix) {
byte[] completeEmpty() {
byte[] complete = new byte[prefix.length + emptyArray.length + suffix.length];
System.arraycopy(prefix, 0, complete, 0, prefix.length);
System.arraycopy(emptyArray, 0, complete, prefix.length, emptyArray.length);
System.arraycopy(suffix, 0, complete, prefix.length + emptyArray.length, suffix.length);
return complete;
}
}
}

View File

@@ -2,11 +2,18 @@ package org.hswebframework.web.crud.web;
import lombok.Getter;
import lombok.Setter;
import org.reactivestreams.Publisher;
import org.springframework.core.MethodParameter;
import org.springframework.core.ReactiveAdapter;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.core.ResolvableType;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.EncoderHttpMessageWriter;
import org.springframework.http.codec.HttpMessageEncoder;
import org.springframework.http.codec.HttpMessageWriter;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.lang.NonNull;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeType;
@@ -16,115 +23,226 @@ import org.springframework.web.reactive.HandlerResult;
import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
import org.springframework.web.reactive.result.method.annotation.ResponseBodyResultHandler;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* Wraps annotated reactive controller results in {@link ResponseMessage}.
*
* <p>Multi-value JSON responses use a dedicated writer so the original Publisher remains
* backpressure-aware and is encoded as an incremental array. Streaming protocols such as
* SSE and NDJSON remain unwrapped.</p>
*/
public class ResponseMessageWrapper extends ResponseBodyResultHandler {
public ResponseMessageWrapper(List<HttpMessageWriter<?>> writers,
RequestedContentTypeResolver resolver,
ReactiveAdapterRegistry registry) {
super(writers, resolver, registry);
setOrder(90);
}
private static final String IGNORE_HEADER = "X-Response-Wrapper";
private static MethodParameter param;
private static final MethodParameter RESPONSE_MESSAGE_TYPE =
returnTypeOf("methodForResponseMessage");
static {
try {
param = new MethodParameter(ResponseMessageWrapper.class
.getDeclaredMethod("methodForParams"), -1);
} catch (NoSuchMethodException e) {
e.printStackTrace();
}
}
private static final MethodParameter STREAMING_RESPONSE_MESSAGE_TYPE =
returnTypeOf("methodForStreamingResponseMessage");
private static Mono<ResponseMessage<?>> methodForParams() {
return Mono.empty();
}
private final boolean streamingResponseWriterAvailable;
@Setter
@Getter
private Set<String> excludes = new HashSet<>();
public ResponseMessageWrapper(List<HttpMessageWriter<?>> writers,
RequestedContentTypeResolver resolver,
ReactiveAdapterRegistry registry) {
this(configureWriters(writers), resolver, registry);
}
private ResponseMessageWrapper(WriterConfiguration configuration,
RequestedContentTypeResolver resolver,
ReactiveAdapterRegistry registry) {
super(configuration.writers(), resolver, registry);
this.streamingResponseWriterAvailable = configuration.streamingWriterAvailable();
setOrder(90);
}
@Override
public boolean supports(@NonNull HandlerResult result) {
if (isExcluded(result)) {
return false;
}
if (!CollectionUtils.isEmpty(excludes) && result.getHandler() instanceof HandlerMethod) {
HandlerMethod method = (HandlerMethod) result.getHandler();
ReactiveAdapter adapter = getAdapter(result);
if (adapter == null) {
return false;
}
String typeName = method.getMethod().getDeclaringClass().getName() + "." + method.getMethod().getName();
for (String exclude : excludes) {
if (typeName.startsWith(exclude)) {
return false;
ResolvableType elementType = getElementType(result.getReturnType(), adapter);
Class<?> elementClass = elementType.resolve();
if (elementClass != null &&
(ResponseMessage.class.isAssignableFrom(elementClass) ||
ResponseEntity.class.isAssignableFrom(elementClass))) {
return false;
}
RequestMapping mapping = getRequestMapping(result);
if (mapping == null || hasStreamingProduces(mapping)) {
return false;
}
return super.supports(result);
}
@Override
public Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
Object body = result.getReturnValue();
if ("Ignore".equals(exchange.getRequest().getHeaders().getFirst(IGNORE_HEADER))) {
return writeBody(body, result.getReturnTypeSource(), exchange);
}
ReactiveAdapter adapter = getAdapter(result);
if (adapter == null) {
return writeBody(body, result.getReturnTypeSource(), exchange);
}
ResolvableType elementType = getElementType(result.getReturnType(), adapter);
MediaType selectedMediaType = selectMediaType(
exchange,
() -> getMediaTypesFor(elementType));
if (isStreamingMediaType(selectedMediaType)) {
return writeBody(body, result.getReturnTypeSource(), exchange);
}
Publisher<?> publisher = adapter.toPublisher(body);
if (adapter.isMultiValue()) {
if (!streamingResponseWriterAvailable || !isJson(selectedMediaType)) {
return writeBody(body, result.getReturnTypeSource(), exchange);
}
StreamingResponseMessage streaming = new StreamingResponseMessage(
ResponseMessage.ok(),
publisher,
result.getReturnType(),
elementType);
return writeBody(
Mono.just(streaming),
STREAMING_RESPONSE_MESSAGE_TYPE,
exchange);
}
Mono<?> wrapped = adapter.isNoValue()
? Mono.from(publisher).then(Mono.fromSupplier(ResponseMessage::ok))
: Mono.from(publisher)
.map(ResponseMessage::ok)
.switchIfEmpty(Mono.fromSupplier(ResponseMessage::ok));
return writeBody(wrapped, RESPONSE_MESSAGE_TYPE, exchange);
}
private boolean isExcluded(HandlerResult result) {
if (CollectionUtils.isEmpty(excludes) || !(result.getHandler() instanceof HandlerMethod method)) {
return false;
}
String typeName = method.getMethod().getDeclaringClass().getName() + "." +
method.getMethod().getName();
return excludes.stream().anyMatch(typeName::startsWith);
}
private RequestMapping getRequestMapping(HandlerResult result) {
Method method = result.getReturnTypeSource().getMethod();
return method == null
? null
: AnnotatedElementUtils.findMergedAnnotation(method, RequestMapping.class);
}
private boolean hasStreamingProduces(RequestMapping mapping) {
for (String produce : mapping.produces()) {
if (isStreamingMediaType(MediaType.asMediaType(MimeType.valueOf(produce)))) {
return true;
}
}
return false;
}
private ResolvableType getElementType(ResolvableType returnType, ReactiveAdapter adapter) {
if (adapter.isNoValue()) {
return ResolvableType.forClass(Void.class);
}
ResolvableType generic = returnType.getGeneric();
return generic == ResolvableType.NONE
? ResolvableType.forClass(Object.class)
: generic;
}
private List<MediaType> getMediaTypesFor(ResolvableType elementType) {
List<MediaType> mediaTypes = new ArrayList<>();
for (HttpMessageWriter<?> writer : getMessageWriters()) {
if (writer.canWrite(elementType, null)) {
mediaTypes.addAll(writer.getWritableMediaTypes(elementType));
}
}
return mediaTypes;
}
private boolean isStreamingMediaType(MediaType mediaType) {
if (mediaType == null) {
return false;
}
if (MediaType.TEXT_EVENT_STREAM.isCompatibleWith(mediaType) ||
mediaType.getSubtype().endsWith("+x-ndjson")) {
return true;
}
for (HttpMessageWriter<?> writer : getMessageWriters()) {
if (!(writer instanceof EncoderHttpMessageWriter<?> encoderWriter) ||
!(encoderWriter.getEncoder() instanceof HttpMessageEncoder<?> encoder)) {
continue;
}
for (MediaType streamingType : encoder.getStreamingMediaTypes()) {
if (streamingType.isCompatibleWith(mediaType)) {
return true;
}
}
}
Class<?> gen = result.getReturnType().resolveGeneric(0);
return false;
}
boolean isAlreadyResponse = gen == ResponseMessage.class || gen == ResponseEntity.class;
private boolean isJson(MediaType mediaType) {
return mediaType != null &&
(MediaType.APPLICATION_JSON.isCompatibleWith(mediaType) ||
mediaType.getSubtype().endsWith("+json"));
}
boolean isStream = result.getReturnType().resolve() == Mono.class
|| result.getReturnType().resolve() == Flux.class;
RequestMapping mapping = result.getReturnTypeSource()
.getMethodAnnotation(RequestMapping.class);
if (mapping == null) {
return false;
}
for (String produce : mapping.produces()) {
MimeType mimeType = MimeType.valueOf(produce);
if (MediaType.TEXT_EVENT_STREAM.includes(mimeType) ||
MediaType.APPLICATION_NDJSON.includes(mimeType)) {
return false;
private static WriterConfiguration configureWriters(List<HttpMessageWriter<?>> writers) {
List<HttpMessageWriter<?>> configured = new ArrayList<>(writers);
for (int index = 0; index < configured.size(); index++) {
HttpMessageWriter<?> writer = configured.get(index);
if (writer instanceof EncoderHttpMessageWriter<?> encoderWriter &&
encoderWriter.getEncoder() instanceof Jackson2JsonEncoder encoder) {
configured.add(index, new ResponseMessageJacksonHttpMessageWriter(encoder));
return new WriterConfiguration(configured, true);
}
}
return isStream
&& super.supports(result)
&& !isAlreadyResponse;
return new WriterConfiguration(configured, false);
}
@Override
@SuppressWarnings("all")
public Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
Object body = result.getReturnValue();
List<MediaType> accept = exchange.getRequest().getHeaders().getAccept();
if (accept.contains(MediaType.TEXT_EVENT_STREAM)||
accept.contains(MediaType.APPLICATION_NDJSON)) {
return writeBody(body, result.getReturnTypeSource(), exchange);
private static MethodParameter returnTypeOf(String methodName) {
try {
return new MethodParameter(
ResponseMessageWrapper.class.getDeclaredMethod(methodName),
-1);
} catch (NoSuchMethodException error) {
throw new ExceptionInInitializerError(error);
}
}
String ignoreWrapper = exchange
.getRequest()
.getHeaders()
.getFirst("X-Response-Wrapper");
if ("Ignore".equals(ignoreWrapper)) {
return writeBody(body, result.getReturnTypeSource(), exchange);
}
private static Mono<ResponseMessage<?>> methodForResponseMessage() {
return Mono.empty();
}
if (body instanceof Mono) {
body = ((Mono) body)
.map(ResponseMessage::ok)
.switchIfEmpty(Mono.just(ResponseMessage.ok()));
}
if (body instanceof Flux) {
body = ((Flux) body)
.collectList()
.map(ResponseMessage::ok)
.switchIfEmpty(Mono.just(ResponseMessage.ok()));
}
if (body == null) {
body = Mono.just(ResponseMessage.ok());
}
return writeBody(body, param, exchange);
private static Mono<StreamingResponseMessage> methodForStreamingResponseMessage() {
return Mono.empty();
}
private record WriterConfiguration(List<HttpMessageWriter<?>> writers,
boolean streamingWriterAvailable) {
}
}

View File

@@ -0,0 +1,48 @@
package org.hswebframework.web.crud.web;
import org.reactivestreams.Publisher;
import org.springframework.core.ResolvableType;
/**
* Internal hand-off between {@link ResponseMessageWrapper} and the JSON message writer.
*
* <p>The controller contract remains a regular reactive return type. This marker keeps
* the result Publisher outside the JavaBean so Jackson never attempts to serialize or
* subscribe to a nested reactive property.</p>
*/
final class StreamingResponseMessage {
private final ResponseMessage<?> metadata;
private final Publisher<?> result;
private final ResolvableType actualType;
private final ResolvableType elementType;
StreamingResponseMessage(ResponseMessage<?> metadata,
Publisher<?> result,
ResolvableType actualType,
ResolvableType elementType) {
this.metadata = metadata;
this.result = result;
this.actualType = actualType;
this.elementType = elementType;
}
ResponseMessage<?> getMetadata() {
return metadata;
}
Publisher<?> getResult() {
return result;
}
ResolvableType getActualType() {
return actualType;
}
ResolvableType getElementType() {
return elementType;
}
}

View File

@@ -9,6 +9,7 @@ import org.hswebframework.web.api.crud.entity.QueryOperation;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.authorization.annotation.Authorize;
import org.hswebframework.web.authorization.annotation.QueryAction;
import org.hswebframework.web.crud.query.QueryHelper;
import org.hswebframework.web.exception.NotFoundException;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
@@ -100,22 +101,7 @@ public interface ReactiveQueryController<E, K> {
@QueryAction
@QueryOperation(summary = "使用GET方式分页动态查询")
default Mono<PagerResult<E>> queryPager(@Parameter(hidden = true) QueryParamEntity query) {
if (query.getTotal() != null) {
return getRepository()
.createQuery()
.setParam(query.rePaging(query.getTotal()))
.fetch()
.collectList()
.map(list -> PagerResult.of(query.getTotal(), list, query));
}
return Mono
.zip(
getRepository().createQuery().setParam(query.clone()).count(),
query(query.clone()).collectList(),
(total, data) -> PagerResult.of(total, data, query)
);
return QueryHelper.queryPager(query,getRepository()::createQuery);
}

View File

@@ -95,16 +95,7 @@ public interface ReactiveServiceQueryController<E, K> {
@QueryAction
@QueryOperation(summary = "使用GET方式分页动态查询")
default Mono<PagerResult<E>> queryPager(@Parameter(hidden = true) QueryParamEntity query) {
if (query.getTotal() != null) {
return getService()
.createQuery()
.setParam(query.rePaging(query.getTotal()))
.fetch()
.collectList()
.map(list -> PagerResult.of(query.getTotal(), list, query));
}
return getService().queryPager(query);
}
/**

View File

@@ -11,4 +11,5 @@ error.data.find.not_found=Data not found
error.sql.prepare.failed.IndexOutOfBoundsException=Execute SQL failed, try check config: `easyorm.dialect`.
error.missing_request_body=Required request body is missing
error.duplicate_key=Duplicate Data
error.data_access_failed=Data Access Failed
error.data_access_failed=Data Access Failed
error.page_size_exceeded=pageSize exceeds the maximum allowed value: {1}

View File

@@ -10,4 +10,5 @@ error.data.find.not_found=\u6570\u636E\u4E0D\u5B58\u5728
error.sql.prepare.failed.IndexOutOfBoundsException=SQL\u6267\u884C\u5931\u8D25,\u8BF7\u5C1D\u8BD5\u68C0\u67E5`easyorm.dialect`\u914D\u7F6E.
error.missing_request_body=\u8BF7\u6C42\u4F53\u7F3A\u5931
error.duplicate_key=\u5DF2\u5B58\u5728\u91CD\u590D\u7684\u6570\u636E
error.data_access_failed=\u8BBF\u95EE\u6570\u636E\u5931\u8D25
error.data_access_failed=\u8BBF\u95EE\u6570\u636E\u5931\u8D25
error.page_size_exceeded=pageSize\u8D85\u8FC7\u6700\u5927\u5141\u8BB8\u503C: {1}

View File

@@ -0,0 +1,285 @@
package org.hswebframework.web.crud.query;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.ezorm.rdb.mapping.ReactiveQuery;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.exception.ValidationException;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import reactor.util.context.Context;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class QueryHelperPagerTest {
@Test
public void testWarnPolicyKeepsExplicitLargePageForCompatibility() {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(20);
source.setTotal(100);
List<QueryParamEntity> appliedParams = new ArrayList<>();
ReactiveQuery<Integer> query = mockQuery(
Mono.just(100),
Flux.range(0, 100),
appliedParams);
StepVerifier
.create(QueryHelper.queryPager(
source,
() -> query,
new PagerQueryPolicy(10, PagerQueryPolicy.OverflowPolicy.WARN)))
.assertNext(result -> assertPage(result, 100, 20, 20))
.verifyComplete();
assertEquals(20, appliedParams.get(0).getPageSize());
assertEquals(20, source.getPageSize());
}
@Test
public void testReactorContextPolicyIsReadAtSubscription() {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(100);
source.setTotal(200);
List<QueryParamEntity> appliedParams = new ArrayList<>();
ReactiveQuery<Integer> query = mockQuery(
Mono.just(200),
Flux.range(0, 100),
appliedParams);
PagerQueryPolicy policy = PagerQueryPolicy.clamp(11);
StepVerifier
.create(QueryHelper
.queryPager(source, () -> query)
.contextWrite(context -> PagerQueryPolicy.writeTo(context, policy)))
.assertNext(result -> assertPage(result, 200, 11, 11))
.verifyComplete();
assertSame(policy, PagerQueryPolicy.from(
reactor.util.context.Context.of(PagerQueryPolicy.class, policy)));
assertEquals(11, appliedParams.get(0).getPageSize());
}
@Test
public void testKnownTotalUsesCustomMaxAndDoesNotMutateSource() {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(100);
source.setTotal(200);
List<QueryParamEntity> appliedParams = new ArrayList<>();
AtomicBoolean cancelled = new AtomicBoolean();
ReactiveQuery<Integer> query = mockQuery(
Mono.just(200),
Flux.range(0, 100).doOnCancel(() -> cancelled.set(true)),
appliedParams);
StepVerifier
.create(QueryHelper.queryPager(source, () -> query, 10))
.assertNext(result -> assertPage(result, 200, 10, 10))
.verifyComplete();
assertEquals(100, source.getPageSize());
assertEquals(1, appliedParams.size());
assertEquals(10, appliedParams.get(0).getPageSize());
assertTrue(appliedParams.get(0).isPaging());
assertTrue(cancelled.get());
verify(query, never()).count();
}
@Test
public void testNonPositivePageSizeFallsBackToBoundedDefault() {
PagerQueryPolicy policy = PagerQueryPolicy.clamp(100);
int expectedPageSize = Math.min(
Math.max(QueryParam.DEFAULT_PAGE_SIZE, 1),
policy.getMaxPageSize());
QueryParamEntity zero = QueryParamEntity.of();
zero.setPageSize(0);
QueryParamEntity negative = QueryParamEntity.of();
negative.setPageSize(-1);
assertEquals(expectedPageSize, policy.normalize(zero).getPageSize());
assertEquals(expectedPageSize, policy.normalize(negative).getPageSize());
assertEquals(0, zero.getPageSize());
assertEquals(-1, negative.getPageSize());
assertEquals(10, PagerQueryPolicy.clamp(10).normalize(zero).getPageSize());
}
@Test
public void testPageSizeWithinLimitIsKept() {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(50);
QueryParamEntity normalized = PagerQueryPolicy.clamp(100).normalize(source);
assertEquals(50, normalized.getPageSize());
assertEquals(50, source.getPageSize());
assertSame(PagerQueryPolicy.defaults(), PagerQueryPolicy.from(Context.empty()));
}
@Test
public void testNoPagingIsBoundedEvenWithWarnPolicy() {
QueryParamEntity source = QueryParamEntity.of().noPaging();
source.setPageSize(Integer.MAX_VALUE);
source.setTotal(200);
List<QueryParamEntity> appliedParams = new ArrayList<>();
ReactiveQuery<Integer> query = mockQuery(
Mono.just(200),
Flux.range(0, 100),
appliedParams);
StepVerifier
.create(QueryHelper.queryPager(
source,
() -> query,
new PagerQueryPolicy(12, PagerQueryPolicy.OverflowPolicy.WARN)))
.assertNext(result -> assertPage(result, 200, 12, 12))
.verifyComplete();
assertFalse(source.isPaging());
assertTrue(appliedParams.get(0).isPaging());
assertEquals(12, appliedParams.get(0).getPageSize());
}
@Test
public void testParallelPagerIsBounded() {
QueryParamEntity source = QueryParamEntity.of();
source.setParallelPager(true);
source.setPageSize(100);
List<QueryParamEntity> appliedParams = new ArrayList<>();
ReactiveQuery<Integer> query = mockQuery(
Mono.just(200),
Flux.range(0, 100),
appliedParams);
StepVerifier
.create(QueryHelper.queryPager(source, () -> query, 15))
.assertNext(result -> assertPage(result, 200, 15, 15))
.verifyComplete();
assertEquals(2, appliedParams.size());
assertTrue(appliedParams.stream().allMatch(param -> param.getPageSize() == 15));
verify(query).count();
}
@Test
public void testSequentialPagerIsBoundedAndMapsData() {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(Integer.MAX_VALUE);
List<QueryParamEntity> appliedParams = new ArrayList<>();
ReactiveQuery<Integer> query = mockQuery(
Mono.just(200),
Flux.range(0, 100),
appliedParams);
StepVerifier
.create(QueryHelper.queryPager(source, () -> query, Object::toString, 20))
.assertNext(result -> {
assertPage(result, 200, 20, 20);
assertEquals("0", result.getData().get(0));
assertEquals("19", result.getData().get(19));
})
.verifyComplete();
assertEquals(2, appliedParams.size());
assertTrue(appliedParams.stream().allMatch(param -> param.getPageSize() == 20));
assertEquals(Integer.MAX_VALUE, source.getPageSize());
}
@Test
public void testZeroTotalSkipsDataQuery() {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(10);
List<QueryParamEntity> appliedParams = new ArrayList<>();
ReactiveQuery<Integer> query = mockQuery(
Mono.just(0),
Flux.error(new AssertionError("fetch should not be called")),
appliedParams);
StepVerifier
.create(QueryHelper.queryPager(source, () -> query, 100))
.assertNext(result -> assertPage(result, 0, 10, 0))
.verifyComplete();
assertEquals(1, appliedParams.size());
verify(query, never()).fetch();
}
@Test
@SuppressWarnings("unchecked")
public void testRejectPolicyReturnsReactiveValidationError() {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(101);
AtomicBoolean queryCreated = new AtomicBoolean();
StepVerifier
.create(QueryHelper.queryPager(
source,
() -> {
queryCreated.set(true);
return mock(ReactiveQuery.class);
},
new PagerQueryPolicy(100, PagerQueryPolicy.OverflowPolicy.REJECT)))
.verifyErrorSatisfies(error -> {
assertTrue(error instanceof ValidationException);
ValidationException validation = (ValidationException) error;
assertEquals("error.page_size_exceeded", validation.getI18nCode());
assertEquals("pageSize", validation.getDetails().get(0).getProperty());
});
assertFalse(queryCreated.get());
assertEquals(101, source.getPageSize());
}
@Test
@SuppressWarnings("unchecked")
public void testInvalidCustomMaxFailsFast() {
QueryParamEntity source = QueryParamEntity.of();
for (int invalidMax : new int[]{0, -1}) {
try {
QueryHelper.queryPager(source, () -> mock(ReactiveQuery.class), invalidMax);
fail("Expected invalid max page size to fail: " + invalidMax);
} catch (IllegalArgumentException error) {
assertTrue(error.getMessage().contains("maxPageSize"));
}
}
}
private static void assertPage(PagerResult<?> result,
int total,
int pageSize,
int dataSize) {
assertEquals(total, result.getTotal());
assertEquals(pageSize, result.getPageSize());
assertEquals(dataSize, result.getData().size());
}
@SuppressWarnings("unchecked")
private static ReactiveQuery<Integer> mockQuery(Mono<Integer> count,
Flux<Integer> data,
List<QueryParamEntity> appliedParams) {
ReactiveQuery<Integer> query = mock(ReactiveQuery.class);
when(query.setParam(any(QueryParam.class))).thenAnswer(invocation -> {
QueryParam param = invocation.getArgument(0);
appliedParams.add(QueryParamEntity.of(param));
return query;
});
when(query.count()).thenReturn(count);
when(query.fetch()).thenReturn(data);
return query;
}
}

View File

@@ -0,0 +1,170 @@
package org.hswebframework.web.crud.service;
import org.hswebframework.ezorm.core.param.QueryParam;
import org.hswebframework.ezorm.rdb.mapping.ReactiveQuery;
import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.hswebframework.web.crud.query.PagerQueryPolicy;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
import reactor.util.context.ContextView;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class ReactiveCrudServicePagerPolicyTest {
@Test
public void testDefaultServiceUsesReactorContextAfterAsyncBoundary() {
QueryParamEntity source = query(100);
List<QueryParamEntity> appliedParams = new ArrayList<>();
TestService service = new TestService(mockQuery(appliedParams));
PagerQueryPolicy contextPolicy = PagerQueryPolicy.clamp(15);
StepVerifier
.create(Mono
.just(source)
.publishOn(Schedulers.parallel())
.flatMap(service::queryPager)
.contextWrite(context -> PagerQueryPolicy.writeTo(context, contextPolicy)))
.assertNext(result -> assertPage(result, 15))
.verifyComplete();
assertEquals(15, appliedParams.get(0).getPageSize());
}
@Test
public void testServicePolicyOverridesReactorContext() {
QueryParamEntity source = query(100);
List<QueryParamEntity> appliedParams = new ArrayList<>();
CustomPolicyService service = new CustomPolicyService(
mockQuery(appliedParams),
PagerQueryPolicy.clamp(7));
StepVerifier
.create(service
.queryPager(source)
.contextWrite(context -> PagerQueryPolicy.writeTo(
context,
PagerQueryPolicy.clamp(20))))
.assertNext(result -> assertPage(result, 7))
.verifyComplete();
assertEquals(1, service.getResolutionCount());
assertEquals(7, appliedParams.get(0).getPageSize());
}
@Test
public void testExplicitPolicyOverridesServiceAndReactorContext() {
QueryParamEntity source = query(100);
List<QueryParamEntity> appliedParams = new ArrayList<>();
CustomPolicyService service = new CustomPolicyService(
mockQuery(appliedParams),
PagerQueryPolicy.clamp(7));
StepVerifier
.create(service
.queryPager(source, Object::toString, PagerQueryPolicy.clamp(13))
.contextWrite(context -> PagerQueryPolicy.writeTo(
context,
PagerQueryPolicy.clamp(20))))
.assertNext(result -> {
assertPage(result, 13);
assertEquals("0", result.getData().get(0));
})
.verifyComplete();
assertEquals(0, service.getResolutionCount());
assertEquals(13, appliedParams.get(0).getPageSize());
}
@Test
public void testExplicitPolicyConvenienceOverload() {
QueryParamEntity source = query(100);
List<QueryParamEntity> appliedParams = new ArrayList<>();
TestService service = new TestService(mockQuery(appliedParams));
StepVerifier
.create(service.queryPager(source, PagerQueryPolicy.clamp(9)))
.assertNext(result -> assertPage(result, 9))
.verifyComplete();
assertEquals(9, appliedParams.get(0).getPageSize());
}
private static QueryParamEntity query(int pageSize) {
QueryParamEntity source = QueryParamEntity.of();
source.setPageSize(pageSize);
source.setTotal(200);
return source;
}
private static void assertPage(PagerResult<?> result, int pageSize) {
assertEquals(200, result.getTotal());
assertEquals(pageSize, result.getPageSize());
assertEquals(pageSize, result.getData().size());
}
@SuppressWarnings("unchecked")
private static ReactiveQuery<Integer> mockQuery(List<QueryParamEntity> appliedParams) {
ReactiveQuery<Integer> query = mock(ReactiveQuery.class);
when(query.setParam(any(QueryParam.class))).thenAnswer(invocation -> {
appliedParams.add(QueryParamEntity.of(invocation.getArgument(0)));
return query;
});
when(query.fetch()).thenReturn(Flux.range(0, 200));
return query;
}
private static class TestService implements ReactiveCrudService<Integer, Integer> {
private final ReactiveQuery<Integer> query;
private TestService(ReactiveQuery<Integer> query) {
this.query = query;
}
@Override
public ReactiveRepository<Integer, Integer> getRepository() {
throw new UnsupportedOperationException();
}
@Override
public ReactiveQuery<Integer> createQuery() {
return query;
}
}
private static final class CustomPolicyService extends TestService {
private final PagerQueryPolicy policy;
private final AtomicInteger resolutionCount = new AtomicInteger();
private CustomPolicyService(ReactiveQuery<Integer> query,
PagerQueryPolicy policy) {
super(query);
this.policy = policy;
}
@Override
public PagerQueryPolicy resolvePagerQueryPolicy(ContextView contextView) {
resolutionCount.incrementAndGet();
return policy;
}
private int getResolutionCount() {
return resolutionCount.get();
}
}
}

View File

@@ -0,0 +1,69 @@
package org.hswebframework.web.crud.web;
import org.hswebframework.web.crud.query.PagerQueryPolicy;
import org.junit.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner;
import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
import org.springframework.mock.web.server.MockServerWebExchange;
import org.springframework.web.server.WebFilter;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
public class PagerQueryConfigurationTest {
private final ReactiveWebApplicationContextRunner contextRunner =
new ReactiveWebApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(CommonWebFluxConfiguration.class))
.withPropertyValues("hsweb.webflux.response-wrapper.enabled=false");
@Test
public void testPropertiesCreatePolicyAndFilterPropagatesIt() {
contextRunner
.withPropertyValues(
"hsweb.web.pageable.max-page-size=321",
"hsweb.web.pageable.overflow-policy=reject")
.run(context -> {
PagerQueryPolicy policy = context.getBean(PagerQueryPolicy.class);
assertEquals(321, policy.getMaxPageSize());
assertEquals(
PagerQueryPolicy.OverflowPolicy.REJECT,
policy.getOverflowPolicy());
WebFilter filter = context.getBean(
"pagerQueryPolicyWebFilter",
WebFilter.class);
MockServerWebExchange exchange = MockServerWebExchange.from(
MockServerHttpRequest.get("/pager"));
StepVerifier
.create(filter.filter(
exchange,
ignored -> Mono
.just(1)
.publishOn(Schedulers.parallel())
.then(Mono.deferContextual(contextView -> {
assertSame(policy, PagerQueryPolicy.from(contextView));
return Mono.empty();
}))))
.verifyComplete();
});
}
@Test
public void testCustomPolicyBeanBacksOffAutoConfiguration() {
PagerQueryPolicy custom = new PagerQueryPolicy(
77,
PagerQueryPolicy.OverflowPolicy.CLAMP);
contextRunner
.withBean(PagerQueryPolicy.class, () -> custom)
.run(context -> assertSame(
custom,
context.getBean(PagerQueryPolicy.class)));
}
}

View File

@@ -0,0 +1,200 @@
package org.hswebframework.web.crud.web;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Test;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.MediaType;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import reactor.test.publisher.TestPublisher;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class ResponseMessageJacksonHttpMessageWriterTest {
private final ObjectMapper mapper = new ObjectMapper();
private final ResponseMessageJacksonHttpMessageWriter writer =
new ResponseMessageJacksonHttpMessageWriter(new Jackson2JsonEncoder(mapper));
@Test
public void testResponseMessageJsonContract() {
StreamingResponseMessage message = message(Flux.just(
new TestEntity("device-001"),
new TestEntity("device-002"),
new TestEntity("device-003")));
encode(message)
.as(DataBufferUtils::join)
.map(this::readAndRelease)
.map(this::readTree)
.as(StepVerifier::create)
.assertNext(json -> {
assertEquals("success", json.get("message").asText());
assertEquals(200, json.get("status").asInt());
assertEquals(123L, json.get("timestamp").asLong());
assertEquals(3, json.get("result").size());
assertEquals("device-001", json.get("result").get(0).get("id").asText());
assertEquals("device-003", json.get("result").get(2).get("id").asText());
})
.verifyComplete();
}
@Test
public void testEmptyResult() {
encode(message(Flux.empty()))
.as(DataBufferUtils::join)
.map(this::readAndRelease)
.map(this::readTree)
.as(StepVerifier::create)
.assertNext(json -> assertTrue(json.get("result").isEmpty()))
.verifyComplete();
}
@Test
public void testResponseMessageExtensionFieldIsPreserved() {
ExtendedResponseMessage metadata = new ExtendedResponseMessage();
metadata.setMessage("success");
metadata.setStatus(200);
metadata.setTimestamp(123L);
StreamingResponseMessage message = new StreamingResponseMessage(
metadata,
Flux.just(new TestEntity("device-001")),
ResolvableType.forClassWithGenerics(Flux.class, TestEntity.class),
ResolvableType.forClass(TestEntity.class));
encode(message)
.as(DataBufferUtils::join)
.map(this::readAndRelease)
.map(this::readTree)
.as(StepVerifier::create)
.assertNext(json -> assertEquals("trace-001", json.get("traceId").asText()))
.verifyComplete();
}
@Test
public void testFirstChunkArrivesBeforeSourceCompletes() {
TestPublisher<TestEntity> source = TestPublisher.create();
StepVerifier
.create(encode(message(source.flux())), 0)
.thenRequest(1)
.then(() -> source.next(new TestEntity("device-001")))
.assertNext(buffer -> {
String json = readAndRelease(buffer);
assertTrue(json.startsWith("{\"message\":\"success\",\"result\":[{"));
assertTrue(json.contains("device-001"));
})
.thenRequest(2)
.then(source::complete)
.assertNext(buffer -> assertEquals("]", readAndRelease(buffer)))
.assertNext(buffer -> assertTrue(readAndRelease(buffer).contains("\"status\":200")))
.verifyComplete();
}
@Test
public void testErrorBeforeFirstItemDoesNotEmitEnvelope() {
IllegalStateException error = new IllegalStateException("source failed");
StepVerifier
.create(encode(message(Flux.error(error))))
.expectErrorMatches(actual -> actual == error)
.verify();
}
@Test
public void testErrorAfterFirstItemDoesNotAppendSuccessSuffix() {
IllegalStateException error = new IllegalStateException("source failed");
StepVerifier
.create(encode(message(Flux.concat(
Flux.just(new TestEntity("device-001")),
Flux.error(error)))))
.assertNext(buffer -> {
String json = readAndRelease(buffer);
assertTrue(json.startsWith("{\"message\":\"success\",\"result\":[{"));
assertTrue(!json.contains("\"status\":200"));
})
.expectErrorMatches(actual -> actual == error)
.verify();
}
@Test
public void testCancellationPropagates() {
AtomicBoolean cancelled = new AtomicBoolean();
Flux<TestEntity> source = Flux.create(sink -> {
sink.onCancel(() -> cancelled.set(true));
sink.next(new TestEntity("device-001"));
});
StepVerifier
.create(encode(message(source)))
.assertNext(this::readAndRelease)
.thenCancel()
.verify();
assertTrue(cancelled.get());
}
private Flux<DataBuffer> encode(StreamingResponseMessage message) {
return writer.encode(
message,
new DefaultDataBufferFactory(),
MediaType.APPLICATION_JSON,
Collections.emptyMap());
}
private StreamingResponseMessage message(Flux<TestEntity> source) {
return new StreamingResponseMessage(
ResponseMessage.of("success", null, 200, null, 123L),
source,
ResolvableType.forClassWithGenerics(Flux.class, TestEntity.class),
ResolvableType.forClass(TestEntity.class));
}
private String readAndRelease(DataBuffer buffer) {
try {
return buffer.toString(StandardCharsets.UTF_8);
} finally {
DataBufferUtils.release(buffer);
}
}
private JsonNode readTree(String json) {
try {
return mapper.readTree(json);
} catch (Exception error) {
throw new AssertionError("Invalid JSON: " + json, error);
}
}
public static class TestEntity {
private final String id;
public TestEntity(String id) {
this.id = id;
}
public String getId() {
return id;
}
}
public static class ExtendedResponseMessage extends ResponseMessage<Object> {
public String getTraceId() {
return "trace-001";
}
}
}

View File

@@ -0,0 +1,267 @@
package org.hswebframework.web.crud.web;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Test;
import org.springframework.core.MethodParameter;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
import org.springframework.mock.web.server.MockServerWebExchange;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.method.HandlerMethod;
import org.springframework.web.reactive.HandlerResult;
import org.springframework.web.reactive.accept.RequestedContentTypeResolverBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import reactor.test.publisher.TestPublisher;
import java.lang.reflect.Method;
import java.util.Set;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class ResponseMessageWrapperTest {
private final ObjectMapper mapper = new ObjectMapper();
private final ResponseMessageWrapper wrapper = createWrapper();
@Test
public void testFluxIsWrappedAsStreamingResultArray() throws Exception {
HandlerResult result = handlerResult(
"flux",
Flux.just(new TestEntity("device-001"), new TestEntity("device-002")));
MockServerWebExchange exchange = exchange(MediaType.APPLICATION_JSON);
assertTrue(wrapper.supports(result));
wrapper
.handleResult(exchange, result)
.then(Mono.defer(exchange.getResponse()::getBodyAsString))
.map(this::readTree)
.as(StepVerifier::create)
.assertNext(json -> {
if (!json.has("message") || !json.has("result")) {
throw new AssertionError("Actual JSON: " + json);
}
assertEquals("success", json.get("message").asText());
assertEquals(2, json.get("result").size());
assertEquals("device-001", json.get("result").get(0).get("id").asText());
})
.verifyComplete();
}
@Test
public void testEmptyFluxHasEmptyResultArray() throws Exception {
HandlerResult result = handlerResult("flux", Flux.empty());
MockServerWebExchange exchange = exchange(MediaType.APPLICATION_JSON);
wrapper
.handleResult(exchange, result)
.then(Mono.defer(exchange.getResponse()::getBodyAsString))
.map(this::readTree)
.as(StepVerifier::create)
.assertNext(json -> {
if (!json.has("result")) {
throw new AssertionError("Actual JSON: " + json);
}
assertTrue(json.get("result").isEmpty());
})
.verifyComplete();
}
@Test
public void testMonoKeepsSingleResultContract() throws Exception {
HandlerResult result = handlerResult("mono", Mono.just(new TestEntity("device-001")));
MockServerWebExchange exchange = exchange(MediaType.APPLICATION_JSON);
wrapper
.handleResult(exchange, result)
.then(Mono.defer(exchange.getResponse()::getBodyAsString))
.map(this::readTree)
.as(StepVerifier::create)
.assertNext(json -> assertEquals(
"device-001",
json.get("result").get("id").asText()))
.verifyComplete();
}
@Test
public void testNdjsonIsNotWrapped() throws Exception {
HandlerResult result = handlerResult(
"flux",
Flux.just(new TestEntity("device-001"), new TestEntity("device-002")));
MockServerWebExchange exchange = exchange(MediaType.APPLICATION_NDJSON);
wrapper
.handleResult(exchange, result)
.then(Mono.defer(exchange.getResponse()::getBodyAsString))
.as(StepVerifier::create)
.assertNext(json -> {
assertFalse(json.contains("\"message\""));
assertTrue(json.contains("device-001"));
assertTrue(json.contains("device-002"));
})
.verifyComplete();
}
@Test
public void testSseAcceptIsNotWrapped() throws Exception {
HandlerResult result = handlerResult(
"flux",
Flux.just(new TestEntity("device-001"), new TestEntity("device-002")));
MockServerWebExchange exchange = exchange(MediaType.TEXT_EVENT_STREAM);
assertTrue(wrapper.supports(result));
wrapper
.handleResult(exchange, result)
.then(Mono.defer(exchange.getResponse()::getBodyAsString))
.as(StepVerifier::create)
.assertNext(body -> {
assertFalse(body.contains("\"message\""));
assertTrue(body.contains("data:"));
assertTrue(body.contains("device-001"));
})
.verifyComplete();
}
@Test
public void testStreamingProducesIsNotSupported() throws Exception {
HandlerResult result = handlerResult(
"sse",
Flux.just(new TestEntity("device-001")));
assertFalse(wrapper.supports(result));
}
@Test
public void testExcludesStillBypassWrapper() throws Exception {
wrapper.setExcludes(Set.of(TestController.class.getName() + ".flux"));
assertFalse(wrapper.supports(handlerResult(
"flux",
Flux.just(new TestEntity("device-001")))));
}
@Test
public void testCancellationPropagatesFromWrapper() throws Exception {
TestPublisher<TestEntity> source = TestPublisher.create();
HandlerResult result = handlerResult("flux", source.flux());
StepVerifier
.create(wrapper.handleResult(exchange(MediaType.APPLICATION_JSON), result))
.then(() -> source.assertSubscribers(1))
.then(() -> source.next(new TestEntity("device-001")))
.thenCancel()
.verify();
source.assertCancelled();
}
@Test
public void testIgnoreHeaderBypassesWrapper() throws Exception {
HandlerResult result = handlerResult(
"flux",
Flux.just(new TestEntity("device-001"), new TestEntity("device-002")));
MockServerWebExchange exchange = MockServerWebExchange.from(
MockServerHttpRequest
.get("/test")
.accept(MediaType.APPLICATION_JSON)
.header("X-Response-Wrapper", "Ignore"));
wrapper
.handleResult(exchange, result)
.then(Mono.defer(exchange.getResponse()::getBodyAsString))
.map(this::readTree)
.as(StepVerifier::create)
.assertNext(json -> {
assertTrue(json.isArray());
assertEquals(2, json.size());
})
.verifyComplete();
}
@Test
public void testExplicitResponseMessageIsNotSupported() throws Exception {
HandlerResult result = handlerResult(
"response",
Mono.just(ResponseMessage.ok(new TestEntity("device-001"))));
assertFalse(wrapper.supports(result));
}
private ResponseMessageWrapper createWrapper() {
RequestedContentTypeResolverBuilder resolver = new RequestedContentTypeResolverBuilder();
resolver.headerResolver();
ResponseMessageWrapper result = new ResponseMessageWrapper(
ServerCodecConfigurer.create().getWriters(),
resolver.build(),
ReactiveAdapterRegistry.getSharedInstance());
assertTrue(result.getMessageWriters().stream()
.anyMatch(ResponseMessageJacksonHttpMessageWriter.class::isInstance));
return result;
}
private HandlerResult handlerResult(String methodName, Object value) throws Exception {
Method method = TestController.class.getDeclaredMethod(methodName);
HandlerMethod handler = new HandlerMethod(new TestController(), method);
return new HandlerResult(handler, value, new MethodParameter(method, -1));
}
private MockServerWebExchange exchange(MediaType accept) {
return MockServerWebExchange.from(
MockServerHttpRequest.get("/test").accept(accept));
}
private JsonNode readTree(String json) {
try {
return mapper.readTree(json);
} catch (Exception error) {
throw new AssertionError("Invalid JSON: " + json, error);
}
}
@Controller
@ResponseBody
public static class TestController {
@RequestMapping("/flux")
public Flux<TestEntity> flux() {
return Flux.empty();
}
@RequestMapping("/mono")
public Mono<TestEntity> mono() {
return Mono.empty();
}
@RequestMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<TestEntity> sse() {
return Flux.empty();
}
@RequestMapping("/response")
public Mono<ResponseMessage<TestEntity>> response() {
return Mono.empty();
}
}
public static class TestEntity {
private final String id;
public TestEntity(String id) {
this.id = id;
}
public String getId() {
return id;
}
}
}

View File

@@ -23,6 +23,11 @@
<artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>context-propagation</artifactId>
</dependency>
<dependency>
<groupId>org.hswebframework.web</groupId>
<artifactId>hsweb-commons-crud</artifactId>
@@ -41,6 +46,12 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor.netty</groupId>
<artifactId>reactor-netty-http</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-test-autoconfigure</artifactId>
@@ -71,6 +82,14 @@
<scope>test</scope>
</dependency>
<!-- EasyORM会预先初始化全部内置方言。保留其可选PostgreSQL类型到测试运行时
直到EasyORM改为仅加载实际选择的方言。 -->
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>r2dbc-postgresql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.glassfish.expressly</groupId>
<artifactId>expressly</artifactId>
@@ -88,4 +107,4 @@
<!-- <scope>compile</scope>-->
<!-- </dependency>-->
</dependencies>
</project>
</project>

View File

@@ -1,369 +1,114 @@
package org.hswebframework.web.starter.jackson;
import lombok.AllArgsConstructor;
import org.hswebframework.web.authorization.Authentication;
import org.hswebframework.web.authorization.AuthenticationHolder;
import org.hswebframework.web.authorization.simple.SimpleAuthentication;
import org.hswebframework.web.i18n.LocaleUtils;
import org.springframework.http.codec.json.Jackson2CodecSupport;
import java.io.IOException;
import java.lang.annotation.Annotation;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.function.Function;
import com.fasterxml.jackson.core.JsonEncoding;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.util.ByteArrayBuilder;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.SequenceWriter;
import com.fasterxml.jackson.databind.exc.InvalidDefinitionException;
import io.micrometer.context.ContextSnapshot;
import io.micrometer.context.ContextSnapshotFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.MethodParameter;
import org.reactivestreams.Subscription;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.CodecException;
import org.springframework.core.codec.EncodingException;
import org.springframework.core.codec.Hints;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.log.LogFormatUtils;
import org.springframework.http.MediaType;
import org.springframework.http.codec.HttpMessageEncoder;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Operators;
import reactor.util.context.Context;
import javax.annotation.Nonnull;
import java.util.Map;
import java.util.function.Function;
/**
* Base class providing support methods for Jackson 2.9 encoding. For non-streaming use
* cases, {@link Flux} elements are collected into a {@link List} before serialization for
* performance reason.
* Jackson encoder that delegates JSON framing to Spring and restores hsweb's
* registered thread-local context while Jackson synchronously serializes each value.
*
* <p>The context bridge only decorates Reactive Streams signals. It does not subscribe,
* buffer, or alter demand and cancellation semantics. All registered Micrometer
* {@code ThreadLocalAccessor} values are captured once per subscription, with Reactor
* Context values overriding same-key compatibility values from the subscribing thread.</p>
*
* @author Sebastien Deleuze
* @author Arjen Poutsma
* @since 5.0
*/
public class CustomJackson2jsonEncoder extends Jackson2CodecSupport implements HttpMessageEncoder<Object> {
private static final byte[] NEWLINE_SEPARATOR = {'\n'};
private static final Map<MediaType, byte[]> STREAM_SEPARATORS;
private static final Map<String, JsonEncoding> ENCODINGS;
static {
STREAM_SEPARATORS = new HashMap<>(4);
STREAM_SEPARATORS.put(MediaType.APPLICATION_NDJSON, NEWLINE_SEPARATOR);
STREAM_SEPARATORS.put(MediaType.parseMediaType("application/stream+x-jackson-smile"), new byte[0]);
ENCODINGS = new HashMap<>(JsonEncoding.values().length + 1);
for (JsonEncoding encoding : JsonEncoding.values()) {
ENCODINGS.put(encoding.getJavaName(), encoding);
}
ENCODINGS.put("US-ASCII", JsonEncoding.UTF8);
}
private final List<MediaType> streamingMediaTypes = new ArrayList<>(1);
public class CustomJackson2jsonEncoder extends Jackson2JsonEncoder {
private static final ContextSnapshotFactory SNAPSHOT_FACTORY = ContextSnapshotFactory
.builder()
.clearMissing(true)
.build();
/**
* Constructor with a Jackson {@link ObjectMapper} to use.
* Constructor with the application configured {@link ObjectMapper}.
*
* @param mapper mapper shared with Spring WebFlux
* @param mimeTypes optional supported mime types
*/
protected CustomJackson2jsonEncoder(ObjectMapper mapper, MimeType... mimeTypes) {
super(mapper, mimeTypes);
streamingMediaTypes.add(MediaType.APPLICATION_NDJSON);
}
/**
* Configure "streaming" media types for which flushing should be performed
* automatically vs at the end of the stream.
* <p>By default this is set to {@link MediaType#APPLICATION_STREAM_JSON}.
*
* @param mediaTypes one or more media types to add to the list
* @see HttpMessageEncoder#getStreamingMediaTypes()
*/
public void setStreamingMediaTypes(List<MediaType> mediaTypes) {
this.streamingMediaTypes.clear();
this.streamingMediaTypes.addAll(mediaTypes);
}
@Override
public boolean canEncode(ResolvableType elementType, @Nullable MimeType mimeType) {
Class<?> clazz = elementType.toClass();
if (!supportsMimeType(mimeType)) {
return false;
public Flux<DataBuffer> encode(Publisher<?> inputStream,
DataBufferFactory bufferFactory,
ResolvableType elementType,
@Nullable MimeType mimeType,
@Nullable Map<String, Object> hints) {
return Flux.deferContextual(contextView -> super.encode(
restoreContext(inputStream, SNAPSHOT_FACTORY.captureAll(contextView)),
bufferFactory,
elementType,
mimeType,
hints));
}
private <T> Publisher<T> restoreContext(Publisher<T> inputStream, ContextSnapshot snapshot) {
Function<? super Publisher<T>, ? extends Publisher<T>> lifter = Operators.liftPublisher(
(publisher, subscriber) -> new ContextRestoringSubscriber<>(subscriber, snapshot));
if (inputStream instanceof Mono<?>) {
return Mono.from(inputStream).transform(lifter);
}
if (mimeType != null && mimeType.getCharset() != null) {
Charset charset = mimeType.getCharset();
if (!ENCODINGS.containsKey(charset.name())) {
return false;
return Flux.from(inputStream).transform(lifter);
}
private static final class ContextRestoringSubscriber<T> implements CoreSubscriber<T> {
private final CoreSubscriber<? super T> actual;
private final ContextSnapshot snapshot;
private ContextRestoringSubscriber(CoreSubscriber<? super T> actual,
ContextSnapshot snapshot) {
this.actual = actual;
this.snapshot = snapshot;
}
@Override
public void onSubscribe(Subscription subscription) {
actual.onSubscribe(subscription);
}
@Override
public void onNext(T value) {
// The downstream onNext performs Spring's synchronous Jackson encoding.
try (ContextSnapshot.Scope ignored = snapshot.setThreadLocals()) {
actual.onNext(value);
}
}
return (Object.class == clazz ||
(!String.class.isAssignableFrom(elementType.resolve(clazz)) && getObjectMapper().canSerialize(clazz)));
}
@Override
@Nonnull
public Flux<DataBuffer> encode(@Nonnull Publisher<?> inputStream, @Nonnull DataBufferFactory bufferFactory,
@Nonnull ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
Assert.notNull(inputStream, "'inputStream' must not be null");
Assert.notNull(bufferFactory, "'bufferFactory' must not be null");
Assert.notNull(elementType, "'elementType' must not be null");
if (inputStream instanceof Mono) {
return Mono
.zip(
currentContext(hints),
Mono.from(inputStream),
(ctx, value) -> ctx
.execute(() -> encodeValue(value, bufferFactory, elementType, mimeType, hints))
)
.flux();
} else {
byte[] separator = streamSeparator(mimeType);
if (separator != null) { // streaming
try {
ObjectWriter writer = createObjectWriter(elementType, mimeType, hints);
ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer
.getFactory()
._getBufferRecycler());
JsonEncoding encoding = getJsonEncoding(mimeType);
JsonGenerator generator = getObjectMapper()
.getFactory()
.createGenerator(byteBuilder, encoding);
SequenceWriter sequenceWriter = writer.writeValues(generator);
return currentContext(hints)
.flatMapMany(ctx -> ctx
.transform(inputStream,
value -> this
.encodeStreamingValue(value,
bufferFactory,
hints,
sequenceWriter,
byteBuilder,
separator)))
.doAfterTerminate(() -> {
try {
byteBuilder.release();
generator.close();
} catch (IOException ex) {
logger.error("Could not close Encoder resources", ex);
}
});
} catch (IOException ex) {
return Flux.error(ex);
}
} else { // non-streaming
ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
return currentContext(hints)
.flatMapMany(ctx -> ctx
.transform(Flux.from(inputStream).collectList(),
value -> encodeValue(value, bufferFactory, listType, mimeType, hints)));
}
}
}
@Override
@Nonnull
public DataBuffer encodeValue(@Nonnull Object value,@Nonnull DataBufferFactory bufferFactory,
@Nonnull ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
ObjectWriter writer = createObjectWriter(valueType, mimeType, hints);
ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer.getFactory()._getBufferRecycler());
try {
JsonEncoding encoding = getJsonEncoding(mimeType);
logValue(hints, value);
try (JsonGenerator generator = getObjectMapper().getFactory().createGenerator(byteBuilder, encoding)) {
writer.writeValue(generator, value);
generator.flush();
} catch (InvalidDefinitionException ex) {
throw new CodecException("Type definition error: " + ex.getType(), ex);
} catch (JsonProcessingException ex) {
throw new EncodingException("JSON encoding error: " + ex.getOriginalMessage(), ex);
} catch (IOException ex) {
throw new IllegalStateException("Unexpected I/O error while writing to byte array builder", ex);
}
byte[] bytes = byteBuilder.toByteArray();
DataBuffer buffer = bufferFactory.allocateBuffer(bytes.length);
buffer.write(bytes);
return buffer;
} finally {
byteBuilder.release();
}
}
private DataBuffer encodeStreamingValue(Object value, DataBufferFactory bufferFactory, @Nullable Map<String, Object> hints,
SequenceWriter sequenceWriter, ByteArrayBuilder byteArrayBuilder, byte[] separator) {
logValue(hints, value);
try {
sequenceWriter.write(value);
sequenceWriter.flush();
} catch (InvalidDefinitionException ex) {
throw new CodecException("Type definition error: " + ex.getType(), ex);
} catch (JsonProcessingException ex) {
throw new EncodingException("JSON encoding error: " + ex.getOriginalMessage(), ex);
} catch (IOException ex) {
throw new IllegalStateException("Unexpected I/O error while writing to byte array builder", ex);
@Override
public void onError(Throwable throwable) {
actual.onError(throwable);
}
byte[] bytes = byteArrayBuilder.toByteArray();
byteArrayBuilder.reset();
int offset;
int length;
if (bytes.length > 0 && bytes[0] == ' ') {
// SequenceWriter writes an unnecessary space in between values
offset = 1;
length = bytes.length - 1;
} else {
offset = 0;
length = bytes.length;
}
DataBuffer buffer = bufferFactory.allocateBuffer(length + separator.length);
buffer.write(bytes, offset, length);
buffer.write(separator);
return buffer;
}
private void logValue(@Nullable Map<String, Object> hints, Object value) {
if (!Hints.isLoggingSuppressed(hints)) {
LogFormatUtils.traceDebug(logger, traceOn -> {
String formatted = LogFormatUtils.formatValue(value, !traceOn);
return Hints.getLogPrefix(hints) + "Encoding [" + formatted + "]";
});
}
}
private ObjectWriter createObjectWriter(ResolvableType valueType, @Nullable MimeType mimeType,
@Nullable Map<String, Object> hints) {
JavaType javaType = getJavaType(valueType.getType(), null);
Class<?> jsonView = (hints != null ? (Class<?>) hints.get(Jackson2CodecSupport.JSON_VIEW_HINT) : null);
ObjectWriter writer = (jsonView != null ?
getObjectMapper().writerWithView(jsonView) : getObjectMapper().writer());
if (javaType.isContainerType()) {
writer = writer.forType(javaType);
@Override
public void onComplete() {
actual.onComplete();
}
return customizeWriter(writer, mimeType, valueType, hints);
}
protected ObjectWriter customizeWriter(ObjectWriter writer, @Nullable MimeType mimeType,
ResolvableType elementType, @Nullable Map<String, Object> hints) {
return writer;
}
@Nullable
private byte[] streamSeparator(@Nullable MimeType mimeType) {
for (MediaType streamingMediaType : this.streamingMediaTypes) {
if (streamingMediaType.isCompatibleWith(mimeType)) {
return STREAM_SEPARATORS.getOrDefault(streamingMediaType, NEWLINE_SEPARATOR);
}
}
return null;
}
/**
* Determine the JSON encoding to use for the given mime type.
*
* @param mimeType the mime type as requested by the caller
* @return the JSON encoding to use (never {@code null})
* @since 5.0.5
*/
protected JsonEncoding getJsonEncoding(@Nullable MimeType mimeType) {
if (mimeType != null && mimeType.getCharset() != null) {
Charset charset = mimeType.getCharset();
JsonEncoding result = ENCODINGS.get(charset.name());
if (result != null) {
return result;
}
}
return JsonEncoding.UTF8;
}
// HttpMessageEncoder
@Override
public List<MimeType> getEncodableMimeTypes() {
return getMimeTypes();
}
@Override
public List<MediaType> getStreamingMediaTypes() {
return Collections.unmodifiableList(this.streamingMediaTypes);
}
@Override
public Map<String, Object> getEncodeHints(@Nullable ResolvableType actualType, ResolvableType elementType,
@Nullable MediaType mediaType, ServerHttpRequest request, ServerHttpResponse response) {
return (actualType != null ? getHints(actualType) : Hints.none());
}
// Jackson2CodecSupport
@Override
protected <A extends Annotation> A getAnnotation(MethodParameter parameter, Class<A> annotType) {
return parameter.getMethodAnnotation(annotType);
}
static final SimpleAuthentication ANONYMOUS = new SimpleAuthentication();
static Mono<EncodingContext> currentContext(Map<String, Object> hints) {
return Mono
.zip(Authentication.currentReactive().defaultIfEmpty(ANONYMOUS),
LocaleUtils.currentReactive(), EncodingContext::new);
}
@AllArgsConstructor
static class EncodingContext {
private final Authentication authentication;
private final Locale locale;
private <T, R> Flux<T> transform(Publisher<R> source, Function<R, T> transformer) {
return Flux
.from(source)
.map((val) -> execute(() -> transformer.apply(val)));
}
private <T> T execute(Callable<T> callable) {
if (authentication == null || authentication == ANONYMOUS) {
return LocaleUtils.doWith(locale, callable);
}
return AuthenticationHolder
.executeWith(authentication, () -> LocaleUtils.doWith(locale, callable));
@Override
public Context currentContext() {
return actual.currentContext();
}
}
}
}

View File

@@ -6,6 +6,9 @@ import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.hswebframework.web.dict.EnumDict;
import org.hswebframework.web.authorization.Authentication;
import org.hswebframework.web.authorization.AuthenticationHolder;
import org.hswebframework.web.authorization.simple.SimpleAuthentication;
import org.hswebframework.web.i18n.LocaleUtils;
import org.hswebframework.web.i18n.MessageSourceInitializer;
import org.junit.Before;
@@ -15,15 +18,19 @@ import org.springframework.context.i18n.SimpleLocaleContext;
import org.springframework.context.support.ResourceBundleMessageSource;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.MediaType;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import reactor.test.publisher.TestPublisher;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Locale;
import java.util.function.Predicate;
import java.util.concurrent.atomic.AtomicBoolean;
public class CustomJackson2jsonEncoderTest {
@@ -44,6 +51,137 @@ public class CustomJackson2jsonEncoderTest {
}
@Test
public void testJsonFluxStreamsAsArray() {
CustomJackson2jsonEncoder encoder = new CustomJackson2jsonEncoder(new ObjectMapper());
TestPublisher<TestEntity> source = TestPublisher.create();
Flux<String> encoded = encoder
.encode(source.flux(),
new DefaultDataBufferFactory(),
ResolvableType.forType(TestEntity.class),
MediaType.APPLICATION_JSON,
Collections.emptyMap())
.map(this::readAndRelease);
StepVerifier
.create(encoded, 0)
.thenRequest(1)
.then(() -> source.next(new TestEntity(TestEnum.e1)))
.expectNextMatches(json -> json.startsWith("[{") && json.contains("\"e1\""))
.thenRequest(1)
.then(() -> source.next(new TestEntity(TestEnum.e2)))
.expectNextMatches(json -> json.startsWith(",{") && json.contains("\"e2\""))
.thenRequest(1)
.then(source::complete)
.expectNext("]")
.verifyComplete();
}
@Test
public void testEmptyJsonFlux() {
CustomJackson2jsonEncoder encoder = new CustomJackson2jsonEncoder(new ObjectMapper());
encoder
.encode(Flux.empty(),
new DefaultDataBufferFactory(),
ResolvableType.forType(TestEntity.class),
MediaType.APPLICATION_JSON,
Collections.emptyMap())
.as(DataBufferUtils::join)
.map(this::readAndRelease)
.as(StepVerifier::create)
.expectNext("[]")
.verifyComplete();
}
@Test
public void testAuthenticationAvailableDuringEncoding() {
CustomJackson2jsonEncoder encoder = new CustomJackson2jsonEncoder(new ObjectMapper());
Authentication authentication = new SimpleAuthentication();
AuthenticationHolder.executeWith(authentication, () -> {
encoder
.encode(Mono.just(new AuthenticationAwareEntity()),
new DefaultDataBufferFactory(),
ResolvableType.forType(AuthenticationAwareEntity.class),
MediaType.APPLICATION_JSON,
Collections.emptyMap())
.as(DataBufferUtils::join)
.map(this::readAndRelease)
.as(StepVerifier::create)
.expectNextMatches(json -> json.contains("\"authenticated\":true"))
.verifyComplete();
org.junit.Assert.assertSame(authentication, Authentication.current().orElse(null));
return null;
});
}
@Test
public void testReactorContextOverridesSameKeyThreadLocal() {
CustomJackson2jsonEncoder encoder = new CustomJackson2jsonEncoder(new ObjectMapper());
Authentication threadLocalAuthentication = new SimpleAuthentication();
Authentication reactorAuthentication = new SimpleAuthentication();
AuthenticationHolder.executeWith(threadLocalAuthentication, () -> {
encoder
.encode(Mono.just(new ExpectedAuthenticationEntity(reactorAuthentication)),
new DefaultDataBufferFactory(),
ResolvableType.forType(ExpectedAuthenticationEntity.class),
MediaType.APPLICATION_JSON,
Collections.emptyMap())
.as(DataBufferUtils::join)
.map(this::readAndRelease)
.contextWrite(context -> context.put(Authentication.class, reactorAuthentication))
.as(StepVerifier::create)
.expectNextMatches(json -> json.contains("\"reactorContextAuthentication\":true"))
.verifyComplete();
org.junit.Assert.assertSame(
threadLocalAuthentication,
Authentication.current().orElse(null));
return null;
});
}
@Test
public void testCancellationPropagatesToSource() {
CustomJackson2jsonEncoder encoder = new CustomJackson2jsonEncoder(new ObjectMapper());
AtomicBoolean cancelled = new AtomicBoolean();
Flux<DataBuffer> encoded = encoder.encode(
Flux.create(sink -> {
sink.onCancel(() -> cancelled.set(true));
sink.next(new TestEntity(TestEnum.e1));
}),
new DefaultDataBufferFactory(),
ResolvableType.forType(TestEntity.class),
MediaType.APPLICATION_JSON,
Collections.emptyMap());
StepVerifier
.create(encoded)
.assertNext(DataBufferUtils::release)
.thenCancel()
.verify();
org.junit.Assert.assertTrue(cancelled.get());
}
@Test
public void testFirstErrorIsNotConvertedToJson() {
CustomJackson2jsonEncoder encoder = new CustomJackson2jsonEncoder(new ObjectMapper());
IllegalStateException error = new IllegalStateException("source failed");
encoder
.encode(Flux.error(error),
new DefaultDataBufferFactory(),
ResolvableType.forType(TestEntity.class),
MediaType.APPLICATION_JSON,
Collections.emptyMap())
.as(StepVerifier::create)
.expectErrorMatches(actual -> actual == error)
.verify();
}
public void doTest(TestEntity entity, Locale locale, Predicate<String> verify){
CustomJackson2jsonEncoder encoder = new CustomJackson2jsonEncoder(new ObjectMapper());
@@ -55,13 +193,20 @@ public class CustomJackson2jsonEncoderTest {
Collections.emptyMap())
.as(DataBufferUtils::join)
.map(buf -> buf.toString(StandardCharsets.UTF_8))
.doOnNext(System.out::println)
.contextWrite(LocaleUtils.useLocale(locale))
.as(StepVerifier::create)
.expectNextMatches(verify)
.verifyComplete();
}
private String readAndRelease(DataBuffer buffer) {
try {
return buffer.toString(StandardCharsets.UTF_8);
} finally {
DataBufferUtils.release(buffer);
}
}
@Getter
@Setter
@AllArgsConstructor
@@ -71,6 +216,23 @@ public class CustomJackson2jsonEncoderTest {
private TestEnum testEnum;
}
public static class AuthenticationAwareEntity {
public boolean isAuthenticated() {
return Authentication.current().isPresent();
}
}
@AllArgsConstructor
public static class ExpectedAuthenticationEntity {
private final Authentication expected;
public boolean isReactorContextAuthentication() {
return Authentication.current().orElse(null) == expected;
}
}
@Getter
@AllArgsConstructor
@@ -86,4 +248,4 @@ public class CustomJackson2jsonEncoderTest {
}
}
}
}

View File

@@ -0,0 +1,564 @@
package org.hswebframework.web.starter.jackson;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.core.async.ByteArrayFeeder;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.micrometer.context.ContextRegistry;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaders;
import org.hswebframework.web.authorization.Authentication;
import org.hswebframework.web.authorization.simple.SimpleAuthentication;
import org.hswebframework.web.crud.web.ResponseMessageWrapper;
import org.hswebframework.web.i18n.LocaleUtils;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.ReactiveAdapterRegistry;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.accept.RequestedContentTypeResolver;
import org.springframework.web.reactive.config.EnableWebFlux;
import org.springframework.web.reactive.config.WebFluxConfigurer;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.netty.DisposableServer;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.server.HttpServer;
import reactor.test.StepVerifier;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Locale;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* End-to-end verification through Spring WebFlux and a real Reactor Netty connection.
*/
public class ResponseMessageStreamingIntegrationTest {
private static final Duration VERIFY_TIMEOUT = Duration.ofSeconds(30);
private static final MediaType VENDOR_JSON =
MediaType.parseMediaType("application/vnd.hsweb+json");
private static final String CORRELATION_CONTEXT_KEY =
ResponseMessageStreamingIntegrationTest.class.getName() + ".correlationId";
private static final ThreadLocal<String> CORRELATION_CONTEXT = new ThreadLocal<>();
private static final Authentication TEST_AUTHENTICATION = new SimpleAuthentication();
private static final Locale TEST_LOCALE = Locale.JAPANESE;
private static AnnotationConfigApplicationContext context;
private static DisposableServer server;
private static HttpClient client;
private static TestController controller;
private static ObjectMapper mapper;
@BeforeClass
public static void startServer() {
ContextRegistry
.getInstance()
.registerThreadLocalAccessor(CORRELATION_CONTEXT_KEY, CORRELATION_CONTEXT);
context = new AnnotationConfigApplicationContext(TestConfiguration.class);
controller = context.getBean(TestController.class);
mapper = context.getBean(ObjectMapper.class);
HttpHandler handler = WebHttpHandlerBuilder
.applicationContext(context)
.build();
server = HttpServer
.create()
.host("127.0.0.1")
.port(0)
.handle(new ReactorHttpHandlerAdapter(handler))
.bindNow(VERIFY_TIMEOUT);
client = HttpClient
.create()
.baseUrl("http://127.0.0.1:" + server.port())
.responseTimeout(VERIFY_TIMEOUT);
}
@AfterClass
public static void stopServer() {
if (server != null) {
server.disposeNow(VERIFY_TIMEOUT);
}
if (context != null) {
context.close();
}
ContextRegistry.getInstance().removeThreadLocalAccessor(CORRELATION_CONTEXT_KEY);
CORRELATION_CONTEXT.remove();
}
@Test
public void testFirstHttpChunkArrivesBeforePublisherCompletes() throws Exception {
ControlledStream stream = controller.openControlledStream();
Mono<String> firstItemResponse = streamResponse(
"/stream/controlled",
MediaType.APPLICATION_JSON,
false)
.scanWith(StringBuilder::new, (builder, chunk) -> builder.append(chunk))
.filter(builder -> builder.indexOf("\"id\":1") >= 0)
.map(StringBuilder::toString)
.next();
StepVerifier
.create(firstItemResponse)
.then(() -> assertEquals(
Sinks.EmitResult.OK,
stream.emit(new TestEntity(1))))
.assertNext(json -> {
assertTrue(json.startsWith("{\"message\":\"success\""));
assertTrue(json.contains("\"result\":[{\"id\":1}"));
assertFalse(json.contains("\"status\":200"));
})
.expectComplete()
.verify(VERIFY_TIMEOUT);
assertFalse(stream.isCompleted());
assertTrue("client cancellation was not propagated to the source",
stream.awaitCancellation(VERIFY_TIMEOUT));
}
@Test
public void testLargeResponseIsParsedIncrementally() {
int expected = 100_000;
StreamingJsonProbe probe = new StreamingJsonProbe(mapper);
Mono<Void> response = client
.headers(headers -> headers.set(
HttpHeaderNames.ACCEPT,
MediaType.APPLICATION_JSON_VALUE))
.get()
.uri("/stream/large?count=" + expected)
.response((httpResponse, content) -> {
assertEquals(200, httpResponse.status().code());
return content
.asByteArray()
.doOnNext(probe::feed)
.then(Mono.fromRunnable(probe::complete));
})
.then();
StepVerifier
.create(response)
.expectComplete()
.verify(VERIFY_TIMEOUT);
assertEquals(expected, probe.getItemCount());
assertEquals(200, probe.getStatus());
assertTrue(probe.hasResultArray());
assertTrue(probe.hasCompleteRootObject());
assertTrue(probe.getByteCount() > expected * 8L);
}
@Test
public void testVendorJsonAndIgnoreHeaderCompatibility() {
StepVerifier
.create(receive("/stream/finite", VENDOR_JSON, false))
.assertNext(response -> {
assertEquals(200, response.status());
assertTrue(response.contentType().startsWith(VENDOR_JSON.toString()));
JsonNode json = readTree(response.body());
assertEquals("success", json.get("message").asText());
assertEquals(3, json.get("result").size());
})
.expectComplete()
.verify(VERIFY_TIMEOUT);
StepVerifier
.create(receive("/stream/finite", MediaType.APPLICATION_JSON, true))
.assertNext(response -> {
JsonNode json = readTree(response.body());
assertTrue(json.isArray());
assertEquals(3, json.size());
})
.expectComplete()
.verify(VERIFY_TIMEOUT);
}
@Test
public void testNdjsonAndSseRemainUnwrapped() {
StepVerifier
.create(receive("/stream/finite", MediaType.APPLICATION_NDJSON, false))
.assertNext(response -> {
assertFalse(response.body().contains("\"message\""));
long lines = response
.body()
.lines()
.filter(line -> !line.isBlank())
.count();
assertEquals(3, lines);
})
.expectComplete()
.verify(VERIFY_TIMEOUT);
StepVerifier
.create(receive("/stream/finite", MediaType.TEXT_EVENT_STREAM, false))
.assertNext(response -> {
assertFalse(response.body().contains("\"message\""));
assertTrue(response.body().contains("data:{\"id\":1}"));
assertTrue(response.body().contains("data:{\"id\":3}"));
})
.expectComplete()
.verify(VERIFY_TIMEOUT);
}
@Test
public void testRegisteredContextsAreRestoredAcrossAsyncBoundary() {
StepVerifier
.create(receive("/stream/async-context", MediaType.APPLICATION_JSON, false))
.assertNext(response -> {
JsonNode result = readTree(response.body()).get("result").get(0);
assertEquals("request-001", result.get("correlationId").asText());
assertEquals(TEST_LOCALE.toLanguageTag(), result.get("locale").asText());
assertTrue(result.get("authenticated").asBoolean());
assertTrue(result.get("thread").asText().contains("response-context-test"));
})
.expectComplete()
.verify(VERIFY_TIMEOUT);
StepVerifier
.create(controller.readAsyncThreadContext())
.assertNext(state -> {
assertNull(state.correlationId());
assertFalse(state.authenticated());
})
.expectComplete()
.verify(VERIFY_TIMEOUT);
}
private static Flux<String> streamResponse(String uri,
MediaType accept,
boolean ignoreWrapper) {
return client
.headers(headers -> configureHeaders(headers, accept, ignoreWrapper))
.get()
.uri(uri)
.response((response, content) -> {
assertEquals(200, response.status().code());
return content.asString(StandardCharsets.UTF_8);
});
}
private static Mono<ReceivedResponse> receive(String uri,
MediaType accept,
boolean ignoreWrapper) {
return client
.headers(headers -> configureHeaders(headers, accept, ignoreWrapper))
.get()
.uri(uri)
.responseSingle((response, content) -> content
.asString(StandardCharsets.UTF_8)
.defaultIfEmpty("")
.map(body -> new ReceivedResponse(
response.status().code(),
Objects.requireNonNullElse(
response.responseHeaders().get(HttpHeaderNames.CONTENT_TYPE),
""),
body)));
}
private static void configureHeaders(HttpHeaders headers,
MediaType accept,
boolean ignoreWrapper) {
headers.set(HttpHeaderNames.ACCEPT, accept.toString());
if (ignoreWrapper) {
headers.set("X-Response-Wrapper", "Ignore");
}
}
private static JsonNode readTree(String json) {
try {
return mapper.readTree(json);
} catch (IOException error) {
throw new AssertionError("Invalid JSON: " + json, error);
}
}
@Configuration
@EnableWebFlux
static class TestConfiguration implements WebFluxConfigurer {
private final ObjectMapper objectMapper = new ObjectMapper();
@Bean
ObjectMapper objectMapper() {
return objectMapper;
}
@Bean(destroyMethod = "dispose")
Scheduler responseContextScheduler() {
return Schedulers.newSingle("response-context-test");
}
@Bean
TestController testController(Scheduler responseContextScheduler) {
return new TestController(responseContextScheduler);
}
@Bean
WebFilter asyncContextWebFilter() {
return (exchange, chain) -> {
Mono<Void> result = chain.filter(exchange);
if (!exchange.getRequest().getPath().value().equals("/stream/async-context")) {
return result;
}
return result.contextWrite(context -> context
.put(CORRELATION_CONTEXT_KEY, "request-001")
.put(Authentication.class, TEST_AUTHENTICATION)
.put(Locale.class, TEST_LOCALE));
};
}
@Bean
ResponseMessageWrapper responseMessageWrapper(ServerCodecConfigurer codecConfigurer,
RequestedContentTypeResolver resolver,
ReactiveAdapterRegistry registry) {
return new ResponseMessageWrapper(codecConfigurer.getWriters(), resolver, registry);
}
@Override
public void configureHttpMessageCodecs(ServerCodecConfigurer configurer) {
configurer
.defaultCodecs()
.jackson2JsonEncoder(new CustomJackson2jsonEncoder(objectMapper));
}
}
@RestController
static class TestController {
private final AtomicReference<ControlledStream> controlled = new AtomicReference<>();
private final Scheduler responseContextScheduler;
TestController(Scheduler responseContextScheduler) {
this.responseContextScheduler = responseContextScheduler;
}
ControlledStream openControlledStream() {
ControlledStream stream = new ControlledStream();
controlled.set(stream);
return stream;
}
@GetMapping("/stream/controlled")
Flux<TestEntity> controlled() {
return Objects.requireNonNull(controlled.get(), "controlled stream not initialized")
.flux();
}
@GetMapping("/stream/finite")
Flux<TestEntity> finite() {
return Flux.just(new TestEntity(1), new TestEntity(2), new TestEntity(3));
}
@GetMapping("/stream/large")
Flux<TestEntity> large(@RequestParam int count) {
return Flux.range(0, count).map(TestEntity::new);
}
@GetMapping("/stream/async-context")
Flux<AsyncContextEntity> asyncContext() {
return Flux
.just(new AsyncContextEntity())
.publishOn(responseContextScheduler);
}
Mono<ThreadContextState> readAsyncThreadContext() {
return Mono
.fromCallable(() -> new ThreadContextState(
CORRELATION_CONTEXT.get(),
Authentication.current().isPresent()))
.subscribeOn(responseContextScheduler);
}
}
record TestEntity(int id) {
}
static class AsyncContextEntity {
public String getCorrelationId() {
return CORRELATION_CONTEXT.get();
}
public String getLocale() {
return LocaleUtils.current().toLanguageTag();
}
public boolean isAuthenticated() {
return Authentication.current().orElse(null) == TEST_AUTHENTICATION;
}
public String getThread() {
return Thread.currentThread().getName();
}
}
private record ReceivedResponse(int status, String contentType, String body) {
}
private record ThreadContextState(String correlationId, boolean authenticated) {
}
private static final class ControlledStream {
private final Sinks.Many<TestEntity> sink = Sinks
.many()
.unicast()
.onBackpressureBuffer();
private final CountDownLatch cancelled = new CountDownLatch(1);
private final AtomicBoolean completed = new AtomicBoolean();
Flux<TestEntity> flux() {
return sink
.asFlux()
.doOnCancel(cancelled::countDown)
.doOnComplete(() -> completed.set(true));
}
Sinks.EmitResult emit(TestEntity entity) {
return sink.tryEmitNext(entity);
}
boolean isCompleted() {
return completed.get();
}
boolean awaitCancellation(Duration timeout) throws InterruptedException {
return cancelled.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
}
}
private static final class StreamingJsonProbe {
private final JsonParser parser;
private final ByteArrayFeeder feeder;
private int itemCount;
private int status;
private int objectDepth;
private long byteCount;
private String currentField;
private boolean resultArray;
private boolean completeRootObject;
private StreamingJsonProbe(ObjectMapper mapper) {
try {
this.parser = mapper.getFactory().createNonBlockingByteArrayParser();
this.feeder = (ByteArrayFeeder) parser.getNonBlockingInputFeeder();
} catch (IOException error) {
throw new UncheckedIOException(error);
}
}
void feed(byte[] bytes) {
try {
assertTrue("parser still has unread input", feeder.needMoreInput());
byteCount += bytes.length;
feeder.feedInput(bytes, 0, bytes.length);
drain();
} catch (IOException error) {
throw new UncheckedIOException(error);
}
}
void complete() {
try {
feeder.endOfInput();
drain();
parser.close();
} catch (IOException error) {
throw new UncheckedIOException(error);
}
}
private void drain() throws IOException {
JsonToken token;
while ((token = parser.nextToken()) != null && token != JsonToken.NOT_AVAILABLE) {
if (token == JsonToken.FIELD_NAME) {
currentField = parser.currentName();
} else if (token == JsonToken.START_OBJECT) {
objectDepth++;
} else if (token == JsonToken.END_OBJECT) {
objectDepth--;
if (objectDepth == 0) {
completeRootObject = true;
}
} else if (token == JsonToken.START_ARRAY && "result".equals(currentField)) {
resultArray = true;
} else if (token == JsonToken.VALUE_NUMBER_INT) {
if ("id".equals(currentField)) {
itemCount++;
} else if ("status".equals(currentField)) {
status = parser.getIntValue();
}
}
}
}
int getItemCount() {
return itemCount;
}
int getStatus() {
return status;
}
long getByteCount() {
return byteCount;
}
boolean hasResultArray() {
return resultArray;
}
boolean hasCompleteRootObject() {
return completeRootObject;
}
}
}