/*
 * Copyright © 2015 Cask Data, Inc.
 *
 * 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 co.cask.cdap.data.runtime;

import co.cask.cdap.api.dataset.module.DatasetModule;
import co.cask.cdap.common.runtime.RuntimeModule;
import co.cask.cdap.data2.dataset2.DatasetFramework;
import co.cask.cdap.data2.dataset2.lib.file.FileSetModule;
import co.cask.cdap.data2.dataset2.lib.partitioned.PartitionedFileSetModule;
import co.cask.cdap.data2.dataset2.lib.partitioned.TimePartitionedFileSetModule;
import co.cask.cdap.data2.dataset2.lib.table.CoreDatasetsModule;
import co.cask.cdap.data2.dataset2.lib.table.ObjectMappedTableModule;
import co.cask.cdap.data2.dataset2.module.lib.hbase.HBaseMetricsTableModule;
import co.cask.cdap.data2.dataset2.module.lib.hbase.HBaseTableModule;
import co.cask.cdap.data2.dataset2.module.lib.inmemory.InMemoryMetricsTableModule;
import co.cask.cdap.data2.dataset2.module.lib.inmemory.InMemoryTableModule;
import co.cask.cdap.data2.dataset2.module.lib.leveldb.LevelDBMetricsTableModule;
import co.cask.cdap.data2.dataset2.module.lib.leveldb.LevelDBTableModule;
import com.google.common.collect.Maps;
import com.google.inject.AbstractModule;
import com.google.inject.Module;
import com.google.inject.multibindings.MapBinder;
import com.google.inject.name.Names;

import java.util.Map;

/**
 * Provides guice bindings for {@link DatasetModule} that are by default available in the system.
 * The guice modules provided by this class are not intended to be used directly, but rather for providing
 * injections to {@link DatasetFramework}, hence for installing from other guice module.
 * This class is separated out so that combining different {@link DatasetFramework} and {@link DatasetModule} is
 * easier, especially for unit-test.
 */
public class DefaultDatasetRuntimeModule extends RuntimeModule {

  @Override
  public Module getInMemoryModules() {
    return new AbstractModule() {
      @Override
      protected void configure() {
        MapBinder<String, DatasetModule> mapBinder = MapBinder.newMapBinder(binder(), String.class, DatasetModule.class,
                                                                            Names.named("defaultDatasetModules"));
        // NOTE: order is important due to dependencies between modules
        mapBinder.addBinding("orderedTable-memory").toInstance(new InMemoryTableModule());
        mapBinder.addBinding("metricsTable-memory").toInstance(new InMemoryMetricsTableModule());
        mapBinder.addBinding("core").toInstance(new CoreDatasetsModule());
        mapBinder.addBinding("fileSet").toInstance(new FileSetModule());
        mapBinder.addBinding("timePartitionedFileSet").toInstance(new TimePartitionedFileSetModule());
        mapBinder.addBinding("partitionedFileSet").toInstance(new PartitionedFileSetModule());
        mapBinder.addBinding("objectMappedTable").toInstance(new ObjectMappedTableModule());
      }
    };
  }

  @Override
  public Module getStandaloneModules() {
    return new AbstractModule() {
      @Override
      protected void configure() {
        MapBinder<String, DatasetModule> mapBinder = MapBinder.newMapBinder(binder(), String.class, DatasetModule.class,
                                                                            Names.named("defaultDatasetModules"));
        // NOTE: order is important due to dependencies between modules
        mapBinder.addBinding("orderedTable-leveldb").toInstance(new LevelDBTableModule());
        mapBinder.addBinding("metricsTable-leveldb").toInstance(new LevelDBMetricsTableModule());
        mapBinder.addBinding("core").toInstance(new CoreDatasetsModule());
        mapBinder.addBinding("fileSet").toInstance(new FileSetModule());
        mapBinder.addBinding("timePartitionedFileSet").toInstance(new TimePartitionedFileSetModule());
        mapBinder.addBinding("partitionedFileSet").toInstance(new PartitionedFileSetModule());
        mapBinder.addBinding("objectMappedTable").toInstance(new ObjectMappedTableModule());
      }
    };
  }

