title: Socket watchdog in rx-java
body: <p>I am currently struggling trying to implement a tcp watchdog/retry system using rx, your help will be greatly appreciated.</p>
  
  <p>Having an Observable, I'd like to have an Observable resulting from periodically check whether we still can write to the socket. Easy enough, I can do something like this:</p>
  
  <pre><code>class SocketSubscribeFunc implements Observable.OnSubscribeFunc&lt;Socket&gt; {
    private final String hostname;
    private final int port;
    private Socket socket;
  
    SocketSubscribeFunc(String hostname, int port) {
      this.hostname = hostname;
      this.port = port;
    }
  
    public Subscription onSubscribe(final Observer&lt;? super Socket&gt; observer) {
      try {
        log.debug("Trying to connect...");
        socket = new Socket(hostname, port);
        observer.onNext(socket);
      } catch (IOException e) {
        observer.onError(e);
      }
      return new Subscription() {
        public void unsubscribe() {
          try {
            socket.close();
          } catch (IOException e) {
            e.printStackTrace();
          }
        }
      };
    }
  }
  
  Observable&lt;Socket&gt; socketObservable = Observable.create(new SocketSubscribeFunc(hostname,port));
  Observable&lt;Boolean&gt; watchdog = Observable.combineLatest(socketObservable, Observable.interval(1, TimeUnit.SECONDS), new Func2&lt;Socket, Long, Boolean&gt;() {
  
    public Boolean call(final Socket socket, final Long aLong) {
      try {
        socket.getOutputStream().write("ping\n".getBytes());
        return true;
      } catch (IOException e) {
       return false;
      }
    }
  });
  </code></pre>
  
  <p>Now, I want to retry connecting if the socket can be acquired (server/link is down at creation) or becomes unwritable (server/link is unreachable after successful connection).
  Ideally by resubscribing to the socket Observable whose OnSubscribeFunc creates the connection using the retry operator.
  As you can see this would introduce a circular dependency between the socket and the watchdog Observables.
  I toyed for a while with switchMap/materialize... in order to propagate the eventual error to no avail.</p>
  
  <p>I am close to abandon this idea and using Subjects from side-effecting code. But there should be a better way somewhere in the global mind :)</p>
  
  <p>Thanks in advance!</p>