DatastoreInputPlugin.kt in embulk-input-datastore-0.1.0

- old
+ new

@@ -9,16 +9,18 @@
 import org.embulk.config.TaskSource
 import org.embulk.spi.*
 import org.embulk.spi.type.Types
 import org.msgpack.value.ValueFactory
 import java.io.FileInputStream
+import java.util.Base64
 
-class DatastoreInputPlugin : InputPlugin {
+class DatastoreInputPlugin(doLogging: Boolean = true) : InputPlugin {
     // number of run() method calls
     private val TASK_COUNT = 1
 
-    private val logger = Exec.getLogger(javaClass)
+    private val logger = if (doLogging) { Exec.getLogger(javaClass) } else { null }
+    private val b64encoder = Base64.getEncoder()
 
     override fun transaction(config: ConfigSource,
                              control: InputPlugin.Control): ConfigDiff {
         val task: PluginTask = config.loadConfig(PluginTask::class.java)
 
@@ -57,14 +59,14 @@
         val datastore = createDatastoreClient(task)
         val col = pageBuilder.schema.getColumn(0)
 
         datastore.run(query)
                 .forEach { entity ->
-                    logger.debug(entity.toString())
+                    logger?.debug(entity.toString())
 
                     val json = entityToJsonObject(entity)
-                    logger.debug(json)
+                    logger?.debug(json)
 
                     pageBuilder.setJson(col, ValueFactory.newString(json))
                     pageBuilder.addRecord()
                 }
 
@@ -132,21 +134,21 @@
      *  e.g.) '"name":"value"'
      *
      */
     private fun valueToField(dsValue: Value<Any>): String? {
         return when (dsValue.type) {
-            ValueType.BLOB -> (dsValue.get() as ByteArray).toString()
+            ValueType.BLOB -> "\"${b64encoder.encodeToString((dsValue.get() as Blob).toByteArray())}\""
             ValueType.BOOLEAN -> (dsValue.get() as Boolean).toString()
             ValueType.DOUBLE -> (dsValue.get() as Double).toString()
             ValueType.ENTITY -> entityToJsonObject(dsValue.get() as FullEntity<*>)
             ValueType.KEY -> (dsValue.get() as Key).toString()
             ValueType.LAT_LNG -> (dsValue.get() as LatLngValue).toString()
             ValueType.LIST -> listToJsonArray(dsValue.get() as List<*>)
             ValueType.LONG -> (dsValue.get() as Long).toString()
             ValueType.NULL -> "null"
             ValueType.RAW_VALUE -> (dsValue.get() as RawValue).toString()
             ValueType.STRING -> "\"${dsValue.get() as String}\""
-            ValueType.TIMESTAMP -> (dsValue.get() as Timestamp).toString()
+            ValueType.TIMESTAMP -> "\"${dsValue.get() as Timestamp}\""
             else -> null // NOTE: unexpected or unsupported type
         }
     }
 }