Try to integrate libraries

Author: syucream

.scalafmt.conf

@@ -0,0 +1,13 @@
+version=2.3.2
+project.git = true
+project.excludeFilters = [
+  scalafmt-benchmarks/src/resources,
+  sbt-test
+  bin/issue
+]
+align = none
+# Disabled in default since this operation is potentially
+# dangerous if you define your own stripMargin with different
+# semantics from the stdlib stripMargin.
+assumeStandardLibraryStripMargin = true
+onTestFailure = "To fix this, run ./scalafmt from the project root directory"

build.gradle

@@ -2,6 +2,7 @@ plugins {
     id 'scala'
     id 'com.github.jruby-gradle.base' version '1.6.0'
     id 'com.github.johnrengelman.shadow' version '5.2.0'
+    id 'cz.alenkacz.gradle.scalafmt' version '1.10.0'
 }
 import com.github.jrubygradle.JRubyExec
 
@@ -11,17 +12,24 @@ configurations {
 
 repositories {
     mavenCentral()
+    mavenLocal()
     jcenter()
 }
 
 // Relocate Guava packages since it's incompatible with Guava's version from Embulk
 shadowJar {
     classifier 'shadow'
-    exclude 'org/embulk/plugin/**'
 
     dependencies {
-        include 'com.google.guava:guava'
-        include 'com.google.cloud:google-cloud-firestore'
+        include dependency('com.google.guava:guava')
+
+        include dependency('com.google.cloud:google-cloud-firestore')
+        include dependency('com.google.firebase:firebase-admin')
+        include dependency('com.syucream:firesql')
+
+        include dependency('io.grpc:grpc-netty-shaded')
+        include dependency('io.grpc:grpc-core')
+        include dependency('io.grpc:grpc-api')
     }
 
     relocate 'com.google.common', 'relocated.com.google.common'
@@ -36,7 +44,9 @@ dependencies {
     compile 'org.embulk:embulk-core:0.9.12'
     provided 'org.embulk:embulk-core:0.9.12'
 
-    compile 'com.google.cloud:google-cloud-firestore:1.28.0'
+    // Should be shadowed
+    compile 'com.google.firebase:firebase-admin:6.11.0'
+    compile 'com.syucream:firesql:0.0.1'
 }
 
 task classpath(type: Copy, dependsOn: ['jar', 'shadowJar']) {
@@ -45,8 +55,8 @@ task classpath(type: Copy, dependsOn: ['jar', 'shadowJar']) {
     from (configurations.runtime
             - configurations.provided
             + configurations.shadow
-            + files(shadowJar.archiveFile)
-            + files(jar.archiveFile))
+            - files(shadowJar.getIncludedDependencies())
+            + files(shadowJar.archiveFile))
 
     into 'classpath'
 }

examples/firestore2stdout.yaml

@@ -2,7 +2,7 @@ in:
   type: firestore
   project_id: "syucream-firebase-dev"
   json_keyfile: credential.json
-  collection: users
+  sql: "SELECT name FROM users"
 
 out:
   type: stdout

src/main/scala/org/embulk/input/firestore/FirestoreInputPlugin.scala

@@ -1,27 +1,99 @@
 package org.embulk.input.firestore
 
+import java.io.FileInputStream
 import java.util
 
+import com.google.auth.oauth2.GoogleCredentials
+import com.google.cloud.firestore.Firestore
+import com.google.firebase.cloud.FirestoreClient
+import com.google.firebase.{FirebaseApp, FirebaseOptions}
+import com.syucream.firesql.FireSQL
 import org.embulk.config.{ConfigDiff, ConfigSource, TaskReport, TaskSource}
+import org.embulk.spi.`type`.Types
+import org.embulk.spi.PageBuilder
+import org.embulk.spi.json.JsonParser
 import org.embulk.spi.{Exec, InputPlugin, PageOutput, Schema}
 
+import scala.util.Try
+
 case class FirestoreInputPlugin() extends InputPlugin {
+  private val TASK_COUNT = 1
+
+  private val jsonParser = new JsonParser()
+
+  override def transaction(
+      config: ConfigSource,
+      control: InputPlugin.Control
+  ): ConfigDiff = {
+    val task = config.loadConfig(classOf[PluginTask])
+
+    // TODO runtime validations
 
-  override def transaction(config: ConfigSource, control: InputPlugin.Control): ConfigDiff = {
-    null
+    val schema = Schema
+      .builder()
+      .add(task.getJsonColumnName, Types.JSON)
+      .build()
+
+    resume(task.dump(), schema, TASK_COUNT, control)
   }
 
-  override def resume(taskSource: TaskSource, schema: Schema, taskCount: Int, control: InputPlugin.Control): ConfigDiff = {
-    null
+  override def resume(
+      taskSource: TaskSource,
+      schema: Schema,
+      taskCount: Int,
+      control: InputPlugin.Control
+  ): ConfigDiff = {
+    // XXX unimplemented
+    control.run(taskSource, schema, taskCount)
+    Exec.newConfigDiff()
   }
 
-  override def cleanup(taskSource: TaskSource, schema: Schema, taskCount: Int, successTaskReports: util.List[TaskReport]): Unit = {
+  override def cleanup(
+      taskSource: TaskSource,
+      schema: Schema,
+      taskCount: Int,
+      successTaskReports: util.List[TaskReport]
+  ): Unit = {
+    // nothing to do
   }
 
-  override def run(taskSource: TaskSource, schema: Schema, taskIndex: Int, output: PageOutput): TaskReport = {
-    null
+  override def run(
+      taskSource: TaskSource,
+      schema: Schema,
+      taskIndex: Int,
+      output: PageOutput
+  ): TaskReport = {
+    val task = taskSource.loadTask(classOf[PluginTask])
+    val allocator = task.getBufferAllocator
+    val pageBuilder = new PageBuilder(allocator, schema, output)
+
+    val firestore = createFirestore(task.getJsonKeyfile).get
+    val firesql = new FireSQL(firestore)
+    val query = firesql.query(task.getSql)
+
+    val col = pageBuilder.getSchema.getColumn(0)
+
+    query.get.getDocuments.forEach { d =>
+      pageBuilder.setJson(col, jsonParser.parse(d.toString))
+      pageBuilder.addRecord()
+    }
+    pageBuilder.finish()
+
+    Exec.newTaskReport()
   }
 
   override def guess(config: ConfigSource): ConfigDiff =
     Exec.newConfigDiff()
+
+  private def createFirestore(pathToCredJson: String): Try[Firestore] = {
+    Try {
+      val serviceAccount = new FileInputStream(pathToCredJson)
+      val credentials = GoogleCredentials.fromStream(serviceAccount)
+      val options =
+        new FirebaseOptions.Builder().setCredentials(credentials).build
+      FirebaseApp.initializeApp(options)
+
+      FirestoreClient.getFirestore
+    }
+  }
 }

src/main/scala/org/embulk/input/firestore/PluginTask.scala

@@ -0,0 +1,23 @@
+package org.embulk.input.firestore
+
+import org.embulk.config.{Config, ConfigDefault, ConfigInject, Task}
+import org.embulk.spi.BufferAllocator
+
+trait PluginTask extends Task {
+
+  @Config("project_id")
+  def getProjectId: String
+
+  @Config("json_keyfile")
+  def getJsonKeyfile: String
+
+  @Config("json_column_name")
+  @ConfigDefault("\"record\"")
+  def getJsonColumnName: String
+
+  @Config("sql")
+  def getSql: String
+
+  @ConfigInject
+  def getBufferAllocator: BufferAllocator
+}