วิธีการออกแบบระบบแบบกระจายที่ควบคุมการเคลื่อนไหวของวัตถุโดยใช้ RxJx, Node และ WebSockets

ในบทความก่อนหน้าของฉัน วิธีคิดเชิงโต้ตอบและสร้างภาพเคลื่อนไหวให้กับวัตถุที่กำลังเคลื่อนที่โดยใช้ RxJs ฉันอธิบายวิธีสร้างคลาส MobileObject ที่จำลองการเคลื่อนไหวของวัตถุที่มีการเร่งความเร็วที่กำหนดโดยตัวควบคุมภายนอก .

ตอนนี้ ฉันต้องการแสดงให้คุณเห็นระบบแบบกระจายที่เรียบง่ายที่ช่วยให้แอป ตัวควบคุม สามารถควบคุมการเคลื่อนไหวของ MobileObject จากระยะไกล แอประยะไกลตัวที่สอง จอภาพ strong> แสดงการเคลื่อนที่ของวัตถุในแผนสองมิติ ที่ศูนย์กลางของระบบจะมี MobileObjectServer ซึ่งเป็นที่ที่ MobileObjects อาศัยอยู่

เป้าหมายของบทความนี้คือการอธิบายว่าการคิดเชิงรับสามารถสร้างการออกแบบที่แมปความต้องการอย่างเป็นธรรมชาติและสร้างวิธีแก้ปัญหาที่เรียบร้อยได้อย่างไร เราจะแก้ไขปัญหาในการสมัครรับข้อมูลเพียงรายการเดียวที่สังเกตได้

เราจะมุ่งเน้นไปที่ส่วนของเซิร์ฟเวอร์ซึ่งเป็นสิ่งที่น่าสนใจที่สุดจากมุมมองนี้

สำหรับการนำไปใช้งาน เราจะใช้ RxJs และ TypeScript เซิร์ฟเวอร์ทำงานบนโหนด ส่วนประกอบทั้งหมดสื่อสารโดยใช้ Web-Sockets

ฐานโค้ดแบบเต็มซึ่งประกอบด้วย Server Controller และ Monitor สามารถพบได้ที่นี่ ที่นี่

สคีมาของระบบแบบกระจาย

สคีมาแบบลอจิคัลของระบบแบบกระจายแสดงอยู่ในแผนภาพต่อไปนี้:

ที่ตรงกลางจะวาง MobileObjectServer ซึ่งเป็นที่ที่อินสแตนซ์ของ MobileObjets ทำงาน MobileObject แต่ละรายการถูกควบคุมโดย ตัวควบคุม ของมัน นั่นคือเว็บแอปพลิเคชันที่เราสามารถออกคำสั่ง (เช่น การเร่งความเร็ว เบรก) ไปยัง MobileObject การเคลื่อนไหวของ MobileObjects ทั้งหมดสามารถเห็นได้บน จอภาพ อย่างน้อยหนึ่งจอ จอภาพ แต่ละตัวจะเป็นเว็บแอปอีกครั้ง

แผนภาพต่อไปนี้แสดงลำดับการโต้ตอบตัวอย่างระหว่าง ตัวควบคุม หนึ่งตัว จอภาพ หนึ่งตัว และ MobileObjectServer

ข้อกำหนดของเซิร์ฟเวอร์ในแง่ของเหตุการณ์

เราสามารถแสดงข้อกำหนดสำหรับส่วนเซิร์ฟเวอร์ของระบบแบบกระจายของเราในแง่ของเหตุการณ์:

