Utilization Streaming

/**
 * Streams Hystrix utilization metrics in text/event-stream format
 * Provides real-time resource utilization for commands and thread pools
 */
public class HystrixUtilizationSseServlet extends HystrixSampleSseServlet {
    
    /**
     * Default constructor using HystrixUtilizationStream and default delay
     */
    public HystrixUtilizationSseServlet();
    
    /**
     * Package-private constructor for testing with custom stream and delay
     * @param sampleStream Observable stream of utilization data
     * @param pausePollerThreadDelayInMs Delay between polling cycles in milliseconds
     */
    HystrixUtilizationSseServlet(Observable<HystrixUtilization> sampleStream, int pausePollerThreadDelayInMs);
    
    /**
     * Returns maximum number of concurrent connections allowed
     * @return Maximum concurrent connections (default: 5)
     */
    protected int getMaxNumberConcurrentConnectionsAllowed();
    
    /**
     * Returns current number of active connections
     * @return Current connection count
     */
    protected int getNumberCurrentConnections();
    
    /**
     * Atomically increments and returns current concurrent connection count
     * @return New connection count after increment
     */
    protected int incrementAndGetCurrentConcurrentConnections();
    
    /**
     * Atomically decrements current concurrent connection count
     */
    protected void decrementCurrentConcurrentConnections();
}

/**
 * Links HystrixUtilizationStream and JSON encoding
 * @deprecated Since 1.5.4 - prefer mapping serialization on HystrixUtilizationStream.observe()
 */
@Deprecated
public class HystrixUtilizationJsonStream {
    
    /**
     * Default constructor using default utilization stream
     */
    public HystrixUtilizationJsonStream();
    
    /**
     * Constructor with custom stream generator
     * @param streamGenerator Function to generate utilization observable
     */
    public HystrixUtilizationJsonStream(Func1<Integer, Observable<HystrixUtilization>> streamGenerator);
    
    /**
     * Convert utilization object to JSON string
     * @param utilization Utilization object to convert
     * @return JSON string representation
     * @throws IOException if JSON generation fails
     */
    protected static String convertToJson(HystrixUtilization utilization) throws IOException;
    
    /**
     * @deprecated Use HystrixUtilizationStream.observe() instead
     */
    @Deprecated
    public Observable<HystrixUtilization> observe(int delay);
    
    /**
     * @deprecated Use HystrixUtilizationStream.observe() and convertToJson() instead
     */
    @Deprecated
    public Observable<String> observeJson(int delay);
}

<servlet>
  <description></description>
  <display-name>HystrixUtilizationSseServlet</display-name>
  <servlet-name>HystrixUtilizationSseServlet</servlet-name>
  <servlet-class>com.netflix.hystrix.contrib.sample.stream.HystrixUtilizationSseServlet</servlet-class>
</servlet>
<servlet-mapping>
  <servlet-name>HystrixUtilizationSseServlet</servlet-name>
  <url-pattern>/hystrix/utilization.stream</url-pattern>
</servlet-mapping>

// Deploy servlet via web.xml configuration
// Access utilization stream
curl http://localhost:8080/app/hystrix/utilization.stream

// Using deprecated JSON stream API
HystrixUtilizationJsonStream stream = new HystrixUtilizationJsonStream();
stream.observeJson(1000).subscribe(json -> {
    System.out.println("Utilization: " + json);
});

data: {
  "type": "HystrixUtilization",
  "commands": {
    "GetUser": {
      "activeCount": 2
    },
    "GetUserPreferences": {
      "activeCount": 0
    },
    "CreateUser": {
      "activeCount": 1
    }
  },
  "threadpools": {
    "UserService": {
      "activeCount": 3,
      "queueSize": 0,
      "corePoolSize": 10,
      "poolSize": 10
    },
    "NotificationService": {
      "activeCount": 0,
      "queueSize": 2,
      "corePoolSize": 5,
      "poolSize": 5
    }
  }
}

/**
 * Write command utilization data to JSON generator
 * @param json JSON generator to write to
 * @param key Command key
 * @param utilization Command utilization data
 * @throws IOException if JSON writing fails
 */
private static void writeCommandUtilizationJson(JsonGenerator json, HystrixCommandKey key, HystrixCommandUtilization utilization) throws IOException;

/**
 * Write thread pool utilization data to JSON generator
 * @param json JSON generator to write to
 * @param threadPoolKey Thread pool key
 * @param utilization Thread pool utilization data
 * @throws IOException if JSON writing fails
 */
private static void writeThreadPoolUtilizationJson(JsonGenerator json, HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolUtilization utilization) throws IOException;

// Monitor utilization via HTTP stream
curl http://localhost:8080/app/hystrix/utilization.stream

// Process utilization data in JavaScript
const eventSource = new EventSource('/hystrix/utilization.stream');
eventSource.onmessage = function(event) {
    const utilization = JSON.parse(event.data);
    
    // Check for overloaded commands
    Object.entries(utilization.commands).forEach(([name, metrics]) => {
        if (metrics.activeCount > 10) {
            console.warn(`Command ${name} has high utilization: ${metrics.activeCount}`);
        }
    });
    
    // Check thread pool usage
    Object.entries(utilization.threadpools).forEach(([name, metrics]) => {
        const utilizationPercent = (metrics.activeCount / metrics.corePoolSize) * 100;
        if (utilizationPercent > 80) {
            console.warn(`Thread pool ${name} is ${utilizationPercent}% utilized`);
        }
    });
};

// Example integration with monitoring system
HystrixUtilizationStream.getInstance()
    .observe()
    .subscribe(utilization -> {
        utilization.getCommandUtilizationMap().forEach((key, commandUtil) -> {
            if (commandUtil.getConcurrentCommandCount() > thresholds.get(key)) {
                alertingService.sendAlert("High command utilization", key.name());
            }
        });
        
        utilization.getThreadPoolUtilizationMap().forEach((key, poolUtil) -> {
            double utilizationPercent = (double) poolUtil.getCurrentActiveCount() / poolUtil.getCurrentCorePoolSize();
            if (utilizationPercent > 0.9) {
                alertingService.sendAlert("High thread pool utilization", key.name());
            }
        });
    });

// Old approach (deprecated)
HystrixUtilizationJsonStream jsonStream = new HystrixUtilizationJsonStream();
jsonStream.observeJson(1000).subscribe(json -> {
    // process json
});

// New approach (recommended)
HystrixUtilizationStream.getInstance()
    .observe()
    .map(SerialHystrixUtilization::toJsonString)
    .subscribe(json -> {
        // process json
    });