adapt to localService (#973)

* update to 5.0.3 
1、use grpc.use_local_subchannel_pool
2、adapt to localService
3、fifo、delay message example

Author: zhouli11

python/example/async_producer_example.py

@@ -31,6 +31,8 @@ def handle_send_result(result_future):
     # if auth enable
     # credentials = Credentials("ak", "sk")
     config = ClientConfiguration(endpoints, credentials)
+    # with namespace
+    # config = ClientConfiguration(endpoints, credentials, "namespace")
     topic = "topic"
     producer = Producer(config, (topic,))
 
@@ -39,10 +41,14 @@ def handle_send_result(result_future):
         try:
             for i in range(10):
                 msg = Message()
+                # topic for the current message
                 msg.topic = topic
                 msg.body = "hello, rocketmq.".encode('utf-8')
+                # secondary classifier of message besides topic
                 msg.tag = "rocketmq-send-message"
+                # key(s) of the message, another way to mark message besides message id
                 msg.keys = "send_async"
+                # user property for the message
                 msg.add_property("send", "async")
                 send_result_future = producer.send_async(msg)
                 send_result_future.add_done_callback(handle_send_result)

python/example/async_simple_consumer_example.py

@@ -35,6 +35,7 @@ def receive_callback(receive_result_future, consumer):
     print(f"{consumer.__str__()} receive {len(messages)} messages.")
     for msg in messages:
         try:
+            # consume message in other thread, don't block the async receive thread
             consume_executor.submit(consume_message, consumer=consumer, message=msg)
         except Exception as exception:
             print(f"receive message raise exception: {exception}")
@@ -46,8 +47,11 @@ def receive_callback(receive_result_future, consumer):
     # if auth enable
     # credentials = Credentials("ak", "sk")
     config = ClientConfiguration(endpoints, credentials)
+    # with namespace
+    # config = ClientConfiguration(endpoints, credentials, "namespace")
     topic = "topic"
-
+    # in most case, you don't need to create too many consumers, singleton pattern is recommended
+    # close the simple consumer when you don't need it anymore
     simple_consumer = SimpleConsumer(config, "consumer-group")
     try:
         simple_consumer.startup()
@@ -58,6 +62,7 @@ def receive_callback(receive_result_future, consumer):
             while True:
                 try:
                     time.sleep(1)
+                    # max message num for each long polling and message invisible duration after it is received
                     future = simple_consumer.receive_async(32, 15)
                     future.add_done_callback(functools.partial(receive_callback, consumer=simple_consumer))
                 except Exception as e:

python/example/delay_producer_example.py

@@ -0,0 +1,50 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import time
+
+from rocketmq import ClientConfiguration, Credentials, Message, Producer
+
+if __name__ == '__main__':
+    endpoints = "foobar.com:8080"
+    credentials = Credentials()
+    # if auth enable
+    # credentials = Credentials("ak", "sk")
+    config = ClientConfiguration(endpoints, credentials)
+    # with namespace
+    # config = ClientConfiguration(endpoints, credentials, "namespace")
+    topic = "delay-topic"
+    producer = Producer(config, (topic,))
+
+    try:
+        producer.startup()
+        try:
+            msg = Message()
+            # topic for the current message
+            msg.topic = topic
+            msg.body = "hello, rocketmq.".encode('utf-8')
+            # secondary classifier of message besides topic
+            msg.tag = "rocketmq-send-delay-message"
+            # delay 10 seconds
+            msg.delivery_timestamp = int(time.time()) + 10
+            res = producer.send(msg)
+            print(f"{producer.__str__()} send message success. {res}")
+            producer.shutdown()
+            print(f"{producer.__str__()} shutdown.")
+        except Exception as e:
+            print(f"normal producer example raise exception: {e}")
+            producer.shutdown()
+    except Exception as e:
+        print(f"{producer.__str__()} startup raise exception: {e}")
+        producer.shutdown()

python/example/fifo_producer_example.py

@@ -0,0 +1,49 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from rocketmq import ClientConfiguration, Credentials, Message, Producer
+
+if __name__ == '__main__':
+    endpoints = "foobar.com:8080"
+    credentials = Credentials()
+    # if auth enable
+    # credentials = Credentials("ak", "sk")
+    config = ClientConfiguration(endpoints, credentials)
+    # with namespace
+    # config = ClientConfiguration(endpoints, credentials, "namespace")
+    topic = "fifo-topic"
+    producer = Producer(config, (topic,))
+
+    try:
+        producer.startup()
+        try:
+            msg = Message()
+            # topic for the current message
+            msg.topic = topic
+            msg.body = "hello, rocketmq.".encode('utf-8')
+            # secondary classifier of message besides topic
+            msg.tag = "rocketmq-send-fifo-message"
+            # message group decides the message delivery order
+            msg.message_group = "your-message-group0"
+            res = producer.send(msg)
+            print(f"{producer.__str__()} send message success. {res}")
+            producer.shutdown()
+            print(f"{producer.__str__()} shutdown.")
+        except Exception as e:
+            print(f"normal producer example raise exception: {e}")
+            producer.shutdown()
+    except Exception as e:
+        print(f"{producer.__str__()} startup raise exception: {e}")
+        producer.shutdown()

python/example/normal_producer_example.py

@@ -21,17 +21,23 @@
     # if auth enable
     # credentials = Credentials("ak", "sk")
     config = ClientConfiguration(endpoints, credentials)
+    # with namespace
+    # config = ClientConfiguration(endpoints, credentials, "namespace")
     topic = "topic"
     producer = Producer(config, (topic,))
 
     try:
         producer.startup()
         try:
             msg = Message()
+            # topic for the current message
             msg.topic = topic
             msg.body = "hello, rocketmq.".encode('utf-8')
+            # secondary classifier of message besides topic
             msg.tag = "rocketmq-send-message"
+            # key(s) of the message, another way to mark message besides message id
             msg.keys = "send_sync"
+            # user property for the message
             msg.add_property("send", "sync")
             res = producer.send(msg)
             print(f"{producer.__str__()} send message success. {res}")

python/example/simple_consumer_example.py

@@ -21,7 +21,11 @@
     # if auth enable
     # credentials = Credentials("ak", "sk")
     config = ClientConfiguration(endpoints, credentials)
+    # with namespace
+    # config = ClientConfiguration(endpoints, credentials, "namespace")
     topic = "topic"
+    # in most case, you don't need to create too many consumers, singleton pattern is recommended
+    # close the simple consumer when you don't need it anymore
     simple_consumer = SimpleConsumer(config, "consumer-group")
     try:
         simple_consumer.startup()
@@ -31,6 +35,7 @@
             # simple_consumer.subscribe(topic, FilterExpression("tag"))
             while True:
                 try:
+                    # max message num for each long polling and message invisible duration after it is received
                     messages = simple_consumer.receive(32, 15)
                     if messages is not None:
                         print(f"{simple_consumer.__str__()} receive {len(messages)} messages.")

python/example/transaction_producer_example.py

@@ -30,6 +30,8 @@ def check(self, message: Message) -> TransactionResolution:
     # if auth enable
     # credentials = Credentials("ak", "sk")
     config = ClientConfiguration(endpoints, credentials)
+    # with namespace
+    # config = ClientConfiguration(endpoints, credentials, "namespace")
     topic = "topic"
     check_from_server = True  # commit message from server check
     producer = Producer(config, (topic,), checker=TestChecker())
@@ -42,18 +44,23 @@ def check(self, message: Message) -> TransactionResolution:
     try:
         transaction = producer.begin_transaction()
         msg = Message()
+        # topic for the current message
         msg.topic = topic
         msg.body = "hello, rocketmq.".encode('utf-8')
+        # secondary classifier of message besides topic
         msg.tag = "rocketmq-send-transaction-message"
+        # key(s) of the message, another way to mark message besides message id
         msg.keys = "send_transaction"
+        # user property for the message
         msg.add_property("send", "transaction")
         res = producer.send(msg, transaction)
         print(f"send message: {res}")
         if check_from_server:
+            # wait for server check in TransactionChecker's check
             input("Please Enter to Stop the Application.\r\n")
             producer.shutdown()
         else:
-            # producer directly commit or rollback
+            # direct commit or rollback
             transaction.commit()
             print(f"producer commit message:{transaction.message_id}")
             # transaction.rollback()

python/rocketmq/v5/client/balancer/queue_selector.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import hashlib
 import random
 
 from rocketmq.v5.exception import IllegalArgumentException
@@ -64,6 +65,12 @@ def select_next_queue(self):
             self.__index.get_and_increment() % len(self.__message_queues)
         ]
 
