You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2021/09/02 03:41:19 UTC
[GitHub] [rocketmq-streams] lizhiboo opened a new issue #47: Is FilterFunction return false indicate this message will be conducted?
lizhiboo opened a new issue #47:
URL: https://github.com/apache/rocketmq-streams/issues/47
AFAIK, FilterFunction return true indicate this message will be conducted in flink.
```java
public <O> DataStream filter(final FilterFunction<O> filterFunction) {
StageBuilder mapUDFOperator = new StageBuilder() {
@Override
protected <T> T operate(IMessage message, AbstractContext context) {
try {
boolean isFilter = filterFunction.filter((O)message.getMessageValue());
if (isFilter) {
context.breakExecute();
}
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
};
ChainStage stage = this.mainPipelineBuilder.createStage(mapUDFOperator);
this.mainPipelineBuilder.setTopologyStages(currentChainStage, stage);
return new DataStream(this.mainPipelineBuilder, this.otherPipelineBuilders, stage);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-streams] duhenglucky closed issue #47: Is FilterFunction return false indicate this message will be conducted?
Posted by GitBox <gi...@apache.org>.
duhenglucky closed issue #47:
URL: https://github.com/apache/rocketmq-streams/issues/47
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org