0

I have an application with multiple nodes in a cluster. Each node writes log files to its local disk. I have implemented a log search function that searches logs across each node. The owner node that receives the search request from the browser submits the log search jobs to the other nodes, then the other nodes pass search results to the original node. The client web browser uses long polling to get the search results from the node. This seems like a good fit for RxJava because each node is an event stream, and the client gets a consolidated event stream from all nodes. (Let's assume that a stingy operations team won't let us use Splunk or some other commercial logging solution).

The client polls the REST API on the original node, which collects search results. My ideal logic for the REST API is as follows:

  • The server should make the client wait at most 15 seconds for a response.
  • The response can be empty if no results were generated within 15 seconds. The client will see this and send a new polling request.
  • If there will be no more results (i.e. the search is finished), send a special response to the client to instruct it not to poll any more.
  • If a result was generated, wait at most a further 100ms for additional results to save the network overhead of an additional poll.
  • The client should not get any duplicate results.

I've written the following sample code to simulate this scenario:

public static void main(String[] args) throws Exception {
    ExecutorService executorService = Executors.newFixedThreadPool(5);

    /* Each searchTask represents the results of a search job running on a
     * node in the cluster */
    Subject<String,String> searchTask1 = PublishSubject.create();
    Subject<String,String> searchTask2 = PublishSubject.create();

    // Limit max number of search results
    Observable<String> searchResults = 
        Observable.merge(searchTask1, searchTask2).take( 1000 );

    /* Add a 100ms buffer window to collect nearby responses together.  
     * Filter out any empty buffers to eliminate unnecessary
     * responses to the browser. */
    BlockingObservable<List<String>> blocking = 
        searchResults.buffer(100, TimeUnit.MILLISECONDS)
            .filter(results -> !results.isEmpty()).toBlocking();
    Iterator<List<String>> it = blocking.getIterator();

    /* Each call to searchTask.onNext represents a search result pushed
     * to the owner node from another node.  This code would be called 
     * from the REST endpoint. */
    executorService.submit( () -> {
        searchTask1.onNext("1");
        try { Thread.sleep(1200); } catch ( Exception ignored ) { }
        searchTask1.onNext("2");
        searchTask1.onCompleted();
    });
    executorService.submit( () -> {
        searchTask2.onNext("a");
        try { Thread.sleep(500); } catch ( Exception ignored ) { }
        searchTask2.onNext("b");
        searchTask2.onCompleted();
    });

    executorService.submit( () -> {
        /* Each iteration of this loop represents a polling request from
         * the browser and the results that are sent back to it. */
        for ( int i = 0; i < 5; i++ ) { 
            it.forEachRemaining(results -> System.out.println(results));
        }
    });

    Thread.sleep(1500);
    System.out.println("exit");
}

What should the logic in the for-loop be to ensure that a response will always be sent back to the client after at most 15 seconds (even if the response is empty)?

EDIT: I've updated the sample code with more comments and to show my current solution, but I still can't get the max response time of 15 seconds that I'm looking for. We have network devices that will close a HTTP connection that is idle for too long, so I want to guarantee that the client will always get a response after at most 15 seconds.

3
  • Could you elaborate more, such as how to poll the further results when receiving a part of results, and what's 300ms timeout?
    – zsxwing
    Sep 18, 2015 at 15:34
  • @zsxwing I've updated the example code with more details.
    – Nathan
    Sep 20, 2015 at 11:15
  • Not an answer to your questions, but have you looked at Kibana? Sep 20, 2015 at 12:18

1 Answer 1

0

All the RxJava articles I could find seem very focused on client-side code rather than server-side. However, after wrangling with Observable operators for a bit, I came up with the following solution is close to what I wanted. The heartbeat happens every 15 seconds rather than 15 seconds after the previous result. That means that the server might send a heartbeat response to the client just after it sent a result, but that's close enough for me.

I created a 15s interval observable and merged it with the searchResults observable that I already had. I used a subject so that I can stop the interval observable when the stream of results stops (otherwise it would keep going indefinitely).

    /* Add a heartbeat that ensures we don't wait too long between
     * sending responses and some network device kills our connection */
    final Observable<String> heartbeat =
        Observable.interval( 1, TimeUnit.SECONDS ).map(el -> "heartbeat");
    final PublishSubject<String> stopHeartbeat = PublishSubject.create();
    searchResults.subscribe( el -> {}, ex -> {}, () -> stopHeartbeat.onNext( null ) );
    final Observable<String> searchResultsWithHeartbeat =
        searchResults.mergeWith( heartbeat.takeUntil( stopHeartbeat ) );

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.