• เหตุการณ์ 1 — เมื่อ ตัวควบคุม เชื่อมต่อ => สร้าง MobileObject
• เหตุการณ์2 — เมื่อ ตัวควบคุม ได้รับคำสั่ง => ส่งต่อคำสั่งไปยัง MobileObject ที่ควบคุมโดย ตัวควบคุม
• เหตุการณ์ 3 — เมื่อ ตัวควบคุม ยกเลิกการเชื่อมต่อ =› ลบ MobileObjectที่ควบคุมโดย ตัวควบคุม
• เหตุการณ์ 4 — เมื่อ จอภาพ เชื่อมต่อ => เริ่มส่งข้อมูลไดนามิกของ MobileObjects ที่ทำงานอยู่ทั้งหมดไปยัง จอภาพ ที่เชื่อมต่อใหม่
• เหตุการณ์5 — เมื่อมีการเพิ่ม MobileObject => เริ่มส่งข้อมูลไดนามิกไปยัง จอภาพ ที่เชื่อมต่อทั้งหมด
• เหตุการณ์ 6 — เมื่อ จอภาพ ยกเลิกการเชื่อมต่อ =› หยุดส่งข้อมูลไดนามิกสำหรับ MobileObjects ทั้งหมดไปยัง จอภาพ นั้น

การคิดเชิงรับจะทำให้เกิดการออกแบบที่สอดคล้องกับความต้องการที่แสดงออกมาในลักษณะนี้

องค์ประกอบที่ประกอบเป็นเซิร์ฟเวอร์

ส่วนประกอบเซิร์ฟเวอร์ของแอปพลิเคชันแบบกระจายประกอบด้วยองค์ประกอบหลักสองประการ:

• คลาส MobileObject ซึ่งใช้ตรรกะการเคลื่อนไหวแบบไดนามิกโดยใช้ RxJs Observables ซึ่งมีการอธิบายไว้โดยละเอียดแล้ว ที่นี่
• MobileObjectServerคลาสซึ่งจัดการโปรโตคอลเว็บซ็อกเก็ต รับคำสั่งจาก คอนโทรลเลอร์ และส่งไปยัง จอภาพ ข้อมูลทั้งหมดเกี่ยวกับไดนามิกของ MobileObject การใช้งานนี้ได้รับแรงบันดาลใจจาก บทความนี้ จาก Luis Aviles

MobileObject API

มาดูภาพรวมคร่าวๆ ของคลาส MobileObject กัน รายละเอียดทั้งหมดมีอยู่ "ที่นี่" ขณะที่โค้ดอยู่ใน "พื้นที่เก็บข้อมูลนี้"

MobileObject มี API สองกลุ่ม

วิธีแรกคือชุดวิธีการที่ ตัวควบคุม ภายนอกสามารถออกคำสั่งที่ส่งผลต่อไดนามิกของวัตถุได้ (เช่น การเร่งความเร็ว เบรก)

ส่วนที่สองคือกระแสข้อมูลแบบอ่านอย่างเดียวซึ่งสื่อสารกับไคลเอนต์ภายนอก จอภาพ ซึ่งเป็นข้อมูลที่เกี่ยวข้องกับพฤติกรรมแบบไดนามิกของวัตถุ (นั่นคือ ตำแหน่งและความเร็วเมื่อเวลาผ่านไป)

ในการย้ายอินสแตนซ์ของ MobileObject ตัวควบคุม จะต้องเปิดใช้งาน (ด้วยวิธีการ turnOn()) ใช้การเร่งความเร็วที่ต้องการ (ด้วยวิธีการ accelerateX(acc: number) และ accelerateY(acc: number)) แล้วอาจจะเบรก (ด้วยวิธี brake())

เมื่อ จอภาพ เชื่อมต่อกับ MobileObjectServer MobileObjectServerจะสมัครรับ dynamicsObs และ MobileObjects ที่สังเกตได้ที่ทำงานอยู่ เซิฟเวอร์. จากนั้นจะเริ่มส่งข้อมูลที่เกี่ยวข้องกับการเคลื่อนไหวไปยังจอภาพที่เชื่อมต่ออยู่

เพื่อวัตถุประสงค์ของบทความนี้ นี่คือทั้งหมดที่คุณต้องรู้เกี่ยวกับ MobileObject

ซ็อกเก็ตเป็นสิ่งที่สังเกตได้

