import logger from '../logger/Logger'
import BaseProvider from '../core/BaseProvider'
import fs from 'fs'
import es from 'event-stream'

export default class CsvProvider extends BaseProvider {
    headers!: string[]

    connectDataSource(): void {
        let { topic, filePath } = this.params

        fs.createReadStream(filePath)
        .pipe(es.split())
        .pipe(es.mapSync((line: string) => {
            if (!line) {
                return
            }

            let fields = line.split(",").map((item: string) => {
                return item.trim()
            })

            if (!this.headers) {
                this.headers = fields
                return
            }

            let row: any = {}

            for (let i = 0; i < this.headers.length; i++) {
                const header = this.headers[i];
                row[header] = fields[i]
            }

            this.onMessage(topic, row)
        }))
        .on('end', () => {
            this.core.endInput()
            console.log("输入结束")
        })
    }
}