/**
 *                                  Apache License
 *                            Version 2.0, January 2004
 *                         http://www.apache.org/licenses/
 *
 *    TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
 *
 *    1. Definitions.
 *
 *       "License" shall mean the terms and conditions for use, reproduction,
 *       and distribution as defined by Sections 1 through 9 of this document.
 *
 *       "Licensor" shall mean the copyright owner or entity authorized by
 *       the copyright owner that is granting the License.
 *
 *       "Legal Entity" shall mean the union of the acting entity and all
 *       other entities that control, are controlled by, or are under common
 *       control with that entity. For the purposes of this definition,
 *       "control" means (i) the power, direct or indirect, to cause the
 *       direction or management of such entity, whether by contract or
 *       otherwise, or (ii) ownership of fifty percent (50%) or more of the
 *       outstanding shares, or (iii) beneficial ownership of such entity.
 *
 *       "You" (or "Your") shall mean an individual or Legal Entity
 *       exercising permissions granted by this License.
 *
 *       "Source" form shall mean the preferred form for making modifications,
 *       including but not limited to software source code, documentation
 *       source, and configuration files.
 *
 *       "Object" form shall mean any form resulting from mechanical
 *       transformation or translation of a Source form, including but
 *       not limited to compiled object code, generated documentation,
 *       and conversions to other media types.
 *
 *       "Work" shall mean the work of authorship, whether in Source or
 *       Object form, made available under the License, as indicated by a
 *       copyright notice that is included in or attached to the work
 *       (an example is provided in the Appendix below).
 *
 *       "Derivative Works" shall mean any work, whether in Source or Object
 *       form, that is based on (or derived from) the Work and for which the
 *       editorial revisions, annotations, elaborations, or other modifications
 *       represent, as a whole, an original work of authorship. For the purposes
 *       of this License, Derivative Works shall not include works that remain
 *       separable from, or merely link (or bind by name) to the interfaces of,
 *       the Work and Derivative Works thereof.
 *
 *       "Contribution" shall mean any work of authorship, including
 *       the original version of the Work and any modifications or additions
 *       to that Work or Derivative Works thereof, that is intentionally
 *       submitted to Licensor for inclusion in the Work by the copyright owner
 *       or by an individual or Legal Entity authorized to submit on behalf of
 *       the copyright owner. For the purposes of this definition, "submitted"
 *       means any form of electronic, verbal, or written communication sent
 *       to the Licensor or its representatives, including but not limited to
 *       communication on electronic mailing lists, source code control systems,
 *       and issue tracking systems that are managed by, or on behalf of, the
 *       Licensor for the purpose of discussing and improving the Work, but
 *       excluding communication that is conspicuously marked or otherwise
 *       designated in writing by the copyright owner as "Not a Contribution."
 *
 *       "Contributor" shall mean Licensor and any individual or Legal Entity
 *       on behalf of whom a Contribution has been received by Licensor and
 *       subsequently incorporated within the Work.
 *
 *    2. Grant of Copyright License. Subject to the terms and conditions of
 *       this License, each Contributor hereby grants to You a perpetual,
 *       worldwide, non-exclusive, no-charge, royalty-free, irrevocable
 *       copyright license to reproduce, prepare Derivative Works of,
 *       publicly display, publicly perform, sublicense, and distribute the
 *       Work and such Derivative Works in Source or Object form.
 *
 *    3. Grant of Patent License. Subject to the terms and conditions of
 *       this License, each Contributor hereby grants to You a perpetual,
 *       worldwide, non-exclusive, no-charge, royalty-free, irrevocable
 *       (except as stated in this section) patent license to make, have made,
 *       use, offer to sell, sell, import, and otherwise transfer the Work,
 *       where such license applies only to those patent claims licensable
 *       by such Contributor that are necessarily infringed by their
 *       Contribution(s) alone or by combination of their Contribution(s)
 *       with the Work to which such Contribution(s) was submitted. If You
 *       institute patent litigation against any entity (including a
 *       cross-claim or counterclaim in a lawsuit) alleging that the Work
 *       or a Contribution incorporated within the Work constitutes direct
 *       or contributory patent infringement, then any patent licenses
 *       granted to You under this License for that Work shall terminate
 *       as of the date such litigation is filed.
 *
 *    4. Redistribution. You may reproduce and distribute copies of the
 *       Work or Derivative Works thereof in any medium, with or without
 *       modifications, and in Source or Object form, provided that You
 *       meet the following conditions:
 *
 *       (a) You must give any other recipients of the Work or
 *           Derivative Works a copy of this License; and
 *
 *       (b) You must cause any modified files to carry prominent notices
 *           stating that You changed the files; and
 *
 *       (c) You must retain, in the Source form of any Derivative Works
 *           that You distribute, all copyright, patent, trademark, and
 *           attribution notices from the Source form of the Work,
 *           excluding those notices that do not pertain to any part of
 *           the Derivative Works; and
 *
 *       (d) If the Work includes a "NOTICE" text file as part of its
 *           distribution, then any Derivative Works that You distribute must
 *           include a readable copy of the attribution notices contained
 *           within such NOTICE file, excluding those notices that do not
 *           pertain to any part of the Derivative Works, in at least one
 *           of the following places: within a NOTICE text file distributed
 *           as part of the Derivative Works; within the Source form or
 *           documentation, if provided along with the Derivative Works; or,
 *           within a display generated by the Derivative Works, if and
 *           wherever such third-party notices normally appear. The contents
 *           of the NOTICE file are for informational purposes only and
 *           do not modify the License. You may add Your own attribution
 *           notices within Derivative Works that You distribute, alongside
 *           or as an addendum to the NOTICE text from the Work, provided
 *           that such additional attribution notices cannot be construed
 *           as modifying the License.
 *
 *       You may add Your own copyright statement to Your modifications and
 *       may provide additional or different license terms and conditions
 *       for use, reproduction, or distribution of Your modifications, or
 *       for any such Derivative Works as a whole, provided Your use,
 *       reproduction, and distribution of the Work otherwise complies with
 *       the conditions stated in this License.
 *
 *    5. Submission of Contributions. Unless You explicitly state otherwise,
 *       any Contribution intentionally submitted for inclusion in the Work
 *       by You to the Licensor shall be under the terms and conditions of
 *       this License, without any additional terms or conditions.
 *       Notwithstanding the above, nothing herein shall supersede or modify
 *       the terms of any separate license agreement you may have executed
 *       with Licensor regarding such Contributions.
 *
 *    6. Trademarks. This License does not grant permission to use the trade
 *       names, trademarks, service marks, or product names of the Licensor,
 *       except as required for reasonable and customary use in describing the
 *       origin of the Work and reproducing the content of the NOTICE file.
 *
 *    7. Disclaimer of Warranty. Unless required by applicable law or
 *       agreed to in writing, Licensor provides the Work (and each
 *       Contributor provides its Contributions) on an "AS IS" BASIS,
 *       WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 *       implied, including, without limitation, any warranties or conditions
 *       of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
 *       PARTICULAR PURPOSE. You are solely responsible for determining the
 *       appropriateness of using or redistributing the Work and assume any
 *       risks associated with Your exercise of permissions under this License.
 *
 *    8. Limitation of Liability. In no event and under no legal theory,
 *       whether in tort (including negligence), contract, or otherwise,
 *       unless required by applicable law (such as deliberate and grossly
 *       negligent acts) or agreed to in writing, shall any Contributor be
 *       liable to You for damages, including any direct, indirect, special,
 *       incidental, or consequential damages of any character arising as a
 *       result of this License or out of the use or inability to use the
 *       Work (including but not limited to damages for loss of goodwill,
 *       work stoppage, computer failure or malfunction, or any and all
 *       other commercial damages or losses), even if such Contributor
 *       has been advised of the possibility of such damages.
 *
 *    9. Accepting Warranty or Additional Liability. While redistributing
 *       the Work or Derivative Works thereof, You may choose to offer,
 *       and charge a fee for, acceptance of support, warranty, indemnity,
 *       or other liability obligations and/or rights consistent with this
 *       License. However, in accepting such obligations, You may act only
 *       on Your own behalf and on Your sole responsibility, not on behalf
 *       of any other Contributor, and only if You agree to indemnify,
 *       defend, and hold each Contributor harmless for any liability
 *       incurred by, or claims asserted against, such Contributor by reason
 *       of your accepting any such warranty or additional liability.
 *
 *    END OF TERMS AND CONDITIONS
 *
 *    APPENDIX: How to apply the Apache License to your work.
 *
 *       To apply the Apache License to your work, attach the following
 *       boilerplate notice, with the fields enclosed by brackets "{}"
 *       replaced with your own identifying information. (Don't include
 *       the brackets!)  The text should be enclosed in the appropriate
 *       comment syntax for the file format. We also recommend that a
 *       file or class name and description of purpose be included on the
 *       same "printed page" as the copyright notice for easier
 *       identification within third-party archives.
 *
 *    Copyright 2014 Edgar Espina
 *
 *    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 org.jooby.mongodb;

import com.google.inject.Binder;
import com.google.inject.Key;
import com.google.inject.name.Names;
import com.mongodb.ConnectionString;
import com.mongodb.ReadConcern;
import com.mongodb.ReadPreference;
import com.mongodb.WriteConcern;
import com.mongodb.async.client.MongoClientSettings;
import com.mongodb.connection.ClusterSettings;
import com.mongodb.connection.ClusterType;
import com.mongodb.connection.ConnectionPoolSettings;
import com.mongodb.connection.ServerSettings;
import com.mongodb.connection.SocketSettings;
import com.mongodb.connection.SslSettings;
import com.mongodb.rx.client.AggregateObservable;
import com.mongodb.rx.client.DistinctObservable;
import com.mongodb.rx.client.FindObservable;
import com.mongodb.rx.client.ListCollectionsObservable;
import com.mongodb.rx.client.ListDatabasesObservable;
import com.mongodb.rx.client.MapReduceObservable;
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 com.mongodb.rx.client.MongoObservable;
import com.mongodb.rx.client.ObservableAdapter;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.typesafe.config.ConfigValueFactory;
import static java.util.Objects.requireNonNull;
import org.bson.Document;
import org.bson.codecs.configuration.CodecRegistry;
import org.jooby.Env;
import org.jooby.Jooby.Module;
import org.jooby.Route;
import org.jooby.funzy.Throwing;
import org.jooby.funzy.Try;
import static org.jooby.funzy.When.when;
import org.jooby.rx.Rx;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;

import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * <h1>mongodb-rx</h1>
 * <p>
 * <a href="http://mongodb.github.io/mongo-java-driver-rx/">MongoDB RxJava Driver: </a> provides
 * composable asynchronous and event-based observable sequences for MongoDB.
 * </p>
 *
 * <p>
 * A MongoDB based driver providing support for <a href="http://reactivex.io">ReactiveX (Reactive
 * Extensions)</a> by using the <a href="https://github.com/ReactiveX/RxJava">RxJava library</a>.
 * All database calls return an
 * <a href="http://reactivex.io/documentation/observable.html">Observable</a> allowing for efficient
 * execution, concise code, and functional composition of results.
 * </p>
 *
 * <p>
 * This module depends on {@link Rx} module, please read the {@link Rx} documentation before using
 * this module.
 * </p>
 *
 * <h2>exports</h2>
 * <ul>
 * <li>{@link MongoClient}</li>
 * <li>{@link MongoDatabase} (when mongo connection string has a database)</li>
 * <li>{@link MongoCollection} (when mongo connection string has a collection)</li>
 * <li>{@link Route.Mapper} for mongo observables</li>
 * </ul>
 *
 * <h2>depends on</h2>
 * <ul>
 * <li>{@link Rx rx module}</li>
 * </ul>
 *
 * <h2>usage</h2>
 * <pre>{@code
 *
 * import org.jooby.mongodb.MongoRx;
 *
 * {
 *   // required by MongoRx
 *   use(new Rx());
 *
 *   use(new MongoRx());
 *
 *   get("/", req -> {
 *     MongoClient client = req.require(MongoClient.class);
 *     // work with client:
 *   });
 * }
 * }</pre>
 *
 * <p>
 * The <code>mongo-rx</code> module connects to <code>mongodb://localhost</code>. You can change the
 * connection string by setting the <code>db</code> property in your
 * <code>application.conf</code> file:
 * </p>
 *
 * <pre>{@code
 * db = "mongodb://localhost/mydb"
 * }</pre>
 *
 * <p>
 * Or at creation time:
 * </p>
 * <pre>{@code
 * {
 *   // required by MongoRx
 *   use(new Rx());
 *
 *   use(new MongoRx("mongodb://localhost/mydb"));
 * }
 * }</pre>
 *
 * <p>
 * If your connection string has a database, then you can require a {@link MongoDatabase} object:
 * </p>
 *
 * <pre>{@code
 * {
 *   // required by MongoRx
 *   use(new Rx());
 *
 *   use(new MongoRx("mongodb://localhost/mydb"));
 *
 *   get("/", req -> {
 *     MongoDatabase mydb = req.require(MongoDatabase.class);
 *     return mydb.listCollections();
 *   });
 * }
 * }</pre>
 *
 * <p>
 * And if your connection string has a collection:
 * </p>
 *
 * <pre>{@code
 * {
 *   // required by MongoRx
 *   use(new Rx());
 *
 *   use(new MongoRx("mongodb://localhost/mydb.mycol"));
 *
 *   get("/", req -> {
 *     MongoCollection mycol = req.require(MongoCollection.class);
 *     return mycol.find();
 *   });
 * }
 * }</pre>
 *
 * <h2>query the collection</h2>
 * <p>
 * The module let you return {@link MongoObservable} directly as route responses:
 * </p>
 *
 * <pre>{@code
 * {
 *   // required by MongoRx
 *   use(new Rx());
 *
 *   use(new MongoRx()
 *      .observableAdapter(observable -> observable.observeOn(Scheduler.io())));
 *
 *   get("/pets", req -> {
 *     MongoDatabase db = req.require(MongoDatabase.class);
 *     return db.getCollection("pets")
 *        .find();
 *   });
 * }
 * }</pre>
 *
 * <p>
 * Previous example will list all the <code>Pets</code> from a collection. Please note you don't
 * have to deal with {@link MongoObservable}, instead the module converts {@link MongoObservable} to
 * Jooby async semantics.
 * </p>
 *
 * <h2>multiple databases</h2>
 * <p>
 * Multiple databases are supported by adding multiple {@link MongoRx} instances to your
 * application:
 * </p>
 *
 * <pre>{@code
 * {
 *   // required by MongoRx
 *   use(new Rx());
 *
 *   use(new MongoRx("db1"));
 *
 *   use(new MongoRx("db2"));
 *
 *   get("/do-with-db1", req -> {
 *     MongoDatabase db1 = req.require("db1", MongoDatabase.class);
 *   });
 *
 *   get("/do-with-db2", req -> {
 *     MongoDatabase db2 = req.require("db2", MongoDatabase.class);
 *   });
 * }
 * }</pre>
 *
 * The keys <code>db1</code> and <code>db2</code> are connection strings in your
 * <code>application.conf</code>:
 *
 * <pre>{@code
 *   db1 = "mongodb://localhost/db1"
 *
 *   db2 = "mongodb://localhost/db2"
 * }</pre>
 *
 * <h2>observable adapter</h2>
 * <p>
 * {@link ObservableAdapter} provides a simple way to adapt all Observables returned by the driver.
 * On such use case might be to use a different Scheduler after returning the results from MongoDB
 * therefore freeing up the connection thread.
 * </p>
 *
 * <pre>{@code
 * {
 *   // required by MongoRx
 *   use(new Rx());
 *
 *   use(new MongoRx().observableAdapter(o -> o.observeOn(Schedulers.io())));
 * }
 * }</pre>
 *
 * <p>
 * Any computations on Observables returned by the {@link MongoDatabase} or {@link MongoCollection}
 * will use the IO scheduler, rather than blocking the MongoDB Connection thread.
 * </p>
 *
 * <p>
 * Please note the {@link #observableAdapter(Function)} works if (and only if) your connection
 * string points to a database. It won't work on <code>mongo://localhost</code> connection string
 * because there is no database in it.
 * </p>
 *
 * <h2>driver options</h2>
 * <p>
 * Driver options are available via
 * <a href="https://docs.mongodb.com/v3.0/reference/connection-string/">connection string</a>.
 * </p>
 *
 * <p>
 * It is also possible to configure specific options:
 * </p>
 *
 * <pre>{@code
 *
 * db = "mongodb://localhost/pets"
 *
 * mongo {
 *   readConcern: default
 *   writeConcern: ACKNOWLEDGED
 *   cluster {
 *     replicaSetName: name
 *     requiredClusterType: REPLICA_SET
 *   }
 *   pool {
 *     maxSize: 100
 *     minSize: 10
 *   }
 * }
 * }</pre>
 *
 * <p>
 * Each option matches a {@link MongoClientSettings} method.
 * </p>
 *
 *
 * @author edgar
 * @since 1.0.0.CR4
 */
