// Path: src/main/java/shopJava/model/User.java
// public class User {
// 
//     private static final String ID = "_id";
//     private static final String NAME = "name";
//     private static final String PASSWORD = "password";
// 
//     public final String name;
//     public final String password;
// 
//     public User(final String name, final String password) {
//         this.name = name;
//         this.password = password;
//     }
// 
//     public User(Document doc) {
//         this(doc.getString(ID), doc.getString(PASSWORD));
//     }
// 
//     @Override
//     public String toString() {
//         return "User{" +
//                 NAME + "='" + name + '\'' +
//                 ", " + PASSWORD + "='" + password + '\'' +
//                 '}';
//     }
// 
//     public void print() {
//         System.out.print(this);
//     }
// 
//     public void println() {
//         System.out.println(this);
//     }
// 
//     public Document toDocument() {
//         return new Document(ID, name).append(PASSWORD, password);
//     }
// }
// 
// Path: src/main/java/shopJava/util/Util.java
// public static void sleep(final int seconds) {
//     try {
//         Thread.sleep(seconds * 1000L);
//     } catch (InterruptedException e) {
//         e.printStackTrace();
//     }
// }


import com.mongodb.rx.client.MongoClient;
import com.mongodb.rx.client.MongoClients;
import com.mongodb.rx.client.MongoCollection;
import com.mongodb.rx.client.MongoDatabase;
import org.bson.Document;
import rx.Observable;
import rx.Observer;
import shopJava.model.User;
import static shopJava.util.Constants.MONGODB_URI;
import static shopJava.util.Constants.SHOP_DB_NAME;
import static shopJava.util.Constants.USERS_COLLECTION_NAME;
import static shopJava.util.Util.sleep;
    private Streams() {

        Observable<String> obsSimpsons = nonblockingIOWithStreams_GetDataFromDB();

        obsSimpsons.subscribe(new Observer<String>() {

            @Override
            public void onNext(String simpson) {
                System.out.println(simpson);
            }

            @Override
            public void onCompleted() {
                System.out.println("----- DONE -----");
            }

            @Override
            public void onError(Throwable t) {
                t.printStackTrace();
            }
        });

        sleep(2);
    }

    private Observable<String> nonblockingIOWithStreams_GetDataFromDB() {

        return usersCollection
                .find()
                .toObservable()
// Path: src/main/java/shopJava/model/User.java
// public class User {
// 
//     private static final String ID = "_id";
//     private static final String NAME = "name";
//     private static final String PASSWORD = "password";
// 
//     public final String name;
//     public final String password;
// 
//     public User(final String name, final String password) {
//         this.name = name;
//         this.password = password;
//     }
// 
//     public User(Document doc) {
//         this(doc.getString(ID), doc.getString(PASSWORD));
//     }
// 
//     @Override
//     public String toString() {
//         return "User{" +
//                 NAME + "='" + name + '\'' +
//                 ", " + PASSWORD + "='" + password + '\'' +
//                 '}';
//     }
// 
//     public void print() {
//         System.out.print(this);
//     }
// 
//     public void println() {
//         System.out.println(this);
//     }
// 
//     public Document toDocument() {
//         return new Document(ID, name).append(PASSWORD, password);
//     }
// }
// 
// Path: src/main/java/shopJava/util/Util.java
// public static void sleep(final int seconds) {
//     try {
//         Thread.sleep(seconds * 1000L);
//     } catch (InterruptedException e) {
//         e.printStackTrace();
//     }
// }


// Path: src/main/java/shopJava/async_reactive/Streams.java
import com.mongodb.rx.client.MongoClient;
import com.mongodb.rx.client.MongoClients;
import com.mongodb.rx.client.MongoCollection;
import com.mongodb.rx.client.MongoDatabase;
import org.bson.Document;
import rx.Observable;
import rx.Observer;
import shopJava.model.User;
import static shopJava.util.Constants.MONGODB_URI;
import static shopJava.util.Constants.SHOP_DB_NAME;
import static shopJava.util.Constants.USERS_COLLECTION_NAME;
import static shopJava.util.Util.sleep;

    private Streams() {

        Observable<String> obsSimpsons = nonblockingIOWithStreams_GetDataFromDB();

        obsSimpsons.subscribe(new Observer<String>() {

            @Override
            public void onNext(String simpson) {
                System.out.println(simpson);
            }

            @Override
            public void onCompleted() {
                System.out.println("----- DONE -----");
            }

            @Override
            public void onError(Throwable t) {
                t.printStackTrace();
            }
        });

        sleep(2);
    }

    private Observable<String> nonblockingIOWithStreams_GetDataFromDB() {

        return usersCollection
                .find()
                .toObservable()
                .map(doc -> new User(doc))