Designing an asynchronous polling service in Java

Question

I am currently writing a Java bridge to a eyetracking library written in C. The whole thing works very well, but building an actual application with it is hard, since eyetracking is done by polling for new data.

Polling becomes a problem as soon as you have a graphical user interface, as a poll-loop blocks the main event loop, freezing the gui. This also makes it impossible to stop the polling loop through user input.

My goal is to build a polling service that is generic enough to be used for a lot of different applications. It has to be asynchronous and has to have some kind of callback mechanism to use any relevant data.

This is my idea (pseudocode):

PollService {
  // The filterFunction decides if a eyetracking event is relevant or can be ignored
  setFilterFunction(filterFunction);

  // The callbackFunction is executed if a relevant eyetracking event is polled
  setCallbackFunction(callbackFunction);

  start();
  stop();
}

The PollService itself only manages the asynchronous polling process. It is supplied a couple of functions to perform the two basic steps:

1. Filter if a eyetracking event is relevant, skip event if not
2. Perform some kind of operation on any event that is relevant

Possible callback operations might be:

• Perform a long running computation with the eyetracking event data
• Print something to the screen
• Store something to a database
• Communicate the event to another thread (using a messaging queue for example)
• Update a GUI (using Platform.runLater() for example)

The filterFunction and callbackFunction would be implemented as functional interfaces, so Java 8 users can easily supply these functions using lambdas or method references.

What do you think about this design? Are there any obvious flaws i am not seeing?

Answer 1

This sounds like the kind of work reactive streams are made for.

I have only used RxJava so I will use it for my comments here, but there are other implementations.

Your polling service sounds like a publisher, or a subject that emits events when there is something available in the event channel. You want to broadcast those events to all interested subscribers.

It is not clear in your requirements if you want to broadcast the event regardless of whether you have subscribers or not (the distinction between hot and cold observables)

There are a few aspects of the design that would be relatively easy to implement with this pattern:

For example, interested objects only need to subscribe to the event stream:

eventService.getEventStream()
  .subscribe(event -> {
      //handle your event here
   });

You could easily filter which events your subscriber is actually interested in:

eventService.getEventStream()
  .filter(event -> isAnEventICareAbout(event))
  .subscribe(event -> {
      //handle only events you care about here
   });

You can easily control threads and make sure that event computations happen in a different thread than that of the UI event loop.

eventService.getEventStream()
  .filter(event -> isAnEventICareAbout(event))
  .observeOn(Schedulers.io())
  .subscribe(event -> {
      //handle your event in an io thread here
   });

You could easily wrap your UI event loop into another event stream and make sure that all events it process only happen within the UI event loop.

PublishingSubject<Event> subject = Platform.getUIEventSubject().observeOn(eventLoop());
subject.onNext(new UIEvent()); //event handling occurs in eventloop 

You can easily control subscriptions of the interested event listeners. Once a listener is no longer interested in receiving events, it can unsubscribe and when there are no subscribers, your polling service can automatically stop producing and resume when new subscribers arrive.

    Subscription subs = eventService.getEventStream()
      .subscribe(event -> {
          //handle your event here
       });

   //later, no longer interested
   subs.unsubscribe();
   subs = null

Your polling service could determine when there are no more subscriptions and stop polling and resume when a new subscription arrives. You can also control/avoid creating new pollers per subscriber and use a global poller for all subscribers (e.g. publish.refCount()), etc. etc.

In general, the problem you're trying to solve just sound like a good fit for reactive programming and it has this functional programming style that you seek.

You may want to consider to give a look at the book Reactive Programming with RxJava which I think is great and I think it would help a great deal if you decide to follow this approach.