public class MongoRx implements Module {

  private static final AtomicInteger instances = new AtomicInteger(0);

  /** The logging system. */
  private final Logger log = LoggerFactory.getLogger(getClass());

  private BiConsumer<MongoClientSettings.Builder, Config> configurer;

  private Optional<ObservableAdapter> adapter = Optional.empty();

  private Optional<CodecRegistry> codecRegistry = Optional.empty();

  private String db;

  /**
   * Creates a new {@link MongoRx} module.
   *
   * @param db A connection string or a property key.
   */
  public MongoRx(final String db) {
    this.db = requireNonNull(db, "Connection String/Database key is required.");
  }

  /**
   * Creates a new {@link MongoRx} module that connects to <code>localhost</code> unless you
   * define/override the <code>db</code> property in your <code>application.conf</code> file.
   */
  public MongoRx() {
    this("db");
  }

  /**
   * Allow further configuration on the {@link MongoClientSettings}.
   *
   * @param configurer Configurer callback.
   * @return This module.
   */
  public MongoRx doWith(final BiConsumer<MongoClientSettings.Builder, Config> configurer) {
    this.configurer = requireNonNull(configurer, "Configurer is required.");
    return this;
  }

  /**
   * Allow further configuration on the {@link MongoClientSettings}.
   *
   * @param configurer Configurer callback.
   * @return This module.
   */
  public MongoRx doWith(final Consumer<MongoClientSettings.Builder> configurer) {
    requireNonNull(configurer, "Configurer is required.");
    return doWith((s, c) -> configurer.accept(s));
  }