+    def select_queue_by_hash_key(self, key):
+        hash_object = hashlib.sha256(key.encode('utf-8'))
+        hash_code = int.from_bytes(hash_object.digest(), byteorder='big')
+        print(f"hashcode: {hash_code}")
+        return self.__message_queues[hash_code % len(self.__message_queues)]
+
     def all_queues(self):
         index = self.__index.get_and_increment() % len(self.__message_queues)
         return self.__message_queues[index:] + self.__message_queues[:index]

python/rocketmq/v5/client/connection/rpc_channel.py

@@ -88,7 +88,7 @@ def __facade(self):
             or len(self.__addresses) == 0
             or self.__scheme == AddressScheme.ADDRESS_SCHEME_UNSPECIFIED
         ):
-            return ""
+            return "", ""
 
         prefix = "dns:"
         if self.__scheme == AddressScheme.IPv4:
@@ -227,6 +227,7 @@ def __create_aio_channel(self):
                     ("grpc.enable_retries", 0),
                     ("grpc.max_send_message_length", -1),
                     ("grpc.max_receive_message_length", -1),
+                    ("grpc.use_local_subchannel_pool", 1),
                 ]
                 if self.__tls_enabled:
                     self.__async_channel = aio.secure_channel(
@@ -237,7 +238,7 @@ def __create_aio_channel(self):
                         self.__endpoints.facade, options
                     )
                 self.__async_stub = MessagingServiceStub(self.__async_channel)
-                logger.debug(
+                logger.info(
                     f"create_aio_channel to [{self.__endpoints.__str__()}] success. channel state:{self.__async_channel.get_state()}"
                 )
         except Exception as e: