private void handleRequest ( HttpServletRequest request , final HttpServletResponse response ) throws ServletException , IOException { final AtomicBoolean moreDataWillBeSent = new AtomicBoolean ( true ) ; Subscription sampleSubscription = null ; /* ensure we aren't allowing more connections than we want */ int numberConnections = incrementAndGetCurrentConcurrentConnections ( ) ; try { int maxNumberConnectionsAllowed = getMaxNumberConcurrentConnectionsAllowed ( ) ; //may change at runtime, so look this up for each request if ( numberConnections > maxNumberConnectionsAllowed ) { response . sendError ( 503 , "MaxConcurrentConnections reached: " + maxNumberConnectionsAllowed ) ; } else { /* initialize response */ response . setHeader ( "Content-Type" , "text/event-stream;charset=UTF-8" ) ; response . setHeader ( "Cache-Control" , "no-cache, no-store, max-age=0, must-revalidate" ) ; response . setHeader ( "Pragma" , "no-cache" ) ; final PrintWriter writer = response . getWriter ( ) ; //since the sample stream is based on Observable.interval, events will get published on an RxComputation thread //since writing to the servlet response is blocking, use the Rx IO thread for the write that occurs in the onNext sampleSubscription = sampleStream . observeOn ( Schedulers . io ( ) ) . subscribe ( new Subscriber < String > ( ) { @ Override public void onCompleted ( ) { logger . error ( "HystrixSampleSseServlet: ({}) received unexpected OnCompleted from sample stream" , getClass ( ) . getSimpleName ( ) ) ; moreDataWillBeSent . set ( false ) ; } @ Override public void onError ( Throwable e ) { moreDataWillBeSent . set ( false ) ; } @ Override public void onNext ( String sampleDataAsString ) { if ( sampleDataAsString != null ) { try { // avoid concurrent writes with ping synchronized ( responseWriteLock ) { writer . print ( "data: " + sampleDataAsString + "\n\n" ) ; // explicitly check for client disconnect - PrintWriter does not throw exceptions if ( writer . checkError ( ) ) { moreDataWillBeSent . set ( false ) ; } writer . flush ( ) ; } } catch ( Exception ex ) { moreDataWillBeSent . set ( false ) ; } } } } ) ; while ( moreDataWillBeSent . get ( ) && ! isDestroyed ) { try { Thread . sleep ( pausePollerThreadDelayInMs ) ; //in case stream has not started emitting yet, catch any clients which connect/disconnect before emits start // avoid concurrent writes with sample synchronized ( responseWriteLock ) { writer . print ( "ping: \n\n" ) ; // explicitly check for client disconnect - PrintWriter does not throw exceptions if ( writer . checkError ( ) ) { moreDataWillBeSent . set ( false ) ; } writer . flush ( ) ; } } catch ( Exception ex ) { moreDataWillBeSent . set ( false ) ; } } } } finally { decrementCurrentConcurrentConnections ( ) ; if ( sampleSubscription != null && ! sampleSubscription . isUnsubscribed ( ) ) { sampleSubscription . unsubscribe ( ) ; } } }
- maintain an open connection with the client - on initial connection send latest data of each requested event type - subsequently send all changes for each requested event type