Stream statuses/filter

        :param \*\*params: Parameters to send with your stream request

        Accepted params found at:
        https://developer.twitter.com/en/docs/tweets/filter-realtime/api-reference/post-statuses-filter
181349
def filter(self, **params):
        
        url = 'https://stream.twitter.com/%s/statuses/filter.json' \
              % self.streamer.api_version
        self.streamer._request(url, 'POST', params=params)
165279
def stream_events(self, filter: Callable[[Event], bool] = None, *, max_queue_size: int = 0):
        """"""
        return stream_events([self], filter, max_queue_size=max_queue_size)
1350032
def match_tweet(self, tweet, user_stream):
        
        if user_stream:
            if len(self.track) > 0:
                return self.is_tweet_match_track(tweet)

            return True

        return self.is_tweet_match_track(tweet) or self.is_tweet_match_follow(tweet)
1162883
def handle(send, msg, args):
    
    worker = args["handler"].workers
    result = worker.run_pool(get_urls, [msg])
    try:
        urls = result.get(5)
    except multiprocessing.TimeoutError:
        worker.restart_pool()
        send("Url regex timed out.", target=args["config"]["core"]["ctrlchan"])
        return
    for url in urls:
        # Prevent botloops
        if (args["db"].query(Urls).filter(Urls.url == url, Urls.time > datetime.now() - timedelta(seconds=10)).count() > 1):
            return

        if url.startswith("https://twitter.com"):
            tid = url.split("/")[-1]
            twitter_api = get_api(args["config"])
            status = twitter_api.GetStatus(tid)
            text = status.text.replace("\n", " / ")
            send("** {} (@{}) on Twitter: {}".format(status.user.name, status.user.screen_name, text))
            return

        imgkey = args["config"]["api"]["googleapikey"]
        title = urlutils.get_title(url, imgkey)

        shortkey = args["config"]["api"]["bitlykey"]
        short = urlutils.get_short(url, shortkey)

        last = args["db"].query(Urls).filter(Urls.url == url).order_by(Urls.time.desc()).first()
        if args["config"]["feature"].getboolean("linkread"):
            if last is not None:
                lasttime = last.time.strftime("%H:%M:%S on %Y-%m-%d")
                send("Url %s previously posted at %s by %s -- %s" % (short, lasttime, last.nick, title))
            else:
                send("** %s - %s" % (title, short))
        args["db"].add(Urls(url=url, title=title, nick=args["nick"], time=datetime.now()))
159870
function(params, cb) {
    if (CONSOLE_MESSAGES) console.log(new Date().toISOString(),'S3Handler.getObject params:',params);

    S3Handler.s3.getObject({Bucket: S3Handler.options.bucket, Key: params.Key}, function(error, responseData) {
      if (error) {
        if (CONSOLE_MESSAGES) console.log(new Date().toISOString(),"S3.getObject Error:", error);
        cb(error, null);
      } else {
        // decode response body according to params.type
        switch(params.type.toLowerCase()) {
          case 'json':
            var responseData = JSON.parse(responseData.Body);
          break;
          case 'stream':
            var responseData = responseData.Body.toString();
          break;
        }

        if (CONSOLE_MESSAGES) console.log(new Date().toISOString(),'S3Handler.getObject responseData:',responseData);

        cb(null, responseData);
      }
    });
  }
1153346
def _send_dweet(payload, url, params=None, session=None):
    
    data = json.dumps(payload)
    headers = {'Content-type': 'application/json'}
    return _request('post', url, data=data, headers=headers, params=params, session=session)
918740
function(tweet, streamChannels){
    helpers.postprocessTweet(tweet, streamChannels);
    return helpers.emitPosprocessedTweet(tweet, streamChannels);
  }