  /**
   * Set a {@link ObservableAdapter} to the {@link MongoDatabase} created by this module.
   *
   * @param adapter An {@link ObservableAdapter}.
   * @return This module.
   */
  @SuppressWarnings("rawtypes")
  public MongoRx observableAdapter(final Function<Observable, Observable> adapter) {
    this.adapter = toAdapter(requireNonNull(adapter, "Adapter is required."));
    return this;
  }

  /**
   * Set a {@link CodecRegistry} to the {@link MongoDatabase} created by this module.
   *
   * @param codecRegistry A codec registry.
   * @return This module.
   */
  public MongoRx codecRegistry(final CodecRegistry codecRegistry) {
    this.codecRegistry = Optional.of(codecRegistry);
    return this;
  }

  @Override
  public Config config() {
    return ConfigFactory.empty(MongoRx.class.getName())
        .withValue("db", ConfigValueFactory.fromAnyRef("mongodb://localhost"));
  }

  @SuppressWarnings({"rawtypes", "unchecked"})
  @Override
  public void configure(final Env env, final Config conf, final Binder binder) {
    /** connection string */
    ConnectionString cstr = Try.apply(() -> new ConnectionString(db))
        .orElseGet(() -> new ConnectionString(conf.getString(db)));

    log.debug("Starting {}", cstr);

    boolean first = instances.getAndIncrement() == 0;
    Throwing.Function3<Class, String, Object, Void> bind = (type, name, value) -> {
      binder.bind(Key.get(type, Names.named(name))).toInstance(value);
      if (first) {
        binder.bind(Key.get(type)).toInstance(value);
      }
      return null;
    };

    /** settings */
    MongoClientSettings.Builder settings = settings(cstr, dbconf(db, conf));
    if (configurer != null) {
      configurer.accept(settings, conf);
    }
    MongoClient client = MongoClients.create(settings.build());
    bind.apply(MongoClient.class, db, client);

    /** bind database */
    Optional.ofNullable(cstr.getDatabase()).ifPresent(dbname -> {
      // observable adapter
      MongoDatabase predb = adapter
          .map(a -> client.getDatabase(dbname).withObservableAdapter(a))
          .orElseGet(() -> client.getDatabase(dbname));
      // codec registry
      MongoDatabase database = codecRegistry
          .map(predb::withCodecRegistry)
          .orElse(predb);

      bind.apply(MongoDatabase.class, dbname, database);

      /** bind collection */
      Optional.ofNullable(cstr.getCollection()).ifPresent(cname -> {
        MongoCollection<Document> collection = database.getCollection(cname);
        bind.apply(MongoCollection.class, cname, collection);
      });
    });

    /** mapper */
    env.router()
        .map(mapper());

    log.info("Started {}", cstr);

    env.onStop(() -> {
      log.debug("Stopping {}", cstr);
      client.close();
      log.info("Stopped {}", cstr);
    });
  }

