Quarkus Mutiny 上下文传播和服务器发送事件

Question

我正在尝试建立一个 SSE 通道，根据订阅客户的身份来过滤事件。然而，订阅时收到的 SecurityIdentity 不会在调度时传播，因此 channel.send 调用会失败并出现 UnauthorizedException。

这就是我所拥有的，正如预期的那样，当服务器调用channel.send(event)时，没有要注入的SecurityIdentity并且

assert ! _identity.isAnonymous();

（事实上，@Authenticated注释被检测到并导致调度失败）。

如何“保存”订阅 REST 调用的安全身份以便稍后在调度时使用它？

谢谢！

休息资源：

@GET
@Path("stream")
@Produces(MediaType.SERVER_SENT_EVENTS)
@RestStreamElementType(MediaType.APPLICATION_JSON)
public Multi<PassportEvent> stream(@QueryParam("order") Integer orderId) {
    return _passportService.stream(orderId);
}

服务：

@Inject
@Channel("passports")
Multi<PassportEvent> passportEvents;
    
@Inject
SecurityIdentity _identity;

@Authenticated
public Multi<PassportEvent> stream(Integer orderId)
{
    assert ! _identity.isAnonymous();

    return passportEvents
        .filter(event -> Objects.equals(event.tenantId, getTenantId())
            && (orderId == null || Objects.equals(orderId, event.orderId)));
}

发射服务：

@Inject
@Channel("passports")
Emitter<PassportEvent> passportEvents;

// no security identity here
createPassport(...) {
    ...
    if (passportEvents.hasRequests()) {
       PassportEvent event = ...;
            passportEvents.send(event);
        }
    }
}

Answer 1

我通过使用一个身份发射器解决了这个问题（这是我的用例所需的）：

public Multi<PaymentEvent> stream(int personId)
    {
        return Multi.createFrom().emitter(e -> {
            _emitters.put(personId, new TenantEmitter<>(e, 
getTenantId()));
            e.onTermination(() ->
                _emitters.remove(personId));
        });
    }

    void dispatchEvent(PaymentEvent event)
    {
        TenantEmitter<PaymentEvent> emitter = _emitters.get(event.getPersondId());
        if (emitter != null && Objects.equals(emitter.getTenantId(), getTenantId())) {
            emitter.getEmitter().emit(event);
        }
    }