// Path: marketdata-common/src/main/java/com/handson/dto/Trade.java
// public class Trade {
// 
//     public final String code;
//     public final int quantity;
//     public final double nominal;
// 
//     public Trade(String code, int quantity, double nominal) {
//         this.code = code;
//         this.quantity = quantity;
//         this.nominal = nominal;
//     }
// 
//     public static Trade fromJson(String input) {
//         return new Gson().fromJson(input, Trade.class);
//     }
// 
// 
//     public String toJson() {
//         return new Gson().toJson(this);
//     }
// 
//     @Override
//     public String toString() {
//         return toJson();
//     }
// }
// 
// Path: marketdata-common/src/main/java/com/handson/dto/Vwap.java
// public class Vwap {
// 
//     public final String code;
//     public final double vwap;
//     public final double volume;
// 
//     public Vwap(String code) {
//         this.code = code;
//         this.vwap = 0;
//         this.volume = 0;
//     }
// 
//     public Vwap(String code, double vwap, double volume) {
//         this.code = code;
//         this.vwap = vwap;
//         this.volume = volume;
//     }
// 
//     public Vwap addTrade(Trade t) {
//         double volume = this.volume + t.quantity;
//         double vwap = (this.volume * this.vwap + t.nominal) / volume;
//         return new Vwap(t.code, vwap, volume);
//     }
// 
//     public static Vwap fromJson(String input) {
//         return new Gson().fromJson(input, Vwap.class);
//     }
// 
// 
//     public String toJson() {
//         return new Gson().toJson(this);
//     }
// 
//     @Override
//     public String toString() {
//         return toJson();
//     }
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/EventStreamClient.java
// public interface EventStreamClient {
// 
//     Observable<String> readServerSideEvents();
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/HttpRequest.java
// public class HttpRequest {
// 
//     private final Map<String, List<String>> parameters;
// 
//     public HttpRequest(Map<String, List<String>> parameters) {
//         this.parameters = parameters;
//     }
// 
//     public String getParameter(String name) {
//         if (parameters.get(name) == null) {
//             return null;
//         }
//         return parameters.get(name).get(0);
//     }
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/RxNettyEventServer.java
// public abstract class RxNettyEventServer<T> {
// 
// 
//     private final int port;
// 
//     public RxNettyEventServer(int port) {
//         this.port = port;
//     }
// 
//     public HttpServer<ByteBuf, ServerSentEvent> createServer() {
//         HttpServer<ByteBuf, ServerSentEvent> server = RxNetty.createHttpServer(port,
//                 (request, response) -> {
//                     response.getHeaders().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
//                     response.getHeaders().set(CACHE_CONTROL, "no-cache");
//                     response.getHeaders().set(CONNECTION, "keep-alive");
//                     response.getHeaders().set(CONTENT_TYPE, "text/event-stream");
//                     return getIntervalObservable(request, response);
//                 }, PipelineConfigurators.<ByteBuf>serveSseConfigurator());
//         System.out.println("HTTP Server Sent Events server started...");
//         return server;
//     }
// 
//     private Observable<Void> getIntervalObservable(HttpServerRequest<?> request, final HttpServerResponse<ServerSentEvent> response) {
//         HttpRequest simpleRequest = new HttpRequest(request.getQueryParameters());
//         return getEvents(simpleRequest)
//                 .flatMap(event -> {
//                     System.out.println("Writing SSE event: " + event);
//                     ByteBuf data = response.getAllocator().buffer().writeBytes(( event + "\n").getBytes());
//                     ServerSentEvent sse = new ServerSentEvent(data);
//                     return response.writeAndFlush(sse);
//                 }).materialize()
//                 .takeWhile(notification -> {
//                     if (notification.isOnError()) {
//                         System.out.println("Write to client failed, stopping response sending.");
//                         notification.getThrowable().printStackTrace(System.err);
//                     }
//                     return !notification.isOnError();
//                 })
//                 .map((Func1<Notification<Void>, Void>) notification -> null);
//     }
// 
// 
//     protected abstract Observable<T> getEvents(HttpRequest request);
// 
// 
// }


import com.handson.dto.Trade;
import com.handson.dto.Vwap;
import com.handson.infra.EventStreamClient;
import com.handson.infra.HttpRequest;
import com.handson.infra.RxNettyEventServer;
import rx.Observable;
import rx.Scheduler;
import java.util.concurrent.TimeUnit;
package com.handson.rx;




public class VwapServer extends RxNettyEventServer<Vwap> {

    private final EventStreamClient tradeEventStreamClient;
    private final Scheduler scheduler;

    public VwapServer(int port, EventStreamClient tradeEventStreamClient, Scheduler scheduler) {
        super(port);
        this.tradeEventStreamClient = tradeEventStreamClient;
        this.scheduler = scheduler;
    }