  @SuppressWarnings("rawtypes")
  static Route.Mapper mapper() {
    return Route.Mapper.create("mongo-rx", v -> {
      if (v instanceof FindObservable) {
        return ((FindObservable) v).toObservable().toList();
      } else if (v instanceof ListCollectionsObservable) {
        return ((ListCollectionsObservable) v).toObservable().toList();
      } else if (v instanceof ListDatabasesObservable) {
        return ((ListDatabasesObservable) v).toObservable().toList();
      } else if (v instanceof AggregateObservable) {
        return ((AggregateObservable) v).toObservable().toList();
      } else if (v instanceof DistinctObservable) {
        return ((DistinctObservable) v).toObservable().toList();
      } else if (v instanceof MapReduceObservable) {
        return ((MapReduceObservable) v).toObservable().toList();
      } else if (v instanceof MongoObservable) {
        return ((MongoObservable) v).toObservable();
      }
      return v;
    });
  }

  static MongoClientSettings.Builder settings(final ConnectionString cstr, final Config conf) {
    MongoClientSettings.Builder settings = MongoClientSettings.builder();

    settings.clusterSettings(cluster(cstr, conf));
    settings.connectionPoolSettings(pool(cstr, conf));
    settings.heartbeatSocketSettings(socket("heartbeat", cstr, conf));

    withStr("readConcern", conf, v -> settings.readConcern(readConcern(v)));

    withStr("readPreference", conf,
        v -> settings.readPreference(ReadPreference.valueOf(v)));

    settings.serverSettings(server(conf));
    settings.socketSettings(socket("socket", cstr, conf));
    settings.sslSettings(ssl(cstr, conf));

    withStr("writeConcern", conf, v -> settings.writeConcern(writeConcern(v)));

    return settings;
  }