MobileObjectServer เริ่มดำเนินการบางอย่างเมื่อไคลเอ็นต์ ไม่ว่าจะเป็น ตัวควบคุม หรือ จอภาพ เปิดการเชื่อมต่อ websocket เมื่อเวลาผ่านไป MobileObjectServerสามารถรับคำขอจำนวนมากเพื่อเปิดการเชื่อมต่อจากไคลเอ็นต์จำนวนมาก

ดูเหมือนว่าซ็อกเก็ตที่สังเกตได้ นี่คือวิธีการรับมันโดยใช้ไลบรารี socket.io:

ผ่านฟังก์ชัน sockets เราสร้าง Observable ของ SocketObs (เราจะเห็นการใช้งานคลาสนี้ในภายหลัง) เมื่อใดก็ตามที่เซิร์ฟเวอร์ websocket ได้รับคำขอ เชื่อมต่อ และสร้าง ซ็อกเก็ต ใหม่ Observable ที่ส่งคืนโดยฟังก์ชันนี้จะปล่อยอินสแตนซ์ของ SocketObs ซึ่งล้อมรอบ socket เพิ่งสร้างขึ้น

ข้อความบนซ็อกเก็ตเป็นสิ่งที่สังเกตได้

สามารถใช้ซ็อกเก็ตเพื่อส่งข้อความจากไคลเอนต์ไปยังเซิร์ฟเวอร์และในทางกลับกัน ด้วยไลบรารี socket.io เราสามารถส่งข้อความโดยใช้เมธอด emit

SocketIO.Socket.emit(event: string, …args: any[]): SocketIO.Socket

พารามิเตอร์ event สามารถเห็นได้เป็นตัวระบุประเภทของข้อความที่เราต้องการส่ง พารามิเตอร์ …args สามารถใช้เพื่อส่งข้อมูลเฉพาะสำหรับข้อความเดียว

ใครก็ตามที่สนใจข้อความบางประเภท (หรือเหตุการณ์ที่ต้องการใช้คำศัพท์ socket.io) สามารถเริ่มฟังบนซ็อกเก็ตโดยใช้วิธี on

SocketIO.Emitter.on(event: string, fn: Function): SocketIO.Emitter

ขอย้ำอีกครั้งว่าลำดับของข้อความที่ผู้รับได้รับจะมีลักษณะเหมือนกับสิ่งที่สังเกตได้ นี่คือวิธีที่เราสามารถสร้าง Observables ที่เปล่งออกมาจริง ๆ ทุกครั้งที่ได้รับข้อความบางประเภท

เมธอด onMessageType เป็นเมธอดที่ใช้กลอุบาย ส่งคืน Observable ซึ่งส่งเสียงทุกครั้งที่ได้รับข้อความประเภท messageType

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

กำหนดลักษณะของลูกค้า

มีไคลเอนต์สองประเภทที่สามารถเชื่อมต่อกับ MobileObjectServer อันหนึ่งคือตัวควบคุม และอีกอันคือจอภาพ ก่อนอื่น MobileObjectServer จำเป็นต้องพิจารณาว่าไคลเอ็นต์ประเภทใดที่จะจัดการกับซ็อกเก็ตเฉพาะ

วิธีที่เราเลือกใช้ตรรกะดังกล่าวคือให้ ผู้ควบคุม และ ผู้ตรวจสอบ ส่งข้อความประเภทต่างๆ เป็นข้อความแรก

• ตัวควบคุม ส่งข้อความประเภท BIND_CONTROLLER
• จอภาพ ส่งข้อความประเภท BIND_MONITOR

ขึ้นอยู่กับประเภทของข้อความแรกที่ได้รับบนซ็อกเก็ต MobileObjectServer สามารถระบุได้ว่ากำลังสื่อสารกับ คอนโทรลเลอร์ หรือ จอภาพ .

