evyos-center-server/prod-wag-backend-automate-services
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

black shift

Browse Source
This commit is contained in:
berkay
2025-04-22 11:10:29 +03:00
parent d7f1da8de8
commit e5f88f2eb4
30 changed files with 671 additions and 521 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
Controllers/Redis/Broadcast/actions.py
View File

@@ -8,17 +8,17 @@ from Controllers.Redis.response import RedisResponse
class RedisPublisher:
"""Redis Publisher class for broadcasting messages to channels."""
def __init__(self, redis_client=redis_cli):
self.redis_client = redis_client
def publish(self, channel: str, message: Union[Dict, List, str]) -> RedisResponse:
"""Publish a message to a Redis channel.
Args:
channel: The channel to publish to
message: The message to publish (will be JSON serialized if dict or list)
Returns:
RedisResponse with status and message
"""
@@ -26,113 +26,124 @@ class RedisPublisher:
# Convert dict/list to JSON string if needed
if isinstance(message, (dict, list)):
message = json.dumps(message)
# Publish the message
recipient_count = self.redis_client.publish(channel, message)
return RedisResponse(
status=True,
message=f"Message published successfully to {channel}.",
data={"recipients": recipient_count}
data={"recipients": recipient_count},
)
except Exception as e:
return RedisResponse(
status=False,
message=f"Failed to publish message to {channel}.",
error=str(e)
error=str(e),
)
class RedisSubscriber:
"""Redis Subscriber class for listening to channels."""
def __init__(self, redis_client=redis_cli):
self.redis_client = redis_client
self.pubsub = self.redis_client.pubsub()
self.active_threads = {}
def subscribe(self, channel: str, callback: Callable[[Dict], None]) -> RedisResponse:
def subscribe(
self, channel: str, callback: Callable[[Dict], None]
) -> RedisResponse:
"""Subscribe to a Redis channel with a callback function.
Args:
channel: The channel to subscribe to
callback: Function to call when a message is received
Returns:
RedisResponse with status and message
"""
try:
# Subscribe to the channel
self.pubsub.subscribe(**{channel: self._message_handler(callback)})
return RedisResponse(
status=True,
message=f"Successfully subscribed to {channel}."
status=True, message=f"Successfully subscribed to {channel}."
)
except Exception as e:
return RedisResponse(
status=False,
message=f"Failed to subscribe to {channel}.",
error=str(e)
status=False, message=f"Failed to subscribe to {channel}.", error=str(e)
)
def psubscribe(self, pattern: str, callback: Callable[[Dict], None]) -> RedisResponse:
def psubscribe(
self, pattern: str, callback: Callable[[Dict], None]
) -> RedisResponse:
"""Subscribe to Redis channels matching a pattern.
Args:
pattern: The pattern to subscribe to (e.g., 'user.*')
callback: Function to call when a message is received
Returns:
RedisResponse with status and message
"""
try:
# Subscribe to the pattern
self.pubsub.psubscribe(**{pattern: self._message_handler(callback)})
return RedisResponse(
status=True,
message=f"Successfully pattern-subscribed to {pattern}."
status=True, message=f"Successfully pattern-subscribed to {pattern}."
)
except Exception as e:
return RedisResponse(
status=False,
message=f"Failed to pattern-subscribe to {pattern}.",
error=str(e)
error=str(e),
)
def _message_handler(self, callback: Callable[[Dict], None]):
"""Create a message handler function for the subscription."""
def handler(message):
# Skip subscription confirmation messages
if message['type'] in ('subscribe', 'psubscribe'):
if message["type"] in ("subscribe", "psubscribe"):
return
# Parse JSON if the message is a JSON string
data = message['data']
data = message["data"]
if isinstance(data, bytes):
data = data.decode('utf-8')
data = data.decode("utf-8")
try:
data = json.loads(data)
except json.JSONDecodeError:
# Not JSON, keep as is
pass
# Call the callback with the message data
callback({
'channel': message.get('channel', b'').decode('utf-8') if isinstance(message.get('channel', b''), bytes) else message.get('channel', ''),
'pattern': message.get('pattern', b'').decode('utf-8') if isinstance(message.get('pattern', b''), bytes) else message.get('pattern', ''),
'data': data
})
callback(
{
"channel": (
message.get("channel", b"").decode("utf-8")
if isinstance(message.get("channel", b""), bytes)
else message.get("channel", "")
),
"pattern": (
message.get("pattern", b"").decode("utf-8")
if isinstance(message.get("pattern", b""), bytes)
else message.get("pattern", "")
),
"data": data,
}
)
return handler
def start_listening(self, in_thread: bool = True) -> RedisResponse:
"""Start listening for messages on subscribed channels.
Args:
in_thread: If True, start listening in a separate thread
Returns:
RedisResponse with status and message
"""
@@ -140,50 +151,41 @@ class RedisSubscriber:
if in_thread:
thread = Thread(target=self._listen_thread, daemon=True)
thread.start()
self.active_threads['listener'] = thread
self.active_threads["listener"] = thread
return RedisResponse(
status=True,
message="Listening thread started successfully."
status=True, message="Listening thread started successfully."
)
else:
# This will block the current thread
self._listen_thread()
return RedisResponse(
status=True,
message="Listening started successfully (blocking)."
status=True, message="Listening started successfully (blocking)."
)
except Exception as e:
return RedisResponse(
status=False,
message="Failed to start listening.",
error=str(e)
status=False, message="Failed to start listening.", error=str(e)
)
def _listen_thread(self):
"""Thread function for listening to messages."""
self.pubsub.run_in_thread(sleep_time=0.01)
def stop_listening(self) -> RedisResponse:
"""Stop listening for messages."""
try:
self.pubsub.close()
return RedisResponse(
status=True,
message="Successfully stopped listening."
)
return RedisResponse(status=True, message="Successfully stopped listening.")
except Exception as e:
return RedisResponse(
status=False,
message="Failed to stop listening.",
error=str(e)
status=False, message="Failed to stop listening.", error=str(e)
)
def unsubscribe(self, channel: Optional[str] = None) -> RedisResponse:
"""Unsubscribe from a channel or all channels.
Args:
channel: The channel to unsubscribe from, or None for all channels
Returns:
RedisResponse with status and message
"""
@@ -194,24 +196,21 @@ class RedisSubscriber:
else:
self.pubsub.unsubscribe()
message = "Successfully unsubscribed from all channels."
return RedisResponse(
status=True,
message=message
)
return RedisResponse(status=True, message=message)
except Exception as e:
return RedisResponse(
status=False,
message=f"Failed to unsubscribe from {'channel' if channel else 'all channels'}.",
error=str(e)
error=str(e),
)
def punsubscribe(self, pattern: Optional[str] = None) -> RedisResponse:
"""Unsubscribe from a pattern or all patterns.
Args:
pattern: The pattern to unsubscribe from, or None for all patterns
Returns:
RedisResponse with status and message
"""
@@ -222,24 +221,21 @@ class RedisSubscriber:
else:
self.pubsub.punsubscribe()
message = "Successfully unsubscribed from all patterns."
return RedisResponse(
status=True,
message=message
)
return RedisResponse(status=True, message=message)
except Exception as e:
return RedisResponse(
status=False,
message=f"Failed to unsubscribe from {'pattern' if pattern else 'all patterns'}.",
error=str(e)
error=str(e),
)
class RedisPubSub:
"""Singleton class that provides both publisher and subscriber functionality."""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(RedisPubSub, cls).__new__(cls)

