from rocketmq.client import Producer, Message

# Initialize the producer and set the producer group information. Use the full group name, for example, rocketmq-xxx|namespace_python%group1. During message production, the client does not need to be re-created.
producer = Producer(groupName)
# Set the service address.
producer.set_name_server_address(nameserver)
# Set permissions (role name and token).
producer.set_session_credentials(
    accessKey,  # Role token
    secretKey,  # Role name
    ''
)
# Start the producer.
producer.start()

# Assemble messages.
# Concatenate the topic with the namespace.
msg = Message(namespace + '%' + topicName)
# Set keys.
msg.set_keys(TAGS)
# Set tags.
msg.set_tags(KEYS)
# Message content
msg.set_body('This is a new message.')

# Send synchronous messages.
ret = producer.send_sync(msg)
print(ret.status, ret.msg_id, ret.offset)
# Release resources.
producer.shutdown()

import time

from rocketmq.client import PushConsumer, ConsumeStatus


# Message processing callback.
def callback(msg):
    # Simulate the business logic.
    print('Received message. messageId: ', msg.id, ' body: ', msg.body)
    # If consumption is successful, return CONSUME_SUCCESS.
    return ConsumeStatus.CONSUME_SUCCESS
    # Return message status after successful consumption.
    # return ConsumeStatus.RECONSUME_LATER


# Initialize the consumer and set consumer group information.
consumer = PushConsumer(namespace + '%' + groupName)
# Set the service address.
consumer.set_name_server_address(nameserver)
# Set permissions (role name and token).
consumer.set_session_credentials(
    accessKey,  # Role token.
    secretKey,  # Role name.
    ''
)
# Subscribe to a topic.
consumer.subscribe(namespace + '%' + topicName, callback, TAGS)
print(' [Consumer] Waiting for messages.')
# Start the consumer.
consumer.start()

while True:
    time.sleep(3600)
# Release resources.
consumer.shutdown()