नेटटी/एनआईओ श्रोताओं के साथ स्कैला निरंतरता का उपयोग

Question

मैं नेटटी लाइब्रेरी (गिटहब से संस्करण 4) का उपयोग कर रहा हूं। यह स्कैला में बहुत अच्छा काम करता है, लेकिन मुझे उम्मीद है कि मेरी लाइब्रेरी असीमित प्रतीक्षा के लिए निरंतरता उत्तीर्ण शैली का उपयोग करने में सक्षम होगी।

Netty साथ परंपरागत रूप से आप (एक उदाहरण अतुल्यकालिक आपरेशन कनेक्ट) कुछ इस तरह करना होगा:

//client is a ClientBootstrap 
val future:ChannelFuture = client.connect(remoteAddr); 
future.addListener(new ChannelFutureListener { 
    def operationComplete (f:ChannelFuture) = { 
     //here goes the code that happens when the connection is made 
    } 
}) 

आप किसी लाइब्रेरी (जो मैं कर रहा हूँ) तो आप मूल रूप से उपयोगकर्ता की अनुमति के लिए तीन सरल विकल्प हैं कार्यान्वित कर रहे हैं तो कनेक्शन के बाद सामान करने के लिए लाइब्रेरी का:

1. बस अपनी कनेक्ट विधि से चैनलफ्यूचर वापस करें और उपयोगकर्ता को इसके साथ सौदा करने दें - यह नेटटी से बहुत अधिक अमूर्तता प्रदान नहीं करता है।
2. एक चैनलफ्यूचरलिस्टर को अपनी कनेक्ट विधि के पैरामीटर के रूप में लें और इसे चैनलफ़्यूचर में श्रोता के रूप में जोड़ें।
3. अपने कनेक्ट विधि का एक पैरामीटर के रूप में एक कॉलबैक फ़ंक्शन वस्तु ले लो और फोन है कि ChannelFutureListener आपके द्वारा बनाए गए के भीतर से

मैं क्या कर रहा हूँ (यह कुछ हद तक Node.js की तरह एक कॉलबैक चालित शैली के लिए होगा) करने का प्रयास चौथा विकल्प है; मैंने इसे ऊपर की गिनती में शामिल नहीं किया क्योंकि यह आसान नहीं है।

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

class MyLibraryClient { 
    def connect(remoteAddr:SocketAddress) = { 
     shift { retrn: (Unit => Unit) => { 
       val future:ChannelFuture = client.connect(remoteAddr); 
       future.addListener(new ChannelFutureListener { 
        def operationComplete(f:ChannelFuture) = { 
         retrn(); 
        } 
       }); 
      } 
     } 
    } 
} 

कल्पना कीजिए अन्य पढ़ें/लिखें आपरेशनों में लागू किया जा रहा वही फैशन इस किया जा रहा है के लक्ष्य को उपयोगकर्ता के कोड अधिक इस तरह दिख सकता है कि:

reset { 
    val conn = new MyLibraryClient(); 
    conn.connect(new InetSocketAddress("127.0.0.1", 1337)); 
    println("This will happen after the connection is finished"); 
} 

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

जिस समस्या में मैं दौड़ रहा हूं वह यह है कि मैं पूरी तरह से समझ नहीं पा रहा हूं कि सीमित निरंतरता का प्रकार कैसे काम करता है। जब मैं इसे उपर्युक्त तरीके से कार्यान्वित करने का प्रयास करता हूं, तो संकलक शिकायत करता है कि मेरा operationComplete कार्यान्वयन वास्तव में Unit के बजाय देता है। मुझे लगता है कि स्कैला के सीपीएस में एक "गॉचाचा" है जिसमें आपको shift विधि के रिटर्न प्रकार को @suspendable के साथ एनोटेट करना होगा, जो reset तक कॉल स्टैक पास कर लेता है, लेकिन ऐसा लगता है कि मेल करने का कोई तरीका नहीं है कि एक पूर्व-मौजूदा जावा लाइब्रेरी के साथ जिसमें सीमित निरंतरता की कोई अवधारणा नहीं है।

मुझे लगता है कि वास्तव में इसके आसपास एक रास्ता होना चाहिए - अगर स्वार निरंतर क्रमबद्ध कर सकता है और नेटवर्क पर उन्हें कहीं और गणना करने के लिए जाम कर सकता है, तो बस पूर्व-मौजूदा जावा क्लास से निरंतरता को कॉल करना संभव है। लेकिन मैं यह नहीं समझ सकता कि यह कैसे किया जा सकता है। ऐसा करने के लिए मुझे स्कैला में नेटटी के पूरे हिस्सों को फिर से लिखना होगा?

Answer 1

मुझे शुरू होने पर Scala's continuations के इस स्पष्टीकरण को बहुत उपयोगी पाया गया। विशेष रूप से उन हिस्सों पर ध्यान दें जहां वह shift[A, B, C] और reset[B, C] बताते हैं। operationComplete के अंतिम विवरण के रूप में एक डमी null जोड़ना चाहिए।

बीटीडब्ल्यू, आपको retrn() को अन्य reset के अंदर डालने की आवश्यकता है यदि इसमें shift इसके अंदर घोंसला हो।

संपादित करें: यहाँ एक काम उदाहरण

import scala.util.continuations._ 
import java.util.concurrent.Executors 

object Test { 

    val execService = Executors.newFixedThreadPool(2) 

    def main(args: Array[String]): Unit = { 
    reset { 
     val conn = new MyLibraryClient(); 
     conn.connect("127.0.0.1"); 
     println("This will happen after the connection is finished"); 
    } 
    println("Outside reset"); 
    } 
} 

class ChannelFuture { 
    def addListener(listener: ChannelFutureListener): Unit = { 
    val future = this 
    Test.execService.submit(new Runnable { 
     def run(): Unit = { 
     listener.operationComplete(future) 
     } 
    }) 
    } 
} 

trait ChannelFutureListener { 
    def operationComplete(f: ChannelFuture): Unit 
} 

class MyLibraryClient { 
    def connect(remoteAddr: String): Unit@cps[Unit] = { 
    shift { 
     retrn: (Unit => Unit) => { 
     val future: ChannelFuture = new ChannelFuture() 
     future.addListener(new ChannelFutureListener { 
      def operationComplete(f: ChannelFuture): Unit = { 
      println("operationComplete starts") 
      retrn(); 
      null 
      } 
     }); 
     } 
    } 
    } 
} 

एक संभव उत्पादन के साथ

है:

Outside reset 
operationComplete starts 
This will happen after the connection is finished