  private static ReadConcern readConcern(final String value) {
    return when(value.toUpperCase())
        .is("DEFAULT", ReadConcern.DEFAULT)
        .is("LOCAL", ReadConcern.LOCAL)
        .is("MAJORITY", ReadConcern.MAJORITY)
        .orElseThrow(() -> new IllegalArgumentException("readConcern: " + value));
  }

  private static WriteConcern writeConcern(final String value) {
    return when(value.toUpperCase())
        .is("W1", WriteConcern.W1)
        .is("W2", WriteConcern.W2)
        .is("W3", WriteConcern.W3)
        .is("ACKNOWLEDGED", WriteConcern.ACKNOWLEDGED)
        .is("MAJORITY", WriteConcern.MAJORITY)
        .orElseThrow(() -> new IllegalArgumentException("writeConcern: " + value));
  }

  static SslSettings ssl(final ConnectionString cstr, final Config conf) {
    SslSettings.Builder ssl = SslSettings.builder().applyConnectionString(cstr);
    withConf("ssl", conf, c -> {
      withBool("enabled", c, ssl::enabled);
      withBool("invalidHostNameAllowed", c, ssl::invalidHostNameAllowed);
    });
    return ssl.build();
  }

  static ServerSettings server(final Config dbconf) {
    ServerSettings.Builder server = ServerSettings.builder();
    withConf("server", dbconf, c -> {
      withMs("heartbeatFrequency", c,
          s -> server.heartbeatFrequency(s.intValue(), TimeUnit.MILLISECONDS));
      withMs("minHeartbeatFrequency", c,
          s -> server.minHeartbeatFrequency(s.intValue(), TimeUnit.MILLISECONDS));
    });
    return server.build();
  }