ทันทีที่มีการสร้างซ็อกเก็ต MobileObjectServer จะต้องเริ่มฟังข้อความทั้งสองประเภท BIND_CONTROLLER และ BIND_MONITOR คนแรกที่เกิดขึ้นจะเป็นผู้ชนะ มันคือ race ระหว่างสอง Observables ซึ่งแมปข้อความสองประเภทที่แตกต่างกัน

ตรรกะดังกล่าวจะต้องทำซ้ำทุกครั้งที่มีการสร้างซ็อกเก็ตใหม่ นั่นคือเวลาที่ Observable ส่งคืนโดยฟังก์ชัน sockets ปล่อยเสียง ดังนั้นเราจึงต้องรวมการแข่งขันทั้งหมดที่ชนะการแข่งขัน เราจำเป็นต้องใช้ตัวดำเนินการ mergeMap ซึ่งรวมเหตุการณ์ทั้งหมดที่เกิดขึ้นโดย Observables ที่เกี่ยวข้อง และทำให้ผลลัพธ์เรียบลงใน Observable ใหม่ (mergeMap เดิมเรียกว่า flatMap)

รหัสเพื่อให้ได้ผลลัพธ์นี้มีดังต่อไปนี้:

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

เหตุการณ์ที่เกี่ยวข้องกับจอภาพ

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

การเพิ่มและการลบ MobileObjects

เหตุการณ์ที่เกี่ยวข้องครั้งแรกคือ:

• มีการเพิ่ม MobileObject แล้ว => MobileObject ปรากฏบน จอภาพ
• MobileObject ถูกลบออก => MobileObject ถูกลบออกจาก จอภาพ

MobileObjects จะถูกเพิ่มหรือลบออกเมื่อเวลาผ่านไป ดังนั้นเหตุการณ์ดังกล่าวสามารถสร้างแบบจำลองด้วย Observables สองรายการ:

• Observable ซึ่งส่งเสียงเมื่อมีการเพิ่ม MobileObject
• Observable ซึ่งส่งเสียงเมื่อ MobileObject ถูกลบออก

เมื่อเชื่อมต่อ จอภาพ แล้ว MobileObjectServer จะเริ่มสนใจ Observables ทั้งสองรายการ ดังนั้นจึงต้องมี merge รายการ:

เช่นเดียวกับที่เราเคยเห็นมาก่อน เราจำเป็นต้องทำซ้ำตรรกะดังกล่าวทุกครั้งที่มีการเพิ่ม จอภาพ ดังนั้นเราจึงจำเป็นต้อง mergeMap สิ่งที่สังเกตได้ทั้งหมดซึ่งเป็นผลมาจาก merge ของ 'วัตถุเคลื่อนที่ที่ถูกเพิ่ม' ที่สังเกตได้พร้อมกับ 'วัตถุมือถือที่ถูกลบออก' ที่สังเกตได้

นี่คือโค้ดที่จะได้รับ Observable ซึ่งจะส่งเสียงทุกครั้งที่มีการเพิ่มหรือลบ MobileObject ออกจากทุก จอภาพ:

เราได้แนะนำบางสิ่งด้วยโค้ดนี้ซึ่งควรค่าแก่การแสดงความคิดเห็นที่นี่

เราได้สร้างคลาส MobileObjectServer ซึ่งจะเป็นที่ที่เราจะเขียนโค้ดตรรกะเซิร์ฟเวอร์ทั้งหมดของเรานับจากนี้เป็นต้นไป

เมธอด handleMonitorsObs ซึ่งเราจะปรับปรุงในภายหลัง จะส่งคืนเพียง merge ของ Observables สองรายการ คือ mobileObjectAdded และ mobileObjectRemoved ซึ่งเป็น Subjects นี่คือ “ภายใน” merge ดังภาพด้านบน

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

จุดสุดท้ายเกี่ยวข้องกับโค้ดที่เราเพิ่มในเมธอด startSocketServer:

race(
   socket.onMessageType(MessageType.BIND_MONITOR)
   .pipe(
      map(() => (sObs: SocketObs) => this.handleMonitorObs(sObs))
   ),
   socket.onMessageType(MessageType.BIND_CONTROLLER)
   // something will be added here soon to make this logic work
)
.pipe(
   mergeMap(handler => handler(socket))
)

