Beginner problems

[Nutelac]

I'm new to reactive programming and I have some troubles. I have a device with N sensors (temperature, pressure, time, ...). I think it would be great idea if I represent sensors as streams. So I end up with this code:

function temperatureSensor(pin) {
  let stream = Kefir.emitter()

  if(pin > 12) {
    stream.error(new Error(`Board has no pin ${pin}.`))
  }

  setInterval(() => {
    stream.emit(readTemperature(pin))
  }, 500)

  return stream
}

Kefir.zip([
  temperatureSensor(1),
  temperatureSensor(42) // Invalid pin, should call .error
]).onValue((state) => {
  // do something with sensor values
}).onError(() => {
  console.error('An error occurred!')
})

There is few problems with this code: 1, .onError will never be called (.onError is attached after .error is called). Maybe use property instead of stream, but I don't know how... 2, .zip is not a ideal function, I need something like that:

// Synchronize sensors, call .onValue, then .onValue every time when sensor value change

a:    ----20------------21--------------
b:    -------25--------------------24---
ab:   --------•----------•----------•---
       [20, 25]   [21, 25]   [21, 24]

Thank you for any advice.

[rpominov]

Hi, first issue when .onError callback never called is because you use Kefir.emitter(). General advice here is to simply avoid using it. In almost any case when you use Kefir.emitter() there is a better alternative.

I'd write temperatureSensor something like this:

function temperatureSensor(pin) {
  if (pin > 12) {
    return Kefir.constantError(new Error(`Board has no pin ${pin}.`));
  } else {
    return Kefir.fromPoll(500, () => readTemperature(pin));
  }
}

Instead of .zip it looks like you need .combine. Looks like beginners often use .zip, but I'm personally never used it, not even once :)

Closing, but feel free to ask further questions.

[asaaki]

Just another example with .merge + .slidingWindow:

function temperatureSensor(pin) {
  if (pin <= 12) {
    return Kefir.fromPoll(500, function () {
      return readTemperature(pin)
    });
  } else {
    return Kefir.constantError(
      new Error('Board has no pin' + pin + '.')
    );
  }
};

var sensors = [
  temperatureSensor(1),
  temperatureSensor(9),
  temperatureSensor(14)
];
var sLength = sensors.length;
// assuming you want slices with same amount like your observed sensors

Kefir
  .merge(sensors)
  .slidingWindow(sLength, 0)
  .onError(function (error) {
    // had to put .onError first, otherwise the error never popped up
    console.log('Error is', error);
  })
  .onValue(function (state) {
    console.log('Current state is', JSON.stringify(state));
  })

(Does not preserve the sensor order in the onValue array.)

But I guess .combine is what needed here.

[rpominov]

had to put .onError first, otherwise the error never popped up

Ah, yes. To solve this you also need to add .toProperty after .zip, .combine, .merge etc. This described in the docs http://pozadi.github.io/kefir/#current-in-streams

Have to say, the fact, that one need to know all tiny things like that to use this lib effectively, doesn't make me happy. I am trying to find a way to make it more easy to learn, but haven't get there yet. https://github.com/pozadi/kefir/issues/43 — here is discussion of one of problems like that.

[asaaki]

Ah, thank you for this explanation. :+1:

[Nutelac]

Thank you very much, your responses are really helpful. I see that I choose the right framework!

[Nutelac]

I have one more question. I have small JSON file where I store all my persistent configuration. I need to do some work, when configuration loads or change. It's possible and good idea to create stream from JSON file?

[rpominov]

Yes this is totally possible, and may make sense. The code can look something like this:

var fileChanges = Kefir.fromBinder(function(emitter) {
  subscribeToFileUpdates(filename, emitter.emit);
  return function() {
    unsubscribeFromFileUpdates(filename, emitter.emit);
  };
});

// if you going to read file synchronously, 
// and don't care about errors, you can use .map instead of .flatMapLatest
var config = fileChanges.flatMapLatest(function() {
  return Kefir.fromNodeCallback(function(cb) {
    getJsonFromFile(filename, cb);
  });
}).toProperty();

config.onValue(function(configObj) {
  // here you have latest config, 
  // or you can combine `config` with other observables
});

[Nutelac]

Thank you very much again!

[rpominov]

No problem :)

[spro]

Old issue but I ran into a similar problem (.onError not being called with .zip) and it seems like the fix was simply to swap the order of .onError and .onValue. Any ideas why that's necessary?

[mAAdhaTTah]

Need to call toProperty, the merge/zip/combine converts properties -> streams, which means the current value in the stream (the error) is lost because it's been "emitted" before you subscribed.