ghermeto / kafka-observable Goto Github PK

KafkaObservable.fromTopic(...)
  .filter(Boolean)
  .kafkaMessage() // <-- these work
  .flatMap(() => ...)
  .otherKafkaThing() // <-- just fine!
  .subscribe()

const { Observable, Subscriber } = require('rxjs');

// Extend Observable, and override `lift` to return instances of KafkaObservable
// All the Rx operators all call `this.lift(some_operator)` just like I do in `kafkaMessage`
export class KafkaObservable extends Observable {
  lift(operator) {
    const kObs = new KafkaObservable();
    kObs.source = this;
    kObs.operator = operator;
    return kObs;
  }
  kafkaMessage(mapper = x => x) {
    return this.lift(function(source) {
      // the `this` context is the destination Subscriber
      const sink = this;
      return source.subscribe(new KafkaMessageSubscriber(sink, mapper));
    });
  }
}

class KafkaMessageSubscriber extends Subscriber {
  constructor(sink, mapper) {
    super(sink);
    this.mapper = mapper;
  }
  next(x) {
    super.next(this.mapper(x.message.toString('utf8'));
  }
}