684563
def connect(path, options = {}, &block)
      stream_parameters, callbacks = connection_options(path, options)

      @stream = EM::Twitter::Client.connect(stream_parameters)
      @stream.each do |item|
        begin
          hash = MultiJson.decode(item, :symbolize_keys => true)
        rescue MultiJson::DecodeError
          invoke_callback(callbacks['error'], "MultiJson::DecodeError occured in stream: #{item}")
          next
        end

        unless hash.is_a?(::Hash)
          invoke_callback(callbacks['error'], "Unexpected JSON object in stream: #{item}")
          next
        end

        respond_to(hash, callbacks, &block)

        yield_message_to(callbacks['anything'], hash)
      end

      @stream.on_close { invoke_callback(callbacks['close']) }

      @stream.on_error do |message|
        invoke_callback(callbacks['error'], message)
      end

      @stream.on_unauthorized { invoke_callback(callbacks['unauthorized']) }

      @stream.on_enhance_your_calm { invoke_callback(callbacks['enhance_your_calm']) }

      @stream.on_reconnect do |timeout, retries|
        invoke_callback(callbacks['reconnect'], timeout, retries)
      end

      @stream.on_max_reconnects do |timeout, retries|
        fail TweetStream::ReconnectError.new(timeout, retries)
      end

      @stream.on_no_data_received { invoke_callback(callbacks['no_data_received']) }

      @stream
    end
211840
def tweet(args = {})
      options = {
        num_hashtags: [0, rand(1..10) - 6].max,
        num_mentions: [0, rand(1..10) - 8].max,
        reply: (rand(1..10) == 1),
        body_length: rand(20..140)
      }.merge(args)

      my_reply = options[:reply] ? "#{mention} " : ''
      my_mentions = options[:num_mentions] > 0 ? "#{mentions(options[:num_mentions])} " : ''
      my_tags = tags(options[:num_hashtags])

      remaining = [
        options[:body_length],
        140 - (my_reply.size + my_mentions.size + my_tags.size)
      ].min

      "#{my_reply}#{body(remaining)}#{my_mentions}#{my_tags}"
    end
1274402
def _send(self, params):
        
        # Get a response
        try:
            r = requests.get(self.url, params=params)
        # catch errors, print then exit.
        except requests.exceptions.RequestException as e:
            print(e)
        return r.json(strict=False)
558930
def search_for_tweets_about(user_id, params):
    
    url = "https://api.twitter.com/1.1/search/tweets.json"
    response = make_twitter_request(url, user_id, params)
    return process_tweets(response.json()["statuses"])
30243
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);
    }
984344
function applyFilter (tweenable, filterName) {
      var filters = Tweenable.prototype.filter;
      var args = tweenable._filterArgs;

      each(filters, function (name) {
        if (typeof filters[name][filterName] !== 'undefined') {
          filters[name][filterName].apply(tweenable, args);
        }
      });
    }
478228
def get_params(self):
        
        params = {}
        complex = False

        for name, opt in self.filter_options.items():
            if opt.ignored:
                continue
            if self.set_param(params, name):
                complex = True
        return params, complex
