solace-pubsubplus

import os import time import threading from solace_pubsubplus import SolaceSession, SolaceSessionProperties, Message, MessageBuilder, DeliveryMode, Subscription, SubscribeFlags, SolaceLogLevel # Configuration from environment variables SOLACE_HOST = os.environ.get("SOLACE_HOST", "localhost:55555") # e.g., 'tcp://your_broker_host:55555' SOLACE_VPN = os.environ.get("SOLACE_VPN", "default") SOLACE_USERNAME = os.environ.get("SOLACE_USERNAME", "username") SOLACE_PASSWORD = os.environ.get("SOLACE_PASSWORD", "password") TOPIC_NAME = "python/quickstart/hello" # Global lock for thread-safe printing print_lock = threading.Lock() def on_message_received(message: Message): """Callback function for received messages.""" with print_lock: print(f"Received message on topic '{message.get_destination_as_topic()}'") if message.get_data(): print(f"Message data: {message.get_data().decode('utf-8')}") def run_quickstart(): # 1. Configure Solace Session Properties session_properties = SolaceSessionProperties() session_properties.host = SOLACE_HOST session_properties.vpn_name = SOLACE_VPN session_properties.username = SOLACE_USERNAME session_properties.password = SOLACE_PASSWORD session_properties.connect_retries = 3 session_properties.reconnect_retries = 3 # Set log level for more detailed output (optional) SolaceSession.set_log_level(SolaceLogLevel.INFO) session = None try: # 2. Create and Connect the Solace Session session = SolaceSession(session_properties) session.connect() with print_lock: print(f"Connected to Solace PubSub+ broker at {SOLACE_HOST}") # 3. Set up Message Callback session.set_message_receive_callback(on_message_received) # 4. Subscribe to a Topic subscription = Subscription(TOPIC_NAME) session.subscribe(subscription, SubscribeFlags.WAIT_FOR_CONFIRM) with print_lock: print(f"Subscribed to topic '{TOPIC_NAME}'") # 5. Publish a Message message_builder = MessageBuilder() message = message_builder \ .with_delivery_mode(DeliveryMode.DIRECT) \ .with_topic(TOPIC_NAME) \ .with_property("user-property", "value") \ .build() message.set_data(f"Hello from Python Quickstart! Timestamp: {time.time()}".encode('utf-8')) session.publish(message) with print_lock: print(f"Published message to topic '{TOPIC_NAME}'") # 6. Wait briefly for the message to be received (if publishing to self) time.sleep(2) except Exception as e: with print_lock: print(f"An error occurred: {e}") finally: if session: # 7. Disconnect the Session with print_lock: print("Disconnecting Solace Session...") session.disconnect() with print_lock: print("Disconnected.") if __name__ == "__main__": run_quickstart()