// Path: reactive-lab-services/src/main/java/io/reactivex/lab/services/metrics/HystrixMetricsStreamHandler.java
// public class HystrixMetricsStreamHandler<I, O> implements RequestHandler<I, O> {
// 
//     public static final String DEFAULT_HYSTRIX_PREFIX = "/hystrix.stream";
// 
//     public static final int DEFAULT_INTERVAL = 2000;
// 
//     private static final byte[] HEADER = "data: ".getBytes(Charset.defaultCharset());
//     private static final byte[] FOOTER = { 10, 10 };
//     private static final int EXTRA_SPACE = HEADER.length + FOOTER.length;
// 
//     private final String hystrixPrefix;
//     private final long interval;
//     private final RequestHandler<I, O> appHandler;
// 
//     private Metrics metrics;
// 
//     public HystrixMetricsStreamHandler(Metrics metrics, String hystrixPrefix, long interval, RequestHandler<I, O> appHandler) {
//         this.metrics = metrics;
//         this.hystrixPrefix = hystrixPrefix;
//         this.interval = interval;
//         this.appHandler = appHandler;
//     }
// 
//     @Override
//     public Observable<Void> handle(HttpServerRequest<I> request, HttpServerResponse<O> response) {
//         if (request.getPath().endsWith(hystrixPrefix)) {
//             return handleHystrixRequest(response);
//         }
//         return appHandler.handle(request, response);
//     }
// 
//     private Observable<Void> handleHystrixRequest(final HttpServerResponse<O> response) {
//         writeHeaders(response);
// 
//         final Subject<Void, Void> subject = PublishSubject.create();
//         final MultipleAssignmentSubscription subscription = new MultipleAssignmentSubscription();
//         Subscription actionSubscription = Observable.timer(0, interval, TimeUnit.MILLISECONDS, Schedulers.computation())
//                 .subscribe(new Action1<Long>() {
//                     @Override
//                     public void call(Long tick) {
//                         if (!response.getChannel().isOpen()) {
//                             subscription.unsubscribe();
//                             return;
//                         }
//                         try {
//                             writeMetric(JsonMapper.toJson(metrics), response);
//                         } catch (Exception e) {
//                             subject.onError(e);
//                         }
//                     }
//                 });
//         subscription.set(actionSubscription);
//         return subject;
//     }
// 
//     private void writeHeaders(HttpServerResponse<O> response) {
//         response.getHeaders().add("Content-Type", "text/event-stream;charset=UTF-8");
//         response.getHeaders().add("Cache-Control", "no-cache, no-store, max-age=0, must-revalidate");
//         response.getHeaders().add("Pragma", "no-cache");
//     }
// 
//     @SuppressWarnings("unchecked")
//     private void writeMetric(String json, HttpServerResponse<O> response) {
//         byte[] bytes = json.getBytes(Charset.defaultCharset());
//         ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.buffer(bytes.length + EXTRA_SPACE);
//         byteBuf.writeBytes(HEADER);
//         byteBuf.writeBytes(bytes);
//         byteBuf.writeBytes(FOOTER);
//         response.writeAndFlush((O) byteBuf);
//     }
// }


import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.lab.services.metrics.HystrixMetricsStreamHandler;
import io.reactivex.lab.services.metrics.Metrics;
import io.reactivex.netty.RxNetty;
import io.reactivex.netty.pipeline.PipelineConfigurators;
import io.reactivex.netty.protocol.http.server.HttpServer;
import io.reactivex.netty.protocol.http.server.HttpServerRequest;
import io.reactivex.netty.protocol.http.server.HttpServerResponse;
import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.UUID;
import rx.Observable;
import com.netflix.eureka2.client.EurekaClient;
import com.netflix.eureka2.registry.InstanceInfo;
import com.netflix.eureka2.registry.ServicePort;
import com.netflix.eureka2.registry.datacenter.BasicDataCenterInfo;
package io.reactivex.lab.services.impls;





/**
 * Common base for the service impls
 */
public abstract class AbstractMiddleTierService {

    private EurekaClient client;
    private HttpServer<ByteBuf, ServerSentEvent> server;
    protected final String eurekaVipAddress;
    private final Metrics metrics;

    protected AbstractMiddleTierService(String eurekaVipAddress, EurekaClient client) {
        this.eurekaVipAddress = eurekaVipAddress;
        this.client = client;
        this.metrics = new Metrics(eurekaVipAddress);
    }

