RxJava का उपयोग कर श्रोताओं को प्रतिक्रियाशील (पर्यवेक्षण) में सही तरीके से कैसे परिवर्तित करें?

Question

मैं एक मल्टीप्लेयर गेम क्लाइंट का उपयोग कर रहा हूं जिसे ऐपवार्प (http://appwarp.shephertz.com) कहा जाता है, जहां आप ईवेंट श्रोताओं को घटना के होने पर वापस बुला सकते हैं, मान लीजिए कि हम कनेक्शन श्रोता के बारे में बात करेंगे, जहां आपको इस इंटरफेस को लागू करने की आवश्यकता है :

public interface ConnectionRequestListener { 
    void onConnectDone(ConnectEvent var1); 
    void onDisconnectDone(ConnectEvent var1); 
    void onInitUDPDone(byte var1); 
} 

मेरे यहाँ लक्ष्य मुख्य रूप से इस ग्राहक के एक प्रतिक्रियाशील संस्करण बनाने के लिए ग्राहकों का ही उपयोग करते हुए सीधे (मैं भी बाद में के बजाय सिर्फ पर निर्भर करता है इंटरफेस पर भरोसा करता हूँ की आंतरिक बजाय मेरी Apps में प्रयोग की जाने वाली है उदाहरण के रूप में WarpClient स्वयं, लेकिन यह महत्वपूर्ण बात नहीं है, कृपया मेरे प्रश्न को बहुत अंत में पढ़ें)।

तो इस प्रकार है क्या मैंने किया है:

public class RxConnectionEvent { 
    // This is the original connection event from the source client 
    private final ConnectEvent connectEvent; 
    // this is to identify if it was Connection/Disconnection 
    private final int eventType; 

    public RxConnectionEvent(ConnectEvent connectEvent, int eventType) { 
     this.connectEvent = connectEvent; 
     this.eventType = eventType; 
    } 

    public ConnectEvent getConnectEvent() { 
     return connectEvent; 
    } 

    public int getEventType() { 
     return eventType; 
    } 
} 

2:

1) मैं एक नई घटना है, यह RxConnectionEvent (नाम कौन सा मुख्य रूप से समूहों कनेक्शन से संबंधित घटनाओं) के रूप में इस प्रकार की शुरुआत की) किसी घटना प्रकार निर्मित इस प्रकार है:

public class RxEventType { 
    // Connection Events 
    public final static int CONNECTION_CONNECTED = 20; 
    public final static int CONNECTION_DISCONNECTED = 30; 
} 

public abstract class BaseObservable<T> implements Observable.OnSubscribe<T> { 

    protected WarpClient warpClient; 

    protected BaseObservable (WarpClient warpClient) 
    { 
     this.warpClient = warpClient; 
    } 

    @Override 
    public abstract void call(Subscriber<? super T> subscriber); 

    protected abstract void onUnsubscribed(WarpClient warpClient); 
} 

मेरा प्रश्न मुख्य रूप से है: सही ऊपर मेरी कार्यान्वयन ७६१९६०३२१० 3) नमूदार जिसके बाद मेरी नई RxConnectionEvent

import com.shephertz.app42.gaming.multiplayer.client.WarpClient; 
import com.shephertz.app42.gaming.multiplayer.client.events.ConnectEvent; 
import rx.Observable; 
import rx.Subscriber; 
import rx.functions.Action0; 
import rx.subscriptions.Subscriptions; 

public class ConnectionObservable extends BaseObservable<RxConnectionEvent> { 

    private ConnectionRequestListener connectionListener; 

    // This is going to be called from my ReactiveWarpClient (Factory) Later. 
    public static Observable<RxConnectionEvent> createConnectionListener(WarpClient warpClient) { 
     return Observable.create(new ConnectionObservable(warpClient)); 
    } 

    private ConnectionObservable(WarpClient warpClient) { 
     super(warpClient); 
    } 

    @Override 
    public void call(final Subscriber<? super RxConnectionEvent> subscriber) { 
     subscriber.onStart(); 
     connectionListener = new ConnectionRequestListener() { 
      @Override 
      public void onConnectDone(ConnectEvent connectEvent) { 
       super.onConnectDone(connectEvent); 
       callback(new RxConnectionEvent(connectEvent, RxEventType.CONNECTION_CONNECTED)); 
      } 

      @Override 
      public void onDisconnectDone(ConnectEvent connectEvent) { 
       super.onDisconnectDone(connectEvent); 
       callback(new RxConnectionEvent(connectEvent, RxEventType.CONNECTION_DISCONNECTED)); 
      } 

      // not interested in this method (for now) 
      @Override 
      public void onInitUDPDone(byte var1) { } 

      private void callback(RxConnectionEvent rxConnectionEvent) 
      { 
       if (!subscriber.isUnsubscribed()) { 
        subscriber.onNext(rxConnectionEvent); 
       } else { 
        warpClient.removeConnectionRequestListener(connectionListener); 
       } 
      } 
     }; 

     warpClient.addConnectionRequestListener(connectionListener); 
     subscriber.add(Subscriptions.create(new Action0() { 
      @Override 
      public void call() { 
       onUnsubscribed(warpClient); 
      } 
     })); 
    } 

