// 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/Stock.java
// public class Stock {
// 
//     public final String code;
// 
//     public final String companyName;
// 
//     public final String market;
// 
//     public Stock(String code, String companyName, String market) {
//         this.code = code;
//         this.companyName = companyName;
//         this.market = market;
//     }
// 
//     public static Stock fromJson(String input) {
//         return new Gson().fromJson(input, Stock.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/RequestReplyClient.java
// public interface RequestReplyClient {
// 
//     Observable<String> request(String parameter);
// }


import com.handson.dto.Quote;
import com.handson.dto.Stock;
import com.handson.infra.EventStreamClient;
import com.handson.infra.RequestReplyClient;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import rx.Observable;
import rx.marble.junit.MarbleRule;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static rx.marble.MapHelper.of;
import static rx.marble.junit.MarbleRule.expectObservable;
import static rx.marble.junit.MarbleRule.hot;
        // given
        Observable<String> quoteSource
                = hot("--f-s-t--", of("f", new Quote("GOOGL", 705.8673).toJson(),
                                      "s", new Quote("GOOGL", 705.8912).toJson(),
                                      "t", new Quote("IBM", 106.344).toJson()
        ));
        // when
        StockServer server = createServer(quoteSource);
        // then
        expectObservable(server.getEvents(null).map(s -> s.companyName).concatWith(Observable.never()))
                .toBe("--g---i--", of("g", "Alphabet Inc",
                                      "i", "International Business Machines Corp."));
    }

    /**
     * Test 9
     */
    @Test
    @Ignore
    public void should_stop_stream_after_10_seconds() {
        // given
        Observable<String> quoteSource = Observable.never();
        // when
        StockServer server = createServer(quoteSource);
        // then
        expectObservable(server.getEvents(null)).toBe("----------|");
    }

    public StockServer createServer(Observable<String> quoteSource) {
        RequestReplyClient stockClient = mock(RequestReplyClient.class);
// 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/Stock.java
// public class Stock {
// 
//     public final String code;
// 
//     public final String companyName;
// 
//     public final String market;
// 
//     public Stock(String code, String companyName, String market) {
//         this.code = code;
//         this.companyName = companyName;
//         this.market = market;
//     }
// 
//     public static Stock fromJson(String input) {
//         return new Gson().fromJson(input, Stock.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/RequestReplyClient.java
// public interface RequestReplyClient {
// 
//     Observable<String> request(String parameter);
// }


// Path: marketdata-web/src/test/java/com/handson/rx/StockServerTest.java
import com.handson.dto.Quote;
import com.handson.dto.Stock;
import com.handson.infra.EventStreamClient;
import com.handson.infra.RequestReplyClient;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import rx.Observable;
import rx.marble.junit.MarbleRule;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static rx.marble.MapHelper.of;
import static rx.marble.junit.MarbleRule.expectObservable;
import static rx.marble.junit.MarbleRule.hot;

        // given
        Observable<String> quoteSource
                = hot("--f-s-t--", of("f", new Quote("GOOGL", 705.8673).toJson(),
                                      "s", new Quote("GOOGL", 705.8912).toJson(),
                                      "t", new Quote("IBM", 106.344).toJson()
        ));
        // when
        StockServer server = createServer(quoteSource);
        // then
        expectObservable(server.getEvents(null).map(s -> s.companyName).concatWith(Observable.never()))
                .toBe("--g---i--", of("g", "Alphabet Inc",
                                      "i", "International Business Machines Corp."));
    }

    /**
     * Test 9
     */
    @Test
    @Ignore
    public void should_stop_stream_after_10_seconds() {
        // given
        Observable<String> quoteSource = Observable.never();
        // when
        StockServer server = createServer(quoteSource);
        // then
        expectObservable(server.getEvents(null)).toBe("----------|");
    }

    public StockServer createServer(Observable<String> quoteSource) {
        RequestReplyClient stockClient = mock(RequestReplyClient.class);
        EventStreamClient stockQuoteClient = mock(EventStreamClient.class);