Skip to content

Commit

Permalink
Deprecated HystrixRequestEventsJsonStream to use common serialization…
Browse files Browse the repository at this point in the history 
… mechanism
  • Loading branch information
Matt Jacobs committed Jun 22, 2016
1 parent 62de321 commit f1bb9c9
Show file tree
Hide file tree
Showing 4 changed files with 61 additions and 176 deletions.
4 changes: 0 additions & 4 deletions ...ain/java/com/netflix/hystrix/contrib/metrics/eventstream/HystrixMetricsStreamServlet.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@
import com.netflix.hystrix.contrib.sample.stream.HystrixSampleSseServlet;
import com.netflix.hystrix.metric.consumer.HystrixDashboardStream;
import com.netflix.hystrix.metric.serial.SerialHystrixDashboardData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.functions.Func1;

Expand Down Expand Up @@ -52,8 +50,6 @@ public class HystrixMetricsStreamServlet extends HystrixSampleSseServlet {

private static final long serialVersionUID = -7548505095303313237L;

private static final Logger logger = LoggerFactory.getLogger(HystrixMetricsStreamServlet.class);

/* used to track number of connections and throttle */
private static AtomicInteger concurrentConnections = new AtomicInteger(0);
private static DynamicIntProperty maxConcurrentConnections =
Expand Down
7 changes: 7 additions & 0 deletions ...main/java/com/netflix/hystrix/contrib/requests/stream/HystrixRequestEventsJsonStream.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,13 @@
import java.util.List;
import java.util.Map;

/**
* Stream that converts HystrixRequestEvents into JSON. This isn't needed anymore, as it more straightforward
* to consider serialization completely separately from the domain object stream
*
* @deprecated Instead, prefer mapping your preferred serialization on top of {@link HystrixRequestEventsStream#observe()}.
*/
@Deprecated //since 1.5.4
public class HystrixRequestEventsJsonStream {
private static final JsonFactory jsonFactory = new JsonFactory();

Expand Down
193 changes: 26 additions & 167 deletions ...main/java/com/netflix/hystrix/contrib/requests/stream/HystrixRequestEventsSseServlet.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,198 +17,57 @@

import com.netflix.config.DynamicIntProperty;
import com.netflix.config.DynamicPropertyFactory;
import com.netflix.hystrix.contrib.sample.stream.HystrixSampleSseServlet;
import com.netflix.hystrix.metric.HystrixRequestEvents;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Subscriber;
import rx.Subscription;
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.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import com.netflix.hystrix.metric.HystrixRequestEventsStream;
import com.netflix.hystrix.metric.serial.SerialHystrixRequestEvents;
import rx.Observable;
import rx.functions.Func1;
import java.util.concurrent.atomic.AtomicInteger;

/**
*/
public class HystrixRequestEventsSseServlet extends HttpServlet {

private static final Logger logger = LoggerFactory.getLogger(HystrixRequestEventsSseServlet.class);

private static volatile boolean isDestroyed = false;
public class HystrixRequestEventsSseServlet extends HystrixSampleSseServlet {

private static final String DELAY_REQ_PARAM_NAME = "delay";
private static final int DEFAULT_DELAY_IN_MILLISECONDS = 10000;
private static final int DEFAULT_QUEUE_DEPTH = 1000;
private static final String PING = "\n: ping\n";
private static final long serialVersionUID = 6389353893099737870L;

/* used to track number of connections and throttle */
private static AtomicInteger concurrentConnections = new AtomicInteger(0);
private static DynamicIntProperty maxConcurrentConnections =
DynamicPropertyFactory.getInstance().getIntProperty("hystrix.requests.stream.maxConcurrentConnections", 5);

private final LinkedBlockingQueue<HystrixRequestEvents> requestQueue = new LinkedBlockingQueue<HystrixRequestEvents>(DEFAULT_QUEUE_DEPTH);
private final HystrixRequestEventsJsonStream requestEventsJsonStream;
DynamicPropertyFactory.getInstance().getIntProperty("hystrix.config.stream.maxConcurrentConnections", 5);

public HystrixRequestEventsSseServlet() {
requestEventsJsonStream = new HystrixRequestEventsJsonStream();
this(HystrixRequestEventsStream.getInstance().observe(), DEFAULT_PAUSE_POLLER_THREAD_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);
}
}

/* package-private */
int getDelayFromHttpRequest(HttpServletRequest req) {
try {
String delay = req.getParameter(DELAY_REQ_PARAM_NAME);
if (delay != null) {
return Math.max(Integer.parseInt(delay), 1);
/* package-private */ HystrixRequestEventsSseServlet(Observable<HystrixRequestEvents> sampleStream, int pausePollerThreadDelayInMs) {
super(sampleStream.map(new Func1<HystrixRequestEvents, String>() {
@Override
public String call(HystrixRequestEvents requestEvents) {
return SerialHystrixRequestEvents.toJsonString(requestEvents);
}
} catch (Throwable ex) {
//silently fail
}
return DEFAULT_DELAY_IN_MILLISECONDS;
}), pausePollerThreadDelayInMs);
}

/**
* 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;
@Override
protected int getMaxNumberConcurrentConnectionsAllowed() {
return maxConcurrentConnections.get();
}

@Override
public void init() throws ServletException {
isDestroyed = false;
protected int getNumberCurrentConnections() {
return concurrentConnections.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();
protected int incrementAndGetCurrentConcurrentConnections() {
return concurrentConnections.incrementAndGet();
}



/**
* - 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 requestsSubscription = null;

/* ensure we aren't allowing more connections than we want */
int numberConnections = concurrentConnections.incrementAndGet();
try {
int maxNumberConnectionsAllowed = maxConcurrentConnections.get();
if (numberConnections > maxNumberConnectionsAllowed) {
response.sendError(503, "MaxConcurrentConnections reached: " + maxNumberConnectionsAllowed);
} 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();

//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
requestsSubscription = requestEventsJsonStream
.getStream()
.observeOn(Schedulers.io())
.subscribe(new Subscriber<HystrixRequestEvents>() {
@Override
public void onCompleted() {
logger.error("HystrixRequestEventsSseServlet received unexpected OnCompleted from request stream");
moreDataWillBeSent.set(false);
}

@Override
public void onError(Throwable e) {
moreDataWillBeSent.set(false);
}

@Override
public void onNext(HystrixRequestEvents requestEvents) {
if (requestEvents != null) {
requestQueue.offer(requestEvents);
}
}
});

while (moreDataWillBeSent.get() && !isDestroyed) {
try {
if (requestQueue.isEmpty()) {
try {
writer.print(PING);
writer.flush();
} catch (Throwable t) {
throw new IOException("Exception while writing ping");
}

if (writer.checkError()) {
throw new IOException("io error");
}
} else {
List<HystrixRequestEvents> l = new ArrayList<HystrixRequestEvents>();
requestQueue.drainTo(l);
String requestEventsAsStr = HystrixRequestEventsJsonStream.convertRequestsToJson(l);
if (requestEventsAsStr != null) {
try {
writer.print("data: " + requestEventsAsStr + "\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);
}
}
}
Thread.sleep(delay);
} catch (InterruptedException e) {
moreDataWillBeSent.set(false);
}
}
}
} finally {
concurrentConnections.decrementAndGet();
if (requestsSubscription != null && !requestsSubscription.isUnsubscribed()) {
requestsSubscription.unsubscribe();
}
}
@Override
protected void decrementCurrentConcurrentConnections() {
concurrentConnections.decrementAndGet();
}
}


33 changes: 28 additions & 5 deletions ...ta-stream/src/main/java/com/netflix/hystrix/metric/serial/SerialHystrixRequestEvents.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.StringWriter;
import java.util.List;
import java.util.Map;

Expand All @@ -32,8 +33,34 @@ public static byte[] toBytes(HystrixRequestEvents requestEvents) {

try {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
JsonGenerator json = cborFactory.createGenerator(bos);
JsonGenerator cbor = cborFactory.createGenerator(bos);

serializeRequestEvents(requestEvents, cbor);

retVal = bos.toByteArray();
} catch (Exception e) {
throw new RuntimeException(e);
}

return retVal;
}

public static String toJsonString(HystrixRequestEvents requestEvents) {
StringWriter jsonString = new StringWriter();

try {
JsonGenerator json = jsonFactory.createGenerator(jsonString);

serializeRequestEvents(requestEvents, json);
} catch (Exception e) {
throw new RuntimeException(e);
}

return jsonString.getBuffer().toString();
}

private static void serializeRequestEvents(HystrixRequestEvents requestEvents, JsonGenerator json) {
try {
json.writeStartArray();

for (Map.Entry<HystrixRequestEvents.ExecutionSignature, List<Integer>> entry: requestEvents.getExecutionsMappedToLatencies().entrySet()) {
Expand All @@ -42,13 +69,9 @@ public static byte[] toBytes(HystrixRequestEvents requestEvents) {

json.writeEndArray();
json.close();

retVal = bos.toByteArray();
} catch (Exception e) {
throw new RuntimeException(e);
}

return retVal;
}

private static void convertExecutionToJson(JsonGenerator json, HystrixRequestEvents.ExecutionSignature executionSignature, List<Integer> latencies) throws IOException {
Expand Down

0 comments on commit f1bb9c9

Please sign in to comment.