Cài đặt RxJS

Cài được package chưa có nghĩa là môi trường đã sẵn sàng để học RxJS. Bạn còn cần biết TypeScript có resolve đúng kiểu dữ liệu không, lệnh nào thật sự chạy file, và vì sao một chương trình dùng interval có thể không chịu thoát.

Trang này dựng một playground Node.js nhỏ, chạy một stream từ đầu đến cuối, rồi dùng chính output để đọc subscribe, complete và teardown. Nếu bạn đã có project TypeScript hoặc framework, hãy xem phần Đưa RxJS vào project hiện có thay vì tạo project mới.

Mục lụcCopy Anchor Link

• Bạn sẽ dựng gì
• Mental model: cài đặt rồi chạy
• Chuẩn bị
• Tạo playground TypeScript
  • Khởi tạo project và cài package
  • Cấu hình ESM và TypeScript
  • Viết stream đầu tiên
  • Type-check và chạy
• Đọc lifecycle từ output
• Teardown khi hủy thủ công
• Ví dụ thực tế: dừng worker bằng Ctrl+C
• Đưa RxJS vào project hiện có
  • Project TypeScript hoặc bundler
  • Project Angular
  • Chọn range phiên bản
• Pitfalls thường gặp
• Xử lý lỗi cài đặt
• Tiếp tục lộ trình
• Nguồn tham khảo

Bạn sẽ dựng gìCopy Anchor Link

Sau các bước bên dưới, thư mục thực hành có dạng:

rxjs-playground/
├── node_modules/
├── src/
│   └── index.ts
├── package-lock.json
├── package.json
└── tsconfig.json

Bạn sẽ có:

• rxjs trong dependencies, vì code chạy thật sự cần runtime của thư viện;
• typescript, tsx và @types/node trong devDependencies, vì chúng chỉ phục vụ việc viết, kiểm tra và chạy source TypeScript;
• một lệnh type-check riêng và một lệnh chạy riêng;
• một stream hữu hạn để chương trình tự complete, cùng một stream sống lâu để luyện teardown.

Mình khuyên tạo playground riêng thay vì cài thêm package vào repo tài liệu hoặc project đang làm việc. Một môi trường nhỏ giúp bạn nhìn rõ lỗi thuộc về RxJS, TypeScript hay cấu hình project.

Mental model: cài đặt rồi chạyCopy Anchor Link

npm install chỉ đặt runtime JavaScript và declaration TypeScript của RxJS vào node_modules. Nó không tự tạo Observable, không chạy operator và cũng không tự subscribe.

npm install
    │
    ▼
node_modules/rxjs
(runtime JavaScript + type declarations)
    │
    ├──► tsc đọc declarations để kiểm tra kiểu
    │
    └──► tsx nạp runtime để chạy src/index.ts
                         │
                         ▼
                    subscribe()
                         │
           source ─► operators ─► Observer
                         │
                         ▼
              complete / error / unsubscribe
                         │
                         ▼
                      teardown

Hãy tách ba việc trong đầu:

1. Cài package làm cho module rxjs có mặt trong dependency graph.
2. Type-check xác nhận import và kiểu dữ liệu hợp lệ, nhưng không chạy stream.
3. Thực thi chạy file; với phần lớn cold Observable, source chỉ bắt đầu khi có subscribe().

Cách tách này đặc biệt hữu ích khi debug: tsx chạy được chưa chắc code đã qua type-check, còn tsc --noEmit thành công chưa chứng minh pipeline có output đúng.

Chuẩn bịCopy Anchor Link

Bạn cần một bản Node.js còn được hỗ trợ và npm đi kèm. Playground trong bài dùng Node.js 20 trở lên để có môi trường hiện đại, nhưng đây là lựa chọn cho bài học, không phải tuyên bố về phiên bản Node.js tối thiểu của mọi ứng dụng RxJS.

Kiểm tra công cụ trước khi tạo project:

node --version
npm --version

Nếu terminal không nhận ra node hoặc npm, hãy cài Node.js trước rồi mở terminal mới. Đừng dùng sudo npm install để chữa lỗi quyền truy cập; cách đó dễ tạo file do root sở hữu trong project. Nên sửa cách cài Node.js hoặc dùng một version manager phù hợp với hệ điều hành.

