[ISSUE #1199] Implement the 1.0.0 openmessaging new consumer API for rocketmq oms module#1240
[ISSUE #1199] Implement the 1.0.0 openmessaging new consumer API for rocketmq oms module#1240zongtanghu merged 8 commits intoapache:feature_oms_1.0.0from odbozhou:feature_oms_1.0.0
Conversation
|
Hi @odbozhou do you verify this pr in your local enviroment? |
Yes，Sending and consuming messages have been validated |
| public void run() { | ||
| consumer.shutdown(); | ||
| messagingAccessPoint.shutdown(); | ||
| consumer.stop(); |
There was a problem hiding this comment.
Could you make a clarify why remove messagingAccessPoint lifecyle management in the new API，just for simplicity?
There was a problem hiding this comment.
There are startup and shutdown methods in oms-java-0.3.1 alpha MessagingAccessPoint, but there is no content in old rocketmq-open messaging implementation. In openmessaging-java-1.0.0, the MessagingAccessPoint lifecycle API has been removed, not in the new rocketmq-openmessaging implementation.
| @Override | ||
| public void routing(String sourceQueue, String targetQueue) { | ||
| @Override | ||
| public ResourceManager resourceManager() { |
There was a problem hiding this comment.
I could not find the usage of resourceManager in your new code.
There was a problem hiding this comment.
The current PR is too big, and resourceManager will be implemented in the next pr
| import io.openmessaging.rocketmq.config.DefaultQueueMetaData; | ||
| import io.openmessaging.rocketmq.domain.ConsumeRequest; | ||
| import io.openmessaging.rocketmq.domain.NonStandardKeys; | ||
| import io.openmessaging.rocketmq.utils.OMSUtil; |
There was a problem hiding this comment.
Name has been modified
|
Just quickly look through, it needs more tests to make the code reliable in addition to what I have pointed in code review. But if you could do more followup, I would like to help to review again and again... I think it is a good polish for integration with OpenMessaging. |
I will add more unit tests and look forward to contributing more to the community |
| messageExt2 = messageExt; | ||
| } | ||
| } | ||
| assertThat(messageExt1).isNotNull(); |
There was a problem hiding this comment.
Here,I think it will be better if you can verify the messageExt's detailed content values by using assertEquals method.For example,you can verify body value,UserProperty and topic,other places you can refer to modify if the same as here. @odbozhou
There was a problem hiding this comment.
This has increased verification
| queueMetaData3 = queueMetaData; | ||
| } | ||
| } | ||
| assertThat(queueMetaData1).isNotNull(); |
There was a problem hiding this comment.
This has increased verification
| message2 = message; | ||
| } | ||
| } | ||
| assertThat(message1).isNotNull(); |
There was a problem hiding this comment.
This has increased verification
| message2 = message; | ||
| } | ||
| } | ||
| assertThat(message1).isNotNull(); |
There was a problem hiding this comment.
The same as above,you can adjust codes appropriately!
There was a problem hiding this comment.
This has increased verification
|
@vongosling @duhenglucky @ShannonDing Please help to review this pr again.Thanks very much. |
|
LGTM |
4 participants
What is the purpose of the change
#1199 Implement the 1.0.0 openmessaging new consumer API for rocketmq oms module
Brief changelog
1、Adapt to the new consumer api
2、Adjust consumer example
3、Optimize consumer code implementation
4、Fix bug
Verifying this change
Follow this checklist to help us incorporate your contribution quickly and easily. Notice,
it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.[ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyleto make sure basic checks pass. Runmvn clean install -DskipITsto make sure unit-test pass. Runmvn clean test-compile failsafe:integration-testto make sure integration-test pass.