นี่เป็นวิธีการพูดโดยพื้นฐาน: ทุกครั้งที่ได้รับข้อความ BIND_MONITOR ให้ส่งคืนฟังก์ชัน

(socketObs: SocketObs) => this.handleMonitorObs(socketObs)

ซึ่งจะถูกดำเนินการภายในตัวดำเนินการ mergeMap ที่ไพพ์ไปยังผลลัพธ์ของฟังก์ชัน race ตัวดำเนินการ mergeMap นี้คือ mergeMap ภายนอกที่แสดงในภาพด้านบน

อีกวิธีหนึ่งในการอ่านโค้ดคือ: เหตุการณ์ใดๆ ที่สอดคล้องกับข้อความประเภท BIND_MONITOR จะถูกแปลงโดยลอจิกของ

mergeMap(() => this.handleMonitorObs(socket))

โดยที่ socket คืออินสแตนซ์ประเภท SocketsObs ที่ปล่อยออกมาจากฟังก์ชัน race

เร็วๆ นี้ เราจะเพิ่มสิ่งที่คล้ายกันสำหรับเคส BIND_CONTROLLER เพื่อให้ตรรกะทั้งหมดนี้ใช้งานได้

จัดการการเปลี่ยนแปลงของ MobileObject ที่สังเกตได้

ลองพิจารณาจอภาพหนึ่งตัวที่เชื่อมต่อกับ MobileObjectServer หลังจากการเชื่อมต่อ จะมีการเพิ่ม MobileObject สองสามรายการใน MobileObjectServer

ตอนนี้สำหรับ MobileObject แต่ละรายการ เราต้องเริ่มพิจารณาไดนามิกที่สังเกตได้ที่นำเสนอโดยเป็นส่วนหนึ่งของ API สิ่งที่สังเกตได้เหล่านี้จะปล่อยข้อมูลเกี่ยวกับการเปลี่ยนแปลง (ตำแหน่งและความเร็ว) ของ MobileObject ในช่วงเวลาสม่ำเสมอ หาก mobileObject เก็บการอ้างอิงถึง MobileObject เราจะสามารถรับไดนามิกที่สังเกตได้ผ่านทาง mobileObject.dynamicsObs (ดู MobileObject API)

ขั้นแรก เราต้องแปลงแต่ละเหตุการณ์ที่แสดงถึงข้อเท็จจริงที่ว่า MobileObject ได้ถูกเพิ่มเข้าไปในชุดของเหตุการณ์ที่ปล่อยออกมาโดย dynamicsObs จากนั้นเรา mergeMap ซีรีส์ทั้งหมดเหล่านี้ให้เป็นซิงเกิลใหม่ Observable ซึ่งปล่อยเหตุการณ์ไดนามิกทั้งหมดสำหรับ MobileObjects ทั้งหมดที่เพิ่มเข้ามา

จากนั้นเราจะใช้แจ๊สทั้งหมดนี้กับ จอภาพ ทั้งหมดที่เชื่อมต่อกับ MobileObjectServer ดังนั้นเราจึงจบลงด้วย Observable ใหม่ซึ่งจะปล่อยข้อมูลไดนามิกสำหรับ จอภาพ ทั้งหมด strong> และ MobileObjects ทั้งหมด (รวมถึงเหตุการณ์ทั้งหมดที่เกี่ยวข้องกับข้อเท็จจริงที่ว่า MobileObject ถูกลบแล้ว)

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

เมื่อชัดเจนแล้ว โค้ดก็จะง่ายมาก:

เราเพิ่งนำเสนอการเปลี่ยนแปลงง่ายๆ เพียงครั้งเดียว เราเปลี่ยนวิธีการ handleMonitorObs เพื่อเพิ่มตัวดำเนินการ mergeMap สิ่งนี้จะแปลง mobileObjectAdded Observable เพื่อให้ Observable ใหม่ปล่อยข้อมูลไดนามิกที่เรากำลังมองหา

