     # so that when we pull the value after that we know that all chunks have been processed
     # and we can display a final message
     _(transformStream).reduce null, -> ''
+    .stopOnError (e) => @_die 'Something went wrong while transforming chunks', e
     .pull (err, x) ->
       if err then log.error err
       else log.info service.summaryReport(options.from)

      process.stdin.setEncoding('utf8')
      process.stdin

    # TODO: error handling
    transformStream = _(inputStream.pipe(JSONStream.parse(jsonPath)))
    .batch(options.batch)
    .pipe(transform (chunk, cb) ->
      log.info 'chunk: %j', chunk, {}

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

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