79
Controllers/Redis/Broadcast/implementations.py
View File

@@ -15,6 +15,7 @@ CHANNEL_WRITER = "chain:writer"
# Flag to control the demo
running = True
def generate_mock_data():
"""Generate a mock message with UUID, timestamp, and sample data."""
return {
@@ -24,40 +25,43 @@ def generate_mock_data():
"data": {
"value": f"Sample data {int(time.time())}",
"status": "new",
"counter": 0
}
"counter": 0,
},
}
def reader_function():
"""
First function in the chain.
Generates mock data and publishes to the reader channel.
"""
print("[READER] Function started")
while running:
# Generate mock data
message = generate_mock_data()
start_time = time.time()
message["start_time"] = start_time
# Publish to reader channel
result = redis_pubsub.publisher.publish(CHANNEL_READER, message)
if result.status:
print(f"[READER] {time.time():.6f} | Published UUID: {message['uuid']}")
else:
print(f"[READER] Publish error: {result.error}")
# Wait before generating next message
time.sleep(2)
def processor_function():
"""
Second function in the chain.
Subscribes to reader channel, processes messages, and publishes to processor channel.
"""
print("[PROCESSOR] Function started")
def on_reader_message(message):
# The message structure from the subscriber has 'data' containing our actual message
# If data is a string, parse it as JSON
@@ -68,47 +72,51 @@ def processor_function():
except json.JSONDecodeError as e:
print(f"[PROCESSOR] Error parsing message data: {e}")
return
# Check if stage is 'red' before processing
if data.get("stage") == "red":
# Process the message
data["processor_timestamp"] = datetime.now().isoformat()
data["data"]["status"] = "processed"
data["data"]["counter"] += 1
# Update stage to 'processed'
data["stage"] = "processed"
# Add some processing metadata
data["processing"] = {
"duration_ms": 150, # Mock processing time
"processor_id": "main-processor"
"processor_id": "main-processor",
}
# Publish to processor channel
result = redis_pubsub.publisher.publish(CHANNEL_PROCESSOR, data)
if result.status:
print(f"[PROCESSOR] {time.time():.6f} | Received UUID: {data['uuid']} | Published UUID: {data['uuid']}")
print(
f"[PROCESSOR] {time.time():.6f} | Received UUID: {data['uuid']} | Published UUID: {data['uuid']}"
)
else:
print(f"[PROCESSOR] Publish error: {result.error}")
else:
print(f"[PROCESSOR] Skipped message: {data['uuid']} (stage is not 'red')")
# Subscribe to reader channel
result = redis_pubsub.subscriber.subscribe(CHANNEL_READER, on_reader_message)
if result.status:
print(f"[PROCESSOR] Subscribed to channel: {CHANNEL_READER}")
else:
print(f"[PROCESSOR] Subscribe error: {result.error}")
def writer_function():
"""
Third function in the chain.
Subscribes to processor channel and performs final processing.
"""
print("[WRITER] Function started")
def on_processor_message(message):
# The message structure from the subscriber has 'data' containing our actual message
# If data is a string, parse it as JSON
@@ -119,42 +127,45 @@ def writer_function():
except json.JSONDecodeError as e:
print(f"[WRITER] Error parsing message data: {e}")
return
# Check if stage is 'processed' before processing
if data.get("stage") == "processed":
# Process the message
data["writer_timestamp"] = datetime.now().isoformat()
data["data"]["status"] = "completed"
data["data"]["counter"] += 1
# Update stage to 'completed'
data["stage"] = "completed"
# Add some writer metadata
data["storage"] = {
"location": "main-db",
"partition": "events-2025-04"
}
data["storage"] = {"location": "main-db", "partition": "events-2025-04"}
# Calculate elapsed time if start_time is available
current_time = time.time()
elapsed_ms = ""
if "start_time" in data:
elapsed_ms = f" | Elapsed: {(current_time - data['start_time']) * 1000:.2f}ms"
elapsed_ms = (
f" | Elapsed: {(current_time - data['start_time']) * 1000:.2f}ms"
)
# Optionally publish to writer channel for any downstream listeners
result = redis_pubsub.publisher.publish(CHANNEL_WRITER, data)
if result.status:
print(f"[WRITER] {current_time:.6f} | Received UUID: {data['uuid']} | Published UUID: {data['uuid']}{elapsed_ms}")
print(
f"[WRITER] {current_time:.6f} | Received UUID: {data['uuid']} | Published UUID: {data['uuid']}{elapsed_ms}"
)
else:
print(f"[WRITER] Publish error: {result.error}")
else:
print(f"[WRITER] Skipped message: {data['uuid']} (stage is not 'processed')")
print(
f"[WRITER] Skipped message: {data['uuid']} (stage is not 'processed')"
)
# Subscribe to processor channel
result = redis_pubsub.subscriber.subscribe(CHANNEL_PROCESSOR, on_processor_message)
if result.status:
print(f"[WRITER] Subscribed to channel: {CHANNEL_PROCESSOR}")
else:
@@ -167,18 +178,18 @@ def run_demo():
print("Chain: READER → PROCESSOR → WRITER")
print(f"Channels: {CHANNEL_READER} → {CHANNEL_PROCESSOR} → {CHANNEL_WRITER}")
print("Format: [SERVICE] TIMESTAMP | Received/Published UUID | [Elapsed time]")
# Start the Redis subscriber listening thread
redis_pubsub.subscriber.start_listening()
# Start processor and writer functions (these subscribe to channels)
processor_function()
writer_function()
# Create a thread for the reader function (this publishes messages)
reader_thread = Thread(target=reader_function, daemon=True)
reader_thread.start()
# Keep the main thread alive
try:
while True:

