RX IOBservable كأن خط أنابيب

Question

حاليا، أستخدم إطار RX لتنفيذ خط أنابيب معالجة الرسائل تشبه سير العمل. في الأساس، لدي منتج للرسائل (Deserializes رسائل الشبكة والمكالمات OnNext () في موضوع) ولدي العديد من المستهلكين.

ملاحظة: إذا كانت الطرق وتحويلها هي طرق تمديد، فقد قمت ببساطة بإرجاع Iobservable.

المستهلك يفعل شيئا مثل ما يلي:

 var commerceRequest = messages.Transform(x => GetSomethingFromDatabase(x)
                              .Where(y => y.Value > 5)
                              .Select(y => y.ComplexObject)
                              .If(z => z.IsPaid, respond(z))
                              .Do(z => SendError(z));

commerceRequest ثم يتم استهلاكها من قبل خط أنابيب آخر مماثل وهذا يستمر حتى الأعلى حيث ينتهي مع شخص يدعو Subscribe() على خط الأنابيب النهائي. القضية التي أواجهها هي أن الرسائل من القاعدة لا تنتشر ما لم يتم استدعاء الاشتراك الرسائل مباشرة في مكان ما.

كيف يمكنني دفع الرسائل حتى أعلى المكدس؟ أعلم أن هذا نهج غير تقليدي ولكن أشعر أنه يجعل الرمز بسيطا جدا لفهم ما يحدث لرسالة. هل يمكن لأي شخص أن يقترح طريقة أخرى للقيام بنفس الشيء إذا كنت تشعر أن هذه فكرة فظيعة تماما؟

Answer 1

لماذا يجب أن يذهبوا من خلال خط الأنابيب إذا لم يكن هناك مشتركين؟ إذا كانت إحدى الخطوات المتوسطة مفيدة لآثارها الجانبية (تريد تشغيلها حتى إذا لم يكن هناك مشتركي آخرون)، فيجب عليك إعادة كتابة عملية التأثير الجانبي لتكون مشتركا.

يمكنك أيضا إجراء الخطوة ذات التأثير الجانبي كعملية تمريرة (أو نقطة مؤقت، إذا كنت تريد) إذا كنت ترغب في متابعة السلسلة.