มือถือ

การสร้างแอปที่มีปฏิกิริยาด้วย Redux, RxJS และ Redux-Observable ใน React Native

ในระบบนิเวศที่กำลังเติบโตของเว็บแอปและแอปบนอุปกรณ์เคลื่อนที่ที่หลากหลายและมีประสิทธิภาพมีสถานะที่ต้องจัดการมากขึ้นเรื่อย ๆ เช่นผู้ใช้ปัจจุบันรายการที่โหลดสถานะการโหลดข้อผิดพลาดและอื่น ๆ อีกมากมาย Redux เป็นหนึ่งในการแก้ปัญหานี้โดยการรักษาสถานะไว้ในวัตถุส่วนกลาง

ข้อ จำกัด ประการหนึ่งของ Redux คือไม่รองรับพฤติกรรมแบบอะซิงโครนัสนอกกรอบ ทางออกหนึ่งสำหรับสิ่งนี้คือ redux-observable ซึ่งใช้ RxJS ซึ่งเป็นไลบรารีที่มีประสิทธิภาพสำหรับการเขียนโปรแกรมแบบรีแอคทีฟใน JavaScript RxJS เป็นการใช้งาน ReactiveX ซึ่งเป็น API สำหรับการเขียนโปรแกรมแบบรีแอคทีฟซึ่งมาจาก Microsoft ReactiveX รวมคุณสมบัติที่ทรงพลังที่สุดบางประการของกระบวนทัศน์ปฏิกิริยาการเขียนโปรแกรมเชิงฟังก์ชันรูปแบบผู้สังเกตการณ์และรูปแบบตัววนซ้ำ

ในบทช่วยสอนนี้เราจะเรียนรู้เกี่ยวกับ Redux และการใช้งานกับ React นอกจากนี้เราจะสำรวจการเขียนโปรแกรมแบบตอบสนองโดยใช้ RxJS และวิธีที่สามารถทำให้งานอะซิงโครนัสที่ซับซ้อนและน่าเบื่อเป็นเรื่องง่ายมาก

สุดท้ายนี้เราจะได้เรียนรู้ redux-observable ซึ่งเป็นไลบรารีที่ใช้ประโยชน์จาก RxJS เพื่อทำงานแบบอะซิงโครนัสจากนั้นจะสร้างแอปพลิเคชันใน React Native โดยใช้ Redux และ redux-observable

Redux

ตามที่อธิบายตัวเองบน GitHub Redux คือ“ คอนเทนเนอร์สถานะที่คาดเดาได้สำหรับแอป JavaScript” ให้แอป JavaScript ของคุณมีสถานะทั่วโลกโดยรักษาสถานะและการดำเนินการให้ห่างจากส่วนประกอบของปฏิกิริยา

ในแอปพลิเคชัน React ทั่วไปที่ไม่มี Redux เราต้องส่งข้อมูลจากโหนดรูทไปยังชายด์ผ่านคุณสมบัติหรือ props การไหลของข้อมูลนี้สามารถจัดการได้สำหรับแอปพลิเคชันขนาดเล็ก แต่อาจซับซ้อนมากเมื่อแอปพลิเคชันของคุณเติบโตขึ้น Redux ช่วยให้เรามีส่วนประกอบที่เป็นอิสระจากกันดังนั้นเราจึงสามารถใช้มันเป็นแหล่งเดียวของความจริง

Redux สามารถใช้ใน React โดยใช้ react-redux ซึ่งจัดเตรียมการผูกสำหรับคอมโพเนนต์ React เพื่ออ่านข้อมูลจาก Redux และจัดส่งการดำเนินการเพื่ออัปเดตสถานะ Redux

Redux

Redux สามารถอธิบายได้ว่าเป็นหลักการง่ายๆสามประการ:

1. แหล่งเดียวของความจริง

สถานะ ของแอปพลิเคชันทั้งหมดของคุณจะถูกเก็บไว้ในวัตถุชิ้นเดียว วัตถุนี้ใน Redux ถูกยึดโดยไฟล์ เก็บ . ควรมีร้านเดียวในแอป Redux

» console.log(store.getState()) « { user: {...}, todos: {...} }