  static SocketSettings socket(final String path, final ConnectionString cstr,
      final Config dbconf) {
    SocketSettings.Builder settings = SocketSettings.builder().applyConnectionString(cstr);
    withConf(path, dbconf, c -> {
      withMs("connectTimeout", c,
          s -> settings.connectTimeout(s.intValue(), TimeUnit.MILLISECONDS));
      withBool("keepAlive", c, settings::keepAlive);
      withMs("readTimeout", c,
          s -> settings.readTimeout(s.intValue(), TimeUnit.MILLISECONDS));
      withInt("receiveBufferSize", c, settings::receiveBufferSize);
      withInt("sendBufferSize", c, settings::sendBufferSize);
    });
    return settings.build();
  }

  static ClusterSettings cluster(final ConnectionString cstr, final Config conf) {
    ClusterSettings.Builder cluster = ClusterSettings.builder().applyConnectionString(cstr);
    withConf("cluster", conf, c -> {
      withInt("maxWaitQueueSize", c, cluster::maxWaitQueueSize);
      withStr("replicaSetName", c, cluster::requiredReplicaSetName);
      withStr("requiredClusterType", c,
          v -> cluster.requiredClusterType(ClusterType.valueOf(v.toUpperCase())));
      withMs("serverSelectionTimeout", c,
          s -> cluster.serverSelectionTimeout(s, TimeUnit.MILLISECONDS));
    });
    return cluster.build();
  }

