scalaz-धारा `कैसे लागू करने के लिए पूछना-से-प्रतीक्षा-reply` टीसीपी ग्राहक

Question

मैं एक क्लाइंट ऐप कि पहली बार सर्वर पर एक अनुरोध भेजने के लागू करने के लिए तो अपने उत्तर के लिए प्रतीक्षा करना (http के समान)

मेरे क्लाइंट प्रक्रिया हो सकती है

val topic = async.topic[ByteVector] 
val client = topic.subscribe 

यहाँ एपीआई

trait Client { 
    val incoming = tcp.connect(...)(client) 
    val reqBus = topic.pubsh() 
    def ask(req: ByteVector): Task[Throwable \/ ByteVector] = { 
     (tcp.writes(req).flatMap(_ => tcp.reads(1024))).to(reqBus) 
     ??? 
    } 
} 

फिर, कैसे लागू करने के लिए ask का हिस्सा बने रहते है?

Answer 1

आमतौर पर, कार्यान्वयन सिंक के माध्यम से संदेश प्रकाशित करने के साथ किया जाता है और फिर आपके विषय की तरह कुछ स्रोतों पर कुछ प्रकार के उत्तर का इंतजार कर रहा है। धारा जवाब देने के लिए

def reqRply[I,O,O2](src:Process[Task,I],sink:Sink[Task,I],reply:Process[Task,O])(pf: PartialFunction[O,O2]):Process[Task,O2] = { 
merge.mergeN(Process(reply, (src to sink).drain)).collectFirst(pf) 
} 

अनिवार्य रूप से यह पहली हुक हमारे अनुरोध भेजा की पुष्टि किसी भी जिसके परिणामस्वरूप O का इंतजार करने के लिए:

असल में हम अपने कोड में इस के मुहावरों का एक बहुत कुछ है। फिर हम संदेश I प्रकाशित करते हैं और किसी भी आने वाले O के लिए O2 पर अनुवादित होने के लिए pf से परामर्श लें और फिर समाप्त करें।