code: private void startTwitterStream() {
          logger.info("starting {} twitter stream", streamType);
  
          if (stream == null) {
              logger.debug("creating twitter stream");
  
              stream = new TwitterStreamFactory(buildTwitterConfiguration()).getInstance();
              if (streamType.equals("user")) {
                  stream.addListener(new UserStreamHandler());
              } else {
                  stream.addListener(new StatusHandler());
              }
  
              logger.debug("twitter stream created");
          }
  
          if (riverStatus != RiverStatus.STOPPED && riverStatus != RiverStatus.STOPPING) {
              if (streamType.equals("filter") || filterQuery != null) {
                  stream.filter(filterQuery);
              } else if (streamType.equals("firehose")) {
                  stream.firehose(0);
              } else if (streamType.equals("user")) {
                  stream.user();
              } else {
                  stream.sample();
              }
          }
          logger.debug("{} twitter stream started!", streamType);
      }
docs: Start twitter stream