Spaces:
Sleeping
Sleeping
| from confluent_kafka import Producer | |
| import json | |
| import logging | |
| import os | |
| from utils.dapr_utils import dapr_http_fallback | |
| logger = logging.getLogger(__name__) | |
| def delivery_report(err, msg): | |
| """Callback for reporting message delivery results.""" | |
| if err is not None: | |
| logger.error(f'Message delivery failed: {err}') | |
| else: | |
| logger.info(f'Message delivered to {msg.topic()} [{msg.partition()}]') | |
| def publish_task_event(event_type: str, task_data: dict) -> bool: | |
| """ | |
| Publish a task event to Dapr pub/sub, with Kafka fallback if Dapr not available. | |
| Args: | |
| event_type: Type of event (e.g., 'task_created', 'task_completed', 'task_updated') | |
| task_data: Dictionary containing task information | |
| Returns: | |
| bool: True if event published successfully, False otherwise | |
| """ | |
| # First, try to use Dapr sidecar if available | |
| dapr_response = dapr_http_fallback( | |
| endpoint="/v1.0/publish/task-pubsub/task-events", | |
| method="POST", | |
| data={ | |
| "event_type": event_type, | |
| "task_data": task_data, | |
| "timestamp": task_data.get('updated_at', task_data.get('created_at')) | |
| } | |
| ) | |
| if dapr_response is not None: | |
| # Dapr succeeded | |
| logger.info(f"Published {event_type} event via Dapr for task: {task_data.get('id', 'unknown')}") | |
| return True | |
| else: | |
| # Dapr not available, fall back to Kafka | |
| logger.info("Dapr sidecar not available, falling back to Kafka") | |
| # Fallback to Kafka | |
| try: | |
| # Get Kafka configuration from environment variables | |
| bootstrap_servers = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092') | |
| kafka_username = os.getenv('KAFKA_USERNAME', '') | |
| kafka_password = os.getenv('KAFKA_PASSWORD', '') | |
| # Configure Kafka producer with SASL_SSL and SCRAM-SHA-256 | |
| conf = { | |
| 'bootstrap.servers': bootstrap_servers, | |
| 'security.protocol': 'SASL_SSL', | |
| 'sasl.mechanism': 'SCRAM-SHA-256', | |
| 'sasl.username': kafka_username, | |
| 'sasl.password': kafka_password, | |
| 'acks': 'all' | |
| } | |
| # Create producer | |
| producer = Producer(conf) | |
| # Create the event payload | |
| event_payload = { | |
| "event_type": event_type, | |
| "task_data": task_data, | |
| "timestamp": task_data.get('updated_at', task_data.get('created_at')) | |
| } | |
| # Convert to JSON string | |
| message_value = json.dumps(event_payload) | |
| # Asynchronously produce a message, the delivery report callback | |
| # will be triggered from poll() above, or flush() below | |
| producer.produce('task-events', message_value.encode('utf-8'), callback=delivery_report) | |
| # Wait for any outstanding messages to be delivered and delivery report | |
| # callbacks to be triggered | |
| producer.flush() | |
| logger.info(f"Published {event_type} event via Kafka for task: {task_data.get('id', 'unknown')}") | |
| return True | |
| except Exception as e: | |
| logger.error(f"Error publishing {event_type} event: {str(e)}") | |
| return False |