  static ConnectionPoolSettings pool(final ConnectionString cstr, final Config conf) {
    ConnectionPoolSettings.Builder pool = ConnectionPoolSettings.builder()
        .applyConnectionString(cstr);
    withConf("pool", conf, c -> {
      withMs("maintenanceFrequency", c,
          s -> pool.maintenanceFrequency(s, TimeUnit.MILLISECONDS));
      withMs("maintenanceInitialDelay", c,
          s -> pool.maintenanceInitialDelay(s, TimeUnit.MILLISECONDS));
      withMs("maxConnectionIdleTime", c,
          s -> pool.maxConnectionIdleTime(s, TimeUnit.MILLISECONDS));
      withMs("maxConnectionLifeTime", c,
          s -> pool.maxConnectionLifeTime(s, TimeUnit.MILLISECONDS));
      withInt("maxSize", c, pool::maxSize);
      withInt("maxWaitQueueSize", c, pool::maxWaitQueueSize);
      withMs("maxWaitTime", c,
          s -> pool.maxWaitTime(s, TimeUnit.MILLISECONDS));
      withInt("minSize", c, pool::minSize);
    });
    return pool.build();
  }

  static Config dbconf(final String db, final Config conf) {
    Function<String, Config> ifconf = path -> {
      if (Try.apply(() -> conf.hasPath(path)).orElse(Boolean.FALSE)) {
        return conf.getConfig(path);
      }
      return ConfigFactory.empty();
    };

    // mongdo.db.* < mongo.*
    return ifconf.apply("mongo." + db)
        .withFallback(ifconf.apply("mongo"));
  }

  static <T> void withMs(final String path, final Config conf,
      final Consumer<Long> callback) {
    withPath(path, conf, callback, () -> conf.getDuration(path, TimeUnit.MILLISECONDS));
  }

  static <T> void withInt(final String path, final Config conf,
      final Consumer<Integer> callback) {
    withPath(path, conf, callback, () -> conf.getInt(path));
  }

  static <T> void withStr(final String path, final Config conf,
      final Consumer<String> callback) {
    withPath(path, conf, callback, () -> conf.getString(path));
  }

  static <T> void withBool(final String path, final Config conf,
      final Consumer<Boolean> callback) {
    withPath(path, conf, callback, () -> conf.getBoolean(path));
  }

  static <T> void withConf(final String path, final Config conf,
      final Consumer<Config> callback) {
    withPath(path, conf, callback, () -> conf.getConfig(path));
  }

  static <T> void withPath(final String path, final Config conf, final Consumer<T> callback,
      final Supplier<T> value) {
    if (conf.hasPath(path)) {
      callback.accept(value.get());
    }
  }

  @SuppressWarnings("rawtypes")
  private static Optional<ObservableAdapter> toAdapter(final Function<Observable, Observable> fn) {
    return Optional.of(new ObservableAdapter() {

      @SuppressWarnings("unchecked")
      @Override
      public <T> Observable<T> adapt(final Observable<T> observable) {
        return fn.apply(observable);
      }
    });
  }

}
