/**
 * Copyright 2015 Peti Koch
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package ch.petikoch.examples.rxjava.threading;

import rx.Observable;
import rx.Single;
import rx.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class OneLaneMcDrive {

    private static final Clock clock = new Clock(Time.EIGTH_AM);

    public static void main(String[] args) throws InterruptedException {

        Observable<String> customers = Observable.just("Fred", "John", "Maria", "Ben", "Ken", "Greg", "Peter", "Henry").repeat();
        Observable<Long> arrivals = Observable.interval(0, 10, TimeUnit.SECONDS);

        Observable<CustomerWithArrivalTime> customerArriveStream =
                Observable.zip(
                        customers,
                        arrivals,
                        (customer, eventNumber) -> new CustomerWithArrivalTime(customer + eventNumber, clock.getTime()))
                        .doOnNext(customer -> sysout(customer.name + " arrived")); // debugging

        Observable<CustomerWithArrivalTime> orderFinishedStream = customerArriveStream.flatMap(
                customerWithArrivalTime -> {
                    Single<String> mac = Single.<String>create(singleSubscriber -> {
                        sysout("Starting with mac for " + customerWithArrivalTime);
                        waitSeconds(3);
                        sysout("Mac ready for " + customerWithArrivalTime);
                        singleSubscriber.onSuccess("mac");
                    }).subscribeOn(Schedulers.io());

                    Single<String> fries = Single.<String>create(singleSubscriber -> {
                        sysout("Starting with fries for " + customerWithArrivalTime);
                        waitSeconds(2);
                        sysout("Fries ready for " + customerWithArrivalTime);
                        singleSubscriber.onSuccess("fries");
                    }).subscribeOn(Schedulers.io());

                    Single<String> coke = Single.<String>create(singleSubscriber -> {
                        sysout("Starting with coke for " + customerWithArrivalTime);
                        waitSeconds(1);
                        sysout("Coke ready for " + customerWithArrivalTime);
                        singleSubscriber.onSuccess("coke");
                    }).subscribeOn(Schedulers.io());

                    Single<CustomerWithArrivalTime> finishedOrder = Single.zip(mac, fries, coke, (s, s2, s3) -> customerWithArrivalTime);
                    return finishedOrder.toObservable();
                },
                1 // = one lane McDrive
        );

        orderFinishedStream.subscribe(customerWithArrivalTime -> {
            int timeDifference = clock.getTime().difference(customerWithArrivalTime.arrivalTime);
            sysout(customerWithArrivalTime.name + " order finished after " + timeDifference + " minutes");
            sysout(customerWithArrivalTime.name + " says good bye and leaves.");
        });

        waitSeconds(60);
    }

    private static void sysout(String text) {
        System.out.println(clock.getTime() + " [" + Thread.currentThread().getName() + "] " + text);
    }

    private static void waitSeconds(int seconds) {
        try {
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class CustomerWithArrivalTime {
        String name;
        Time arrivalTime;

        CustomerWithArrivalTime(String name, Time arrivalTime) {
            this.name = name;
            this.arrivalTime = arrivalTime;
        }

        @Override
        public String toString() {
            return name + " (arrived at " + arrivalTime + ")";
        }
    }

    static class Clock {

        private final AtomicInteger internalTimeInMinutes = new AtomicInteger();

        Clock(Time initialTime) {
            internalTimeInMinutes.set(initialTime.internalTimeInMinutes);
            Observable.interval(1, TimeUnit.SECONDS).subscribe(aLong -> {
                internalTimeInMinutes.incrementAndGet();
                System.out.println(getTime());
            });
        }

        public boolean isBefore(Time time) {
            return internalTimeInMinutes.get() < time.internalTimeInMinutes;
        }

        public Time getTime() {
            return new Time(internalTimeInMinutes.get());
        }

    }

    static class Time {

        static final Time EIGTH_AM = new Time(8 * 60);
        static final Time EIGTH_TWO_AM = new Time(8 * 60 + 2);
        static final Time NINE_AM = new Time(9 * 60);
        static final Time FIVE_PM = new Time(17 * 60);

        private final int internalTimeInMinutes;

        Time(int internalTimeInMinutes) {
            this.internalTimeInMinutes = internalTimeInMinutes;
        }

        private static String leftPadded2DigitsString(int number, int maxNumber) {
            if (number < 10) {
                return "0" + number;
            } else if (number >= 10 && number <= maxNumber) {
                return Integer.toString(number);
            } else {
                throw new IllegalStateException("Unhandled: " + number);
            }
        }

        int difference(Time laterOther) {
            int result = laterOther.internalTimeInMinutes - internalTimeInMinutes;
            if (result < 0) {
                result = result * -1;
            }
            return result;
        }

        @Override
        public String toString() {
            return leftPadded2DigitsString(internalTimeInMinutes / 60, 23) + ":" + leftPadded2DigitsString(internalTimeInMinutes % 60, 59);
        }
    }
}

@@ -1910,17 +1910,17 @@
 Seconds(
-3
+8
 );%0A     
@@ -2312,17 +2312,17 @@
 Seconds(
-2
+4
 );%0A     

make OneLaneMcDrive same pace as OneLaneMcDrive2