observable.js

const Observable = subscribe =>
({
  subscribe,
  map: f =>
    Observable(observer =>
      subscribe({next: x => observer.next(f(x))})
    ),
  filter: f =>
    Observable(observer =>
      subscribe({next: x => f(x) ? observer.next(x) : null})
    ),
  reduce(f, seed) {
    let self = this;
    return Observable(observer =>
      subscribe({
        next(x) {
          const result = self.acc !== undefined ? f(self.acc, x) : f(seed, x)
          self.acc = result
          observer.next(result)
        }
      })
    )
  },
  take(amount) {
    let self = this;
    self.taken = 0;
    return Observable(observer =>
      subscribe({
        next(x) {
          self.taken < amount ? observer.next(x) : null
          self.taken++
        }
      })
    )
  },
  do: f =>
    Observable(observer =>
      subscribe({
        next(x) {
          f(x)
          observer.next(x)
        }
      })
    ),
  flatten: () =>
    Observable(observer =>
      subscribe({
        next: observable =>
          observable.subscribe({
            next(x) {
              observer.next(x)
            }
          })
      })
    )
})

const periodic = period =>
  Observable(observer => {
    let i = 0
    setInterval(() => observer.next(i++), period)
  })
//
// periodic(1000)
// .map(x => x + 1)
// .filter(x => x % 2 == 0)
// .do(x => console.log('n: ', x))
// .reduce((acc, x) => acc + x, 0)
// .take(5)
// .subscribe({
//   next(x) {
//     console.log('total: ', x)
//   }
// })


periodic(1000)
.map(x => periodic(1000))
.flatten()
.subscribe({
  next(x) {
    console.log('flat: ', x)
  }
})
//
// periodic(1000)
//   .map(x => x + 1)
//   .filter(x => x % 2 !== 0)
//   .reduce((acc, x) => acc + x, 0)
//   .subscribe({
//     next(x) {
//       console.log(x)
//     }
//   })

// periodic(1000)
//   .take(5)
//   .subscribe({
//     next(x) {
//       console.log(x)
//     }
//   })