[ISSUE#51]: to_avro working

igned-off-by: ashish <ashishpatel0720@gmail.com>
Signed-off-by: ashish <ashishpatel0720@gmail.com>

Author: pateash

src/commands/avro.ts

@@ -1,6 +1,8 @@
 import {Command, flags} from '@oclif/command'
+import {createFileEncoder} from 'avsc'
 import * as avro from 'avsc'
-import * as chalk from 'chalk' // includes all from avro-js and some more
+import * as chalk from 'chalk'
+import * as fs from 'fs' // includes all from avro-js and some more
 
 import Logger from '../utilities/logger'
 import Utilities from '../utilities/utilities'
@@ -12,7 +14,7 @@ export default class Avro extends Command {
     help: flags.help({char: 'h'}),
     file: flags.string({char: 'f' , description: 'input file path'}),
     output: flags.string({char: 'o' , description: 'output file path'}),
-    schema: flags.string({char: 't' , description: 'schema type file path'}),
+    schemaType: flags.string({char: 't' , description: 'schema type file path'}),
 
   }
 
@@ -55,7 +57,7 @@ export default class Avro extends Command {
     avro.createFileDecoder(flags.file)
       .on('metadata', function (type) {
         let output = type.schema()
-        let schemaStr = JSON.stringify(output, null, '\t')
+        let schemaStr = JSON.stringify(output)
         Utilities.writeStringToFile(this, flags.output, schemaStr)
       })
 
@@ -64,20 +66,35 @@ export default class Avro extends Command {
     Utilities.truncateFile(this, flags.output)
     avro.createFileDecoder(flags.file)
       .on('data', function (recordStr) {
-        Utilities.appendStringToFile(this, flags.output, JSON.stringify(recordStr, null, '\t'))
+        Utilities.appendStringToFile(this, flags.output, JSON.stringify(recordStr))
       })
     Logger.success(this, `output written to file: ${chalk.green(flags.output)}`) // this will output error and exit command
   }
+
   private toAvro(flags: any, args: any) {
-    if (!flags.schema)
+    if (!flags.schemaType)
       Logger.error(this, 'Schema file is not provided')
-    this.checkValidSchema(flags, args)
-  }
 
-  private checkValidSchema(flags: any, args: any) {
-    let avroFile = avro.parse(flags.file)
-    // var schemaFile = {name: 'Bob', address: {city: 'Cambridge', zip: '02139'}}
-    // var status = type.isValid(person); // Boolean status.
-    console.log(avroFile)
+    let schema = avro.parse(flags.schemaType)
+    let avroEncoder = new avro.streams.BlockEncoder(schema)
+
+    avroEncoder.pipe(fs.createWriteStream(flags.output))
+
+// We write the records to the block encoder, which will take care of serializing them
+// into an object container file.
+
+    let jsonStr = '[' + Utilities.getInputString(this, flags, args) + ']'
+    jsonStr = jsonStr.replace(/[\s\n]+/mg, '')
+    jsonStr = jsonStr.replace(/\}\{/mg, '},{')
+    let jsonObjects = JSON.parse(jsonStr)
+
+    jsonObjects.forEach( function(data){
+      if (schema.isValid(data)) {
+        avroEncoder.write(data)
+      } else {
+        Logger.warn(this, `${chalk.yellow('[SKIPPING RECORD]')} schema is invalid: ${chalk.yellowBright(JSON.stringify(data))}`)
+      }
+    })
+    avroEncoder.end()
   }
 }

src/utilities/utilities.ts

@@ -46,7 +46,9 @@ export default class Utilities {
       Logger.warn(thisRef, `File already exists: ${chalk.green(filePath)}, ${chalk.yellow('overriding content')}`) // this will output error and exit command
 
     fs.writeFileSync(filePath, string)
-    Logger.success(thisRef, `output written to file: ${chalk.green(filePath)}`) // this will output error and exit command
+
+    if (string !== '') // this condition comes for truncating
+      Logger.success(thisRef, `output written to file: ${chalk.green(filePath)}`) // this will output error and exit command
     // return `${chalk.red(pkg)} ${message}`
 
   }

test/resources/avro/complex.json

@@ -0,0 +1,52 @@
+{
+ "stringArray": [
+  "string1",
+  "string2",
+  "string3",
+  "string4",
+  "string5"
+ ],
+ "longArray": [
+  1,
+  2,
+  3,
+  4,
+  5
+ ],
+ "enumField": "C",
+ "mapOfInts": {
+  "key4": 4,
+  "key3": 3,
+  "key5": 5,
+  "key2": 2,
+  "key1": 1
+ },
+ "unionField": "union value",
+ "fixedField": {
+  "type": "Buffer",
+  "data": [
+   1,
+   2,
+   3,
+   4,
+   1,
+   2,
+   3,
+   4,
+   1,
+   2,
+   3,
+   4,
+   1,
+   2,
+   3,
+   4
+  ]
+ },
+ "recordField": {
+  "longRecordField": 1925639126735,
+  "stringRecordField": "I am a test record",
+  "intRecordField": 666,
+  "floatRecordField": 7171.169921875
+ }
+}

test/resources/avro/output_complex.avsc

@@ -0,0 +1,80 @@
+{
+	"name": "example.avro.Complex",
+	"type": "record",
+	"fields": [
+		{
+			"name": "stringArray",
+			"type": {
+				"type": "array",
+				"items": "string"
+			}
+		},
+		{
+			"name": "longArray",
+			"type": {
+				"type": "array",
+				"items": "long"
+			}
+		},
+		{
+			"name": "enumField",
+			"type": {
+				"name": "example.avro.foo",
+				"type": "enum",
+				"symbols": [
+					"A",
+					"B",
+					"C",
+					"D"
+				]
+			}
+		},
+		{
+			"name": "mapOfInts",
+			"type": {
+				"type": "map",
+				"values": "int"
+			}
+		},
+		{
+			"name": "unionField",
+			"type": [
+				"null",
+				"string"
+			]
+		},
+		{
+			"name": "fixedField",
+			"type": {
+				"name": "example.avro.md5",
+				"type": "fixed",
+				"size": 16
+			}
+		},
+		{
+			"name": "recordField",
+			"type": {
+				"name": "example.avro.TestRecord",
+				"type": "record",
+				"fields": [
+					{
+						"name": "longRecordField",
+						"type": "long"
+					},
+					{
+						"name": "stringRecordField",
+						"type": "string"
+					},
+					{
+						"name": "intRecordField",
+						"type": "int"
+					},
+					{
+						"name": "floatRecordField",
+						"type": "float"
+					}
+				]
+			}
+		}
+	]
+}