ส่วนที่เหลือยังคงไม่ถูกแตะต้อง

สรุปจนถึงตอนนี้

เราได้ทำอะไรไปแล้วบ้าง? เราเพิ่งเปลี่ยน Observables เพื่อรับ Observables ใหม่ซึ่งจะปล่อยเหตุการณ์ทั้งหมดที่ MobileObjectServer สนใจเมื่อต้องจัดการกับ Monitor ไม่มีอะไรอีกแล้ว.

คุณสามารถดูได้ว่าการเปลี่ยนแปลงเหล่านี้สะท้อนให้เห็นอย่างไรในโค้ดในภาพต่อไปนี้:

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

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

เหตุการณ์ที่เกี่ยวข้องกับผู้ควบคุม

เมื่อ ตัวควบคุม เชื่อมต่อกับ MobileObjectServer มีบางสิ่งที่เซิร์ฟเวอร์จำเป็นต้องให้ความสำคัญน้อยลง อย่างน้อยก็มีเหตุการณ์ที่เกี่ยวข้องซ้อนกันเกิดขึ้นน้อยลง

สิ่งที่ MobileObjectServer ต้องใส่ใจคือ:

• คอนโทรลเลอร์ ได้เชื่อมต่อแล้ว ซึ่งในตรรกะง่ายๆ ของเราหมายความว่าเราต้องสร้าง MobileObject ใหม่เอี่ยม
• ตัวควบคุม ได้ส่งคำสั่งสำหรับ MobileObject
• ตัวควบคุม ถูกตัดการเชื่อมต่อแล้ว ในการนำไปใช้งานของเรา หมายความว่าเราต้องลบ MobileObject ที่ควบคุมโดย ตัวควบคุม (เรามีความสัมพันธ์แบบ 1 ต่อ 1 ระหว่าง MobileObject และ ตัวควบคุม)

เรารู้เหตุการณ์แรกแล้ว: เป็นเหตุการณ์ที่ Observable ส่งกลับโดย socket.onMessageType(BIND_CONTROLLER)

ตัวควบคุม จะส่งคำสั่งไปยัง MobileObjectServer ในรูปแบบของข้อความ ดังนั้นเราจึงสามารถสร้างคำสั่งที่สังเกตได้ที่ได้รับผ่าน ซ็อกเก็ต (รับจากคอนโทรลเลอร์บางตัว) เนื่องจากคอนโทรลเลอร์แต่ละตัวมี ซ็อกเก็ตของตัวเอง เราทำสิ่งนี้โดยใช้เพียงแค่ใช้ onMessageType วิธีการ SocketObs

socket.onMessageType(CONTROLLER_COMMAND)

SocketObs ยังเสนอวิธีการ onDisconnect ซึ่งส่งคืน Observable ที่ปล่อยออกมาเมื่อ ซ็อกเก็ต ถูกตัดการเชื่อมต่อ นี่คือสิ่งที่เราต้องการเพื่อจัดการกับเหตุการณ์ที่สาม

เนื่องจากเรากำลังติดต่อกับ ตัวควบคุม มากกว่าหนึ่งตัวที่อาจเชื่อมต่อกับ MobileObjectServer คุณจึงไม่แปลกใจเลยที่ทราบว่าเราจำเป็นต้อง mergeMap ผลลัพธ์ของ merge นี่เป็นการเปลี่ยนแปลงแบบเดียวกับที่เราได้ทำไปแล้วสองสามครั้ง

รหัสไม่ควรแปลกใจเช่นกัน

เราได้เพิ่มวิธีการ handleControllerObs ที่เกี่ยวข้องกับ คำสั่งที่ได้รับ และ ตัดการเชื่อมต่อ ของคอนโทรลเลอร์ เราใช้การแปลง mergeMap กับมันเหมือนที่เราได้ทำไปแล้วกับ handleMonitorObs