6
Controllers/Redis/connection.py
View File

@@ -41,15 +41,15 @@ class RedisConn:
# Add connection pooling settings if not provided
if "max_connections" not in self.config:
self.config["max_connections"] = 50 # Increased for better concurrency
# Add connection timeout settings
if "health_check_interval" not in self.config:
self.config["health_check_interval"] = 30 # Health check every 30 seconds
# Add retry settings for operations
if "retry_on_timeout" not in self.config:
self.config["retry_on_timeout"] = True
# Add connection pool settings for better performance
if "socket_keepalive" not in self.config:
self.config["socket_keepalive"] = True

78
Controllers/Redis/implementations.py
View File

@@ -113,56 +113,58 @@ def run_all_examples() -> None:
def run_concurrent_test(num_threads=100):
"""Run a comprehensive concurrent test with multiple threads to verify Redis connection handling."""
print(f"\nStarting comprehensive Redis concurrent test with {num_threads} threads...")
print(
f"\nStarting comprehensive Redis concurrent test with {num_threads} threads..."
)
# Results tracking with detailed metrics
results = {
"passed": 0,
"failed": 0,
"passed": 0,
"failed": 0,
"retried": 0,
"errors": [],
"operation_times": [],
"retry_count": 0,
"max_retries": 3,
"retry_delay": 0.1
"retry_delay": 0.1,
}
results_lock = threading.Lock()
def worker(thread_id):
# Track operation timing
start_time = time.time()
retry_count = 0
success = False
error_message = None
while retry_count <= results["max_retries"] and not success:
try:
# Generate unique key for this thread
unique_id = str(uuid.uuid4())[:8]
full_key = f"test:concurrent:{thread_id}:{unique_id}"
# Simple string operations instead of JSON
test_value = f"test-value-{thread_id}-{time.time()}"
# Set data in Redis with pipeline for efficiency
from Controllers.Redis.database import redis_cli
# Use pipeline to reduce network overhead
with redis_cli.pipeline() as pipe:
pipe.set(full_key, test_value)
pipe.get(full_key)
pipe.delete(full_key)
results_list = pipe.execute()
# Check results
set_ok = results_list[0]
retrieved_value = results_list[1]
if isinstance(retrieved_value, bytes):
retrieved_value = retrieved_value.decode('utf-8')
retrieved_value = retrieved_value.decode("utf-8")
# Verify data
success = set_ok and retrieved_value == test_value
if success:
break
else:
@@ -170,26 +172,28 @@ def run_concurrent_test(num_threads=100):
retry_count += 1
with results_lock:
results["retry_count"] += 1
time.sleep(results["retry_delay"] * (2 ** retry_count)) # Exponential backoff
time.sleep(
results["retry_delay"] * (2**retry_count)
) # Exponential backoff
except Exception as e:
error_message = str(e)
retry_count += 1
with results_lock:
results["retry_count"] += 1
# Check if it's a connection error and retry
if "Too many connections" in str(e) or "Connection" in str(e):
# Exponential backoff for connection issues
backoff_time = results["retry_delay"] * (2 ** retry_count)
backoff_time = results["retry_delay"] * (2**retry_count)
time.sleep(backoff_time)
else:
# For other errors, use a smaller delay
time.sleep(results["retry_delay"])
# Record operation time
operation_time = time.time() - start_time
# Update results
with results_lock:
if success:
@@ -200,26 +204,30 @@ def run_concurrent_test(num_threads=100):
else:
results["failed"] += 1
if error_message:
results["errors"].append(f"Thread {thread_id} failed after {retry_count} retries: {error_message}")
results["errors"].append(
f"Thread {thread_id} failed after {retry_count} retries: {error_message}"
)
else:
results["errors"].append(f"Thread {thread_id} failed after {retry_count} retries with unknown error")
results["errors"].append(
f"Thread {thread_id} failed after {retry_count} retries with unknown error"
)
# Create and start threads using a thread pool
start_time = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
futures = [executor.submit(worker, i) for i in range(num_threads)]
concurrent.futures.wait(futures)
# Calculate execution time and performance metrics
execution_time = time.time() - start_time
ops_per_second = num_threads / execution_time if execution_time > 0 else 0
# Calculate additional metrics if we have successful operations
avg_op_time = 0
min_op_time = 0
max_op_time = 0
p95_op_time = 0
if results["operation_times"]:
avg_op_time = sum(results["operation_times"]) / len(results["operation_times"])
min_op_time = min(results["operation_times"])
@@ -227,8 +235,12 @@ def run_concurrent_test(num_threads=100):
# Calculate 95th percentile
sorted_times = sorted(results["operation_times"])
p95_index = int(len(sorted_times) * 0.95)
p95_op_time = sorted_times[p95_index] if p95_index < len(sorted_times) else sorted_times[-1]
p95_op_time = (
sorted_times[p95_index]
if p95_index < len(sorted_times)
else sorted_times[-1]
)
# Print detailed results
print("\nConcurrent Redis Test Results:")
print(f"Total threads: {num_threads}")
@@ -237,17 +249,17 @@ def run_concurrent_test(num_threads=100):
print(f"Operations with retries: {results['retried']}")
print(f"Total retry attempts: {results['retry_count']}")
print(f"Success rate: {(results['passed'] / num_threads) * 100:.2f}%")
print("\nPerformance Metrics:")
print(f"Total execution time: {execution_time:.2f} seconds")
print(f"Operations per second: {ops_per_second:.2f}")
if results["operation_times"]:
print(f"Average operation time: {avg_op_time * 1000:.2f} ms")
print(f"Minimum operation time: {min_op_time * 1000:.2f} ms")
print(f"Maximum operation time: {max_op_time * 1000:.2f} ms")
print(f"95th percentile operation time: {p95_op_time * 1000:.2f} ms")
# Print errors (limited to 10 for readability)
if results["errors"]:
print("\nErrors:")
@@ -255,7 +267,7 @@ def run_concurrent_test(num_threads=100):
print(f"- {error}")
if len(results["errors"]) > 10:
print(f"- ... and {len(results['errors']) - 10} more errors")
# Return results for potential further analysis
return results
@@ -263,6 +275,6 @@ def run_concurrent_test(num_threads=100):
if __name__ == "__main__":
# Run basic examples
run_all_examples()
# Run enhanced concurrent test
run_concurrent_test(10000)