    @Override
// Path: marketdata-common/src/main/java/com/handson/dto/Trade.java
// public class Trade {
// 
//     public final String code;
//     public final int quantity;
//     public final double nominal;
// 
//     public Trade(String code, int quantity, double nominal) {
//         this.code = code;
//         this.quantity = quantity;
//         this.nominal = nominal;
//     }
// 
//     public static Trade fromJson(String input) {
//         return new Gson().fromJson(input, Trade.class);
//     }
// 
// 
//     public String toJson() {
//         return new Gson().toJson(this);
//     }
// 
//     @Override
//     public String toString() {
//         return toJson();
//     }
// }
// 
// Path: marketdata-common/src/main/java/com/handson/dto/Vwap.java
// public class Vwap {
// 
//     public final String code;
//     public final double vwap;
//     public final double volume;
// 
//     public Vwap(String code) {
//         this.code = code;
//         this.vwap = 0;
//         this.volume = 0;
//     }
// 
//     public Vwap(String code, double vwap, double volume) {
//         this.code = code;
//         this.vwap = vwap;
//         this.volume = volume;
//     }
// 
//     public Vwap addTrade(Trade t) {
//         double volume = this.volume + t.quantity;
//         double vwap = (this.volume * this.vwap + t.nominal) / volume;
//         return new Vwap(t.code, vwap, volume);
//     }
// 
//     public static Vwap fromJson(String input) {
//         return new Gson().fromJson(input, Vwap.class);
//     }
// 
// 
//     public String toJson() {
//         return new Gson().toJson(this);
//     }
// 
//     @Override
//     public String toString() {
//         return toJson();
//     }
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/EventStreamClient.java
// public interface EventStreamClient {
// 
//     Observable<String> readServerSideEvents();
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/HttpRequest.java
// public class HttpRequest {
// 
//     private final Map<String, List<String>> parameters;
// 
//     public HttpRequest(Map<String, List<String>> parameters) {
//         this.parameters = parameters;
//     }
// 
//     public String getParameter(String name) {
//         if (parameters.get(name) == null) {
//             return null;
//         }
//         return parameters.get(name).get(0);
//     }
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/RxNettyEventServer.java
// public abstract class RxNettyEventServer<T> {
// 
// 
//     private final int port;
// 
//     public RxNettyEventServer(int port) {
//         this.port = port;
//     }
// 
//     public HttpServer<ByteBuf, ServerSentEvent> createServer() {
//         HttpServer<ByteBuf, ServerSentEvent> server = RxNetty.createHttpServer(port,
//                 (request, response) -> {
//                     response.getHeaders().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
//                     response.getHeaders().set(CACHE_CONTROL, "no-cache");
//                     response.getHeaders().set(CONNECTION, "keep-alive");
//                     response.getHeaders().set(CONTENT_TYPE, "text/event-stream");
//                     return getIntervalObservable(request, response);
//                 }, PipelineConfigurators.<ByteBuf>serveSseConfigurator());
//         System.out.println("HTTP Server Sent Events server started...");
//         return server;
//     }
// 
//     private Observable<Void> getIntervalObservable(HttpServerRequest<?> request, final HttpServerResponse<ServerSentEvent> response) {
//         HttpRequest simpleRequest = new HttpRequest(request.getQueryParameters());
//         return getEvents(simpleRequest)
//                 .flatMap(event -> {
//                     System.out.println("Writing SSE event: " + event);
//                     ByteBuf data = response.getAllocator().buffer().writeBytes(( event + "\n").getBytes());
//                     ServerSentEvent sse = new ServerSentEvent(data);
//                     return response.writeAndFlush(sse);
//                 }).materialize()
//                 .takeWhile(notification -> {
//                     if (notification.isOnError()) {
//                         System.out.println("Write to client failed, stopping response sending.");
//                         notification.getThrowable().printStackTrace(System.err);
//                     }
//                     return !notification.isOnError();
//                 })
//                 .map((Func1<Notification<Void>, Void>) notification -> null);
//     }
// 
// 
//     protected abstract Observable<T> getEvents(HttpRequest request);
// 
// 
// }


// Path: marketdata-web/src/main/java/com/handson/rx/VwapServer.java
import com.handson.dto.Trade;
import com.handson.dto.Vwap;
import com.handson.infra.EventStreamClient;
import com.handson.infra.HttpRequest;
import com.handson.infra.RxNettyEventServer;
import rx.Observable;
import rx.Scheduler;
import java.util.concurrent.TimeUnit;

package com.handson.rx;




public class VwapServer extends RxNettyEventServer<Vwap> {

    private final EventStreamClient tradeEventStreamClient;
    private final Scheduler scheduler;

    public VwapServer(int port, EventStreamClient tradeEventStreamClient, Scheduler scheduler) {
        super(port);
        this.tradeEventStreamClient = tradeEventStreamClient;
        this.scheduler = scheduler;
    }

    @Override
    protected Observable<Vwap> getEvents(HttpRequest request) {