Virtual transport of RocketMQ

[imzs]

#2305
Add virtual transport for RocketMQ

[auvipy]

can you please also fix the failing test

=================================== FAILURES ===================================
____________________________ test_Channel.test_get _____________________________

self = <t.unit.transport.test_rocketmq.test_Channel object at 0x7f75eb63bbf0>

def test_get(self):
    queue = 'new-queue'
    self.channel.basic_consume(queue, True, None, 'cg1')

    self.consumer.receive.return_value = []
    with pytest.raises(Empty):
        self.channel._get(queue)

    rocketmq_message = _message_to_rocketmq_ack_message(_mock_message(topic=queue, queue=queue))

  rocketmq_message.body = str_to_bytes(dumps(self.channel.prepare_message({})))

t/unit/transport/test_rocketmq.py:150:

---

self = <rocketmq.v5.model.message.Message object at 0x7f75d9132c00>
body = b'{"body": {}, "content-encoding": null, "content-type": null, "headers": {}, "properties": {"delivery_info": {}, "priority": 0}}'
Restoring 2 unacknowledged message(s)

@body.setter
def body(self, body):

  if body is None or body.strip() == "":

E BytesWarning: Comparison between bytes and string

.tox/3.12-unit/lib/python3.12/site-packages/rocketmq/v5/model/message.py:175: BytesWarning

[imzs]

can you please also fix the failing test

=================================== FAILURES =================================== ____________________________ test_Channel.test_get _____________________________

self = <t.unit.transport.test_rocketmq.test_Channel object at 0x7f75eb63bbf0>

def test_get(self):
    queue = 'new-queue'
    self.channel.basic_consume(queue, True, None, 'cg1')

    self.consumer.receive.return_value = []
    with pytest.raises(Empty):
        self.channel._get(queue)

    rocketmq_message = _message_to_rocketmq_ack_message(_mock_message(topic=queue, queue=queue))

  rocketmq_message.body = str_to_bytes(dumps(self.channel.prepare_message({})))

t/unit/transport/test_rocketmq.py:150:

self = <rocketmq.v5.model.message.Message object at 0x7f75d9132c00> body = b'{"body": {}, "content-encoding": null, "content-type": null, "headers": {}, "properties": {"delivery_info": {}, "priority": 0}}' Restoring 2 unacknowledged message(s)

@body.setter
def body(self, body):

  if body is None or body.strip() == "":

E BytesWarning: Comparison between bytes and string

.tox/3.12-unit/lib/python3.12/site-packages/rocketmq/v5/model/message.py:175: BytesWarning

hi, I've suppressed the BytesWarning for those test cases, and I'll submit an issue to the SDK community to get it fixed.

[auvipy]

Ok, thanks

[imzs]

@auvipy hi, I've fixed the issue which causes test failure in py313, could you help me trigger the workflow?

[auvipy]

Sure

[imzs]

@auvipy hi, the conflict was fixed and I notice that python-version has changed, so could you help me trigger the workflow?

[codecov]

Codecov Report

❌ Patch coverage is 82.66254% with 56 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.23%. Comparing base (0a5a0bc) to head (dcb38e2).
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
kombu/transport/rocketmq.py	82.66%	42 Missing and 14 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2306      +/-   ##
==========================================
+ Coverage   81.19%   81.23%   +0.04%     
==========================================
  Files          77       78       +1     
  Lines        9735    10058     +323     
  Branches     1098     1133      +35     
==========================================
+ Hits         7904     8171     +267     
- Misses       1625     1667      +42     
- Partials      206      220      +14     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[Copilot]

Pull Request Overview

This PR adds RocketMQ transport support to Kombu, enabling message queuing through Apache RocketMQ as a backend. The implementation follows the virtual transport pattern and provides a comprehensive mapping between AMQP semantics and RocketMQ concepts.

• Implements RocketMQ transport module with Channel, QoS, and Transport classes
• Adds comprehensive test coverage for the new RocketMQ transport
• Includes documentation and dependency configuration for the RocketMQ integration

Reviewed Changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
kombu/transport/rocketmq.py	Main implementation of RocketMQ transport with Channel, QoS, and Transport classes, including message handling, producer/consumer management, and AMQP-to-RocketMQ mapping
t/unit/transport/test_rocketmq.py	Comprehensive unit tests for RocketMQ transport covering QoS operations, channel operations, producer/consumer management, and message handling
requirements/extras/rocketmq.txt	Defines RocketMQ-specific dependencies (rocketmq-python-client and grpcio-tools)
requirements/test-ci.txt	Adds RocketMQ requirements to CI test dependencies
requirements/extras/gcpubsub.txt	Updates protobuf version specification
kombu/transport/init.py	Registers the RocketMQ transport in the transport registry
docs/reference/kombu.transport.rocketmq.rst	Adds Sphinx documentation for the RocketMQ transport module
docs/reference/index.rst	Adds RocketMQ transport to the documentation index

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[imzs]

@auvipy I've fixed the typo, indentation, deprecated method in resolve-conflict branch based on Copilot's suggestions.

[imzs]

@auvipy Hi, is there anything else I should improve or clarify? I’m happy to make any further adjustments needed.