DatastoreInputPlugin.kt in embulk-input-datastore-0.1.2

- old
+ new

@@ -6,21 +6,27 @@
 import org.embulk.config.TaskReport
 import org.embulk.config.ConfigDiff
 import org.embulk.config.ConfigSource
 import org.embulk.config.TaskSource
 import org.embulk.spi.*
+import org.embulk.spi.json.JsonParser
 import org.embulk.spi.type.Types
 import org.msgpack.value.ValueFactory
 import java.io.FileInputStream
 import java.util.Base64
 
 class DatastoreInputPlugin(doLogging: Boolean = true) : InputPlugin {
     // number of run() method calls
     private val TASK_COUNT = 1
 
-    private val logger = if (doLogging) { Exec.getLogger(javaClass) } else { null }
+    private val logger = if (doLogging) {
+        Exec.getLogger(javaClass)
+    } else {
+        null
+    }
     private val b64encoder = Base64.getEncoder()
+    private val jsonParser = JsonParser()
 
     override fun transaction(config: ConfigSource,
                              control: InputPlugin.Control): ConfigDiff {
         val task: PluginTask = config.loadConfig(PluginTask::class.java)
 
@@ -63,11 +69,11 @@
         datastore.run(query)
                 .forEach { entity ->
                     logger?.debug(entity.toString())
 
 
-                    val json = when(entity) {
+                    val json = when (entity) {
                         is FullEntity<*> -> entityToJsonObject(entity)
                         is ProjectionEntity -> entityToJsonObject(entity)
                         else -> null
                     }
 
@@ -75,10 +81,10 @@
                         logger?.debug(json)
                     } ?: run {
                         logger?.error("Unexpected result type")
                     }
 
-                    pageBuilder.setJson(col, ValueFactory.newString(json))
+                    pageBuilder.setJson(col, jsonParser.parse(json))
                     pageBuilder.addRecord()
                 }
 
         pageBuilder.finish()
 