Is it possible to not activate a staled observable when reactivating flatMapLatest?

[wongcy0]

Hi, I am looking for a way to avoid the activation of an observable created from a previous activation of flatMapLatest.

let pool = Kefir.pool();
let foo = pool.toProperty();
let bar = foo.flatMapLatest(v => {
  return Kefir.stream(emitter => {
    console.log("Activation " + v);
    // ...
  });
});

foo.observe();

pool.plug(Kefir.constant(1));

bar.log("bar");
// Activation 1
bar.offLog("bar");

pool.plug(Kefir.constant(2));

bar.log("bar");
// Activation 1 <-- unwanted activation
// Activation 2

Any input would be appreciated. Thanks!

[mAAdhaTTah]

It's not clear why this would be "unwanted", based on the behavior of flatMapLatest. This is probably more of a business logic question than a Kefir question. Could you elaborate on what you're trying to do?

[wongcy0]

Thanks for your reply. Just realized that the snippet failed to illustrate my problem. It's now updated.

I don't want it because it's no longer the observable created from the latest value.

Here's what I am trying to do:

let loggedInUserDetails$ = loggedInUser$.flatMapLatest(user => {
  if (!user) {
    return Kefir.constant(null);
  }
  return Kefir.stream(emitter => {
    user.on('detailsChange', emitter.emit); // <-- It throws an exception if the user is logged out
    return () => user.off('detailsChange', emitter.emit);
  });
});

[mAAdhaTTah]

The new snippet is probs how I'd solve it, tbh (if I understand the issue correctly). Presumably, there's something about the user data you can check to determine whether you should be subscribing to it or not (e.g. user.isLoggedIn or whathaveyou). You may also find flatMapFirst useful, if you don't want a new observable created if the first one is still active. Also, the throwing behavior itself is a bit weird, but perhaps that's out of your control.

[wongcy0]

Thank you. The snippet above looks to work but is actually not:

// initially, loggedInUser has a current value of user A

loggedInUserDetails.log("details");
// flatMapLatest creates a Kefir.stream with user A, so far so good
loggedInUserDetails.offLog("details");

// after some time, user A logs out, loggedInUser emits a `null`

loggedInUserDetails.log("details");
// before creating a Kefir.constant(null), flatMapLatest activates the previous
// Kefir.stream bound to user A, which triggers a throw

The throwing is indeed out of my control. I could wrap the user.on() with try/catch to stop it from exploding. But it feels clumsy having to write the same checking across multiple places.

[mAAdhaTTah]

Ah, so that's caused by the toProperty call in the original snippet. The Property "remembers" the last value it emitted and emits it synchronously on activation. If the loggedInUser "emits null" when nothing is subscribed to it, because the Observable wasn't active at the time, it doesn't actually emit that null.

[wongcy0]

Oh that null from loggedInUser is expected. And loggedInUser does have some other subcriber. Let's replace 1 null to avoid confusion.

let loggedInUserDetails$ = loggedInUser$.flatMapLatest(user => {
  if (!user) {
    return Kefir.constant({});
  }
  // ...
});

What I think is the stream is remembered in _curSources of the AbstractPool after deactivation.

[mAAdhaTTah]

So in general, I think all of this is expected from a "what Kefir is supposed to do" perspective. I'm not sure I can come up with a good solution without a reproducible example, as I'm having some difficulty getting my head around the intended behavior here vs what you're seeing.

[wongcy0]

Here's a reproducible example.

// library stuff

let currentUserName = null;

class User {
  username = '';
  detailsChangeCbs = [];

  constructor (username) {
    this.username = username;
  }
  onDetailsChange = (cb) => {
    if (currentUserName !== this.username) {
      throw new Error("Nope. User " + this.username + " is not logged in.");
    }
    cb({
      username: this.username,
      details: "more details"
    })
    this.detailsChangeCbs.push(cb);
  }
  offDetailsChange = (cb) => {
    this.detailsChangeCbs = this.detailsChangeCbs.filter(x => x !== cb);
  }
}

let userEmitter;
let loggedInUser$ = Kefir.stream(emitter => {
  userEmitter = emitter;
  return () => (userEmitter = undefined);
})
.toProperty(() => currentUserName ? new User(currentUserName) : null);

let login = (username) => {
  currentUserName = username;
  if (userEmitter) {
    userEmitter.emit(new User(username));
  }
};
let logout = () => {
  currentUserName = null;
  if (userEmitter) {
    userEmitter.emit(null);
  }
};


// my stuff

let loggedInUserDetails$ = loggedInUser$.flatMapLatest(user => {
  if (!user) {
    return Kefir.constant({ details: "nil" });
  }
  return Kefir.stream(emitter => {
    try {
      user.onDetailsChange(emitter.emit);
      return () => user.offDetailsChange(emitter.emit);
    } catch (ex) {
      console.error(ex);
    }
  });
});

const onUserChange = user => console.log("Logged in user:", user);
const onUserDetailsChange = details => console.log("Logged in user details:", details);

loggedInUser$.onValue(onUserChange)
loggedInUserDetails$.onValue(onUserDetailsChange);

login("A")

loggedInUserDetails$.offValue(onUserDetailsChange);

setTimeout(() => {
  logout();

  setTimeout(() => {
    loggedInUserDetails$.onValue(onUserDetailsChange);
  }, 100);

}, 100)

The outputs of observables matches what I want. Now I am trying to get rid of the exception without try/catch.

Thanks