Check Topic Filter

Based on 4.7 Topic Names and Topic Filters

@param string $topic_filter
@throws Exception
511368
static public function CheckTopicFilter($topic_filter)
    {
        if (!isset($topic_filter[0]) || isset($topic_filter[65535])) {
            throw new Exception('Topic filter must be at 1~65535 bytes long');
        }
        self::ValidateUTF8($topic_filter);

        if (false !== strpos($topic_filter, chr(0))) {
            throw new Exception('Null character is not allowed in topic');
        }

        $length = strlen($topic_filter);

        /*
         The multi-level wildcard character MUST be specified either on its own or following a topic level separator.
         In either case it MUST be the last character specified in the Topic Filter [MQTT-4.7.1-2].
         */
        if (($p = strpos($topic_filter, '#')) !== false) {
            if ($p != $length - 1) {
                throw new Exception('"#" MUST be the last char in topic filter');
            } else if ($length > 1 && $topic_filter[$length - 2] != '/') {
                throw new Exception('"#" MUST occupy an entire level of the filter');
            }
        }

        $levels = explode('/', $topic_filter);
        foreach ($levels as $l) {
            if ($l == '') {
                continue;
            } else if (strpos($l, '+') !== false && isset($l[1])) {
                /*
                 The single-level wildcard can be used at any level in the Topic Filter, including first and last levels.
                 Where it is used it MUST occupy an entire level of the filter [MQTT-4.7.1-3].
                 */
                throw new Exception('"+" MUST occupy an entire level of the filter');
            }
        }

        if ($topic_filter[0] == '#') {
            Debug::Log(Debug::DEBUG, 'If you want to subscribe topic begin with $, please subscribe both "#" and "$SOMETOPIC/#"');
        }
    }
713251
public function addSub($topic, $uid)
    {
        Utility::CheckTopicFilter($topic);
        ProcessManager::getInstance()->getRpcCall(ClusterProcess::class, true)->my_addSub($topic, $uid);
    }
1082661
def find_topic(category_name, forum_name, topic_title, options = {})
      forum = self.find_forum(category_name, forum_name, options)
      if forum
        self.topics(forum['id'], options).detect {|t| t['title'] == topic_title}
      end
    end
1136978
public static function delete($topic_id)
    {
        // Delete the topic and any redirect topics
        $where_delete_topic = [
            ['id' => $topic_id],
            ['moved_to' => $topic_id]
        ];

        DB::forTable('topics')
            ->where_any_is($where_delete_topic)
            ->delete_many();

        // Delete posts in topic
        DB::forTable('posts')
            ->where('topic_id', $topic_id)
            ->delete_many();

        // Delete any subscriptions for this topic
        DB::forTable('topic_subscriptions')
            ->where('topic_id', $topic_id)
            ->delete_many();
    }
334779
@Then("^A kafka topic named '(.+?)' exists")
    public void kafkaTopicExist(String topic_name) throws KeeperException, InterruptedException {
        assert commonspec.getKafkaUtils().getZkUtils().pathExists("/" + topic_name) : "There is no topic with that name";
    }
1194883
static public function CheckTopicName($topic_name)
    {
        if (!isset($topic_name[0]) || isset($topic_name[65535])) {
            throw new Exception('Topic name must be at 1~65535 bytes long');
        }

        self::ValidateUTF8($topic_name);

        if (false !== strpos($topic_name, chr(0))) {
            throw new Exception('null character is not allowed in topic');
        }
        if (false !== strpos($topic_name, '#')) {
            throw new Exception('# is not allowed in topic');
        }
        if (false !== strpos($topic_name, '+')) {
            throw new Exception('+ is not allowed in topic');
        }

        if ($topic_name[0] == '$') {
            Debug::Log(Debug::WARN, 'Publish to topics beginning with \'$\' may be ignored by server');
        }
    }
262819
private static function determineFilterInOrNotInList(array $filter, QueryBuilder $queryBuilder)
    {
        if (!\in_array($filter['operation'], ['IN', 'NOT IN'])) {
            return null;
        }

        return static::filterInOrNotInList($filter, $queryBuilder);
    }
1119939
public function setFilter($filter)
    {
        parent::setFilter($filter);

        $minValue = $this->getOption('min_value');
        $maxValue = $this->getOption('max_value');

        $this->userOptions['min_value'] = $minValue ? str_replace(',', '', $minValue) : $minValue;
        $this->userOptions['max_value'] = $maxValue ? str_replace(',', '', $maxValue) : $maxValue;
    }
