// Path: marketdata-common/src/main/java/com/handson/dto/Quote.java
// public class Quote {
// 
//     public final String code;
//     public final double quote;
// 
//     public Quote(String code, double quote) {
//         this.code = code;
//         this.quote = quote;
//     }
// 
//     public static Quote fromJson(String input) {
//         return new Gson().fromJson(input, Quote.class);
//     }
// 
//     public String toJson() {
//         return new Gson().toJson(this);
//     }
// 
//     @Override
//     public String toString() {
//         return toJson();
//     }
// }
// 
// 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/infra/RandomSequenceGenerator.java
// public class RandomSequenceGenerator {
// 
//     private final double min;
//     private final double max;
//     private final Random random;
//     private double bias = 0.3;
// 
//     public RandomSequenceGenerator(double min, double max) {
//         this.min = min;
//         this.max = max;
//         this.random = new Random();
//     }
// 
//     public Observable<Double> create(long interval, TimeUnit timeUnit) {
//         double init = (min + max) /2;
//         return Observable
//                 .interval(interval, timeUnit)
//                 .scan(init, (previous, i) -> computeNextNumber(previous));
//     }
// 
//     public Observable<Integer> createIntegerSequence(long interval, TimeUnit timeUnit) {
//         double range = max - min;
//         return Observable
//                 .interval(interval, timeUnit)
//                 .map(i ->  ((Double)(random.nextDouble() * range  + min)).intValue());
//     }
// 
//     public double computeNextNumber(double previous) {
//         double range = (max - min) / 20;
//         double scaled = (random.nextDouble() - 0.5 + bias) * range;
//         double shifted = previous + scaled;
//         if (shifted < min || shifted > max) {
//             shifted = previous - scaled;
//             bias = -bias;
//         }
// 
//         shifted = ((Long)Math.round(shifted * 10000)).doubleValue() /10000;
// 
//         return shifted;
//     }
// 
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/RxNettyEventBroadcaster.java
// public abstract class RxNettyEventBroadcaster<T> extends RxNettyEventServer {
// 
// 
//     private final boolean flaky;
//     private Observable<T> events;
// 
//     public RxNettyEventBroadcaster(int port, boolean flaky) {
//         super(port);
//         this.flaky = flaky;
//     }
// 
//     public HttpServer<ByteBuf, ServerSentEvent> createServer() {
//         if (flaky) {
//             events = SubscriptionLimiter
//                         .limitSubscriptions(1,initializeEventStream());
//         } else {
//             events  = initializeEventStream();
//         }
//         return super.createServer();
//     }
// 
//     protected abstract Observable<T> initializeEventStream();
// 
//     public Observable<T> getEvents(HttpRequest request) {
//         return events;
//     }
// 
// }


import com.handson.dto.Quote;
import com.handson.dto.Trade;
import com.handson.infra.RandomSequenceGenerator;
import com.handson.infra.RxNettyEventBroadcaster;
import rx.Observable;
import java.util.concurrent.TimeUnit;
package com.handson.market;



public class TradeProvider extends RxNettyEventBroadcaster<Trade> {

    private static final int INTERVAL = 50;
    private final StockQuoteProvider stockQuoteProvider;


    public TradeProvider(int port, boolean flaky, StockQuoteProvider stockQuoteProvider) {
        super(port, flaky);
        this.stockQuoteProvider = stockQuoteProvider;
    }

    @Override
    protected Observable<Trade> initializeEventStream() {
        Observable<Quote> quotes
                = stockQuoteProvider.getEvents(null);
        Observable<Integer> quantities
// Path: marketdata-common/src/main/java/com/handson/dto/Quote.java
// public class Quote {
// 
//     public final String code;
//     public final double quote;
// 
//     public Quote(String code, double quote) {
//         this.code = code;
//         this.quote = quote;
//     }
// 
//     public static Quote fromJson(String input) {
//         return new Gson().fromJson(input, Quote.class);
//     }
// 
//     public String toJson() {
//         return new Gson().toJson(this);
//     }
// 
//     @Override
//     public String toString() {
//         return toJson();
//     }
// }
// 
// 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/infra/RandomSequenceGenerator.java
// public class RandomSequenceGenerator {
// 
//     private final double min;
//     private final double max;
//     private final Random random;
//     private double bias = 0.3;
// 
//     public RandomSequenceGenerator(double min, double max) {
//         this.min = min;
//         this.max = max;
//         this.random = new Random();
//     }
// 
//     public Observable<Double> create(long interval, TimeUnit timeUnit) {
//         double init = (min + max) /2;
//         return Observable
//                 .interval(interval, timeUnit)
//                 .scan(init, (previous, i) -> computeNextNumber(previous));
//     }
// 
//     public Observable<Integer> createIntegerSequence(long interval, TimeUnit timeUnit) {
//         double range = max - min;
//         return Observable
//                 .interval(interval, timeUnit)
//                 .map(i ->  ((Double)(random.nextDouble() * range  + min)).intValue());
//     }
// 
//     public double computeNextNumber(double previous) {
//         double range = (max - min) / 20;
//         double scaled = (random.nextDouble() - 0.5 + bias) * range;
//         double shifted = previous + scaled;
//         if (shifted < min || shifted > max) {
//             shifted = previous - scaled;
//             bias = -bias;
//         }
// 
//         shifted = ((Long)Math.round(shifted * 10000)).doubleValue() /10000;
// 
//         return shifted;
//     }
// 
// }
// 
// Path: marketdata-common/src/main/java/com/handson/infra/RxNettyEventBroadcaster.java
// public abstract class RxNettyEventBroadcaster<T> extends RxNettyEventServer {
// 
// 
//     private final boolean flaky;
//     private Observable<T> events;
// 
//     public RxNettyEventBroadcaster(int port, boolean flaky) {
//         super(port);
//         this.flaky = flaky;
//     }
// 
//     public HttpServer<ByteBuf, ServerSentEvent> createServer() {
//         if (flaky) {
//             events = SubscriptionLimiter
//                         .limitSubscriptions(1,initializeEventStream());
//         } else {
//             events  = initializeEventStream();
//         }
//         return super.createServer();
//     }
// 
//     protected abstract Observable<T> initializeEventStream();
// 
//     public Observable<T> getEvents(HttpRequest request) {
//         return events;
//     }
// 
// }


// Path: marketdata-external/src/main/java/com/handson/market/TradeProvider.java
import com.handson.dto.Quote;
import com.handson.dto.Trade;
import com.handson.infra.RandomSequenceGenerator;
import com.handson.infra.RxNettyEventBroadcaster;
import rx.Observable;
import java.util.concurrent.TimeUnit;

package com.handson.market;



public class TradeProvider extends RxNettyEventBroadcaster<Trade> {

    private static final int INTERVAL = 50;
    private final StockQuoteProvider stockQuoteProvider;


    public TradeProvider(int port, boolean flaky, StockQuoteProvider stockQuoteProvider) {
        super(port, flaky);
        this.stockQuoteProvider = stockQuoteProvider;
    }

    @Override
    protected Observable<Trade> initializeEventStream() {
        Observable<Quote> quotes
                = stockQuoteProvider.getEvents(null);
        Observable<Integer> quantities
                = new RandomSequenceGenerator(10, 1000).createIntegerSequence(INTERVAL, TimeUnit.MILLISECONDS).share();