Tạo playground TypeScriptCopy Anchor Link

Khởi tạo project và cài packageCopy Anchor Link

Chạy các lệnh sau trong thư mục dành cho bài tập, không phải trong repo tài liệu này:

mkdir rxjs-playground
cd rxjs-playground
npm init -y
npm install rxjs@7.8.2
npm install --save-dev typescript tsx @types/node
npm pkg set type=module
mkdir src

Việc pin rxjs@7.8.2 giúp output của bài không thay đổi vì một major release trong tương lai. npm cũng tạo package-lock.json; hãy commit lockfile trong ứng dụng để các máy cài cùng dependency tree.

Xác nhận package thực sự được resolve từ project hiện tại:

npm ls rxjs

Kết quả mong đợi có dòng tương tự:

rxjs-playground@1.0.0
└── rxjs@7.8.2

Nếu npm ls báo (empty), thường là bạn đã chạy npm install ở thư mục khác. Kiểm tra pwd hoặc vị trí hiện tại của terminal trước khi cài lại.

Cấu hình ESM và TypeScriptCopy Anchor Link

Tạo tsconfig.json:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "noEmit": true,
    "types": ["node"],
    "skipLibCheck": true
  },
  "include": ["src/**/*.ts"]
}

"type": "module" trong package.json và cặp NodeNext trong tsconfig.json cùng mô tả cách Node.js resolve ESM. Giữ chúng nhất quán sẽ tránh tình trạng TypeScript hiểu file theo một module system còn runtime hiểu theo hệ khác.

noEmit có nghĩa tsc chỉ kiểm tra kiểu. tsx sẽ transpile và chạy file trực tiếp trong lúc học, nên bạn chưa cần thư mục dist hay một build pipeline riêng.

Viết stream đầu tiênCopy Anchor Link

Tạo src/index.ts với import đầy đủ từ public entry point rxjs:

import { finalize, interval, map, take } from 'rxjs';

const statuses = ['đã nhận', 'đang xử lý', 'hoàn tất'] as const;

const orderStatus$ = interval(200).pipe(
  take(statuses.length),
  map((index) => statuses[index]),
  finalize(() => console.log('teardown: đã dọn timer')),
);

const subscription = orderStatus$.subscribe({
  next: (status) => console.log('next:', status),
  error: (error: unknown) => console.error('error:', error),
  complete: () => console.log('complete'),
});

console.log('ngay sau subscribe:', subscription.closed);

setTimeout(() => {
  console.log('sau khi kết thúc:', subscription.closed);
}, 750);

Tên biến có hậu tố $ là convention để báo rằng biến chứa Observable; đây không phải cú pháp bắt buộc của RxJS. Ví dụ mô phỏng trạng thái một đơn hàng, nhưng timer chỉ để bạn quan sát lifecycle rõ ràng, không phải cách lưu trạng thái đơn hàng production.

Type-check và chạyCopy Anchor Link

Trước hết, kiểm tra kiểu mà không sinh file JavaScript:

npx tsc --noEmit

Nếu lệnh không in lỗi và trả exit code 0, chạy chương trình:

npx tsx src/index.ts

Output xuất hiện theo thời gian như sau:

ngay sau subscribe: false
next: đã nhận
next: đang xử lý
next: hoàn tất
complete
teardown: đã dọn timer
sau khi kết thúc: true

interval(200) không phát 0 ngay lập tức; emission đầu tiên đến sau một period. Thời điểm thực tế có thể lệch đôi chút do event loop, nhưng thứ tự các dòng trên không đổi trong ví dụ này.

Đọc lifecycle từ outputCopy Anchor Link

Khi khai báo orderStatus$, bạn mới tạo một công thức. Timer bắt đầu khi subscribe() nối Observer vào pipeline.

subscribe
    │
    ▼
interval: 0 ─────► 1 ─────► 2 ─────► ...
    │              │         │
    └──────────────┴─────────┴──► take(3)
                                  │
                                  ├──► map: trạng thái
                                  ├──► Observer.complete()
                                  └──► unsubscribe upstream
                                             │
                                             ▼
                                          finalize

Lifecycle cụ thể là:

