[Feature] Event Filter

### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/eventmesh/issues?q=is%3Aissue) and found no similar issues.

### Feature Request

### 1. Scenario

Traditional, when consumer want to filter interested messages, consumer need to rely on specified MQ client and code upon it to accomplish. But when it comes to apache EventMesh, users only need to **submit customized expressions which defines message filter logic to EventMesh**, EventMesh will do the filter and return filtered messages to users.

### 2. Implementation

#### 2.1 Implementation Plan One
Do stream processing in EventMesh, it's MQ independent, but have some impact in performance. And we can maintain the expression engine ourself.
Concrete implementation, define a set of EventMesh api to integrate and implement filter rule engine, combine with EQL and Json path. And we can also consider integrating the research rules engine (DataSonnet).

#### 2.2 Implementation Plan Two
Bind to RocketMq, use RocketMq as engine to realize filter.
Concrete implementation, implement openmessaging-api interface, and use RocketMq to realize filter ability.

#### 2.3 Implementation Conclusion
At present, it is **inclined to plan 2.1**. The ability of filter has nothing to do with specific MQ components. The specific plan still needs to be investigated.

#### 2.4 related problems:
1. Can Kafka originally support filter?
2. Does RocketMq's filter support multiple thread, currently support.
3. Consider multi-tenant segregation.
4. The storage problem of rules supports plug-in implementation.
5. How about integrate the rules engine, such as Drools or EasyRule?

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

Start by investigating EventMesh's existing APIs and message-processing flow, then compare the proposed MQ-independent approach with RocketMQ and Kafka filtering. Review the open questions around expression engines, multi-tenant rule isolation, and pluggable rule storage; the work is done only when a concrete design and implementation scope are established.

Written by the indexing model from the issue text.