  @Override
  public Module getDistributedModules() {
    return new AbstractModule() {
      @Override
      protected void configure() {
        MapBinder<String, DatasetModule> mapBinder = MapBinder.newMapBinder(binder(), String.class, DatasetModule.class,
                                                                            Names.named("defaultDatasetModules"));

        // NOTE: order is important due to dependencies between modules
        Map<String, DatasetModule> defaultModules = Maps.newLinkedHashMap();
        mapBinder.addBinding("orderedTable-hbase").toInstance(new HBaseTableModule());
        mapBinder.addBinding("metricsTable-hbase").toInstance(new HBaseMetricsTableModule());
        mapBinder.addBinding("core").toInstance(new CoreDatasetsModule());
        mapBinder.addBinding("fileSet").toInstance(new FileSetModule());
        mapBinder.addBinding("timePartitionedFileSet").toInstance(new TimePartitionedFileSetModule());
        mapBinder.addBinding("partitionedFileSet").toInstance(new PartitionedFileSetModule());
        mapBinder.addBinding("objectMappedTable").toInstance(new ObjectMappedTableModule());
      }
    };
  }
}

@@ -1601,47 +1601,8 @@
 le;%0A
-import com.google.common.collect.Maps;%0A
 impo
@@ -1763,31 +1763,8 @@
 s;%0A%0A
-import java.util.Map;%0A%0A
 /**%0A
@@ -2922,436 +2922,36 @@
     
-mapBinder.addBinding(%22core%22).toInstance(new CoreDatasetsModule());%0A        mapBinder.addBinding(%22fileSet%22).toInstance(new FileSetModule());%0A        mapBinder.addBinding(%22timePartitionedFileSet%22).toInstance(new TimePartitionedFileSetModule());%0A        mapBinder.addBinding(%22partitionedFileSet%22).toInstance(new PartitionedFileSetModule());%0A        mapBinder.addBinding(%22objectMappedTable%22).toInstance(new ObjectMappedTableModule()
+bindDefaultModules(mapBinder
 );%0A 
@@ -3616,436 +3616,36 @@
     
-mapBinder.addBinding(%22core%22).toInstance(new CoreDatasetsModule());%0A        mapBinder.addBinding(%22fileSet%22).toInstance(new FileSetModule());%0A        mapBinder.addBinding(%22timePartitionedFileSet%22).toInstance(new TimePartitionedFileSetModule());%0A        mapBinder.addBinding(%22partitionedFileSet%22).toInstance(new PartitionedFileSetModule());%0A        mapBinder.addBinding(%22objectMappedTable%22).toInstance(new ObjectMappedTableModule()
+bindDefaultModules(mapBinder
 );%0A 
@@ -4115,85 +4115,8 @@
 les%0A
-        Map%3CString, DatasetModule%3E defaultModules = Maps.newLinkedHashMap();%0A
     
@@ -4288,32 +4288,247 @@
 eModule());%0A    
+    bindDefaultModules(mapBinder);%0A      %7D%0A    %7D;%0A  %7D%0A%0A  /**%0A   * Add bindings for Dataset modules that are available by default%0A   */%0A  private void bindDefaultModules(MapBinder%3CString, DatasetModule%3E mapBinder) %7B%0A
     mapBinder.ad
@@ -4574,36 +4574,32 @@
 asetsModule());%0A
-    
     mapBinder.ad
@@ -4643,36 +4643,32 @@
 leSetModule());%0A
-    
     mapBinder.ad
@@ -4742,36 +4742,32 @@
 leSetModule());%0A
-    
     mapBinder.ad
@@ -4837,36 +4837,32 @@
 tModule());%0A    
-   
- 
 mapBinder.addBin
@@ -4930,29 +4930,15 @@
 ule());%0A
-      %7D%0A    %7D;
 %0A  %7D%0A%7D%0A

Refactor to gather common dataset modules setup