1. subscribe() trả về một Subscription; lúc này closed là false.
2. interval phát các số tăng dần. map đổi chỉ số thành trạng thái để Observer nhận qua next.
3. Sau ba giá trị, take(3) gửi complete xuống Observer và hủy subscription phía upstream.
4. Teardown của interval dọn tác vụ định kỳ; callback trong finalize chạy khi chuỗi kết thúc.
5. Lần kiểm tra sau cùng thấy subscription.closed là true.

complete là notification từ pipeline đến Observer. Teardown là quá trình giải phóng tài nguyên của execution. Hai ý liên quan nhưng không đồng nghĩa: một subscription cũng có thể teardown do error hoặc do consumer tự gọi unsubscribe().

Teardown khi hủy thủ côngCopy Anchor Link

Bây giờ thay nội dung src/index.ts bằng một stream không tự complete:

import { finalize, interval } from 'rxjs';

const heartbeat$ = interval(300).pipe(
  finalize(() => console.log('finalize: đã dọn heartbeat')),
);

const subscription = heartbeat$.subscribe({
  next: (tick) => console.log('tick:', tick),
  complete: () => console.log('complete'),
});

setTimeout(() => {
  console.log('hủy subscription');
  subscription.unsubscribe();
  console.log('closed:', subscription.closed);
}, 750);

Output mong đợi:

tick: 0
tick: 1
hủy subscription
finalize: đã dọn heartbeat
closed: true

Bạn sẽ không thấy dòng complete. Gọi unsubscribe() dừng execution và chạy teardown, nhưng không biến thao tác hủy của consumer thành notification complete. Vì thế, đừng đặt cleanup bắt buộc chỉ trong callback complete; finalize hoặc teardown của source mới bao phủ cả complete, error và hủy chủ động.

Nếu bỏ take(...) ở ví dụ trước và cũng không gọi unsubscribe(), timer tiếp tục giữ Node.js process sống. Đây là lý do một bài thử RxJS đôi khi “chạy xong output” nhưng terminal vẫn không trả prompt.

Đọc kỹ hơn về quyền sở hữu subscription tại Subscription và teardown.

Ví dụ thực tế: dừng worker bằng Ctrl+CCopy Anchor Link

Một worker thường có công việc định kỳ và cần dừng gọn khi người vận hành nhấn Ctrl+C. Bạn có thể biểu diễn cả heartbeat lẫn tín hiệu SIGINT thành Observable, rồi để takeUntil nối hai lifecycle lại:

import { finalize, fromEvent, interval, takeUntil, tap } from 'rxjs';

const shutdown$ = fromEvent(process, 'SIGINT');

const worker$ = interval(1_000).pipe(
  tap((tick) => console.log(`heartbeat ${tick}`)),
  takeUntil(shutdown$),
  finalize(() => console.log('teardown: đã dọn timer và listener')),
);

worker$.subscribe({
  complete: () => console.log('worker đã dừng'),
  error: (error: unknown) => console.error('worker lỗi:', error),
});

Chạy file rồi nhấn Ctrl+C sau vài giây. Một phiên minh họa có thể in:

heartbeat 0
heartbeat 1
^Cworker đã dừng
teardown: đã dọn timer và listener

fromEvent đăng ký listener cho SIGINT khi có subscription. Khi listener phát giá trị đầu tiên, takeUntil complete worker$; quá trình teardown gỡ cả timer của interval và listener của shutdown$, nên process có thể thoát. Ví dụ này cho thấy lợi ích thực tế của RxJS: thay vì tự giữ hai handler rồi nhớ dọn từng cái, bạn mô tả quan hệ “chạy cho đến khi có tín hiệu dừng” trong một pipeline.

Đưa RxJS vào project hiện cóCopy Anchor Link

Project TypeScript hoặc bundlerCopy Anchor Link

Nếu project đã có TypeScript, Vite, Next.js hoặc bundler khác, bạn thường chỉ cần:

npm install rxjs@7.8.2

Không cài lại typescript, tsx hay chép nguyên tsconfig.json của playground nếu project đã có toolchain. Cấu hình NodeNext ở trên dành cho việc chạy trực tiếp bằng Node.js; bundler của ứng dụng có thể cần moduleResolution khác.

