// Path: reactive-lab-services/src/main/java/io/reactivex/lab/services/common/SimpleJson.java
// public class SimpleJson {
// 
//     private static final SimpleJson INSTANCE = new SimpleJson();
// 
//     private SimpleJson() {
// 
//     }
// 
//     public static Map<String, Object> jsonToMap(String jsonString) {
//         return INSTANCE._jsonToMap(jsonString);
//     }
// 
//     public static String mapToJson(Map<String, ?> map) {
//         return INSTANCE._mapToJson(map);
//     }
// 
//     private final ObjectMapper objectMapper = new ObjectMapper();
//     private final ObjectReader objectReader = objectMapper.reader(Map.class);
// 
//     private Map<String, Object> _jsonToMap(String jsonString) {
//         try {
//             return objectReader.readValue(jsonString);
//         } catch (IOException e) {
//             throw new RuntimeException("Unable to parse JSON", e);
//         }
//     }
// 
//     private final ObjectWriter objectWriter = objectMapper.writerWithType(Map.class);
// 
//     private String _mapToJson(Map<String, ?> map) {
//         try {
//             return objectWriter.writeValueAsString(map);
//         } catch (IOException e) {
//             throw new RuntimeException("Unable to write JSON", e);
//         }
//     }
// }


import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.lab.services.common.SimpleJson;
import io.reactivex.netty.RxNetty;
import io.reactivex.netty.protocol.http.server.file.ClassPathFileRequestHandler;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.stream.Collectors;
import rx.Observable;
import rx.schedulers.Schedulers;
import com.netflix.eureka2.client.Eureka;
import com.netflix.eureka2.client.EurekaClient;
import com.netflix.eureka2.client.resolver.ServerResolver;
import com.netflix.eureka2.client.resolver.ServerResolvers;
import com.netflix.eureka2.interests.Interests;
import com.netflix.eureka2.registry.NetworkAddress.ProtocolType;
import com.netflix.eureka2.registry.ServicePort;
import com.netflix.eureka2.registry.datacenter.LocalDataCenterInfo;
import com.netflix.eureka2.server.EurekaWriteServer;
import com.netflix.eureka2.server.WriteServerConfig;
import com.netflix.eureka2.transport.EurekaTransports.Codec;

    public static class StaticFileHandler extends ClassPathFileRequestHandler {
        public StaticFileHandler() {
            super(".");
        }
    }

    private static void startEurekaDashboard(final int port, EurekaClient client) {
        final StaticFileHandler staticFileHandler = new StaticFileHandler();

        RxNetty.createHttpServer(port, (request, response) -> {
            if (request.getUri().startsWith("/dashboard")) {
                return staticFileHandler.handle(request, response);
            } else if (request.getUri().startsWith("/data")) {
                response.getHeaders().set(HttpHeaders.Names.CONTENT_TYPE, "text/event-stream");
                response.getHeaders().set(HttpHeaders.Names.CACHE_CONTROL, "no-cache");
                return client.forInterest(Interests.forFullRegistry())
                        .flatMap(notification -> {
                            ByteBuf data = response.getAllocator().buffer();
                            data.writeBytes("data: ".getBytes());
                            Map<String, String> dataAttributes = new HashMap<>();
                            dataAttributes.put("type", notification.getKind().toString());
                            dataAttributes.put("instance-id", notification.getData().getId());
                            dataAttributes.put("vip", notification.getData().getVipAddress());
                            if (notification.getData().getStatus() != null) {
                                dataAttributes.put("status", notification.getData().getStatus().name());
                            }
                            HashSet<ServicePort> servicePorts = notification.getData().getPorts();
                            int port1 = servicePorts.iterator().next().getPort();
                            dataAttributes.put("port", String.valueOf(port1));
// Path: reactive-lab-services/src/main/java/io/reactivex/lab/services/common/SimpleJson.java
// public class SimpleJson {
// 
//     private static final SimpleJson INSTANCE = new SimpleJson();
// 
//     private SimpleJson() {
// 
//     }
// 
//     public static Map<String, Object> jsonToMap(String jsonString) {
//         return INSTANCE._jsonToMap(jsonString);
//     }
// 
//     public static String mapToJson(Map<String, ?> map) {
//         return INSTANCE._mapToJson(map);
//     }
// 
//     private final ObjectMapper objectMapper = new ObjectMapper();
//     private final ObjectReader objectReader = objectMapper.reader(Map.class);
// 
//     private Map<String, Object> _jsonToMap(String jsonString) {
//         try {
//             return objectReader.readValue(jsonString);
//         } catch (IOException e) {
//             throw new RuntimeException("Unable to parse JSON", e);
//         }
//     }
// 
//     private final ObjectWriter objectWriter = objectMapper.writerWithType(Map.class);
// 
//     private String _mapToJson(Map<String, ?> map) {
//         try {
//             return objectWriter.writeValueAsString(map);
//         } catch (IOException e) {
//             throw new RuntimeException("Unable to write JSON", e);
//         }
//     }
// }


// Path: reactive-lab-services/src/main/java/io/reactivex/lab/services/StartEurekaServer.java
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.lab.services.common.SimpleJson;
import io.reactivex.netty.RxNetty;
import io.reactivex.netty.protocol.http.server.file.ClassPathFileRequestHandler;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.stream.Collectors;
import rx.Observable;
import rx.schedulers.Schedulers;
import com.netflix.eureka2.client.Eureka;
import com.netflix.eureka2.client.EurekaClient;
import com.netflix.eureka2.client.resolver.ServerResolver;
import com.netflix.eureka2.client.resolver.ServerResolvers;
import com.netflix.eureka2.interests.Interests;
import com.netflix.eureka2.registry.NetworkAddress.ProtocolType;
import com.netflix.eureka2.registry.ServicePort;
import com.netflix.eureka2.registry.datacenter.LocalDataCenterInfo;
import com.netflix.eureka2.server.EurekaWriteServer;
import com.netflix.eureka2.server.WriteServerConfig;
import com.netflix.eureka2.transport.EurekaTransports.Codec;


    public static class StaticFileHandler extends ClassPathFileRequestHandler {
        public StaticFileHandler() {
            super(".");
        }
    }

    private static void startEurekaDashboard(final int port, EurekaClient client) {
        final StaticFileHandler staticFileHandler = new StaticFileHandler();

        RxNetty.createHttpServer(port, (request, response) -> {
            if (request.getUri().startsWith("/dashboard")) {
                return staticFileHandler.handle(request, response);
            } else if (request.getUri().startsWith("/data")) {
                response.getHeaders().set(HttpHeaders.Names.CONTENT_TYPE, "text/event-stream");
                response.getHeaders().set(HttpHeaders.Names.CACHE_CONTROL, "no-cache");
                return client.forInterest(Interests.forFullRegistry())
                        .flatMap(notification -> {
                            ByteBuf data = response.getAllocator().buffer();
                            data.writeBytes("data: ".getBytes());
                            Map<String, String> dataAttributes = new HashMap<>();
                            dataAttributes.put("type", notification.getKind().toString());
                            dataAttributes.put("instance-id", notification.getData().getId());
                            dataAttributes.put("vip", notification.getData().getVipAddress());
                            if (notification.getData().getStatus() != null) {
                                dataAttributes.put("status", notification.getData().getStatus().name());
                            }
                            HashSet<ServicePort> servicePorts = notification.getData().getPorts();
                            int port1 = servicePorts.iterator().next().getPort();
                            dataAttributes.put("port", String.valueOf(port1));
                            String jsonData = SimpleJson.mapToJson(dataAttributes);