title: RxJava2 - Transfer downstream errors upstream
body: <p>I'm trying to work with a RxJava library for bluetooth on Android. My requirements are:</p>
  
  <ol>
  <li>An observable to monitor the connection status</li>
  <li>The connection automatically restarts on any error</li>
  <li>An observable to listen for messages from the connection</li>
  </ol>
  
  <p>This is what I have so far, but I can't help but think it is convoluted. Primarily, my concern is that the way I forced the errors from downstream to cause a new connection to be created. Specifically, its me using the <code>mCommErrors</code> publish subject down when I receive a read/parsing error. </p>
  
  <p>I think all of this is necessary because my client's will always be connected to the observable created in observeIsConnected() which prevents the <code>refCount()</code> going to 0 upon any error. </p>
  
  <p>Thanks for any insight you might be able to provide. I have much larger problems with a deadlock involving the refCount() operator so Im trying to weed out any of the potential "bad things" I've done. </p>
  
  <pre><code>public class MyBluetoothDevice
  {   
      private final MessageParser mParser;
      private final Observable&lt;MyBluetoothConnection&gt; mConnectionObservable;
      // The comms errors will be piped in so we can restart our connection as needed. These errors
      // can originate post-connection when reading / sending messages
      private final PublishSubject&lt;Throwable&gt; mCommsErrors = PublishSubject.create();
  
  public MyBluetoothDevice(BluetoothDevice device, MessageParser parser, Scheduler scheduler)
  {
      mParser = parser;
  
      mConnectionObservable = mCommsErrors
              .switchMap(err -&gt; Observable.&lt;BluetoothSocket&gt; error(err))
              .mergeWith(createConnection(device, SPP_UUID))
              .subscribeOn(scheduler)
              .map(MyBluetoothConnection::new)
              .retryWhen(err -&gt; err.delay(5, TimeUnit.SECONDS))
              .replay(1)
              .refCount();
  }
  
  // Copy this from RxBluetooth since they forget to close the bluetooth socket if an error occurs
  private Observable&lt;BluetoothSocket&gt; createConnection(final BluetoothDevice device, final UUID uuid)
  {
      return Observable.create(emitter -&gt; {
          try {
              final BluetoothSocket bluetoothSocket = device.createRfcommSocketToServiceRecord(uuid);
              emitter.setCancellable(() -&gt; silentlyCloseSocket(bluetoothSocket));
              // We don't have to handle closing the socket if connect fails, because setCancellable
              // will cause the socket to be closed when the error is emitted
              bluetoothSocket.connect();
              emitter.onNext(bluetoothSocket);
          } catch (IOException e) {
              emitter.onError(e);
          }
      });
  }
  
  private void silentlyCloseSocket(BluetoothSocket socket)
  {
      try {
          socket.close();
      } catch (IOException ignore) {
      }
  }
  
  public Observable&lt;Boolean&gt; observeIsConnected()
  {
      return mConnectionObservable
              .switchMap(MyBluetoothConnection::observeIsConnected);
  }
  
  public Observable&lt;TruPulseMessage&gt; observeMessages()
  {
      return mConnectionObservable
              .switchMap(connection -&gt; connection
                      .observeStringLineStream()
                      .switchMap(mParser::parse)
                      .doOnError(err -&gt; connection.closeConnection())
                      .doOnError(err -&gt; mCommsErrors.onNext(err))
                      .onErrorResumeNext(Observable.empty())
                      );
  }
  
  public Completable send(String command)
  {
      return mConnectionObservable
              .take(1)
              .concatMapCompletable(connection -&gt; connection.sendCommand(command))
              .doOnError(mCommsErrors::onNext);
  }
  }
  </code></pre>
  
  <p>and here is the MyBluetoothConnection</p>
  
  <pre><code>public class MyBluetoothConnection
  {
  private final BluetoothSocket mSocket;
  private final InputStream mInputStream;
  private final OutputStream mOutputStream;
  private final BehaviorSubject&lt;Boolean&gt; mIsConnected;
  private Observable&lt;String&gt; mOutObservable;
  
  MyBluetoothConnection(BluetoothSocket socket) throws Exception
  {
      if (socket == null) {
          throw new InvalidParameterException("Bluetooth socket can't be null");
      }
  
      this.mSocket = socket;
      this.mIsConnected = BehaviorSubject.createDefault(Boolean.TRUE);
  
      try {
          mInputStream = socket.getInputStream();
          mOutputStream = socket.getOutputStream();
      } catch (IOException e) {
          closeConnection();
          throw new IOException("Can't get stream from bluetooth socket", e);
      }
  }
  
  public Observable&lt;Boolean&gt; observeIsConnected()
  {
      return mIsConnected;
  }
  
  public Observable&lt;String&gt; observeStringLineStream()
  {
      if (mOutObservable == null) {
          mOutObservable = Observable.create((ObservableOnSubscribe&lt;String&gt;) emitter -&gt; {
              BufferedReader reader = new BufferedReader(new InputStreamReader(mInputStream));
              try {
                  String line = "";
                  while (!emitter.isDisposed() &amp;&amp; line != null) {
                      line = reader.readLine();
                      if (line != null)
                          emitter.onNext(line);
                      else
                          emitter.onComplete();
                  }
              } catch (IOException e) {
                  closeConnection();
                  if (!emitter.isDisposed())
                      emitter.onError(e);
              }
          }).share();
      }
      return mOutObservable;
  }
  
  public Completable sendCommand(String command)
  {
      return Completable.create(emitter -&gt; {
          if (!mIsConnected.getValue()) {
              emitter.onError(new IOException("BluetoothConnection is disconnected"));
              return;
          }
  
          String line = command + "\r\n";
          try {
              mOutputStream.write(line.getBytes());
              mOutputStream.flush();
              emitter.onComplete();
          } catch (IOException e) {
              // Error occurred. Better to close terminate the connection
              closeConnection();
              if (!emitter.isDisposed())
                  emitter.onError(new IOException("Can't send Bluetooth command", e));
          }
      });
  }
  
  /**
   * Close the streams and socket connection.
   */
  public void closeConnection()
  {
      if (!mIsConnected.getValue())
          return;
  
      try {
          if (mInputStream != null) {
              mInputStream.close();
          }
  
          if (mOutputStream != null) {
              mOutputStream.close();
          }
  
          if (mSocket != null) {
              mSocket.close();
          }
      } catch (IOException ignored) {
      } finally {
          mIsConnected.onNext(Boolean.FALSE);
      }
  }
  }
  </code></pre>