ในการอ่านข้อมูลจาก Redux ในส่วนประกอบ React ของคุณเราใช้ไฟล์ connect ฟังก์ชันจาก react-redux. connect ใช้อาร์กิวเมนต์สี่อาร์กิวเมนต์ซึ่งทั้งหมดนี้เป็นทางเลือก ในตอนนี้เราจะเน้นไปที่อันแรกซึ่งเรียกว่า mapStateToProps

 /* UserTile.js */ import { connect } from 'react-redux'; class UserTile extends React.Component { render() { return { this.props.user.name }
 
 
 } } function mapStateToProps(state) { return { user: state.user } } export default connect(mapStateToProps)(UserTile)

ในตัวอย่างด้านบน mapStateToProps รับสถานะ Global Redux เป็นอาร์กิวเมนต์แรกและส่งคืนวัตถุที่จะรวมเข้ากับอุปกรณ์ประกอบฉากที่ส่งผ่านไปยังองค์ประกอบหลัก

2. สถานะเป็นแบบอ่านอย่างเดียว

สถานะ Redux เป็นแบบอ่านอย่างเดียวสำหรับส่วนประกอบ React และวิธีเดียวที่จะเปลี่ยนสถานะคือการปล่อยไฟล์ หนังบู๊ . การกระทำเป็นวัตถุธรรมดาที่แสดงถึงความตั้งใจที่จะเปลี่ยนสถานะ ทุกวัตถุการกระทำต้องมี type และค่าต้องเป็นสตริง นอกเหนือจากนั้นเนื้อหาของการกระทำขึ้นอยู่กับคุณโดยสิ้นเชิง แต่แอปส่วนใหญ่จะทำตามไฟล์ ฟลักซ์มาตรฐานการกระทำ รูปแบบซึ่ง จำกัด โครงสร้างของการกระทำเพียงสี่ปุ่ม:

type ตัวระบุสตริงใด ๆ สำหรับการดำเนินการ ทุกการกระทำต้องมีเอกลักษณ์
payload ข้อมูลเสริมสำหรับการดำเนินการใด ๆ อาจเป็นได้ตลอดเวลาและมีข้อมูลเกี่ยวกับการดำเนินการ
error คุณสมบัติบูลีนทางเลือกใด ๆ ที่ตั้งค่าเป็น true หากการกระทำแสดงถึงข้อผิดพลาด สิ่งนี้คล้ายคลึงกับการปฏิเสธ Promise. string ตัวระบุสำหรับการดำเนินการ ทุกการกระทำต้องมีเอกลักษณ์ ตามอัตภาพเมื่อ error คือ true, payload ควรเป็นวัตถุข้อผิดพลาด
meta Meta สามารถเป็นค่าประเภทใดก็ได้ มีไว้สำหรับข้อมูลเพิ่มเติมใด ๆ ที่ไม่ได้เป็นส่วนหนึ่งของน้ำหนักบรรทุก

ตัวอย่างการดำเนินการมีสองตัวอย่างดังนี้

 store.dispatch({ type: 'GET_USER', payload: '21', }); store.dispatch({ type: 'GET_USER_SUCCESS', payload: { user: { id: '21', name: 'Foo' } } });

3. สถานะมีการเปลี่ยนแปลงด้วยฟังก์ชันบริสุทธิ์

