C # में कोटलिन फ्लो कार्यान्वयन

सभी को नमस्कार!

हाल के वर्षों में, मैं कोटलिन पर एंड्रॉइड के लिए विकास कर रहा हूं। बहुत समय पहले नहीं, कोटलिन मल्टीप्लेट रिकॉर्डर पर RxJava की कमी के लिए, हमने कॉटलिन और प्रवाह का उपयोग करना शुरू कर दिया - कोटलिन आउट बॉक्स के लिए ठंडी धाराएं। एंड्रॉइड से पहले, मैंने सी # के साथ कई साल बिताए, और वहां बहुत लंबे समय तक मेरे कॉरआउट रहे हैं, केवल उन्हें इस तरह से नहीं बुलाया जाता है। लेकिन मैंने एसिंक्स / वेट पर प्रवाह के एनालॉग के बारे में नहीं सुना। प्रतिक्रियाशील प्रोग्रामिंग का मुख्य उपकरण आरएक्स.नेट है (वास्तव में, आरएक्स का जन्म यहां हुआ था)। इसलिए मैंने उदासीन होने का फैसला किया और एक बाइक को देखने की कोशिश की।

यह आगे समझा गया है कि पाठक उन चीजों से अवगत है जो पिछले पैराग्राफ में चर्चा की गई थीं। अधीर के लिए - तुरंत भंडार से लिंक करें ।

डिस्क्लेमर: यह कोड उत्पादन में उपयोग किए जाने के लिए नहीं है। यह एक अवधारणा है, अधिक कुछ नहीं। कुछ बिल्कुल के रूप में काम नहीं कर सकते।

IFlow और IFlowCollector

ठीक है, चलो माथे पर सी # में फ्लो और फ्लोकोलेक्टर इंटरफेस को फिर से लिखना शुरू करें।
यह था:

interface Flow<out T> { suspend fun collect(collector: FlowCollector<T>) } interface FlowCollector<in T> { suspend fun emit(value: T) } 

यह बन गया:

  public interface IFlow<out T> { Task Collect(IFlowCollector<T> collector); } public interface IFlowCollector<in T> { Task Emit(T item); } 

मेरा मानना ​​है कि अंतर समझने और अतुल्यकालिक के अलग-अलग कार्यान्वयन द्वारा समझाया गया है।

इन इंटरफेस का उपयोग करने के लिए, उन्हें लागू किया जाना चाहिए। यहाँ क्या हुआ:

  internal class Flow<T> : IFlow<T> { private readonly Func<IFlowCollector<T>, Task> _emitter; public Flow(Func<IFlowCollector<T>, Task> emitter) { _emitter = emitter; } public Task Collect(IFlowCollector<T> collector) { return _emitter(collector); } } internal class FlowCollector<T> : IFlowCollector<T> { private readonly Func<T, Task> _handler; public FlowCollector(Func<T, Task> handler) { _handler = handler; } public Task Emit(T item) { return _handler(item); } } 

प्रवाह के निर्माता में, हम एक फ़ंक्शन पास करते हैं जो मूल्यों का उत्सर्जन करेगा। और कलेक्टर के निर्माण के लिए, एक फ़ंक्शन जो प्रत्येक उत्सर्जित मूल्य को संसाधित करेगा।

आप इसे इस तरह से उपयोग कर सकते हैं

 var flow = new Flow<int>(async collector => { await collector.Emit(1); await Task.Delay(1000); await collector.Emit(2); await Task.Delay(1000); await collector.Emit(3); }); var collector = new FlowCollector<int>(async item => Console.WriteLine(item)); await flow.Collect(collector); 

