// Path: src/main/java/shopJava/util/Constants.java
// public interface Constants {
// 
//     int MAX_ORDERS_PER_USER = 10;
// 
//     String MONGODB_URI = "mongodb://localhost:27017";
//     String SHOP_DB_NAME = "shop";
//     String USERS_COLLECTION_NAME = "users";
//     String ORDERS_COLLECTION_NAME = "orders";
// 
//     String HOMER = "homer";
//     String MARGE = "marge";
//     String BART = "bart";
//     String LISA = "lisa";
//     String MAGGIE = "maggie";
// 
//     String[] SIMPSONS = {HOMER, MARGE, BART, LISA, MAGGIE};
// }
// 
// Path: src/main/java/shopJava/util/Util.java
// public static int average(final int totalAmount, final int orderCount) {
//     return orderCount == 0 ? 0 : Math.round((100.0f * totalAmount / orderCount) / 100);
// }
// 
// Path: src/main/java/shopJava/util/Util.java
// public static User checkUserLoggedIn(final Optional<User> optUser, final Credentials credentials) {
// 
//     if (!optUser.isPresent()) {     // replaces if (user != null)
//         throw new RuntimeException(new IllegalAccessException("User unknown: " + credentials.username));
//     }
//     final User user = optUser.get();
//     return checkUserLoggedIn(user, credentials);
// }


import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import org.bson.Document;
import org.reactivestreams.Publisher;
import rx.Observable;
import rx.Single;
import shopJava.model.*;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import static com.mongodb.client.model.Filters.eq;
import static java.lang.Thread.sleep;
import static rx.RxReactiveStreams.toObservable;
import static shopJava.util.Constants.*;
import static shopJava.util.Util.average;
import static shopJava.util.Util.checkUserLoggedIn;
package shopJava.queries;




@SuppressWarnings("Convert2MethodRef")
public class QueryJ08RxStreamsWithObservables {

    public static void main(String[] args) throws Exception {
        new QueryJ08RxStreamsWithObservables();
    }

    private final DAO dao = new DAO();

    private class DAO {

        private final MongoCollection<Document> usersCollection;
        private final MongoCollection<Document> ordersCollection;

        DAO() {
            final MongoClient client = MongoClients.create(MONGODB_URI);
            final MongoDatabase db = client.getDatabase(SHOP_DB_NAME);
            this.usersCollection = db.getCollection(USERS_COLLECTION_NAME);
            this.ordersCollection = db.getCollection(ORDERS_COLLECTION_NAME);
        }

        Publisher<Document> findUserByName(final String name) {
            return usersCollection
                    .find(eq("_id", name))
                    .first();
        }

        Publisher<Document> findOrdersByUsername(final String username) {
            return ordersCollection
                    .find(eq("username", username));
        }
    }   // end DAO


    private Single<String> logIn(final Credentials credentials) {
        return toObservable(dao.findUserByName(credentials.username))
                .toSingle()
                .map(doc -> new User(doc))      // no null check as we don't get null objects in the stream
// Path: src/main/java/shopJava/util/Constants.java
// public interface Constants {
// 
//     int MAX_ORDERS_PER_USER = 10;
// 
//     String MONGODB_URI = "mongodb://localhost:27017";
//     String SHOP_DB_NAME = "shop";
//     String USERS_COLLECTION_NAME = "users";
//     String ORDERS_COLLECTION_NAME = "orders";
// 
//     String HOMER = "homer";
//     String MARGE = "marge";
//     String BART = "bart";
//     String LISA = "lisa";
//     String MAGGIE = "maggie";
// 
//     String[] SIMPSONS = {HOMER, MARGE, BART, LISA, MAGGIE};
// }
// 
// Path: src/main/java/shopJava/util/Util.java
// public static int average(final int totalAmount, final int orderCount) {
//     return orderCount == 0 ? 0 : Math.round((100.0f * totalAmount / orderCount) / 100);
// }
// 
// Path: src/main/java/shopJava/util/Util.java
// public static User checkUserLoggedIn(final Optional<User> optUser, final Credentials credentials) {
// 
//     if (!optUser.isPresent()) {     // replaces if (user != null)
//         throw new RuntimeException(new IllegalAccessException("User unknown: " + credentials.username));
//     }
//     final User user = optUser.get();
//     return checkUserLoggedIn(user, credentials);
// }


// Path: src/main/java/shopJava/queries/QueryJ08RxStreamsWithObservables.java
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import org.bson.Document;
import org.reactivestreams.Publisher;
import rx.Observable;
import rx.Single;
import shopJava.model.*;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import static com.mongodb.client.model.Filters.eq;
import static java.lang.Thread.sleep;
import static rx.RxReactiveStreams.toObservable;
import static shopJava.util.Constants.*;
import static shopJava.util.Util.average;
import static shopJava.util.Util.checkUserLoggedIn;

package shopJava.queries;




@SuppressWarnings("Convert2MethodRef")
public class QueryJ08RxStreamsWithObservables {

    public static void main(String[] args) throws Exception {
        new QueryJ08RxStreamsWithObservables();
    }

    private final DAO dao = new DAO();

    private class DAO {

        private final MongoCollection<Document> usersCollection;
        private final MongoCollection<Document> ordersCollection;

        DAO() {
            final MongoClient client = MongoClients.create(MONGODB_URI);
            final MongoDatabase db = client.getDatabase(SHOP_DB_NAME);
            this.usersCollection = db.getCollection(USERS_COLLECTION_NAME);
            this.ordersCollection = db.getCollection(ORDERS_COLLECTION_NAME);
        }

        Publisher<Document> findUserByName(final String name) {
            return usersCollection
                    .find(eq("_id", name))
                    .first();
        }

        Publisher<Document> findOrdersByUsername(final String username) {
            return ordersCollection
                    .find(eq("username", username));
        }
    }   // end DAO


    private Single<String> logIn(final Credentials credentials) {
        return toObservable(dao.findUserByName(credentials.username))
                .toSingle()
                .map(doc -> new User(doc))      // no null check as we don't get null objects in the stream
                .map(optUser -> checkUserLoggedIn(optUser, credentials))