Detailed explanation of the simple use of RxJS in TypeScript

# Install typescript, rxjs packagenpm install -D typescript @types/node
npm install rxjs

import {from} from 'rxjs'

// Generate subscribeable objects from array// The object type of obser is Observablelet obser = from([1,2,3,4,5])
// Consumption objects// next When there is a value flowing in the pipeline, it will start next// When an error occurs, an error will be triggered// When the array is completed, complete will be called({
  next(item) {
    (item)
  },
  error(err) {
    (err)
  },
  complete() {
    ("Done")
  }
}

import {Observable, from} from 'rxjs'

// Create a new subscribeable objectlet obser = new Observable<string>(productor => {
  // Customize data flow  ("hello")
  ("world")
  setTimeout(()=>{
    ("After 1 Sec")
    ()
  }, 1000)
})

// Consumption({
  next(item) {
    (item)
  },
  error(err) {
    (err)
  },
  complete() {
    ("Done")
  }
}

npm install redis @types/redis

import {RedisClient} from "redis"
import {Observable} from "rxjs"
// The constructor type of redis is "ClientOpt interface", as long as it is an object that meets its field definitionlet reids = new RedisClient({
  host:"localhost",
  port:6379
})
// Redis operations are asynchronous operations. When using callbacks, it is easy to fall into ruining hell("name", "tom", (err, res)=>{
  // Operations that need to be synchronized after set succeeds  if(!err) {
    ("name", (err,res)=> {
      (res)
    })
  }
})

// You can use Promise to encapsulate the above method, but we use rxjs event stream herelet redisObser = new Observable(productor => {
// This column has no practical purpose, only for demonstration  ("name", "jack", (err, res) => {
    (res)
    ()
  })
})
// Consumption code({
  next(key) {
    // Reading or other operations after set    (key, (err, res) => {
      (res)
    })
  },
  error(err) {
    (err)
  },
  complete() {
    ("Query Finish")
  }
}