EventMachine प्रॉक्सी डेमॉन

इस तथ्य के बावजूद कि EventMachine उच्च प्रदर्शन और अच्छी तरह से स्केलेबल नेटवर्क अनुप्रयोगों को लिखने के लिए एक काफी सुविधाजनक ढांचा है, इंटरनेट इसके उपयोग और परीक्षण के उदाहरणों की एक बहुतायत के साथ कृपया नहीं करता है। और वे उदाहरण जो मौजूद हैं, उदाहरण के लिए, एक हब पर, सही ढंग से काम नहीं करेगा, क्योंकि वे डेटा ट्रांसफर की विशेषताओं को ध्यान में नहीं रखते हैं (किसी कारण से वे उस डेटा को ध्यान में नहीं रखते हैं, सामान्य रूप से, भागों में प्रेषित होता है)। वास्तव में, यह लेख उन लोगों के लिए है, जिन्होंने EM के मूल सिद्धांतों से खुद को परिचित किया है, उदाहरण के लिए, रूबी और EventMachine लेख में , और सीखना चाहते हैं कि इसके आधार पर कुछ और अधिक जटिल कैसे लिखा जाए और परिणामस्वरूप प्राप्त कोड का परीक्षण कैसे करें।

हाल ही में, उन्होंने मुझे एक परीक्षण कार्य भेजा, जिसका सार ईएम पर एक प्रॉक्सी डेमॉन लिखना था, जो कि यूनिक्स डोमेन सॉकेट के माध्यम से ग्राहकों से कनेक्शन को अनिवार्य रूप से स्वीकार करेगा, उन्हें कतारबद्ध करेगा और इन आदेशों को कुछ सार प्रणाली से जुड़े सॉकेट में रीडायरेक्ट करेगा, और, प्राप्त कर रहा है इस प्रणाली से एक प्रतिक्रिया उन्हें ग्राहकों को वापस भेज देगी। क्लाइंट संदेश का प्रारूप {आईडी: 1, पाठ: "req1"} है, जिसे सर्वर द्वारा प्रतिक्रिया में परिवर्तित किया जाना चाहिए - {आईडी: 1, पाठ: "उत्तर 1"}।

कार्य को आसान बनाने के लिए, क्लाइंट और अमूर्त प्रणाली जिसे मैंने EM का उपयोग करके भी अनुकरण किया है, जो अंतर्निहित डेटा ट्रांसफर प्रोटोकॉल EM :: P :: ObjectProtocol का उपयोग करने की अनुमति देता है, जो इस बात को ध्यान में रखता है कि डेटा को रिएक्टर के एक भी टिक के लिए नहीं लिया जा सकता है, और ट्रांसमिशन के लिए डेटा को क्रमबद्ध भी करता है।

क्लाइंट कोड लिखकर शुरू करते हैं। जब कोई क्लाइंट प्रॉक्सी डेमॉन के साथ संबंध स्थापित करता है, तो पोस्ट_इनिट फ़ंक्शन को कहा जाता है , जिसमें क्लाइंट send_object का उपयोग करके एक हैश संदेश भेजता है।
फिर यह डेमॉन से प्रतिक्रिया की प्रतीक्षा करता है, और इसे कंसोल में प्रदर्शित करता है। चूंकि एक बार में कई ग्राहकों के कनेक्शन का अनुकरण करना आवश्यक था, इसलिए चर को उन कनेक्शनों की संख्या और पहले से ही एक प्रतिक्रिया प्राप्त करने वाले ग्राहकों की संख्या पर डेटा संग्रहीत किया गया था और डेमॉन से काट दिया गया था। कनेक्शन की कुल संख्या निरंतर TOTAL_CONNECTIONS में संग्रहीत की जाती है, जो क्लाइंट शुरू होने पर सेट की जाती है। जब क्लाइंट सर्वर से डिस्कनेक्ट होता है, तो एक अनबाइंड कॉल किया जाता है । जब सभी ग्राहकों को एक प्रतिक्रिया मिलती है, तो रिएक्टर बंद हो जाता है।

module EMClient<br/> include EM::P::ObjectProtocol<br/> <br/> @@connection_number = 0 <br/> @@dissconnected = 0 <br/> <br/> #send request as the connection has been established <br/> def post_init <br/> @@connection_number += 1 <br/> send_object({ 'id' => rand ( 10 ), 'text' => "req#{@@connection_number}" })<br/> end <br/> <br/> <br/> def receive_object (obj)<br/> #display response from server <br/> p obj.inspect<br/> end <br/> <br/> def unbind <br/> @@dissconnected += 1 <br/> #stop reactor after all requests have been processed <br/> EM.stop if @@dissconnected == TOTAL_CONNECTIONS<br/> end <br/> end <br/>