मुझे लगता है कि उपरोक्त कोड में सब कुछ स्पष्ट है। पहले हम एक फ्लो बनाते हैं, फिर एक कलेक्टर बनाते हैं (प्रत्येक तत्व के लिए हैंडलर)। फिर हम फ्लो शुरू करते हैं, उस पर एक कलेक्टर "हस्ताक्षरित" करते हैं। यदि आप थोड़ी सी चीनी (जीथूब देखें) जोड़ते हैं, तो हमें कुछ ऐसा मिलता है:

 await Flow<int>(async collector => { await collector.Emit(1); await Task.Delay(1000); await collector.Emit(2); await Task.Delay(1000); await collector.Emit(3); }) .Collect(Console.WriteLine); 

कोटलिन पर यह इस तरह दिखता है:

 scope.launch{ flow{ emit(1) delay(1000) … }.collect{ printl(it) } } 

व्यक्तिगत रूप से, मैं सभी को एक प्रवाह बनाते समय स्पष्ट रूप से तत्व के प्रकार को इंगित करने के लिए शार्प पर विकल्प पसंद नहीं करता हूं। लेकिन यहाँ बात यह नहीं है कि कोटलिन में टाइप इंट्रेंस बहुत अधिक है। प्रवाह समारोह इस तरह दिखता है:

 public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block) 

जैसा कि हम देख सकते हैं, ब्लॉक पैरामीटर को बिल्डरइंटरनेशन एनोटेशन के साथ चिह्नित किया गया है, जो कंपाइलर को बताता है कि इस पैरामीटर से टाइप लिया जाना चाहिए। क्या किसी को पता है कि क्या रोजलिन पर C # के लिए समान दर्ज करना संभव है?

CancellationToken

आरएक्स में एक सदस्यता है जिसमें से आप सदस्यता समाप्त कर सकते हैं। कोटलिन फ्लो में, नौकरी रद्द करने के लिए ज़िम्मेदार है, जिसे बिल्डर, या कॉरटीन स्कोप द्वारा लौटाया जाता है। हमें निश्चित रूप से एक उपकरण की आवश्यकता है जो फ्लो को जल्दी पूरा करने की अनुमति देता है। C # में, एसिंक्रोनस ऑपरेशंस को रद्द करने के लिए, मैं इस शब्द से नहीं डरता, कैंसिलेशन टोकन पैटर्न का उपयोग किया जाता है। कैंसेलेशनटोकन एक ऐसा वर्ग है, जिसका ऑब्जेक्ट एसिंक्रोनस ऑपरेशन को जानकारी प्रदान करता है जिसे इसे रद्द कर दिया गया है। यह स्टार्टअप पर एक अतुल्यकालिक ऑपरेशन में खुद को फेंकता है, और यह ऑपरेशन खुद अपनी स्थिति को देखता है। और राज्य बाहर से बदल जाता है।

संक्षेप में, हमें अपने प्रवाह और फ्लोकोलेक्टर में कैंसेलेशनटोकन को फेंकने की आवश्यकता है।

  public interface IFlow<out T> { Task Collect(IFlowCollector<T> collector, CancellationToken cancellationToken = default); } public interface IFlowCollector<in T> { Task Emit(T item, CancellationToken cancellationToken = default); } 

मैं यहां कार्यान्वयन को नहीं चिपकाऊंगा - जीथब देखें।
परीक्षण अब इस तरह दिखेगा:

  var cts = new CancellationTokenSource(); var flowTask = Flow<int>(async (collector, cancellationToken) => { await collector.Emit(1); await Task.Delay(2000, cancellationToken); await collector.Emit(2); await Task.Delay(2000, cancellationToken); await collector.Emit(3); }) .Collect(item => Log(item), cts.Token); var cancelationTask = Task.Run(async () => { await Task.Delay(3000); cts.Cancel(); }); await flowTask; 

बात यह है। फ्लो के समानांतर, हम एक ऑपरेशन शुरू करते हैं जो 3 सेकंड के बाद इसे रद्द कर देगा। नतीजतन, फ्लो तीसरे तत्व का उत्सर्जन करने का प्रबंधन नहीं करता है और टास्ककैन्लेड ​​अपवाद के साथ समाप्त होता है, जो आवश्यक व्यवहार है।

