def process_tweets(opts):
    """Set up and process incoming streams."""
    try:
        auth = tweepy.OAuthHandler(os.environ['CONSUMER_KEY'], os.environ['CONSUMER_SECRET'])
        auth.set_access_token(os.environ['ACCESS_KEY'], os.environ['ACCESS_SECRET'])
    except KeyError as e:
        logger.error('You must set the %s API key environment variable.', e)
        return

    logger.debug('Init tweepy.Stream()')
    logger.debug(opts)
    listener = StreamListener(opts=opts, logger=logger)
    streamer = tweepy.Stream(auth=auth, listener=listener, retry_count=9999,
        retry_time=1, buffer_size=16000)

    try:
        kwargs = make_filter_args(opts, auth)
    except ValueError as e:
        listener.running = False
        sys.stderr.write("%s: error: %s\n" % (__file__, e.message))
        return

    while listener.running:
        try:
            try:
                logger.debug('streamer.filter(%s)' % kwargs)
                streamer.filter(**kwargs)
            except TypeError as e:
                if 'stall_warnings' in e.message:
                    logger.warn("Installed Tweepy version doesn't support stall_warnings parameter.  Restarting without stall_warnings parameter.")
                    del kwargs['stall_warnings']
                    streamer.filter(**kwargs)
                else:
                    raise

            logger.debug('Returned from streaming...')
        except IOError:
            if opts.terminate_on_error:
                listener.running = False
            logger.exception('Caught IOError')
        except KeyboardInterrupt:
            # Stop the listener loop.
            listener.running = False
        except Exception:
            listener.running = False
            logger.exception("Unexpected exception.")

        if listener.running:
            logger.debug('Sleeping...')
            time.sleep(5)
dataset/ETHPy150Open inactivist/twitter-streamer/streamer/streamer.py/process_tweets/original