สรุปการเปลี่ยนแปลงที่ใช้กับตัวควบคุม

แผนภาพต่อไปนี้แสดงการเปลี่ยนแปลงทั้งหมดที่เราใช้ โดยเริ่มต้นจาก Observable ที่ปล่อยออกมาเมื่อ คอนโทรลเลอร์ เชื่อมต่อ

สิ่งสุดท้ายที่สังเกตได้

หากเรารวบรวมการเปลี่ยนแปลงที่เราทำสำหรับทั้ง จอภาพ และ ตัวควบคุม สิ่งที่เราได้รับคือสิ่งที่สังเกตได้ขั้นสุดท้ายต่อไปนี้

เพียงสมัครรับรายการ Observable สุดท้ายนี้ กิจกรรมทั้งหมดก็จะถูกเปิดเผย

ผลข้างเคียง

แผนผังกิจกรรมที่สวยงามที่เราสร้างขึ้นโดยการสมัครรับข้อมูล Final Observable ไม่ได้ทำอะไรเลย แต่มันทำงานได้ดีในการแมป เหตุการณ์ ที่เราระบุในขณะที่อธิบายข้อกำหนดของเซิร์ฟเวอร์ในตอนต้นของบทความนี้

โดยพื้นฐานแล้วมันจะบอกเราอย่างชัดเจนเมื่อเราต้องทำบางสิ่งบางอย่าง

บางสิ่งนี้คือสิ่งที่เราเรียกว่าผลข้างเคียง

เมื่อคอนโทรลเลอร์เชื่อมต่อและยกเลิกการเชื่อมต่อ เราจะสร้างหรือลบ MobileObject ตามลำดับ ผลข้างเคียงของการกระทำเหล่านี้ก็คือ เราเพิ่มเหตุการณ์ “MobileObject added” และ “MobileObject ถูกลบ” โดยใช้หัวข้อ mobileObjectAdded และ mobileObjectRemoved ที่เราแนะนำบางส่วน ย่อหน้าที่ผ่านมา

วิธีใช้ผลข้างเคียง

ใน RxJs มีหลายวิธีในการใช้ ผลข้างเคียง

ผู้สังเกตการณ์เป็นหนึ่งเดียว เราสามารถเพิ่มผู้สังเกตการณ์ได้ในขณะที่เรา subscribe ใช้ตัวดำเนินการ tap (เดิมเรียกว่า do)

อีกวิธีหนึ่งคือการฉีดมันเข้าไปในฟังก์ชันใดๆ ที่เราส่งผ่านไปยังตัวดำเนินการ RxJs

เราจะใช้ tap เป็นหลัก เนื่องจากช่วยให้เราวางผลข้างเคียงได้ทั่วทั้งแผนผังเหตุการณ์ แต่เรายังจะวางผลข้างเคียงไว้ภายในฟังก์ชันที่เราส่งไปยังตัวดำเนินการ RxJs โดยตรงด้วย

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

หวังว่า ณ จุดนี้โค้ดจะพูดเพื่อตัวมันเอง

สุดท้ายแต่ไม่ท้ายสุด: การเสร็จสิ้น Observables

มีสิ่งหนึ่งที่เรายังต้องทำเพื่อให้การออกแบบของเราเสร็จสมบูรณ์: หยุดกระแสของเหตุการณ์ หรือดำเนินการ Observables ให้เสร็จสิ้น เมื่อ ตัวควบคุม หรือ จอภาพ ยกเลิกการเชื่อมต่อ

เมื่อคอนโทรลเลอร์ตัดการเชื่อมต่อ

เมื่อตัวควบคุมยกเลิกการเชื่อมต่อ เราจะลบ MobileObject ที่ตัวควบคุมนั้นควบคุม ในส่วนหนึ่งของการลบ สิ่งสำคัญคือต้องแน่ใจว่า MobileObjectServer หยุดส่งข้อมูลไดนามิกที่เกี่ยวข้องกับ MobileObject นี้ไปยังจอภาพที่เชื่อมต่อ ซึ่งหมายความว่าเราต้องดำเนินการ Observable ต่อไปนี้ให้เสร็จสิ้น:

