       process.stdin.setEncoding('utf8')
       process.stdin
 
-    # TODO: error handling
     transformStream = _(inputStream.pipe(JSONStream.parse(jsonPath)))
+    .stopOnError (e) => @_die 'Cannot parse chunk as JSON', e
     .batch(options.batch)
     .pipe(transform (chunk, cb) ->
       log.info 'chunk: %j', chunk, {}

    @_validateOptions(@program)
    @_process(@program)

  @_validateOptions: (options) ->
    errors = []
    errors.push('Missing required option: type') unless options.type

        service = new StockImport(log, {})
        @_stream(options, service, 'stocks.*')
      else
        @_die "Unsupported type: #{type}"

  @_stream: (options, service, jsonPath) ->
    inputStream = if options.from