थोड़ा सा अभ्यास

आइए प्रयोग करने का प्रयास करें कि व्यवहार में क्या हुआ। उदाहरण के लिए, हमारे फ़्लो में कुछ ईवेंट लपेटें। Rx.Net में इसके लिए एक पुस्तकालय विधि FromEventPattern भी है।

वास्तविक UI के साथ गड़बड़ न करने के लिए, मैंने ClicksEmulator वर्ग लिखा, जो यादृच्छिक अंतराल पर सशर्त माउस क्लिक उत्पन्न करता है।

  public class ClicksEmulator { public enum Button { Left, Right } public class ClickEventArgs : EventArgs { //… public int X { get; } public int Y { get; } public Button Button { get; } } public event EventHandler<ClickEventArgs> ButtonClick; public async Task Start(CancellationToken cancellationToken = default) {… } } 

मैंने कार्यान्वयन को छोड़ दिया वह यहां बहुत महत्वपूर्ण नहीं है। मुख्य बात ईवेंट बटनक्लिक है, जिसे हम फ्लो में बदलना चाहते हैं। इसके लिए हम एक विस्तार विधि लिखते हैं

 public static IFlow<ClicksEmulator.ClickEventArgs> Clicks(this ClicksEmulator emulator) { return FlowFactory.Flow<ClicksEmulator.ClickEventArgs>(async (collector, cancellationToken) => { void clickHandler(object sender, ClicksEmulator.ClickEventArgs args) => collector.Emit(args); emulator.ButtonClick += clickHandler; cancellationToken.Register(() => { emulator.ButtonClick -= clickHandler; }); await Task.Delay(-1, cancellationToken); }); } 

सबसे पहले, हम एक ईवेंट हैंडलर घोषित करते हैं, जो कलेक्टर के पास ईवेंट के तर्क के अलावा कुछ नहीं करता है। तब हम ईवेंट रद्द करने और प्रवाह को रद्द करने (पूरा होने) के मामले में एक सदस्यता समाप्त करने के लिए सदस्यता लेते हैं। ठीक है, तो हम अंतहीन इंतजार करते हैं और रद्द करने तक बटन क्लिक करें घटनाओं को सुनें।

यदि आपने कॉलबैकफ्लो या चैनलफ्लो का उपयोग कोटलिन में किया है या आरएक्स में श्रोताओं से ठंडा ऑब्जर्वेबल बनाया है, तो आप देखेंगे कि कोड संरचना सभी मामलों में बहुत समान है। यह ठीक है, लेकिन सवाल उठता है - इस मामले में एक क्रूड घटना की तुलना में फ्लो से बेहतर क्या है? जेट स्ट्रीम की सारी शक्ति उन ऑपरेटरों में निहित है जो उन पर विभिन्न परिवर्तन करते हैं: फ़िल्टरिंग, मैपिंग, और कई अन्य, अधिक जटिल। लेकिन हमारे पास उन्हें अभी तक नहीं है। आइए इसके बारे में कुछ करने की कोशिश करें।

फ़िल्टर, मानचित्र, OnNext

आइए सबसे सरल ऑपरेटरों में से एक के साथ शुरू करें - फ़िल्टर। जैसा कि नाम से ही स्पष्ट है कि यह दिए गए विधेय के अनुसार प्रवाह तत्वों को फ़िल्टर करेगा। यह मूल प्रवाह और केवल फ़िल्टर्ड तत्वों के साथ वापसी प्रवाह के लिए लागू एक विस्तार विधि होगी। यह पता चला है कि हमें मूल प्रवाह से प्रत्येक तत्व को लेने की जरूरत है, जांच करें और आगे निकल जाएं यदि विधेय सही है। तो चलिए इसे करते हैं:

  public static IFlow<T> Filter<T>(this IFlow<T> source, Func<T, bool> predicate) => FlowFactory.Flow<T>((collector, cancellationToken) => source.Collect(item => { if (predicate(item)) collector.Emit(item); }, cancellationToken) ); 

अब, यदि हमें केवल बाईं माउस बटन पर क्लिक करना है, तो हम इसे लिख सकते हैं:

 emulator .Clicks() .Filter(click => click.Button == ClicksEmulator.Button.Left) .Collect(item => Log($"{item.Button} {item.X} {item.Y}"), cts.Token); 

सादृश्य से, हम ऑपरेटरों मानचित्र और OnNext लिखते हैं। पहले पास किए गए मैपर फ़ंक्शन का उपयोग करके मूल प्रवाह के प्रत्येक तत्व को दूसरे में परिवर्तित करता है। दूसरा मूल के समान तत्वों के साथ वापस आ जाएगा, लेकिन प्रत्येक पर एक कार्रवाई कर रहा है (आमतौर पर लॉगिंग)।

  public static IFlow<R> Map<T, R>(this IFlow<T> source, Func<T, R> mapper) => FlowFactory.Flow<R>((collector, cancellationToken) => source.Collect( item => collector.Emit(mapper(item)), cancellationToken ) ); public static IFlow<T> OnNext<T>(this IFlow<T> source, Action<T> action) => FlowFactory.Flow<T>((collector, cancellationToken) => source.Collect(item => { action(item); collector.Emit(item); }, cancellationToken) ); 

और उपयोग का एक उदाहरण:

 emulator .Clicks() .OnNext(click => Log($"{click.Button} {click.X} {click.Y}")) .Map(click => click.Button == ClicksEmulator.Button.Left ? 0 : 1) .Collect(item => Log($"{item}"), cts.Token); 

सामान्य तौर पर, जेट स्ट्रीम के लिए बहुत सारे ऑपरेटरों का आविष्कार किया गया था, उन्हें पाया जा सकता है, उदाहरण के लिए, यहां ।

और कुछ भी नहीं IF IF के लिए उनमें से किसी को भी लागू करने से रोकता है।

Rx.Net से परिचित लोग जानते हैं कि IObservable के लिए नए और विशिष्ट ऑपरेटरों के अलावा, Linq-to-ऑब्जेक्ट्स की एक्सटेंशन विधियों का उपयोग किया जाता है, और यह आपको धाराओं को "घटनाओं के संग्रह" के रूप में विचार करने और सामान्य Linq के साथ हेरफेर करने की अनुमति देता है। -methods। क्यों, खुद बयान लिखने के बजाय, इफ्लो को लिनक रेल पर रखने की कोशिश न करें?

IAsyncEnumerable

C # 8 में, IEnumerable का एक अतुल्यकालिक संस्करण पेश किया गया था - IAsyncEnumerable - एक संग्रह इंटरफ़ेस जिसे अतुल्यकालिक रूप से पुनरावृत्त किया जा सकता है। IAsyncEnumerable और प्रतिक्रियाशील धाराओं (IObservable और IFlow) के बीच मूलभूत अंतर यह है। IAsyncEnumerable, IEnumerable की तरह, एक पुल मॉडल है। हम इस संग्रह पर पुनरावृत्ति करते हैं कि हमें कितनी और कब जरूरत है और हम खुद से तत्वों को आकर्षित करते हैं। धाराएँ धक्का हैं। हम आयोजनों की सदस्यता लेते हैं और जब वे आते हैं तो उन्हें "प्रतिक्रिया" देते हैं - इसके लिए वे प्रतिक्रियाशील होते हैं। हालांकि, पुल मॉडल से धक्का-जैसा व्यवहार प्राप्त किया जा सकता है। इसे लॉन्ग पोलिंग https://en.wikipedia.org/wiki/Push_technology#Long_polling कहा जाता है। सार यह है: हम संग्रह पर पुनरावृति करते हैं, इसके अगले तत्व का अनुरोध करते हैं और जब तक हम इसे पसंद करते हैं, तब तक प्रतीक्षा करते हैं, जब तक कि संग्रह हमें वापस नहीं करता, अर्थात। अगली घटना आने तक। IAsyncEnumerable, IEnumerable के विपरीत, हमें अतुल्यकालिक रूप से प्रतीक्षा करने की अनुमति देगा। संक्षेप में, हमें किसी तरह से इफ्लो पर IAsyncEnumerable खींचने की जरूरत है।

जैसा कि आप जानते हैं, IAsyncEnumerator इंटरफ़ेस IAsyncEnumerable संग्रह के वर्तमान तत्व को वापस करने और अगले तत्व पर जाने के लिए जिम्मेदार है। इस मामले में, हमें IFlow से तत्वों को लेने की आवश्यकता है, और IFlowCollector ऐसा करता है। यह पता चला है कि एक वस्तु है जो इन इंटरफेस को लागू करती है:

 internal class FlowCollectorEnumerator<T> : IFlowCollector<T>, IAsyncEnumerator<T> { private readonly SemaphoreSlim _backpressureSemaphore = new SemaphoreSlim(0, 1); private readonly SemaphoreSlim _longPollingSemaphore = new SemaphoreSlim(0, 1); private bool _isFinished; public T Current { get; private set; } public async ValueTask DisposeAsync() { } public async Task Emit(T item, CancellationToken cancellationToken) { await _backpressureSemaphore.WaitAsync(cancellationToken); Current = item; _longPollingSemaphore.Release(); } public async Task Finish() { await _backpressureSemaphore.WaitAsync(); _isFinished = true; _longPollingSemaphore.Release(); } public async ValueTask<bool> MoveNextAsync() { _backpressureSemaphore.Release(); await _longPollingSemaphore.WaitAsync(); return !_isFinished; } } 

यहाँ मुख्य विधियाँ Emit , Finish, और MoveNextAsync हैं ।
शुरुआत में एमिट उस पल की प्रतीक्षा कर रहा है जब संग्रह से अगले आइटम का अनुरोध किया जाएगा। यानी किसी वस्तु का तब तक उत्सर्जन नहीं करता, जब तक कि उसकी जरूरत न हो। इसे बैकस्पेस कहा जाता है, इसलिए सेमाफोर का नाम। फिर वर्तमान आइटम सेट किया गया है और यह बताया गया है कि एक लंबे मतदान अनुरोध का परिणाम मिल सकता है।
MoveNextAsync को उस समय कहा जाता है जब संग्रह से कोई अन्य आइटम खींचा जाता है। वह _backpressureSemaphore जारी करता है और अगले तत्व को ट्रिगर करने के लिए फ़्लो का इंतजार करता है। फिर यह एक संकेत देता है कि संग्रह समाप्त हो गया है। यह ध्वज फिनिश विधि सेट करता है।

समाप्त एमिट के समान सिद्धांत पर काम करता है, केवल अगले तत्व के बजाय यह संग्रह के अंत का संकेत सेट करता है।

अब हमें इस वर्ग का उपयोग करने की आवश्यकता है।

 public static class AsyncEnumerableExtensions { public static IAsyncEnumerable<T> CollectEnumerable<T>(this IFlow<T> flow, CancellationToken cancellationToken = default) { var collector = new FlowCollectorEnumerator<T>(); flow .Collect(collector, cancellationToken) .ContinueWith(_ => collector.Finish(), cancellationToken); return new FlowEnumerableAdapter<T>(collector); } } internal class FlowEnumerableAdapter<T> : IAsyncEnumerable<T> { private readonly IAsyncEnumerator<T> _enumerator; public FlowEnumerableAdapter(IAsyncEnumerator<T> enumerator) { _enumerator = enumerator; } public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return _enumerator; } } 

IFlow के लिए ExtensionEnumerable विस्तार विधि एक FlowCollectorEnumerator बनाता है और उस पर एक प्रवाह पर हस्ताक्षर करता है, जिसके बाद समाप्त () विधि कहा जाता है। और यह एक FlowEnumerableAdapter को लौटाता है, जो कि IEAsumerator के रूप में FlowCollectorEnumerator का उपयोग करके IAsyncEnumerable का सबसे सरल कार्यान्वयन है।
हम कोशिश करते हैं कि क्या हुआ।

 var clicks = emulator .Clicks() .OnNext(item => Log($"{item.Button} {item.X} {item.Y}")) .CollectEnumerable(cts.Token) .Where(click => click.Button == ClicksEmulator.Button.Right) .Select(click => click.Y < 540 ? "TOP" : "LEFT"); await foreach (var click in clicks) { Log($"Clicked at: {click}"); } 

यहां हमें फ़्लो क्लिक मिलते हैं (), प्रत्येक क्लिक को लॉग इन करें, फिर IFAs को IAsyncEnumerable में बदल दें। फिर यह प्रसिद्ध लिनक-ऑपरेटरों पर लागू होता है: हम केवल राइट-क्लिक छोड़ते हैं और हम स्क्रीन के किस हिस्से में बने होते हैं।

अगला, एक उदाहरण पर अधिक जटिल विचार करें। हम राइट क्लिक को डबल लेफ्ट वाले से बदल देंगे। यानी हमें प्रत्येक तत्व को किसी दूसरे से नहीं, बल्कि संग्रह में ले जाना होगा। या फ्लो में, एक संग्रह में परिवर्तित हो गया।

 var clicks = emulator .Clicks() .OnNext(item => Log($"Original: {item.Button} {item.X} {item.Y}")) .CollectEnumerable(cts.Token) .Select(click => click.Button == ClicksEmulator.Button.Left ? Flow<ClicksEmulator.ClickEventArgs>(collector => collector.Emit(click)) : Flow<ClicksEmulator.ClickEventArgs>(async collector => { var changedClick = new ClicksEmulator.ClickEventArgs(click.X, click.Y, ClicksEmulator.Button.Left); await collector.Emit(changedClick); await Task.Delay(200); await collector.Emit(changedClick); }) ) .SelectMany(flow => flow.CollectEnumerable()); await foreach (var click in clicks) { Log($"Changed: {click.Button} {click.X} {click.Y}"); } 

ऐसा करने के लिए, Linq में SelectMany ऑपरेटर है। जेट स्ट्रीम में इसका समकक्ष फ्लैटपाइप है। सबसे पहले, IFlow के प्रत्येक क्लिक को मैप करें: बाएं क्लिक के लिए - इस एक क्लिक के साथ प्रवाह करें, दाएं के लिए - उन दोनों के बीच देरी के साथ दो बाएं क्लिक से प्रवाह करें। और फिर SelectMany में हम IFlow को IAyncEnumerable में बदल देते हैं।

और यह काम करता है! यानी कई ऑपरेटरों को IFlow के लिए लागू करने की आवश्यकता नहीं है - आप Linq का उपयोग कर सकते हैं।

निष्कर्ष

Rx.Net - C # में घटनाओं के अतुल्यकालिक अनुक्रमों के साथ काम करते समय मुख्य उपकरण बना हुआ है। लेकिन कोड आकार के मामले में यह एक बड़ा पुस्तकालय है। जैसा कि हमने देखा, समान कार्यक्षमता को बहुत सरल बनाया जा सकता है - बस दो इंटरफेस और कुछ बाध्यकारी। यह भाषा सुविधाओं के उपयोग के लिए संभव है - async / प्रतीक्षा। जब Rx का जन्म हुआ, तब इस सुविधा को C # में नहीं लाया गया था।

आपका ध्यान के लिए धन्यवाद!