mobObjInfo.mobObj.dynamicsObs
.pipe(
  tap(dynamics => socket.send(MessageType.DYNAMICS_INFO, dynamics)),
)

เราสามารถบรรลุสิ่งนี้ได้อย่างง่ายดายเพียงแค่ใช้ตัวดำเนินการ takeUntil ร่วมกับ mobileObjectRemoved Observable ที่เรารู้อยู่แล้ว:

mobObjInfo.mobObj.dynamicsObs
.pipe(
  tap(dynamics => socket.send(MessageType.DYNAMICS_INFO, dynamics)),
  takeUntil(this.mobileObjectRemoved.pipe(
    filter(id => id === mobObjInfo.mobObjId)
  ))
)

takeUntil ทำให้แน่ใจว่า Observable จะเสร็จสมบูรณ์เมื่อ Observable ส่งผ่านเป็นพารามิเตอร์ไปที่ takeUntil

mobileObjectRemoved ส่งเสียงทุกครั้งที่ MobileObject ถูกลบ อย่างไรก็ตาม สิ่งที่เราต้องการคือการหยุดส่งข้อมูลไดนามิกเมื่อ MobileObject เฉพาะเจาะจงซึ่งระบุด้วยรหัสนั้นถูกลบออก ดังนั้นเราจึงเพิ่มตรรกะ filter

เมื่อจอภาพตัดการเชื่อมต่อ

ในกรณีนี้ เรายังสามารถใช้ takeUntil ได้ด้วย

เรารู้ว่าเมื่อจอภาพตัดการเชื่อมต่อเนื่องจาก socket ประเภท SocketObs ที่เกี่ยวข้องกับมันส่งเสียงผ่าน socket.onDisconnect() Observable ดังนั้นสิ่งที่เราต้องทำคือหยุดส่งข้อมูลไดนามิกเมื่อ socket.onDisconnect() ส่งเสียง

ดังนั้นตรรกะสุดท้ายที่จะควบคุมความสมบูรณ์ของสิ่งที่สังเกตได้คือ

mobObjInfo.mobObj.dynamicsObs
.pipe(
  tap(dynamics => socket.send(MessageType.DYNAMICS_INFO, dynamics)),
  takeUntil(this.stopSendDynamics(socket, mobObjInfo.mobObjId))
)

ที่ไหน

private stopSendDynamics(socket: SocketObs, mobObjId: string){
  return merge(
            this.mobileObjectRemoved.pipe(
                                       filter(id => id === mobObjId)
                                     ),
            socket.onDisconnect()
  );
}

และนี่คือลักษณะของโค้ดที่ใช้ตรรกะของเรา:

บทสรุป

มันเป็นการเดินทางที่ยาวนานมาก เราได้เห็นการให้เหตุผลบางอย่างที่ขับเคลื่อนโดยการคิดเชิงรับและการนำเหตุผลนี้ไปใช้

เราเริ่มเปลี่ยนกิจกรรม WebSockets ให้เป็น Observables จากนั้น เมื่อใช้การแปลงส่วนเพิ่ม เราก็สร้าง Observable ขึ้นมารายการเดียว ซึ่งเมื่อสมัครรับข้อมูลแล้ว ก็จะเผยให้เห็นกิจกรรมทั้งหมดที่เราสนใจ

ณ จุดนี้ การเพิ่มผลข้างเคียงที่ช่วยให้เราบรรลุเป้าหมายนั้นทำได้ตรงไปตรงมา

กระบวนการออกแบบทางจิตซึ่งเพิ่มขึ้นในตัวเองเป็นความหมายที่ฉันมอบให้กับ "การคิดเชิงรับ"

ฐานโค้ดแบบเต็ม ซึ่งประกอบด้วย Server Controller และ Monitor สามารถพบได้ที่นี่ ที่นี่