Với RxJS 7.8.x, hãy ưu tiên public import từ package root:

import { fromEvent, map } from 'rxjs';

Đừng import từ rxjs/internal/.... Đường dẫn internal là chi tiết triển khai, không phải API contract dành cho ứng dụng. Các entry point chuyên biệt như rxjs/ajax, rxjs/testing hoặc rxjs/webSocket chỉ nên dùng khi bạn thật sự cần module tương ứng.

Project AngularCopy Anchor Link

Angular project thường đã có rxjs trong dependency tree. Trước khi cài thêm, kiểm tra:

npm ls rxjs

Hãy giữ phiên bản trong range mà phiên bản Angular của project hỗ trợ. Đừng ép rxjs@7.8.2 chỉ để giống playground nếu việc đó xung đột với peer dependency của framework; ở đây compatibility của ứng dụng quan trọng hơn việc khớp tuyệt đối phiên bản bài học. Phần tích hợp framework nằm tại RxJS với Angular.

Chọn range phiên bảnCopy Anchor Link

Ba cách cài phục vụ ba mục tiêu khác nhau:

Mặc định cho bài này, mình chọn pin 7.8.2. Với ứng dụng thật, mình thường để range phù hợp chính sách cập nhật của team và dựa vào lockfile cùng CI để kiểm soát thay đổi. Tránh dùng tag prerelease nếu mục tiêu chỉ là học API ổn định của RxJS 7.

Pitfalls thường gặpCopy Anchor Link

• Chạy file nhưng không có output: tạo Observable hoặc gọi pipe(...) chưa khởi động cold source. Kiểm tra xem code đã subscribe() chưa.
• Node.js process không thoát: một interval, event listener hoặc resource khác vẫn còn subscription sống. Dùng operator kết thúc phù hợp như take, takeUntil, hoặc gọi unsubscribe() ở lifecycle owner.
• Chỉ chạy tsx rồi tưởng type-safe: tsx có thể thực thi trong khi tsc vẫn tìm thấy lỗi kiểu. Giữ bước npx tsc --noEmit trong workflow.
• Import từ đường dẫn internal: autocomplete có thể gợi ý đường dẫn sâu, nhưng chúng làm code phụ thuộc vào cấu trúc bên trong thư viện. Chỉ import từ public entry point.
• Cài RxJS cả global lẫn local: code của project resolve dependency local. npm install --global rxjs không thay thế cho dependency trong package.json.
• Tạo nhiều subscription ngoài ý muốn: với cold Observable, mỗi lần subscribe() có thể tạo timer, listener hoặc request riêng. Cài đúng package không tự động chia sẻ execution.
• Tin rằng unsubscribe sẽ gọi complete: hủy chủ động chạy teardown nhưng callback complete của Observer không chạy.

Các khái niệm phía sau những lỗi này được mở rộng tại RxJS là gì?, Subscribe và Observer và Playground và debug.

Xử lý lỗi cài đặtCopy Anchor Link

Nếu lỗi xuất hiện trong một project lớn nhưng không tái hiện ở playground, khác biệt thường nằm ở module resolution, framework plugin hoặc dependency tree của project đó. So sánh package.json, tsconfig.json và kết quả npm ls rxjs trước khi thay operator trong code.

Tiếp tục lộ trìnhCopy Anchor Link

Môi trường đã ổn khi cả type-check lẫn hai ví dụ lifecycle chạy đúng. Bước tiếp theo không phải học thuộc hàng chục operator; hãy nắm chắc ai tạo execution, ai nhận notification và ai chịu trách nhiệm teardown.

Nguồn tham khảoCopy Anchor Link

• RxJS — Installation Instructions — lệnh cài package rxjs qua npm và cách package cung cấp module cho consumer.
• Gói rxjs 7.8.2 trên npm — phiên bản phát hành và metadata của package.
• RxJS 7.8.2 — package.json — public exports, runtime entry points và TypeScript declarations.
• RxJS 7.8.2 — source của interval — emission đầu tiên sau một period và chuỗi số tăng dần.
• RxJS 7.8.2 — source của take — phát tối đa số lượng giá trị đã chọn rồi complete.
• RxJS 7.8.2 — source của finalize — callback chạy khi complete, error hoặc unsubscribe chủ động.