code: @Override
    @SuppressWarnings("unchecked")
    public void start() {
      if (isStopped()) return;
      
      logger().log("Inbound stream listening: for '" + reader.name() + "'");
      
      try {
        reader.openFor(this);
      } catch (Exception e) {
        reader.close();
        e.printStackTrace();
        throw new IllegalStateException(e.getMessage(), e);
      }
      cancellable = this.stage().scheduler().schedule(selfAs(Scheduled.class), null, 1000, probeInterval);
    }
docs: =========================================