    @Override 
    protected void onUnsubscribed(WarpClient warpClient) { 
     warpClient.removeConnectionRequestListener(connectionListener); 
    } 
} 

4 उत्सर्जन करता है) और अंत में मेरी BaseObservable ऐसा दिखाई देता है बनाया गया या मुझे प्रत्येक घटना के लिए अलग-अलग अवलोकन योग्य बनाना चाहिए, लेकिन यदि ऐसा है, तो इस क्लाइंट में 40-50 से अधिक ईवेंट हैं जो मुझे प्रत्येक ईवेंट के लिए अलग-अलग देखने योग्य बनाना है?

मैं भी ऊपर कोड का उपयोग इस प्रकार (एक सरल "गैर अंतिम" एकीकरण परीक्षण में इसका इस्तेमाल किया) के रूप में:

public void testConnectDisconnect() { 
    connectionSubscription = reactiveWarpClient.createOnConnectObservable(client) 
      .subscribe(new Action1<RxConnectionEvent>() { 
       @Override 
       public void call(RxConnectionEvent rxEvent) { 
        assertEquals(WarpResponseResultCode.SUCCESS, rxEvent.getConnectEvent().getResult()); 
        if (rxEvent.getEventType() == RxEventType.CONNECTION_CONNECTED) { 
         connectionStatus = connectionStatus | 0b0001; 
         client.disconnect(); 
        } else { 
         connectionStatus = connectionStatus | 0b0010; 
         connectionSubscription.unsubscribe(); 
         haltExecution = true; 
        } 
       } 
      }, new Action1<Throwable>() { 
       @Override 
       public void call(Throwable throwable) { 
        fail("Unexpected error: " + throwable.getMessage()); 
        haltExecution = true; 
       } 
      }); 

    client.connectWithUserName("test user"); 
    waitForSomeTime(); 
    assertEquals(0b0011, connectionStatus); 
    assertEquals(true, connectionSubscription.isUnsubscribed()); 
} 

Answer 1

मैं तुम्हें सीधे BaseObservable विस्तार के बाद से यह बहुत त्रुटियों की संभावना है से बचने के सुझाव देते हैं। इसके बजाए, उपकरण का उपयोग करने का प्रयास करें आरएक्स स्वयं आपको अपना अवलोकन करने के लिए देता है।

सबसे आसान समाधान PublishSubject का उपयोग कर रहा है, जो एक अवलोकन और सब्सक्राइबर दोनों है। श्रोता को केवल विषय के onNext का आह्वान करने की आवश्यकता होती है, और विषय ईवेंट को छोड़ देगा। यहाँ एक सरल काम कर उदाहरण है:

public class PublishSubjectWarpperDemo { 

    public interface ConnectionRequestListener { 
     void onConnectDone(); 

     void onDisconnectDone(); 

     void onInitUDPDone(); 
    } 

    public static class RxConnectionEvent { 
     private int type; 

     public RxConnectionEvent(int type) { 
      this.type = type; 
     } 

     public int getType() { 
      return type; 
     } 

     public String toString() { 
      return "Event of Type " + type; 
     } 
    } 

    public static class SimpleCallbackWrapper { 
     private final PublishSubject<RxConnectionEvent> subject = PublishSubject.create(); 

     public ConnectionRequestListener getListener() { 
      return new ConnectionRequestListener() { 

       @Override 
       public void onConnectDone() { 
        subject.onNext(new RxConnectionEvent(1)); 
       } 

       @Override 
       public void onDisconnectDone() { 
        subject.onNext(new RxConnectionEvent(2)); 
       } 

       @Override 
       public void onInitUDPDone() { 
        subject.onNext(new RxConnectionEvent(3)); 
       } 
      }; 
     } 

     public Observable<RxConnectionEvent> getObservable() { 
      return subject; 
     } 

    } 

    public static void main(String[] args) throws IOException { 
     SimpleCallbackWrapper myWrapper = new SimpleCallbackWrapper(); 
     ConnectionRequestListener listner = myWrapper.getListener();// Get the listener and attach it to the game here. 
     myWrapper.getObservable().observeOn(Schedulers.newThread()).subscribe(event -> System.out.println(event)); 

     listner.onConnectDone(); // Call the listener a few times, the observable should print the event 
     listner.onDisconnectDone(); 
     listner.onInitUDPDone(); 

     System.in.read(); // Wait for enter 
    } 
} 

एक अधिक जटिल समाधान Observable.create() का उपयोग कर एक नमूदार बनाने के लिए onSubscribe कार्यान्वयन में से एक का उपयोग करने के लिए होगा। उदाहरण के लिए AsyncOnSubscibe। इस समाधान में बैकपरसुर को सही तरीके से संभालने का लाभ है, इसलिए आपका ईवेंट ग्राहक घटनाओं से अभिभूत नहीं होता है। लेकिन आपके मामले में, यह एक असंभव परिदृश्य की तरह लगता है, इसलिए अतिरिक्त जटिलता शायद इसके लायक नहीं है।