908372
function () {
			this.writeDebug('checkQueryStringFilters',arguments);

			// Loop through the filters.
			for(var key in filters) {
				if (filters.hasOwnProperty(key)) {
					var filterVal = this.getQueryString(key);

					// Check for multiple values separated by comma.
					if ( filterVal.indexOf( ',' ) !== -1 ) {
						filterVal = filterVal.split( ',' );
					}

					// Only add the taxonomy id if it doesn't already exist
					if (typeof filterVal !== 'undefined' && filterVal !== '' && filters[key].indexOf(filterVal) === -1) {
						if ( Array.isArray( filterVal ) ) {
							filters[key] = filterVal;
						} else {
							filters[key] = [filterVal];
						}
					}

					// Select the filters indicated in the query string.
					if ( filters[key].length ) {
						this.selectQueryStringFilters( key, filters[key] );
					}
				}
			}

		}
138192
def _del_subscription(self, a_filter, session):
        
        deleted = 0
        try:
            subscriptions = self._subscriptions[a_filter]
            for index, (sub_session, qos) in enumerate(subscriptions):
                if sub_session.client_id == session.client_id:
                    self.logger.debug("Removing subscription on topic '%s' for client %s" %
                                      (a_filter, format_client_message(session=session)))
                    subscriptions.pop(index)
                    deleted += 1
                    break
        except KeyError:
            # Unsubscribe topic not found in current subscribed topics
            pass
        finally:
            return deleted
661136
public function match(Topic $topic)
    {
        try {
            list($routeName, $route, $attributes) = $this->pubSubRouter->match($topic->getId());

            if ($this->logger) {
                $this->logger->debug(
                    sprintf(
                        'Matched route "%s"',
                        $routeName
                    ),
                    $attributes
                );
            }

            return new WampRequest($routeName, $route, new ParameterBag($attributes), $topic->getId());
        } catch (ResourceNotFoundException $e) {
            if ($this->logger) {
                $this->logger->error(
                    sprintf(
                        'Unable to find route for %s',
                        $topic->getId()
                    )
                );
            }

            throw $e;
        }
    }
588924
async def lookup_session(self, topic_name):
        
        res = await self.call("wamp.subscription.lookup", topic_name)
        self.log.info(res)
521887
public void addFilterAt(IRuleFilter filter, Class<? extends IRuleFilter> atFilter) {
        int index = getIndexOfClass(filters, atFilter);
        if (index == -1) {
            throw new FilterAddException("filter " + atFilter.getSimpleName() + " has not been added");
        }
        filters.remove(index);
        filters.add(index, filter);
    }
379423
public function getAllPeriodicTimers(TopicInterface $topic, $name)
    {
        if (!$this->isPeriodicTimerActive($topic, $name)) {
            return false;
        }

        $namespace = $this->getTopicNamespace($topic);

        return $this->registry[$namespace][$name];
    }
1217412
def check_filtering_param(filtering_param, allowed)
      🔞 = filtering_param.keys.map(&:to_sym) - allowed.map(&:to_sym)
      raise InvalidFilterException, "Attributes #{🔞.map(&:to_s).to_sentence} doesn't exists or aren't filterables. Available filters are: #{allowed.to_sentence}" if 🔞.any?
    end
442550
private function getParams($filter)
    {
        $params = [];
        if ($filter) {
            $filter = (string) $filter;
            $filter = strtolower($filter);
            if ($filter !== 'all') {
                $validate = ['completed', 'incomplete', 'late', 'upcoming'];
                if (!in_array($filter, $validate)) {
                    throw new Exception('Invalid value for param filter');
                }
                $params['find'] = $filter;
            }
        }
        return $params;
    }
807817
private static function cleanAndValidateFilter($token, $filter)
	{
		$filter_should_be_scalar = self::shouldBeScalar($token);

		// Format the filter, cutting off the trailing ']' if appropriate
		$filter = $filter_should_be_scalar ? explode(',', substr($filter, strlen($token))) :
			explode(',', substr($filter, strlen($token), -1));

		if ($filter_should_be_scalar) {
			if (count($filter) > 1) {
				return false;
			}

			// Set to first index if should be scalar
			$filter = $filter[0];
		}

		return $filter;
	}