अगला सर्वर कोड है जो एक अमूर्त प्रणाली के संचालन का अनुकरण करता है। इसका कार्य वस्तु प्राप्त करना, उसे रूपांतरित करना और उसे वापस भेजना है। EM.add_time () का उपयोग करके , आप फ़ंक्शन के लिए दूसरे तर्क द्वारा पारित कोड ब्लॉक के निष्पादन में देरी कर सकते हैं, इस मामले में, डेमन से अनुरोध की प्रतिक्रिया।

module SocketServer<br/> include EM::P::ObjectProtocol<br/> <br/> def receive_object (obj)<br/> #emulation of job on server <br/> EM.add_timer( 1 + rand ( 5 )) do <br/> #validation of obj goes here)) <br/> obj[ 'text' ]. sub !(/req/, 'answ' )<br/> send_object(obj)<br/> end <br/> p "Server received object: #{obj.inspect}" <br/> end <br/> end <br/>

और अब चलो अधिक दिलचस्प भाग पर चलते हैं, संदेशों के लिए एक संदेश कतार और एक सार सर्वर के साथ एक कनेक्शन पूल बनाते हैं। इसके लिए, EM :: कतार वर्ग का उपयोग किया गया था, जिसमें दो विधियाँ पॉप (* a, & b) और पुश (* आइटम) हैं जो आपको कतार में आइटम जोड़ने और उन्हें इससे बाहर निकालने की अनुमति देती हैं। पॉप विधि अंतिम तर्क को कोड के एक ब्लॉक के रूप में लेती है जिसे एक तत्व कतार में प्रकट होने पर निष्पादित किया जाएगा।

सर्वर के साथ संबंध स्थापित करने के लिए, EMConnection मॉड्यूल का उपयोग किया गया था, जिसमें send_request (obj, & block) पद्धति को परिभाषित किया गया था, जिसका सार सर्वर संदेश भेजना था और एक ब्लॉक को संचारित करना था जिसे सर्वर से प्रतिक्रिया मिलने पर निष्पादित किया जाएगा।

कनेक्शन पूल बनाने के लिए ConnectionPool वर्ग जिम्मेदार है। जब इसे प्रारंभ किया जाता है, तो पूल का आकार निर्धारित किया जाता है और कतार को प्रारंभ किया जाता है। फिर, कनेक्शन की निर्दिष्ट संख्या start_queue विधि में सेट की गई है, और प्रत्येक कनेक्शन के लिए कार्यकर्ता ( queue_worker_loop ) लॉन्च किया गया है, जो कि एक खरीद है जो कनेक्शन को तर्क के रूप में लेता है। उनके काम का सार कतार से एक तत्व को प्राप्त करने वाला तत्व है जो सर्वर को भेजा जाना चाहिए, और कोड का एक ब्लॉक जिसे ऑब्जेक्ट प्राप्त करने के बाद निष्पादित किया जाना चाहिए। इसके अलावा, कोड के इस ब्लॉक को क्रियान्वित करने के बाद, स्वयं कॉल करें, इस प्रकार, एक प्रकार का अनंत लूप प्राप्त किया जाता है।

module EMConnection<br/> include EM::P::ObjectProtocol<br/> <br/> def receive_object (obj)<br/> #calling callback on object receiving <br/> @callback. call (obj)<br/> end <br/> <br/> def send_request obj, &block<br/> #sending data to server and setting callback <br/> send_object obj<br/> @callback = block<br/> end <br/> <br/> end <br/> <br/> #simple connection pool using EM queue, default size 10 <br/> class ConnectionPool <br/> <br/> def initialize (conf)<br/> @pool_size = conf[:size] || 10 <br/> @connections = []<br/> @query_queue = EM:: Queue .new<br/> start_queue conf<br/> end <br/> <br/> def queue_worker_loop <br/> proc { |connection|<br/> @query_queue.pop do | request |<br/> connection. send_request ( request [:obj]) do |response|<br/> request [:callback].call response #if request[:callback] <br/> queue_worker_loop .call connection<br/> end <br/> end <br/> }<br/> end <br/> <br/> def start_queue (conf)<br/> @pool_size.times do <br/> connection = EM.connect( '0.0.0.0' , 8080 , EMConnection)<br/> @connections << connection<br/> queue_worker_loop .call connection<br/> end <br/> end <br/> <br/> def request (obj, &block)<br/> @query_queue.push :obj => obj, :callback => block<br/> end <br/> end <br/>

