title: RxJava return observer after others finish
body: <p>I'm new to RxJava, i'm trying to create a function that returns an observer that receives a message from an AMQP queue, I need to assure that there's a connection to the server and that the queue is created, this is what i have:</p>
  
  <pre><code>    public Observable&lt;String&gt; consume(String queueName) {
          Observable&lt;String&gt; consume = Observable.create(new Observable.OnSubscribe&lt;String&gt;() {
              @Override
              public void call(Subscriber&lt;? super String&gt; sub) {
                  try {
                      Consumer consumer = new DefaultConsumer(channel) {
                          @Override
                          public void handleDelivery(String consumerTag, Envelope envelope,
                                  AMQP.BasicProperties properties, byte[] body) throws IOException {
                              String message = new String(body, "UTF-8");
                              sub.onNext(message);
                              // sub.onCompleted();
                          }
                      };
                      channel.basicConsume(queueName, true, consumer);
                  } catch (IOException e) {
                      sub.onError(e);
                  }
  
              }
          });
          return Observable.zip(connect(), queueDeclare(queueName), consume, (conn, queue, consumer) -&gt; {
              return consumer;
          });
      }
  }
  
  server.consume(QUEUE_NAME).subscribe(c -&gt; {
          System.out.println("Received " + c);
      }
  </code></pre>
  
  <p>This works, it connects to the server, creates the queue and create the consumer, and writes when the message arrives. the problem is that it only works the first time, because the subscription is to the "zip" not to the consume object, next message arrives calls on next but doesn't activate the subscriber.</p>
  
  <p>I made it work using:</p>
  
  <pre><code>connectionObserver.subscribe(conn-&gt;{
     queueObserver.subscribe(queue-&gt;{
        consumeObserver.subscribe(message-&gt;{
           System.out.println("Received " + message);
        });
     });
  });
  </code></pre>
  
  <p>But I'd like to think there's a better way, specially since I want to create a function to do this</p>
  
  <p>thanks in advance</p>
  
  <p>EDIT:</p>
  
  <p>this also works and is much cleaner:</p>
  
  <pre><code>public Observable&lt;String&gt; consume(String queueName) {
      return connect().flatMap(conn -&gt; {
          return queueDeclare(queueName);
      }).flatMap(queue -&gt; {
          return Observable.create(new Observable.OnSubscribe&lt;String&gt;() {
              @Override
              public void call(Subscriber&lt;? super String&gt; sub) {
                  try {
                      Consumer consumer = new DefaultConsumer(channel) {
                          @Override
                          public void handleDelivery(String consumerTag, Envelope envelope,
                                  AMQP.BasicProperties properties, byte[] body) throws IOException {
                              String message = new String(body, "UTF-8");
                              sub.onNext(message);
                              // sub.onCompleted();
                          }
                      };
                      channel.basicConsume(queueName, true, consumer);
                  } catch (IOException e) {
                      sub.onError(e);
                  }
              }
          });
      });
  }
  }
  </code></pre>
  
  <p>still open to ideas</p>