Added parent class to encapsulate shared SSE logic

Author: Matt Jacobs

hystrix-contrib/hystrix-metrics-event-stream/src/main/java/com/netflix/hystrix/contrib/sample/stream/HystrixConfigSseServlet.java

@@ -27,22 +27,12 @@
 import com.netflix.hystrix.config.HystrixConfiguration;
 import com.netflix.hystrix.config.HystrixConfigurationStream;
 import com.netflix.hystrix.config.HystrixThreadPoolConfiguration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import rx.Subscriber;
-import rx.Subscription;
+import rx.Observable;
 import rx.functions.Func1;
-import rx.schedulers.Schedulers;
 
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServlet;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
-import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -66,88 +56,54 @@
  * </servlet-mapping>
  * } </pre>
  */
-public class HystrixConfigSseServlet extends HttpServlet {
+public class HystrixConfigSseServlet extends HystrixSampleSseServlet<HystrixConfiguration> {
 
     private static final long serialVersionUID = -3599771169762858235L;
 
-    private static final Logger logger = LoggerFactory.getLogger(HystrixConfigSseServlet.class);
-
-    private static final String DELAY_REQ_PARAM_NAME = "delay";
     private static final int DEFAULT_ONNEXT_DELAY_IN_MS = 10000;
 
-    private final Func1<Integer, HystrixConfigurationStream> createStream;
     private JsonFactory jsonFactory = new JsonFactory();
 
     /* used to track number of connections and throttle */
     private static AtomicInteger concurrentConnections = new AtomicInteger(0);
     private static DynamicIntProperty maxConcurrentConnections = DynamicPropertyFactory.getInstance().getIntProperty("hystrix.config.stream.maxConcurrentConnections", 5);
 
-    private static volatile boolean isDestroyed = false;
-
     public HystrixConfigSseServlet() {
-        this.createStream = new Func1<Integer, HystrixConfigurationStream>() {
+        super(new Func1<Integer, Observable<HystrixConfiguration>>() {
             @Override
-            public HystrixConfigurationStream call(Integer delay) {
-                return new HystrixConfigurationStream(delay);
+            public Observable<HystrixConfiguration> call(Integer delay) {
+                return new HystrixConfigurationStream(delay).observe();
             }
-        };
-    }
-
-    /* package-private */ HystrixConfigSseServlet(Func1<Integer, HystrixConfigurationStream> createStream) {
-        this.createStream = createStream;
+        });
     }
 
-    /**
-     * WebSphere won't shutdown a servlet until after a 60 second timeout if there is an instance of the servlet executing
-     * a request.  Add this method to enable a hook to notify Hystrix to shutdown.  You must invoke this method at
-     * shutdown, perhaps from some other servlet's destroy() method.
-     */
-    public static void shutdown() {
-        isDestroyed = true;
+    /* package-private */ HystrixConfigSseServlet(Func1<Integer, Observable<HystrixConfiguration>> createStream) {
+        super(createStream);
     }
 
     @Override
-    public void init() throws ServletException {
-        isDestroyed = false;
+    int getDefaultDelayInMilliseconds() {
+        return DEFAULT_ONNEXT_DELAY_IN_MS;
     }
 
-    /**
-     * Handle incoming GETs
-     */
     @Override
-    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
-        if (isDestroyed) {
-            response.sendError(503, "Service has been shut down.");
-        } else {
-            handleRequest(request, response);
-        }
+    int getMaxNumberConcurrentConnectionsAllowed() {
+        return maxConcurrentConnections.get();
     }
 
-    /**
-     * Handle servlet being undeployed by gracefully releasing connections so poller threads stop.
-     */
     @Override
-    public void destroy() {
-        /* set marker so the loops can break out */
-        isDestroyed = true;
-        super.destroy();
+    int getNumberCurrentConnections() {
+        return concurrentConnections.get();
     }
 
-    /* package-private */ int getNumberCurrentConnections() {
-        return concurrentConnections.get();
+    @Override
+    protected int incrementAndGetCurrentConcurrentConnections() {
+        return concurrentConnections.incrementAndGet();
     }
 
-    /* package-private */
-    static int getDelayFromHttpRequest(HttpServletRequest req) {
-        try {
-            String delay = req.getParameter(DELAY_REQ_PARAM_NAME);
-            if (delay != null) {
-                return Math.max(Integer.parseInt(delay), 1);
-            }
-        } catch (Throwable ex) {
-            //silently fail
-        }
-        return DEFAULT_ONNEXT_DELAY_IN_MS;
+    @Override
+    protected void decrementCurrentConcurrentConnections() {
+        concurrentConnections.decrementAndGet();
     }
 
     private void writeCommandConfigJson(JsonGenerator json, HystrixCommandKey key, HystrixCommandConfiguration commandConfig) throws IOException {
@@ -215,7 +171,8 @@ private void writeCollapserConfigJson(JsonGenerator json, HystrixCollapserKey co
         json.writeEndObject();
     }
 
-    private String convertToString(HystrixConfiguration config) throws IOException {
+    @Override
+    protected String convertToString(HystrixConfiguration config) throws IOException {
         StringWriter jsonString = new StringWriter();
         JsonGenerator json = jsonFactory.createGenerator(jsonString);
 
@@ -250,95 +207,5 @@ private String convertToString(HystrixConfiguration config) throws IOException {
 
         return jsonString.getBuffer().toString();
     }
-
-    /**
-     * - 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
-     *
-     * @param request  incoming HTTP Request
-     * @param response outgoing HTTP Response (as a streaming response)
-     * @throws javax.servlet.ServletException
-     * @throws java.io.IOException
-     */
-    private void handleRequest(HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException {
-        final AtomicBoolean moreDataWillBeSent = new AtomicBoolean(true);
-        Subscription configSubscription = null;
-
-        /* ensure we aren't allowing more connections than we want */
-        int numberConnections = concurrentConnections.incrementAndGet();
-        try {
-            if (numberConnections > maxConcurrentConnections.get()) {
-                response.sendError(503, "MaxConcurrentConnections reached: " + maxConcurrentConnections.get());
-            } else {
-                int delay = getDelayFromHttpRequest(request);
-
-                /* 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();
-
-                HystrixConfigurationStream configurationStream = createStream.call(delay);
-
-                //since the config 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
-                configSubscription = configurationStream
-                        .observe()
-                        .observeOn(Schedulers.io())
-                        .subscribe(new Subscriber<HystrixConfiguration>() {
-                            @Override
-                            public void onCompleted() {
-                                logger.error("HystrixConfigSseServlet received unexpected OnCompleted from config stream");
-                                moreDataWillBeSent.set(false);
-                            }
-
-                            @Override
-                            public void onError(Throwable e) {
-                                moreDataWillBeSent.set(false);
-                            }
-
-                            @Override
-                            public void onNext(HystrixConfiguration hystrixConfiguration) {
-                                if (hystrixConfiguration != null) {
-                                    String configAsStr = null;
-                                    try {
-                                        configAsStr = convertToString(hystrixConfiguration);
-                                    } catch (IOException ioe) {
-                                        //exception while converting String to JSON
-                                        logger.error("Error converting configuration to JSON ", ioe);
-                                    }
-                                    if (configAsStr != null) {
-                                        try {
-                                            writer.print("data: " + configAsStr + "\n\n");
-                                            // explicitly check for client disconnect - PrintWriter does not throw exceptions
-                                            if (writer.checkError()) {
-                                                throw new IOException("io error");
-                                            }
-                                            writer.flush();
-                                        } catch (IOException ioe) {
-                                            moreDataWillBeSent.set(false);
-                                        }
-                                    }
-                                }
-                            }
-                        });
-
-                while (moreDataWillBeSent.get() && !isDestroyed) {
-                    try {
-                        Thread.sleep(delay);
-                    } catch (InterruptedException e) {
-                        moreDataWillBeSent.set(false);
-                    }
-                }
-            }
-        } finally {
-            concurrentConnections.decrementAndGet();
-            if (configSubscription != null && !configSubscription.isUnsubscribed()) {
-                configSubscription.unsubscribe();
-            }
-        }
-    }
 }