Java Spring / Reactor send / update flux when underlying database changed

user_6318261 0
Viewed 381

I have a data for candidate "likes", which I'd like to send to client every time the "like" number is changed. I think this is achievable using Spring Flux? But I can't find any example for this. Most flux example is based on specific interval (e.g. every second). This might be a waste, because the transaction is not that much, and a candidate might not get likes in many minutes. I just want to create dashboard that subscribe to "likes" change, and get updated when certain candidate "likes" number changed.

What is the way to get this?

This is what I did, and it works, but it based on interval (5 seconds), not based on data change.

    public Flux<Candidate> subscribeItemChange(String id) {
        return Flux.interval(Duration.ofSeconds(5)).map(t -> candidateService.getCandidateDetail(id));
    }

The candidateService.getCandidateDetail is basically query database for certain id, so this is more like polling instead of "update on change". I think I must put something on candidateService.updateLikes() below, but what should I update?

public class CandidateService {

    public Candidate getCandidateDetail(String id) {
        // query candidate from database
        // select * from candidates where id = :id
        // and return it
    }
    
    
    public void updateLikes(String id, int likesCount) {
        // update candidates set likes_count = :likesCount where id = :id
        // ...
        // I think I need to write something here, but what?
    }
    
}
project-reactor spring java
2 Answers

You could make use of a dynamic sink, adding a field similar to:

private Sinks.Many<Candidate> likesSink = Sinks.many().multicast().onBackpressureBuffer();

...then you can:

  • Use sink.tryEmitNext in your updateLikes() method to publish to the sink whenever likes are updated for a candidate;
  • Implement your subscribeItemChange() method which uses likesSink.asFlux(), which can then be filtered if necessary to only return the stream of "like updates" for a particular candidate.
user_5514060

Based on @Michael Berry guide.

    public void updateLikes(String id, int likesCount) {
        Candidate c = getCandidateDetail(id);
        c.setLikesCount(likesCount);

        CandidateDummyDatasource.likesSink.tryEmitNext(c);
    }

On subscriber

    public Flux<Candidate> subscribeItemChange(String id) {
        return CandidateDummyDatasource.likesSink.asFlux()
                .filter(c -> c.getId().equals(id))
                .map(data -> candidateService.getCandidateDetail(id));
    }
user_63182610
Related