The before code shows ServiceA calling ServiceB synchronously, creating tight coupling and chatty calls. The after code uses an EventBus to publish events asynchronously, allowing ServiceB to handle events without direct calls, reducing coupling and chatty communication.
### Before: Chatty synchronous calls causing tight coupling
class ServiceA:
def call_service_b(self):
service_b = ServiceB()
result = service_b.process()
return result
class ServiceB:
def process(self):
# Some processing
return "data"
### After: Asynchronous event-driven communication reducing chatty calls
import asyncio
class EventBus:
def __init__(self):
self.subscribers = {}
def subscribe(self, event_type, handler):
self.subscribers.setdefault(event_type, []).append(handler)
async def publish(self, event_type, data):
handlers = self.subscribers.get(event_type, [])
for handler in handlers:
await handler(data)
class ServiceA:
def __init__(self, event_bus):
self.event_bus = event_bus
async def do_work(self):
await self.event_bus.publish('event_b', {'info': 'start'})
class ServiceB:
async def handle_event(self, data):
# Process event asynchronously
print(f"Processing event with data: {data}")
# Setup
bus = EventBus()
service_b = ServiceB()
bus.subscribe('event_b', service_b.handle_event)
service_a = ServiceA(bus)
# Run
asyncio.run(service_a.do_work())