Broken interoperability with Observables

[josepot]

Following the recipe outlined in here.

The following code:

const rune = observableToRune(
  new Observable((observer) => {
    observer.next("hello")
    observer.next("there!")
    observer.next("how are you?")
  }),
)

const runTest = async (id: string) => {
  for await (const value of rune.iter()) {
    console.log(`${value} from ${id}`)
  }
}

Promise.all([runTest("a"), runTest("b")])

outputs this:

how are you? from a
how are you? from b

As you can see: it has skipped the first 2 emitted values.

Also, I've noticed that the program exited before the Observable completed 🤔 I'm not quite sure if this is a bug or not, TBH... However, if it's not, then I think that it's a behavior worth being documented.

On the other hand, the following code:

class ObservableError extends Error {
  override readonly name = "ObservableError"
  constructor(public value: unknown) {
    super()
  }
}

const rune = observableToRune(
  new Observable((observer) => {
    setTimeout(() => {
      observer.next("hello")
      observer.next("there!")
      observer.next("how are you?")
    }, 0)
  }),
)

const runTest = async (id: string) => {
  for await (const value of rune.iter()) {
    console.log(`${value} from ${id}`)
  }
}

Promise.all([runTest("a"), runTest("b")])

outputs this:

hello from a
hello from b
there! from a
there! from b
how are you? from a
how are you? from b

I know that you may consider this a feature 🤷‍♂️. However, please understand that from my point of view: this is an arbitrary baked-in behavior and it would make my life hell if I had to use Runes. The expected output should have been:

hello from a
there! from a
how are you? from a
hello from b
there! from b
how are you? from b

[tjjfvi]

As you can see: it has skipped the first 2 emitted values.

This is a bug; thanks for catching it. I've opened a PR to resolve this: #1160.

Also, I've noticed that the program exited before the Observable completed 🤔

This is independent of Rune. The program exits when the event queue is empty, even if there are unresolved promises. Because nothing in the code creates a task to call .complete(), the program is not held open for this. Removing Rune from the equation, the following adaption of your example also exits:

new Promise<void>((resolve) => {
  new Observable((observer) => {
    observer.next("hello")
    observer.next("there!")
    observer.next("how are you?")
  }).subscribe({
    next: console.log,
    error: console.error,
    complete: resolve,
  })
}).then(() => {
  console.log("done")
})

hello
there!
how are you?
[exits]

[josepot]

@tjjfvi I'm re-opening this bug because I haven't been able to understand the reason why the following code:

class ObservableError extends Error {
  override readonly name = "ObservableError"
  constructor(public value: unknown) {
    super()
  }
}

const rune = observableToRune(
  new Observable((observer) => {
    setTimeout(() => {
      observer.next("hello")
      observer.next("there!")
      observer.next("how are you?")
    }, 0)
  }),
)

const runTest = async (id: string) => {
  for await (const value of rune.iter()) {
    console.log(`${value} from ${id}`)
  }
}

Promise.all([runTest("a"), runTest("b")])

outputs this:

hello from a
hello from b
there! from a
there! from b
how are you? from a
how are you? from b

rather than outputting this:

hello from a
there! from a
how are you? from a
hello from b
there! from b
how are you? from b

Could you please justify why that is not a bug, and in the event that this is not considered a bug, whether there is a way to opt-out from such an awkward behavior? 🙏