716447
def __stream(self, endpoint, listener, params={}, run_async=False, timeout=__DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=__DEFAULT_STREAM_RECONNECT_WAIT_SEC):
        

        # Check if we have to redirect
        instance = self.instance()
        if "streaming_api" in instance["urls"] and instance["urls"]["streaming_api"] != self.api_base_url:
            # This is probably a websockets URL, which is really for the browser, but requests can't handle it
            # So we do this below to turn it into an HTTPS or HTTP URL
            parse = urlparse(instance["urls"]["streaming_api"])
            if parse.scheme == 'wss':
                url = "https://" + parse.netloc
            elif parse.scheme == 'ws':
                url = "http://" + parse.netloc
            else:
                raise MastodonAPIError(
                        "Could not parse streaming api location returned from server: {}.".format(
                            instance["urls"]["streaming_api"]))
        else:
            url = self.api_base_url

        # The streaming server can't handle two slashes in a path, so remove trailing slashes
        if url[-1] == '/':
            url = url[:-1]
        
        # Connect function (called and then potentially passed to async handler)
        def connect_func():
            headers = {"Authorization": "Bearer " + self.access_token}
            connection = self.session.get(url + endpoint, headers = headers, data = params, stream = True,
                                  timeout=(self.request_timeout, timeout))

            if connection.status_code != 200:
                raise MastodonNetworkError("Could not connect to streaming server: %s" % connection.reason)
            return connection
        connection = None
        
        # Async stream handler
        class __stream_handle():
            def __init__(self, connection, connect_func, reconnect_async, reconnect_async_wait_sec):
                self.closed = False
                self.running = True
                self.connection = connection
                self.connect_func = connect_func
                self.reconnect_async = reconnect_async
                self.reconnect_async_wait_sec = reconnect_async_wait_sec
                self.reconnecting = False
                
            def close(self):
                self.closed = True
                self.connection.close()

            def is_alive(self):
                return self._thread.is_alive()

            def is_receiving(self):
                if self.closed or not self.running or self.reconnecting or not self.is_alive():
                    return False
                else:
                    return True

            def _threadproc(self):
                self._thread = threading.current_thread()
                
                # Run until closed or until error if not autoreconnecting
                while self.running:
                    if not self.connection is None:
                        with closing(self.connection) as r:
                            try:
                                listener.handle_stream(r)
                            except (AttributeError, MastodonMalformedEventError, MastodonNetworkError) as e:
                                if not (self.closed or self.reconnect_async):
                                    raise e
                                else:
                                    if self.closed:
                                        self.running = False

                    # Reconnect loop. Try immediately once, then with delays on error.
                    if (self.reconnect_async and not self.closed) or self.connection is None:
                        self.reconnecting = True
                        connect_success = False
                        while not connect_success:
                            connect_success = True
                            try:
                                self.connection = self.connect_func()
                                if self.connection.status_code != 200:
                                    time.sleep(self.reconnect_async_wait_sec)
                                    connect_success = False
                                    exception = MastodonNetworkError("Could not connect to server.")
                                    listener.on_abort(exception)
                            except:
                                time.sleep(self.reconnect_async_wait_sec)
                                connect_success = False
                        self.reconnecting = False
                    else:
                        self.running = False
                return 0

        if run_async:
            handle = __stream_handle(connection, connect_func, reconnect_async, reconnect_async_wait_sec)
            t = threading.Thread(args=(), target=handle._threadproc)
            t.daemon = True
            t.start()
            return handle
        else:
            # Blocking, never returns (can only leave via exception)
            connection = connect_func()            
            with closing(connection) as r:
                listener.handle_stream(r)
473899
def filter(self, track=None, follow=None, locations=None, event=None,
               record_keepalive=False):
        
        if locations is not None:
            if type(locations) == list:
                locations = ','.join(locations)
            locations = locations.replace('\\', '')

        url = 'https://stream.twitter.com/1.1/statuses/filter.json'
        params = {
            "stall_warning": True,
            "include_ext_alt_text": True
        }
        if track:
            params["track"] = track
        if follow:
            params["follow"] = follow
        if locations:
            params["locations"] = locations
        headers = {'accept-encoding': 'deflate, gzip'}
        errors = 0
        while True:
            try:
                log.info("connecting to filter stream for %s", params)
                resp = self.post(url, params, headers=headers, stream=True)
                errors = 0
                for line in resp.iter_lines(chunk_size=1024):
                    if event and event.is_set():
                        log.info("stopping filter")
                        # Explicitly close response
                        resp.close()
                        return
                    if not line:
                        log.info("keep-alive")
                        if record_keepalive:
                            yield "keep-alive"
                        continue
                    try:
                        yield json.loads(line.decode())
                    except Exception as e:
                        log.error("json parse error: %s - %s", e, line)
            except requests.exceptions.HTTPError as e:
                errors += 1
                log.error("caught http error %s on %s try", e, errors)
                if self.http_errors and errors == self.http_errors:
                    log.warning("too many errors")
                    raise e
                if e.response.status_code == 420:
                    if interruptible_sleep(errors * 60, event):
                        log.info("stopping filter")
                        return
                else:
                    if interruptible_sleep(errors * 5, event):
                        log.info("stopping filter")
                        return
            except Exception as e:
                errors += 1
                log.error("caught exception %s on %s try", e, errors)
                if self.http_errors and errors == self.http_errors:
                    log.warning("too many exceptions")
                    raise e
                log.error(e)
                if interruptible_sleep(errors, event):
                    log.info("stopping filter")
                    return
1072016
public static syslog_params[] get_filtered(nitro_service service, String filter) throws Exception
	{
		syslog_params obj = new syslog_params();
		options option = new options();
		option.set_filter(filter);
		syslog_params[] response = (syslog_params[]) obj.getfiltered(service, option);
		return response;
	}