| | |
|---|
| | | this.credentials = credentials; |
|---|
| | | } |
|---|
| | | |
|---|
| | | /** |
|---|
| | | * 接收推式来源程序的一批载荷：校验来源与令牌，按载荷类别写入落地表，并刷新该来源的心跳。 |
|---|
| | | * |
|---|
| | | * <p>前置条件：来源已登记为启用中的推式实例，端点令牌与本地凭据一致，载荷为合法 JSON 且类别已知。 |
|---|
| | | * 整批在同一事务内落库，任一行失败则整批回滚；重复推送按来源行身份幂等忽略，内容不一致的行记为冲突 |
|---|
| | | * 而不覆盖既有事实。心跳、拓扑与采样都会刷新心跳时间。</p> |
|---|
| | | * |
|---|
| | | * @param adapterCode 来源实例编码，取自接收端点路径 |
|---|
| | | * @param token 端点令牌，取自请求头 |
|---|
| | | * @param body 载荷正文，含 kind 与对应的行数据 |
|---|
| | | * @return 本批的接收计数：成功写入、幂等忽略、内容冲突的行数 |
|---|
| | | * @throws ServiceException 来源不是启用的推式实例、令牌不匹配、载荷格式非法或行数超出单批上限 |
|---|
| | | */ |
|---|
| | | @Override |
|---|
| | | @Transactional(rollbackFor = Exception.class) |
|---|
| | | public IngestResult ingest(String adapterCode, String token, String body) |
|---|
| | | { |
|---|
| | | // STEP 1: 校验来源是启用中的推式实例，并校验端点令牌。 |
|---|
| | | TDeviceAdapter adapter = requirePushSource(adapterCode); |
|---|
| | | requireToken(adapter, token); |
|---|
| | | // STEP 2: 解析载荷并按 kind 分流；未知类别直接拒绝，不按猜测处理。 |
|---|
| | | JSONObject payload = parse(body); |
|---|
| | | String kind = payload.getString("kind"); |
|---|
| | | Instant receivedAt = clock.instant(); |
|---|
| | |
|---|
| | | { |
|---|
| | | throw payloadInvalid("未知的载荷类别 kind：" + kind); |
|---|
| | | } |
|---|
| | | // STEP 3: 刷新心跳时间；每一批成功接收都是"程序还在推"的证据。 |
|---|
| | | // 每一批成功接收都刷新心跳：收到采样行同样是"程序还在推"的证据，缺一次心跳不会误判离线。 |
|---|
| | | touchHeartbeat(adapterCode, receivedAt); |
|---|
| | | return new IngestResult(counters.accepted, counters.ignored, counters.conflicts); |
|---|