अब आइए प्रॉक्सी डेमॉन के लिए जिम्मेदार कोड पर चलते हैं। इसका काम कनेक्शन पूल को शुरू करना है, ज़ाहिर है, यह डेमन कोड के साथ नहीं किया जा सकता है, लेकिन इस तरह से पूल को केवल तब ही आरंभ किया जाएगा जब इसकी आवश्यकता होगी। जब क्लाइंट से कोई ऑब्जेक्ट प्राप्त होता है, तो वह ऑब्जेक्ट और कनेक्शन पूल कतार कोड ब्लॉक से गुजरता है, और जब एब्सट्रैक्ट सर्वर से एक प्रतिक्रिया प्राप्त होती है, तो यह क्लाइंट को वापस संदेश भेजता है और क्लोज_कनेक्शन_ऑनलाइन राइटिंग विधि का उपयोग करके कनेक्शन बंद कर देता है, जिससे यह सुनिश्चित होता है कि संदेश पूरी तरह से क्लाइंट के पास भेजा गया है ( क्लोज़_कनेक्शन विधि के विपरीत) जो तुरंत कनेक्शन बंद कर देता है)।

module DaemonServer<br/> include EM::P::ObjectProtocol<br/> <br/> def post_init <br/> @@connections_pool ||= ConnectionPool. new (:size => 5 )<br/> end <br/> <br/> def receive_object (obj)<br/> @@connections_pool.request obj do |response|<br/> send_object(response)<br/> close_connection_after_writing<br/> end <br/> end <br/> end <br/>

अब सर्वर, क्लाइंट और प्रॉक्सी डेमॉन को शुरू करने के लिए जिम्मेदार स्क्रिप्ट पर चलते हैं।
हम TCPSocket पर सर्वर शुरू करते हैं। यहां सब कुछ बहुत सरल है।

EventMachine.run {<br/> EventMachine.start_server "127.0.0.1" , 8080 , SocketServer<br/>} <br/>


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

tc = ARGV[ 0 ].to_i<br/>TOTAL_CONNECTIONS = tc > 0 ? tc : 25 <br/> <br/>file = File .expand_path( '../tmp/daemon.sock' , __FILE__ )<br/> p "Starting #{TOTAL_CONNECTIONS} client(s)" <br/>EventMachine::run {<br/> TOTAL_CONNECTIONS.times{ EM.connect_unix_domain(file, EMClient) }<br/>} <br/>

एक दानव बनने के लिए छद्म के लिए, निस्संदेह, इसे निस्तारित (कप्तान) करने की आवश्यकता है। इसके लिए मैंने डेमोंस रत्न का इस्तेमाल किया। यदि स्क्रिप्ट -d स्विच से शुरू होती है, तो इसे अतिरिक्त विकल्पों के साथ Daemons.daemonize विधि को कॉल करके डीमॉनेटाइज किया जाता है, जो यह निर्धारित करता है कि लॉग्स को कैसे स्टोर करें और प्रक्रिया पीआईडी ​​युक्त फ़ाइल को हटा दें , ताकि डेमॉन को रोका जा सके।

options = {<br/> :app_name => 'ProxyServer' ,<br/> :backtrace => true ,<br/> :log_output => true ,<br/> :dir_mode => :normal,<br/> :dir => File .expand_path( '../tmp' , __FILE__ )<br/>}<br/> <br/>file = File .expand_path( '../tmp/daemon.sock' , __FILE__ )<br/> File .unlink(file) if File .exists?(file)<br/> <br/>Daemons.daemonize(options) if ARGV.index( '-d' )<br/> <br/>EventMachine::run {<br/> EventMachine::start_unix_domain_server(file, DaemonServer)<br/>} <br/>

मेरा मानना ​​है कि परीक्षणों के बिना लगभग किसी भी कोड की लागत बहुत कम होती है, भले ही यह काम कर रहा हो, क्योंकि यह भविष्य में बहुत सारी समस्याओं और सिरदर्द पैदा कर सकता है, जो इसे लिखा था और जिन्हें इसे संपादित करने की आवश्यकता है।

