title: RxJava observables not emitting events
body: <p>My android application has a <code>FeedDetailFragment</code> that displays <code>Feed</code> details. A feed has basic information and metadata, which are retrieved through two separate calls to the server. The server interface is filled in with Retrofit. I have implemented something that, to my novice Rx knowledge, looks logical. However, as you may have guessed, it doesn't work.</p>
  
  <p>External classes:</p>
  
  <ul>
  <li><code>FeedInfo</code> - parcellable class that contains basic feed info</li>
  <li><code>FeedMetadata</code> - parcellable class that contains metadata about feed</li>
  <li><code>Feed</code> - auxiliary class that combines feed info and metadata, providing some hepler functions</li>
  <li><code>UvClient</code> - server interface implemented with Retrofit</li>
  </ul>
  
  <p>Relevant <code>FeedDetailFragment</code> code:</p>
  
  <pre><code>public class FeedDetailFragment extends Fragment implements OnMapReadyCallback {
      public static final String ARG_FEED_ID       = "feed_id";
      public static final String ARG_FEED_INFO     = "feed_info";
      public static final String ARG_FEED_METADATA = "feed_metadata";
      public static final int INVALID_FEED_ID      = -1;
  
      ...
  
      private class PlaceFeedSubscriber extends Subscriber&lt;Pair&lt;GoogleMap, Feed&gt;&gt; {
          @Override
          public void onNext(Pair&lt;GoogleMap, Feed&gt; pair) {
              Log.i(TAG, String.format("Placing feed %d on [%f, %f] onto map %s",
                      pair.second.getInfo(),
                      pair.second.getMetadata().getSensorLatitude(),
                      pair.second.getMetadata().getSensorLongitude(),
                      pair.first.getMapType()));
  
              pair.first.addMarker(new MarkerOptions()
                      .position(new LatLng(
                              pair.second.getMetadata().getSensorPoint().getCoordinates()[1],
                              pair.second.getMetadata().getSensorPoint().getCoordinates()[0]))
                      .title("Marker"));
              mMapAPI.moveCamera(CameraUpdateFactory.newLatLngZoom(
                      new LatLng(
                              pair.second.getMetadata().getSensorPoint().getCoordinates()[1],
                              pair.second.getMetadata().getSensorPoint().getCoordinates()[0])
                      , 15));
          }
  
          @Override
          public void onCompleted() {
              Log.i(TAG, "Completed drawing of feed");
          }
  
          @Override
          public void onError(Throwable e) {
              Log.e(TAG, "Drawing of feed failed with: " + e);
          }
      }
  
      public FeedDetailFragment() {
          mMapObservable = Observable.empty().subscribeOn(Schedulers.io());
          mFeedIdObservable = Observable.empty().subscribeOn(Schedulers.io());
          mFeedInfoObservable = Observable.empty();
          mFeedMetadataObservable = Observable.empty();
  
          // Start fetching new feed information
          mFeedIdObservable.doOnEach(new Action1&lt;Integer&gt;() {
              @Override
              public void call(Integer feedId) {
                  Log.d(TAG, "Got a new feed id - " + feedId);
                  mFeedInfoObservable.mergeWith(mUvClient.getFeed(feedId));
              }
          });
  
          // Start fetching new feed metadata
          mFeedInfoObservable.doOnEach(new Action1&lt;FeedInfo&gt;() {
              @Override
              public void call(FeedInfo feedInfo) {
                  Log.d(TAG, "Got a new feed info - " + feedInfo.getTitle());
                  mFeedMetadataObservable.mergeWith(mUvClient.getFeedMetadata(feedInfo.getId()));
              }
          });
  
          // Produce a new feed
          mFeedObservable = Observable.combineLatest(mFeedInfoObservable, mFeedMetadataObservable, new Func2&lt;FeedInfo, FeedMetadata, Feed&gt;() {
              @Override
              public Feed call(FeedInfo feedInfo, FeedMetadata feedMetadata) {
                  return new Feed(feedInfo, feedMetadata);
              }
          });
  
          // Render the feed onto map
          Observable.combineLatest(mFeedObservable, mMapObservable, new Func2&lt;Feed, GoogleMap, Pair&lt;GoogleMap, Feed&gt;&gt;() {
              @Override
              public Pair&lt;GoogleMap, Feed&gt; call(Feed feed, GoogleMap map) {
                  return new Pair(map, feed);
              }
          }).observeOn(AndroidSchedulers.mainThread())
                  .subscribe(new PlaceFeedSubscriber());
      }
  
      @Override
      public void onCreate(Bundle savedInstanceState) {
          super.onCreate(savedInstanceState);
  
          Bundle arguments = getArguments();
  
          if (arguments.containsKey(ARG_FEED_ID)) {
              setFeed(arguments.getInt(ARG_FEED_ID));
          }
          else if (arguments.containsKey(ARG_FEED_INFO)) {
              if (arguments.containsKey(ARG_FEED_METADATA)) {
                  setFeed((FeedInfo)Parcels.unwrap(arguments.getParcelable(ARG_FEED_INFO)),
                          (FeedMetadata)Parcels.unwrap(arguments.getParcelable(ARG_FEED_METADATA)));
              }
              else {
                  setFeed((FeedInfo)Parcels.unwrap(arguments.getParcelable(ARG_FEED_INFO)));
              }
          }
      }
  
      ...
  
      @Override
      public void onMapReady(GoogleMap googleMap) {
          mMapAPI = googleMap;
  
          mMapObservable.mergeWith(Observable.just(googleMap));
      }
  
      /**
       * Sets the feed ID to be shown in the fragment. This triggers the chain of fetching feed info
       * and feed metadata, finally displaying it on the map.
       * @param feedId ID of the feed to display in the fragment.
       */
      public void setFeed(int feedId) {
          Log.d(TAG, String.format("Setting new feed ID - %d", feedId));
          mFeedIdObservable.mergeWith(Observable.just(feedId));
      }
  
      /**
       * Sets feed info. This triggers fetching of feed metadata, finally displaying it on the map.
       * @param feedInfo Information of the feed to display on the map.
       */
      public void setFeed(FeedInfo feedInfo) {
          Log.d(TAG, String.format("Setting new feed info - %s", feedInfo.getTitle()));
          mFeedInfoObservable.mergeWith(Observable.just(feedInfo));
      }
  
      /**
       * Displays feed info on the map.
       * @param feedInfo Information of the feed to display on the map.
       * @param feedMetadata Metadata of the feed to display on the map.
       */
      public void setFeed(FeedInfo feedInfo, FeedMetadata feedMetadata) {
          Log.d(TAG, String.format("Setting new feed info and metadata - %s", feedInfo.getTitle()));
          mFeedObservable.mergeWith(Observable.just(new Feed(feedInfo, feedMetadata)));
      }
  }
  </code></pre>
  
  <p>The log output I see is as follows:</p>
  
  <pre><code>Setting new feed info - SampleFeed
  Completed drawing of feed
  </code></pre>
  
  <p>My overall idea was that the observables would emit new data when I merge it in. Some observables are created empty so that they do not emit anything but I can still work with them.</p>
  
  <p>The potential flow could be as follows:</p>
  
  <ul>
  <li>Activity gets a callback from <code>FeedListFragment</code> notifying the <code>id</code> of the feed that was clicked</li>
  <li>Activity checks either gets and passes <code>FeedInfo</code> to <code>FeedDetailFragment.setFeed</code> or invokes <code>FeedDetailFragment.setFeed</code> with the feed's <code>id</code> (lets assume the latter for completeness)</li>
  <li><code>FeedDetailFragment</code> merges new observable with the received feed <code>id</code></li>
  <li>The merge triggers emission of new event on <code>mFeedIdObservable</code></li>
  <li>The <code>.doOnEach</code> of <code>mFeedIdObservable</code> kicks off Retrofit interface to fetch <code>FeedInfo</code></li>
  <li>The <code>.doOnEach</code> of <code>mFeedInfoObservable</code> kick off Retrofit interface to fetch <code>FeedMetadata</code></li>
  <li>The <code>.combineLatest</code> of <code>mFeedObservable</code> fires off when both <code>mFeedInfoObservable</code> and <code>mFeedMetadataObservable</code> return new data</li>
  <li>Finally, getting the <code>GoogleMap</code> and <code>Feed</code>, the call is made to subscriber to draw the feed on the map</li>
  </ul>
  
  <p>This is how the thought was put together in my head. Obviously, it is wrong. Where did I go wrong and how can I fix it? I'd love some pointers and maybe more general ideology/methodology approach teaching. Thanks for any advice!</p>
  
  <p><strong>UPDATE 1</strong></p>
  
  <p>So I've been trying to figure this out. Read more documents... a lot to learn. I've replaced <code>Observable.empty()</code> with <code>Observable.never()</code>. From documentation, I read that <code>empty</code> doesn't emit anything and completes, which is not what I want. On the other hand, <code>never</code> doesn't emit anything but does not complete. As such, I can use it for the purpose I'm seeking. Still not getting what I want but, I hope, one step closer.</p>
  
  <p><strong>UPDATE 2</strong></p>
  
  <p>Getting a bit more hang of it. Looking into source of <code>.never()</code> and <code>.empty()</code>, I see that the former does not call <code>.onNext()</code> and the latter calls <code>.onComplete()</code>. There is nothing in the middle I can choose. Started looking around for alternatives. Basically, my code doesn't execute because, in my previous tries, observable either completed immediately or never proceeded to call next. However, there is nothing to call <code>.onNext()</code> in the beginning. As such, I need a placeholder.</p>
  
  <p>Reading more docs, I came across <code>Subject</code>s. In particular, <code>PublishSubject</code> doesn't emit anything until a subscriber subscribes. This seemed like a viable solution. However, the subscriber must subscribe directly to the subject. This didn't seem to work with <code>.mergeWith()</code> upon the subject.</p>
  
  <p>Will not give up :)</p>
  
  <p><strong>UPDATE 3</strong></p>
  
  <p>Thanks to @dwursteisen, I continued with <code>PublishSubject</code> approach. This is the relevant code that changed:</p>
  
  <pre><code>...
  
  private PublishSubject&lt;GoogleMap&gt; mMapObservable = null;
  private PublishSubject&lt;Feed&gt; mFeedObservable = null;
  private PublishSubject&lt;Integer&gt; mFeedIdObservable = null;
  private PublishSubject&lt;FeedInfo&gt; mFeedInfoObservable = null;
  private PublishSubject&lt;FeedMetadata&gt; mFeedMetadataObservable = null;
  
  ...
  
  public FeedDetailFragment() {
      mMapObservable = PublishSubject.create();
      mFeedObservable = PublishSubject.create();
      mFeedIdObservable = PublishSubject.create();
      mFeedInfoObservable = PublishSubject.create();
      mFeedMetadataObservable = PublishSubject.create();
  
      mMapObservable.subscribe(new Action1&lt;GoogleMap&gt;() {
          @Override
          public void call(GoogleMap googleMap) {
              mMapApi = googleMap;
          }
      });
  
      mFeedMetadataObservable.subscribe(new Action1&lt;FeedMetadata&gt;() {
          @Override
          public void call(FeedMetadata feedMetadata) {
              // no code
          }
      });
  
      mFeedObservable.subscribe(new Action1&lt;Feed&gt;() {
          @Override
          public void call(Feed feed) {
              // no code
          }
      });
  
      // Start fetching new feed information
      mFeedIdObservable.subscribe(new Action1&lt;Integer&gt;() {
          @Override
          public void call(Integer feedId) {
              mUvClient.getFeed(feedId).subscribe(new Action1&lt;FeedInfo&gt;() {
                  @Override
                  public void call(FeedInfo feedInfo) {
                      mFeedInfoObservable.onNext(feedInfo);
                  }
              });
          }
      });
  
      // Start fetching new feed metadata
      mFeedInfoObservable
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe(new Action1&lt;FeedInfo&gt;() {
                  @Override
                  public void call(FeedInfo feedInfo) {
                      mFeedTitle.setText(feedInfo.getTitle());
  
                      mUvClient.getFeedMetadata(feedInfo.getId()).subscribe(new Action1&lt;FeedMetadata&gt;() {
                          @Override
                          public void call(FeedMetadata feedMetadata) {
                              mFeedMetadataObservable.onNext(feedMetadata);
                          }
                      });
                  }
              });
  
      // Produce a new feed
      Observable.combineLatest(mFeedInfoObservable, mFeedMetadataObservable, new Func2&lt;FeedInfo, FeedMetadata, Feed&gt;() {
          @Override
          public Feed call(FeedInfo feedInfo, FeedMetadata feedMetadata) {
              Feed feed = new Feed(feedInfo, feedMetadata);
              return feed;
          }
      }).subscribeOn(Schedulers.io()).subscribe(new Action1&lt;Feed&gt;() {
          @Override
          public void call(Feed feed) {
              mFeedObservable.onNext(feed);
          }
      });
  
      // Render the feed onto map
      Observable.combineLatest(mFeedObservable, mMapObservable, new Func2&lt;Feed, GoogleMap, Pair&lt;GoogleMap, Feed&gt;&gt;() {
          @Override
          public Pair&lt;GoogleMap, Feed&gt; call(Feed feed, GoogleMap map) {
              return new Pair(map, feed);
          }
      }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
              .subscribe(new PlaceFeedSubscriber());
  }
  
  ...
  
  @Override
  public void onMapReady(GoogleMap googleMap) {
      mMapObservable.onNext(googleMap);
  }
  
  /**
   * Sets the feed ID to be shown in the fragment. This triggers the chain of fetching feed info
   * and feed metadata, finally displaying it on the map.
   * @param feedId ID of the feed to display in the fragment.
   */
  public void setFeed(int feedId) {
      mFeedIdObservable.onNext(feedId);
  }
  
  /**
   * Sets feed info. This triggers fetching of feed metadata, finally displaying it on the map.
   * @param feedInfo Information of the feed to display on the map.
   */
  public void setFeed(FeedInfo feedInfo) {
      mFeedInfoObservable.onNext(feedInfo);
  }
  
  /**
   * Displays feed info on the map.
   * @param feedInfo Information of the feed to display on the map.
   * @param feedMetadata Metadata of the feed to display on the map.
   */
  public void setFeed(FeedInfo feedInfo, FeedMetadata feedMetadata) {
      mFeedObservable.onNext(new Feed(feedInfo, feedMetadata));
  }
  </code></pre>
  
  <p>Obviously, now that I got the basics working, I'll go through it and do proper handling of errors, caching, and other conditions. However, I do have one question: is there any way to simplify the following code to directly use Retrofit observable instead of subscribing to it inside the subscribe... maybe Rx operator that would inject its resolution into <code>mFeedInfoObservable</code>?</p>
  
  <pre><code>mFeedIdObservable.subscribe(new Action1&lt;Integer&gt;() {
      @Override
      public void call(Integer feedId) {
          mUvClient.getFeed(feedId).subscribe(new Action1&lt;FeedInfo&gt;() {
              @Override
              public void call(FeedInfo feedInfo) {
                  mFeedInfoObservable.onNext(feedInfo);
              }
          });
      }
  });
  </code></pre>
  
  <p>Also, I would love to hear any comments in the general approach. I'm still wrapping my head around Rx and my implementation is not best, I am sure.</p>