0

我处理反应式编程的主题。我使用 WebFlux 和 R2dbc 来访问数据库。我正在为聊天应用程序开发一个小型后端。所以我有一些关于这个主题的问题，我会很高兴提出建议和改进。

这是服务等级

@Service
public class MessageService implements IMessageService {

    UnicastProcessor<Message> hotProcessor = UnicastProcessor.create();
    private final FluxSink<Message> fluxSink = hotProcessor.sink(FluxSink.OverflowStrategy.LATEST);
    private final Flux<Message> hotFlux = hotProcessor.publish().autoConnect();



    @Autowired
    private MessageRepository messageRepository;

    public Flux<Message> findAll() {
        return this.messageRepository.findAll();
    }


    public Mono<Message> create(Message nachricht) {
        this.fluxSink.next(nachricht);
        return this.messageRepository.save(nachricht);
    }

    // how can I persist messages and to arrange a chat room.
    public Flux<Message> finAllMessagesByChatroomId(Long id) {
        return hotFlux.filter(m->m.getId().intValue() == id.intValue());        
    }
}

DirectProcessor 什么时候应该使用 UnicastProcessor。

更新：
一点修正

public Flux<Message> finAllMessagesByChatroomId(Long id) {
        return hotFlux.filter(m->m.getRoom_id == id);        
}

但是如何从数据库中获取所有消息并将其添加到fluxSink？

4

0 回答 0