How to use oboe - 10 common examples

self.totalParsed++;

                if (self.totalParsed > self.stopAt) {
                    emit(buffer);
                    buffer=[];
                    this.abort(); // tell oboe to shut things down
                    this.emit('end');
                } else {
                    buffer.push(point);
                    if (buffer.length >= self.limit) {
                        emit(buffer);
                        buffer = [];
                    }
                }

                return oboe.drop;
            })
            .on('end', function(){

.node('data[*]', (lesson: TimetableLesson) => {
          // Consume and discard each lesson
          lessonConsumer(lesson);
          return oboe.drop;
        })
        .done((data) => {

'features.*': feature => {
        if (!feature.geometry || !feature.geometry.type) {
          // Only add this error once
          // TODO: Give feedback on which features failed
          if (!this._errors.length) {
            this._errors.push(
              new Error(
                i18n.translate('xpack.fileUpload.patternReader.featuresOmitted', {
                  defaultMessage: 'Some features without geometry omitted',
                })
              )
            );
          }
          return oboe.drop;
        }
        return featurePatternCallback(feature);
      },
      // Handle single feature files

return new Promise((resolve, reject) => {
                this.logger.verbose(`"oboe" start`);

                // TODO replace "oboe" with alternative library that doesn't block the Event Loop so heavily
                //      review the following libraries:
                //      - https://gitlab.com/philbooth/bfj
                //      - https://github.com/ibmruntimes/yieldable-json
                //      or consider moving parsing to a separate Worker/Process
                oboe(readableStream)
                    .done((parsed) => {
                        this.logger.verbose(`"oboe" end`);
                        resolve(parsed);
                    })
                    .fail((error) => {
                        this.logger.error(`"oboe" fail`, error);
                        reject(error);
                    });
            });
        };

new Promise((resolve, reject) => {
      const endpoint = 'classtt/withdate/published';
      const url = new URL(endpoint, config.baseUrl);
      const body = JSON.stringify({ term });

      oboe({
        url: url.href,
        headers,
        body,
        method: 'POST',
      })
        .node('data[*]', (lesson: TimetableLesson) => {
          // Consume and discard each lesson
          lessonConsumer(lesson);
          return oboe.drop;
        })
        .done((data) => {
          // Handle application level errors
          const { code, msg } = data;

          if (code === OKAY) {
            resolve();

let debounceDispatch = debounce(dispatch, 5);

    function sendStoryline() {
      return function(dispatch) {
        dispatch(addPlaces(places));
        dispatch(addStays(stays));
        dispatch(addTrips(trips));

        // Reset data arrays
        places = [];
        stays = [];
        trips = [];
      }
    }

    oboe('/api')
      .node('startAt', function(startAt) {
        return new Date(startAt * 1000);
      })
      .node('endAt', function(endAt) {
        return new Date(endAt * 1000);
      })
      .node('location', function(location) {
        return L.latLng(location.lat, location.lon);
      })
      .node('place', function(place) {
        place = new Place(place);
        places.push(place);

        debounceDispatch(sendStoryline());

        return oboe.drop;

.on('done', function(event){
        //console.log("---------- event: ---------------------------------------------");
        //console.dir(event);
        return oboe.drop;
     })
     .on('fail', function(e){

on('node', process.env.JSON_PATH, () => {
       meter.mark();
       return oboe.drop;
    }).
    fail((err) => console.error(err.stack)).

return function(this: oboe.Oboe, node: any) {
      try {
        callback(node)
        return oboe.drop
      } catch (e) {
        winston.info(`safeOboe callback produced an error, aborting stream`, logInfo)
        this.abort()
        stream.destroy()
        reject(e)
      }
    }
  }

.node(pattern, (item) => {
            this.transform(item)
              .then(this.setNode)
              .catch(reject);
            return oboe.drop;
          })
          .done(resolve);