    public HttpServer<ByteBuf, ServerSentEvent> createServer(int port) {
        System.out.println("Start " + getClass().getSimpleName() + " on port: " + port);

        // declare handler chain (wrapped in Hystrix)
        // TODO create a better way of chaining these (related https://github.com/ReactiveX/RxNetty/issues/232 and https://github.com/ReactiveX/RxNetty/issues/202)
        HystrixMetricsStreamHandler<ByteBuf, ServerSentEvent> handlerChain 
// Path: reactive-lab-services/src/main/java/io/reactivex/lab/services/metrics/HystrixMetricsStreamHandler.java
// public class HystrixMetricsStreamHandler<I, O> implements RequestHandler<I, O> {
// 
//     public static final String DEFAULT_HYSTRIX_PREFIX = "/hystrix.stream";
// 
//     public static final int DEFAULT_INTERVAL = 2000;
// 
//     private static final byte[] HEADER = "data: ".getBytes(Charset.defaultCharset());
//     private static final byte[] FOOTER = { 10, 10 };
//     private static final int EXTRA_SPACE = HEADER.length + FOOTER.length;
// 
//     private final String hystrixPrefix;
//     private final long interval;
//     private final RequestHandler<I, O> appHandler;
// 
//     private Metrics metrics;
// 
//     public HystrixMetricsStreamHandler(Metrics metrics, String hystrixPrefix, long interval, RequestHandler<I, O> appHandler) {
//         this.metrics = metrics;
//         this.hystrixPrefix = hystrixPrefix;
//         this.interval = interval;
//         this.appHandler = appHandler;
//     }
// 
//     @Override
//     public Observable<Void> handle(HttpServerRequest<I> request, HttpServerResponse<O> response) {
//         if (request.getPath().endsWith(hystrixPrefix)) {
//             return handleHystrixRequest(response);
//         }
//         return appHandler.handle(request, response);
//     }
// 
//     private Observable<Void> handleHystrixRequest(final HttpServerResponse<O> response) {
//         writeHeaders(response);
// 
//         final Subject<Void, Void> subject = PublishSubject.create();
//         final MultipleAssignmentSubscription subscription = new MultipleAssignmentSubscription();
//         Subscription actionSubscription = Observable.timer(0, interval, TimeUnit.MILLISECONDS, Schedulers.computation())
//                 .subscribe(new Action1<Long>() {
//                     @Override
//                     public void call(Long tick) {
//                         if (!response.getChannel().isOpen()) {
//                             subscription.unsubscribe();
//                             return;
//                         }
//                         try {
//                             writeMetric(JsonMapper.toJson(metrics), response);
//                         } catch (Exception e) {
//                             subject.onError(e);
//                         }
//                     }
//                 });
//         subscription.set(actionSubscription);
//         return subject;
//     }
// 
//     private void writeHeaders(HttpServerResponse<O> response) {
//         response.getHeaders().add("Content-Type", "text/event-stream;charset=UTF-8");
//         response.getHeaders().add("Cache-Control", "no-cache, no-store, max-age=0, must-revalidate");
//         response.getHeaders().add("Pragma", "no-cache");
//     }
// 
//     @SuppressWarnings("unchecked")
//     private void writeMetric(String json, HttpServerResponse<O> response) {
//         byte[] bytes = json.getBytes(Charset.defaultCharset());
//         ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.buffer(bytes.length + EXTRA_SPACE);
//         byteBuf.writeBytes(HEADER);
//         byteBuf.writeBytes(bytes);
//         byteBuf.writeBytes(FOOTER);
//         response.writeAndFlush((O) byteBuf);
//     }
// }


// Path: reactive-lab-services/src/main/java/io/reactivex/lab/services/impls/AbstractMiddleTierService.java
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.lab.services.metrics.HystrixMetricsStreamHandler;
import io.reactivex.lab.services.metrics.Metrics;
import io.reactivex.netty.RxNetty;
import io.reactivex.netty.pipeline.PipelineConfigurators;
import io.reactivex.netty.protocol.http.server.HttpServer;
import io.reactivex.netty.protocol.http.server.HttpServerRequest;
import io.reactivex.netty.protocol.http.server.HttpServerResponse;
import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.UUID;
import rx.Observable;
import com.netflix.eureka2.client.EurekaClient;
import com.netflix.eureka2.registry.InstanceInfo;
import com.netflix.eureka2.registry.ServicePort;
import com.netflix.eureka2.registry.datacenter.BasicDataCenterInfo;

package io.reactivex.lab.services.impls;





/**
 * Common base for the service impls
 */
public abstract class AbstractMiddleTierService {

    private EurekaClient client;
    private HttpServer<ByteBuf, ServerSentEvent> server;
    protected final String eurekaVipAddress;
    private final Metrics metrics;

    protected AbstractMiddleTierService(String eurekaVipAddress, EurekaClient client) {
        this.eurekaVipAddress = eurekaVipAddress;
        this.client = client;
        this.metrics = new Metrics(eurekaVipAddress);
    }

    public HttpServer<ByteBuf, ServerSentEvent> createServer(int port) {
        System.out.println("Start " + getClass().getSimpleName() + " on port: " + port);

        // declare handler chain (wrapped in Hystrix)
        // TODO create a better way of chaining these (related https://github.com/ReactiveX/RxNetty/issues/232 and https://github.com/ReactiveX/RxNetty/issues/202)
        HystrixMetricsStreamHandler<ByteBuf, ServerSentEvent> handlerChain 
          = new HystrixMetricsStreamHandler<>(metrics, "/hystrix.stream", 1000, (request, response) -> {