This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new af9c7e23 Remove redundant __init__.py (#554)
af9c7e23 is described below
commit af9c7e23fd326c54607bcd217c8f7a5a8eeee566
Author: Aaron Ai <[email protected]>
AuthorDate: Tue Jun 27 17:28:23 2023 +0800
Remove redundant __init__.py (#554)
---
python/rocketmq/__init__.py | 38 --------------------------------------
python/rocketmq/client.py | 2 +-
python/rocketmq/log.py | 3 +--
python/rocketmq/producer.py | 29 +++++++++++++++++++++--------
python/rocketmq/rpc_client.py | 2 +-
5 files changed, 24 insertions(+), 50 deletions(-)
diff --git a/python/rocketmq/__init__.py b/python/rocketmq/__init__.py
deleted file mode 100644
index 3fc459fd..00000000
--- a/python/rocketmq/__init__.py
+++ /dev/null
@@ -1,38 +0,0 @@
-# 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 logging
-import os
-
-logger = logging.getLogger("rocketmqlogger")
-logger.setLevel(logging.DEBUG)
-
-log_path = os.path.join(
- os.path.expanduser("~"), "logs", "rocketmq", "rocketmq-client.log"
-)
-file_handler = logging.FileHandler(log_path)
-file_handler.setLevel(logging.DEBUG)
-
-console_handler = logging.StreamHandler()
-console_handler.setLevel(logging.DEBUG)
-
-formatter = logging.Formatter(
- "%(asctime)s [%(levelname)s] [%(process)d]
[%(filename)s#%(funcName)s:%(lineno)d] %(message)s"
-)
-file_handler.setFormatter(formatter)
-console_handler.setFormatter(formatter)
-
-logger.addHandler(file_handler)
-logger.addHandler(console_handler)
diff --git a/python/rocketmq/client.py b/python/rocketmq/client.py
index 4ccda2b9..0ef32e60 100644
--- a/python/rocketmq/client.py
+++ b/python/rocketmq/client.py
@@ -39,7 +39,7 @@ class Client:
self.sessionsLock = threading.Lock()
self.client_manager = ClientManager(self)
- async def start_up(self):
+ async def start(self):
# get topic route
for topic in self.topics:
self.topic_route_cache[topic] = await self.fetch_topic_route(topic)
diff --git a/python/rocketmq/log.py b/python/rocketmq/log.py
index a9c82e4b..f3e4eae3 100644
--- a/python/rocketmq/log.py
+++ b/python/rocketmq/log.py
@@ -29,8 +29,7 @@ console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter(
- "%(asctime)s [%(levelname)s] [%(process)d] [%(threadName)s] [%(filename)s#\
- %(funcName)s:%(lineno)d] %(message)s"
+ "%(asctime)s [%(levelname)s] [%(process)d] [%(threadName)s]
[%(filename)s#%(funcName)s:%(lineno)d] %(message)s"
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
diff --git a/python/rocketmq/producer.py b/python/rocketmq/producer.py
index c9c0d351..3d604a25 100644
--- a/python/rocketmq/producer.py
+++ b/python/rocketmq/producer.py
@@ -21,6 +21,7 @@ import rocketmq
from rocketmq.client import Client
from rocketmq.client_config import ClientConfig
from rocketmq.definition import TopicRouteData
+from rocketmq.log import logger
from rocketmq.message_id_codec import MessageIdCodec
from rocketmq.protocol.definition_pb2 import Message as ProtoMessage
from rocketmq.protocol.definition_pb2 import Resource, SystemProperties
@@ -92,8 +93,20 @@ class Producer(Client):
self.client_id, self.endpoints, None, 10, topics
)
- async def start_up(self):
- await super().start_up()
+ async def __aenter__(self):
+ await self.start()
+
+ async def __aexit__(self, exc_type, exc_val, exc_tb):
+ await self.shutdown()
+
+ async def start(self):
+ logger.info(f"Begin to start the rocketmq producer,
client_id={self.client_id}")
+ await super().start()
+ logger.info(f"The rocketmq producer starts successfully,
client_id={self.client_id}")
+
+ async def shutdown(self):
+ logger.info(f"Begin to shutdown the rocketmq producer,
client_id={self.client_id}")
+ logger.info(f"Shutdown the rocketmq producer successfully,
client_id={self.client_id}")
async def send_message(self, message):
req = SendMessageRequest()
@@ -107,14 +120,13 @@ class Producer(Client):
async def test():
- creds = SessionCredentials("username", "password")
- creds_provider = SessionCredentialsProvider(creds)
+ credentials = SessionCredentials("username", "password")
+ credentials_provider = SessionCredentialsProvider(credentials)
client_config = ClientConfig(
endpoints=Endpoints("rmq-cn-jaj390gga04.cn-hangzhou.rmq.aliyuncs.com:8080"),
- session_credentials_provider=creds_provider,
+ session_credentials_provider=credentials_provider,
ssl_enabled=True,
)
- producer = Producer(client_config, topics={"normal_topic"})
topic = Resource()
topic.name = "normal_topic"
msg = ProtoMessage()
@@ -123,8 +135,9 @@ async def test():
sysperf = SystemProperties()
sysperf.message_id = MessageIdCodec.next_message_id()
msg.system_properties.CopyFrom(sysperf)
- print(msg)
- await producer.start_up()
+ logger.info(f"{msg}")
+ producer = Producer(client_config, topics={"normal_topic"})
+ await producer.start()
result = await producer.send_message(msg)
print(result)
diff --git a/python/rocketmq/rpc_client.py b/python/rocketmq/rpc_client.py
index a737d3d0..c1f129c4 100644
--- a/python/rocketmq/rpc_client.py
+++ b/python/rocketmq/rpc_client.py
@@ -24,7 +24,7 @@ from functools import reduce
import certifi
from grpc import aio, ssl_channel_credentials
from protocol import service_pb2
-from rocketmq import logger
+from rocketmq.log import logger
from rocketmq.protocol import service_pb2_grpc
from rocketmq.protocol.definition_pb2 import Address as ProtoAddress
from rocketmq.protocol.definition_pb2 import \
