code: public void start() {
          if (log.isDebugEnabled()) {
              log.debug("start - subscriber stream state: {}", (subscriberStream != null ? subscriberStream.getState() : null));
          }
          switch (subscriberStream.getState()) {
              case UNINIT:
                  // allow start if uninitialized and change state to stopped
                  subscriberStream.setState(StreamState.STOPPED);
                  IMessageOutput out = consumerService.getConsumerOutput(subscriberStream);
                  if (msgOutReference.compareAndSet(null, out)) {
                      out.subscribe(this, null);
                  } else if (log.isDebugEnabled()) {
                      log.debug("Message output was already set for stream: {}", subscriberStream);
                  }
                  break;
              default:
                  throw new IllegalStateException(String.format("Cannot start in current state: %s", subscriberStream.getState()));
          }
      }
docs: Start stream