สถานะ Redux ทั่วโลกถูกเปลี่ยนโดยใช้ฟังก์ชันบริสุทธิ์ที่เรียกว่าตัวลด ตัวลดใช้สถานะและการดำเนินการก่อนหน้าและส่งกลับสถานะถัดไป ตัวลดจะสร้างออบเจ็กต์สถานะใหม่แทนการกลายพันธุ์ที่มีอยู่ Redux store อาจมีตัวลดตัวเดียวหรือตัวลดหลายตัวขึ้นอยู่กับขนาดแอพ

 /* store.js */ import { combineReducers, createStore } from 'redux' function user(state = {}, action) { switch (action.type) { case 'GET_USER_SUCCESS': return action.payload.user default: return state } } function todos(state = [], action) { switch (action.type) { case 'ADD_TODO_SUCCESS': return [ ...state, { id: uuid(), // a random uuid generator function text: action.text, completed: false } ] case 'COMPLETE_TODO_SUCCESS': return state.map(todo => { if (todo.id === action.id) { return { ...todo, completed: true } } return todo }) default: return state } } const rootReducer = combineReducers({ user, todos }) const store = createStore(rootReducer)

คล้ายกับการอ่านจากรัฐเราสามารถใช้ connect ฟังก์ชันเพื่อจัดส่งการกระทำ

 /* UserProfile.js */ class Profile extends React.Component { handleSave(user) { this.props.updateUser(user); } } function mapDispatchToProps(dispatch) { return ({ updateUser: (user) => dispatch({ type: 'GET_USER_SUCCESS', user, }), }) } export default connect(mapStateToProps, mapDispatchToProps)(Profile);

RxJS

การเขียนโปรแกรมปฏิกิริยา

การเขียนโปรแกรมแบบปฏิกิริยาเป็นกระบวนทัศน์การเขียนโปรแกรมแบบเปิดเผยที่เกี่ยวข้องกับการไหลของข้อมูลใน“ สตรีม ” และด้วยการขยายพันธุ์และการเปลี่ยนแปลง RxJS ไลบรารีสำหรับการเขียนโปรแกรมแบบรีแอกทีฟใน JavaScript มีแนวคิดของ สังเกตได้ ซึ่งเป็นกระแสข้อมูลที่ผู้สังเกตการณ์สามารถทำได้ ติดตาม ถึงและผู้สังเกตการณ์นี้จะถูกส่งข้อมูลเมื่อเวลาผ่านไป

ผู้สังเกตสิ่งที่สังเกตได้คือวัตถุที่มีฟังก์ชันสามอย่าง: next, error และ complete ฟังก์ชันทั้งหมดนี้เป็นทางเลือก

 observable.subscribe({ next: value => console.log(`Value is ${value}`), error: err => console.log(err), complete: () => console.log(`Completed`), })

.subscribe ฟังก์ชั่นยังสามารถมีสามฟังก์ชันแทนวัตถุ

 observable.subscribe( value => console.log(`Value is ${value}`), err => console.log(err), () => console.log(`Completed`) )

เราสามารถสร้างสิ่งที่สังเกตได้ใหม่โดยการสร้างอ็อบเจกต์ของ observable ส่งผ่านฟังก์ชั่นที่รับผู้ติดตามหรือที่เรียกว่าผู้สังเกตการณ์ สมาชิกมีสามวิธี: next, error และ complete ผู้สมัครสมาชิกสามารถโทรครั้งต่อไปโดยใช้ค่ากี่ครั้งก็ได้ตามต้องการและ complete หรือ error ในตอนท้าย. หลังโทร complete หรือ error สิ่งที่สังเกตได้จะไม่ผลักค่าใด ๆ ลงไปในสตรีม

 import { Observable } from 'rxjs' const observable$ = new Observable(function subscribe(subscriber) { const intervalId = setInterval(() => { subscriber.next('hi'); subscriber.complete() clearInterval(intervalId); }, 1000); }); observable$.subscribe( value => console.log(`Value is ${value}`), err => console.log(err) )

ตัวอย่างด้านบนจะพิมพ์ Value is hi หลังจาก 1,000 มิลลิวินาที

การสร้างสิ่งที่สังเกตได้ด้วยตนเองทุกครั้งอาจกลายเป็นเรื่องละเอียดและน่าเบื่อ ดังนั้น RxJS จึงมีฟังก์ชันมากมายในการสร้างสิ่งที่สังเกตได้ คำที่ใช้บ่อยที่สุด ได้แก่ of, from และ ajax

ของ

of รับลำดับของค่าและแปลงเป็นสตรีม:

import { of } from 'rxjs' of(1, 2, 3, 'Hello', 'World').subscribe(value => console.log(value)) // 1 2 3 Hello World

จาก

from แปลงเกือบทุกอย่างเป็นกระแสของค่า:

ความสามัคคีในหลักการออกแบบ

 import { from } from 'rxjs' from([1, 2, 3]).subscribe(console.log) // 1 2 3 from(new Promise.resolve('Hello World')).subscribe(console.log) // 'Hello World' from(fibonacciGenerator).subscribe(console.log) // 1 1 2 3 5 8 13 21 ...

อาแจ็กซ์

ajax รับ URL สตริงหรือสร้างสิ่งที่สังเกตได้ซึ่งทำให้คำขอ HTTP ajax มีฟังก์ชัน ajax.getJSON ซึ่งส่งคืนเฉพาะอ็อบเจ็กต์การตอบกลับที่ซ้อนกันจากการเรียก AJAX โดยไม่มีคุณสมบัติอื่นที่ส่งคืนโดย ajax():

 import { ajax } from 'rxjs/ajax' ajax('https://jsonplaceholder.typicode.com/todos/1').subscribe(console.log) // {request, response: {userId, id, title, completed}, responseType, status} ajax.getJSON('https://jsonplaceholder.typicode.com/todos/1').subscribe(console.log) // {userId, id, title, completed} ajax({ url, method, headers, body }).subscribe(console.log) // {...}

มีอีกหลายวิธีที่จะทำให้สังเกตได้ (คุณสามารถดูรายการทั้งหมดได้ ที่นี่ ).

ตัวดำเนินการ

Operators เป็นโรงไฟฟ้าที่แท้จริงของ RxJS ซึ่งมีตัวดำเนินการสำหรับเกือบทุกอย่างที่คุณต้องการ เนื่องจาก RxJS 6 ตัวดำเนินการไม่ใช่เมธอดบนวัตถุที่สังเกตได้ แต่ใช้ฟังก์ชันบริสุทธิ์บนวัตถุที่สังเกตได้โดยใช้ .pipe วิธี.

แผนที่

map ใช้ฟังก์ชันอาร์กิวเมนต์เดียวและใช้การฉายภาพกับแต่ละองค์ประกอบในสตรีม:

 import { of } from 'rxjs' import { map } from 'rxjs/operators' of(1, 2, 3, 4, 5).pipe( map(i=> i * 2) ).subscribe(console.log) // 2, 4, 6, 8, 10

แผนที่

กรอง

filter ใช้อาร์กิวเมนต์เดียวและลบค่าออกจากสตรีมซึ่งส่งคืนเท็จสำหรับฟังก์ชันที่กำหนด:

 import { of } from 'rxjs' import { map, filter } from 'rxjs/operators' of(1, 2, 3, 4, 5).pipe( map(i => i * i), filter(i => i % 2 === 0) ).subscribe(console.log) // 4, 16

กรอง

แบนแผนที่

flatMap โอเปอเรเตอร์รับฟังก์ชั่นที่จับคู่ไอเท็มทุกชิ้นในสตีมเข้ากับสตรีมอื่นและแบนค่าทั้งหมดของสตรีมเหล่านี้:

 import { of } from 'rxjs' import { ajax } from 'rxjs/ajax' import { flatMap } from 'rxjs/operators' of(1, 2, 3).pipe( flatMap(page => ajax.toJSON(`https://example.com/blog?size=2&page=${page}`)), ).subscribe(console.log) // [ { blog 1 }, { blog 2 }, { blog 3 }, { blog 4 }, { blog 5 }, { blog 6 } ]

FlatMap

ไป

merge รวมรายการจากสองสตรีมตามลำดับที่มาถึง:

 import { interval, merge } from 'rxjs' import { pipe, take, mapTo } from 'rxjs/operators' merge( interval(150).pipe(take(5), mapTo('A')), interval(250).pipe(take(5), mapTo('B')) ).subscribe(console.log) // A B A A B A A B B B

มีรายชื่อตัวดำเนินการทั้งหมด ที่นี่ .

Redux-Observable

ตามการออกแบบการดำเนินการทั้งหมดใน Redux เป็นแบบซิงโครนัส Redux-observable เป็นมิดเดิลแวร์สำหรับ Redux ที่ใช้สตรีมที่สังเกตได้เพื่อทำงานแบบอะซิงโครนัสจากนั้นส่งการดำเนินการอื่นใน Redux ด้วยผลลัพธ์ของการทำงานแบบอะซิงโครนัสนั้น

Redux-observable ขึ้นอยู่กับแนวคิดของ มหากาพย์ . มหากาพย์คือฟังก์ชันที่รับกระแสของการกระทำและเป็นทางเลือกที่จะสตรีมของสถานะและส่งกลับกระแสของการกระทำ

function (action $: Observable, state $: StateObservable): Observable;

ตามอนุสัญญาทุกตัวแปรที่เป็นสตรีม (_aka _observable) ลงท้ายด้วย $ ก่อนที่เราจะใช้ redux-observable เราต้องเพิ่มเป็นมิดเดิลแวร์ในร้านของเรา เนื่องจากมหากาพย์เป็นกระแสของสิ่งที่สังเกตได้และทุกการกระทำที่ออกจากไอน้ำนี้จะถูกส่งกลับเข้าไปในสตรีมการส่งคืนการกระทำเดิมจะทำให้เกิดการวนซ้ำที่ไม่มีที่สิ้นสุด

 const epic = action$ => action$.pipe( filter(action => action.type === 'FOO'), mapTo({ type: 'BAR' }) // not changing the type of action returned // will also result in an infinite loop ) // or import { ofType } from 'redux-observable' const epic = action$ => action$.pipe( ofType('FOO'), mapTo({ type: BAZ' }) )

คิดว่าสถาปัตยกรรมรีแอคทีฟนี้เป็นระบบของท่อที่เอาท์พุทของท่อทุกท่อป้อนกลับเข้าไปในท่อทุกท่อรวมทั้งตัวมันเองและในตัวลดของ Redux ด้วย เป็นตัวกรองที่อยู่ด้านบนของท่อเหล่านี้ที่กำหนดว่าอะไรจะเข้าและสิ่งที่ถูกปิดกั้น

มาดูกันว่ามหากาพย์ Ping-Pong จะทำงานอย่างไร ต้องใช้ ping ส่งไปยังเซิร์ฟเวอร์และหลังจากที่คำขอเสร็จสมบูรณ์จะส่ง pong กลับไปที่แอป

 const pingEpic = action$ => action$.pipe( ofType('PING'), flatMap(action => ajax('https://example.com/pinger')), mapTo({ type: 'PONG' }) ) Now, we are going to update our original todo store by adding epics and retrieving users. import { combineReducers, createStore } from 'redux' import { ofType, combineEpics, createEpicMiddleware } from 'redux-observable'; import { map, flatMap } from 'rxjs/operators' import { ajax } from 'rxjs/ajax' // ... /* user and todos reducers defined as above */ const rootReducer = combineReducers({ user, todos }) const epicMiddleware = createEpicMiddleware(); const userEpic = action$ => action$.pipe( ofType('GET_USER'), flatMap(() => ajax.getJSON('https://foo.bar.com/get-user')), map(user => ({ type: 'GET_USER_SUCCESS', payload: user })) ) const addTodoEpic = action$ => action$.pipe( ofType('ADD_TODO'), flatMap(action => ajax({ url: 'https://foo.bar.com/add-todo', method: 'POST', body: { text: action.payload } })), map(data => data.response), map(todo => ({ type: 'ADD_TODO_SUCCESS', payload: todo })) ) const completeTodoEpic = action$ => action$.pipe( ofType('COMPLETE_TODO'), flatMap(action => ajax({ url: 'https://foo.bar.com/complete-todo', method: 'POST', body: { id: action.payload } })), map(data => data.response), map(todo => ({ type: 'COMPLEE_TODO_SUCCESS', payload: todo })) ) const rootEpic = combineEpics(userEpic, addTodoEpic, completeTodoEpic) const store = createStore(rootReducer, applyMiddleware(epicMiddleware)) epicMiddleware.run(rootEpic);

_ สิ่งสำคัญ: มหากาพย์ก็เหมือนกับสตรีมที่สังเกตได้อื่น ๆ ใน RxJS อาจอยู่ในสถานะสมบูรณ์หรือผิดพลาด หลังจากสถานะนี้มหากาพย์และแอปของคุณจะหยุดทำงาน ดังนั้นคุณต้องจับทุกข้อผิดพลาดที่อาจเกิดขึ้นในไอน้ำ คุณสามารถใช้ __catchError__ ตัวดำเนินการสำหรับสิ่งนี้ ข้อมูลมากกว่านี้: ข้อผิดพลาดในการจัดการในตัวสำรองที่สังเกตได้ .

แอปสิ่งที่ต้องทำปฏิกิริยา

เมื่อเพิ่ม UI บางส่วนแอปสาธิต (ขั้นต่ำ) จะมีลักษณะดังนี้:

แอปสิ่งที่ต้องทำปฏิกิริยา

ซอร์สโค้ดสำหรับแอพนี้มีให้บน Github . ลองใช้โครงการบน เอ็กซ์โป หรือสแกนโค้ด QR ด้านบนในแอป Expo

สรุปบทช่วยสอน React, Redux และ RxJS

เราได้เรียนรู้ว่าแอปที่ตอบสนองคืออะไร เรายังได้เรียนรู้เกี่ยวกับ Redux , RxJS และ redux-observable และยังสร้างแอป Todo แบบตอบสนองใน Expo ด้วย React Native สำหรับ ตอบสนอง และ ตอบสนองดั้งเดิม นักพัฒนาแนวโน้มปัจจุบันเสนอทางเลือกในการจัดการของรัฐที่มีประสิทธิภาพมาก

อีกครั้งซอร์สโค้ดสำหรับแอปนี้เปิดอยู่ GitHub . อย่าลังเลที่จะแบ่งปันความคิดของคุณเกี่ยวกับการจัดการสถานะสำหรับแอปที่มีปฏิกิริยาในความคิดเห็นด้านล่าง

ทำความเข้าใจพื้นฐาน

Redux คืออะไร?

Redux เป็นคอนเทนเนอร์ระดับโลกที่คาดเดาได้สำหรับแอป JavaScript มันแยกการจัดการสถานะและตรรกะการเผยแพร่การเปลี่ยนแปลงออกจากส่วนประกอบปฏิกิริยาของแอปพลิเคชัน

การเขียนโปรแกรมปฏิกิริยาคืออะไร?

การเขียนโปรแกรมแบบปฏิกิริยาเป็นกระบวนทัศน์การเขียนโปรแกรมแบบเปิดเผยที่เกี่ยวข้องกับข้อมูลเป็นสตรีมและการแพร่กระจายของการเปลี่ยนแปลงสถานะผ่านสตรีมนั้น

ReactiveX คืออะไร?

ReactiveX เป็น API สำหรับการเขียนโปรแกรมแบบอะซิงโครนัสกับสตรีมที่สังเกตได้ เป็นการผสมผสานแนวคิดที่ดีที่สุดจากรูปแบบ Observer รูปแบบ Iterator และการเขียนโปรแกรมเชิงฟังก์ชัน

RxJS คืออะไร?

RxJS เป็นไลบรารีการใช้งานของ ReactiveX API ใน JavaScript โดยใช้ประโยชน์จากลักษณะไดนามิกของ JavaScript พร้อมกับ ReactiveX API ที่กว้างขวางและกว้างขวาง

ใครเป็นผู้สร้าง ReactiveX และ RxJS

Erik Meijer นักวิทยาศาสตร์คอมพิวเตอร์ของ Microsoft ได้สร้าง ReactiveX API และนำไปใช้งานใน. NET RxJS ถูกสร้างขึ้นเป็นพอร์ตของ Rx.NET โดยวิศวกรหลายคน

RxJS น่าเรียนไหม?

RxJS มี API มากมายสำหรับจัดการงานแบบอะซิงโครนัสด้วยการเขียนโปรแกรมเชิงฟังก์ชันซึ่งอำนวยความสะดวกในการใช้การดำเนินการเช่นการลองใหม่การควบคุมปริมาณการสำรองแบบเอ็กซ์โพเนนเชียลการแคชการจัดการข้อผิดพลาดและอื่น ๆ

redux-observable คืออะไร?

redux-observable คือไลบรารีสำหรับจัดการงานอะซิงโครนัสใน Redux ทำงานโดยจับการกระทำที่ส่งมาจาก Redux และทำงานแบบอะซิงโครนัสบางอย่าง หลังจากทำงานเสร็จสิ้นแล้ว 'redux-observable' จะเพิ่มการตอบกลับลงในคอนเทนเนอร์สถานะ Redux โดยส่งการดำเนินการอื่น

เราสามารถใช้ RxJS โดยไม่ต้อง React และ Redux ได้หรือไม่?

RxJS เป็นการนำ ReactiveX มาใช้และสามารถใช้ได้กับการจัดการสถานะหรือไลบรารี UI หรือแม้กระทั่งกับวานิลลา JavaScript มี API ที่มีประสิทธิภาพในการจัดการการทำงานแบบ async ในรูปแบบของสตรีมที่สังเกตได้ สตรีมเหล่านี้สามารถเชื่อมต่อกับไลบรารีอื่น ๆ หรือใช้โดยตรง ภายนอก JavaScript คุณสามารถดูการใช้งาน ReactiveX สำหรับภาษาทั่วไปได้ใน reactivex.io