EM - EMSpec के आधार पर बनाए गए परीक्षण कार्यक्रमों के लिए तैयार समाधान है। लेकिन मैंने इसका उपयोग नहीं किया, और इसलिए मैं यह प्रदर्शित करूंगा कि बिना rspec का उपयोग कैसे किया जाए।
सबसे पहले आपको टेस्ट क्लाइंट बनाने की जरूरत है। इसमें, हम send_request (obj, & block) विधि को परिभाषित करते हैं, जो डेमॉन को प्रॉक्सी अनुरोध भेजने और कॉलबैक को कोड के एक ब्लॉक के रूप में सेट करने की अनुमति देता है जिसे क्लाइंट द्वारा प्रतिक्रिया मिलने पर कॉल किया जाएगा। हम onclose = (proc) विधि भी बनाते हैं, जो कनेक्शन बंद होने पर कॉलबैक को निर्धारित करेगा।

module TestClient<br/> include EM::P::ObjectProtocol<br/> <br/> #on object received callback <br/> def receive_object (obj)<br/> @onresponse. call (obj)<br/> p "Client received object: #{obj.inspect}" <br/> end <br/> <br/> def send_request obj, &block<br/> @onresponse = block<br/> send_object obj<br/> end <br/> <br/> # on disconnect callback <br/> def onclose =( proc )<br/> @onclosed = proc <br/> end <br/> <br/> def unbind <br/> @onclosed.call<br/> end <br/> <br/> end <br/>

अब हम उन तरीकों को बनाने के लिए आगे बढ़ सकते हैं जो हमें लिखित कोड का परीक्षण करने की अनुमति देते हैं। पहला तरीका start_serv सर्वर, टेस्ट क्लाइंट और प्रॉक्सी शुरू करने के लिए जिम्मेदार है, और यह एक तर्क के रूप में एक ब्लॉक लेता है, जिसके पास इसके निपटान में क्लाइंट चर होगा, जो क्लाइंट को हेरफेर करने की अनुमति देता है। यदि कुछ गलत हो जाता है और क्लाइंट को सर्वर से प्रतिक्रिया नहीं मिलती है, तो टाइमर विधि की आवश्यकता होती है, फिर rspec दिखाएगा कि परीक्षण विफल हो गया, और न केवल जमा हुआ। परीक्षणों के लिए आधार सर्वर_टेस्ट विधि होगी, जो सर्वर, क्लाइंट और प्रॉक्सी शुरू करने के लिए उपरोक्त विधियों का उपयोग करती है, यह निर्धारित करती है कि जब कनेक्शन बंद हो जाता है, तो रिएक्टर को रोक दिया जाना चाहिए, और क्लाइंट द्वारा सर्वर को भेजे जाने के लिए एक तर्क भी लेता है।

module HelperMethods<br/> def start_serv <br/> File .unlink(SOCK_FILE) if File .exists?(SOCK_FILE)<br/> EM.run {<br/> EventMachine.start_server "127.0.0.1" , 8080 , SocketServer<br/> EventMachine.start_unix_domain_server(SOCK_FILE, DaemonServer)<br/> client = EM.connect_unix_domain(SOCK_FILE, TestClient)<br/> yield client<br/> }<br/> end <br/> <br/> # if request takes to long it will show fail <br/> def timer start<br/> timeout = 6 <br/> EM.add_timer(timeout){<br/> ( Time .now-start).should be_within( 0 ).of(timeout)<br/> EM.stop<br/> }<br/> end <br/> <br/> #main wrapper for test starts server daemon and client <br/> def server_test request<br/> time = Time .now<br/> start_serv do |client|<br/> client.send_request request do |response|<br/> yield response<br/> end <br/> client.onclose= lambda {EM.stop}<br/> timer (time)<br/> end <br/> end <br/> end <br/>

एक उदाहरण के लिए, मैं परीक्षण की जांच करूंगा कि यह कार्य सही तरीके से किया गया है।

describe "on sending test request" do <br/> include HelperMethods<br/> it "should responsend with right answer" do <br/> server_test({ 'id' => 0 , 'text' => "req1" }) do |response|<br/> response[ 'text' ].should == "answ1" <br/> response[ 'id' ].should == 0 <br/> end <br/> end <br/> end <br/>

यह सब, वास्तव में, मुझे आशा है कि यह लेख किसी के लिए उपयोगी होगा। सभी कोड github पर उपलब्ध है।

PS यह बहुत पाठ निकला, जिसने अंत तक पढ़ने में महारत हासिल की, धन्यवाद।